title: Subjects and resilience logic workflow
body: <p>I would like to get a better understanding of the expected behaviour of Subjects when used with the resiliency operators, namely <a href="https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/retry.md" rel="nofollow">retry</a> and <a href="https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/retrywhen.md" rel="nofollow">retryWhen</a>.</p>
  
  <p><em>The following code samples will differ slightly from their JSBin counterparts (found in example links) in that I used arrow functions and types for easier consumption, this is using version 4.0.0 - 4.0.7</em></p>
  
  <p>My expected resiliency behaviour can be expressed with the following <a href="https://jsbin.com/vamepac/19/edit?html,js,console,output" rel="nofollow">example</a>:</p>
  
  <pre><code>Rx.Observable
    .interval(1000)
    .flatMap( (count:number) =&gt; { 
      return count === 4 ? Rx.Observable.throw('Break') : Rx.Observable.return(count);
    })
    .retry()
    .take(5);
  
   Output 
   // 0
   // 1
   // 2
   // 3 
   // 0 &lt;-- Retry means we start again from scratch (expected)
  </code></pre>
  
  <p>Up until this point everything is consistent, that is after the error has occurred on the fourth notification the entire stream restarts from scratch (win for stateless architecture).</p>
  
  <p>Now comes the scratching my head part if we add a multicast operator and in doing so add an underlying Subject (in my case a ReplaySubject with a buffer of 1), <a href="https://jsbin.com/vamepac/18/edit?html,js,console,output" rel="nofollow">example</a>:</p>
  
  <pre><code>const consumer : Rx.Observable&lt;number&gt; = Rx.Observable
    .interval(1000)
    .flatMap( (count:number) =&gt; { 
      return count === 4 ? Rx.Observable.throw('Break') : Rx.Observable.return(count);
    })
    .shareReplay(1) /* multicast(new Rx.ReplaySubject(1)).refCount() */
    .retry()
    .take(5);
  
  const firstSubscriber : Rx.Disposable = consumer.subscribe( (next:number) =&gt; {
     console.log('first subscriber: ' + next);
  });
  
  setTimeout(() =&gt; {
     firstSubscriber.dispose(); /* Lets start fresh in that refCount === 0 */
     const secondSubscriber : Rx.Disposable = consumer.subscribe( (next) =&gt; {
        console.log('second subscriber: ' + next);
     });
  }, 5000 );
  
  Output (before error is thrown)
  // "first subscriber: 0"
  // "first subscriber: 1"
  // "first subscriber: 2"
  // "first subscriber: 3"
  Output (after error is thrown)
  // "first subscriber: 3"
  // "second subscriber: 3"
  // "second subscriber: 3"
  // "second subscriber: 3"
  // "second subscriber: 3"
  // "second subscriber: 3" 
  </code></pre>
  
  <p>A quick look into a <a href="https://github.com/Reactive-Extensions/RxJS/blob/master/src/core/subjects/replaysubject.js#L61-L63" rel="nofollow">Subject</a> identifies when an error arrises the subject is marked as inError and each future subscriber will get the last notification (Line 46) and straight after a call to onError will be made (Line 50). </p>
  
  <p>So Where does this leave us? In my opinion I don't believe you can ever use a resilience operator when it follows any other operator that contains a Subject (shareReplay, publish etc...).</p>
  
  <p>At this point I am thinking the only way to succeed with this design would be ensuring when an error has occurred and node has been disposed, whenever a subject has been used a new one would need to be created (and down the rabbit hole we start to go)?</p>
  
  <p><a href="https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/multicast.md" rel="nofollow">multicast</a> can take a factory/subjectSelector:</p>
  
  <pre><code>.multicast( () =&gt; new Rx.ReplaySubject(1), (source:Rx.ConnectableObservable) =&gt; source );
  </code></pre>
  
  <p>Looking at the <a href="https://github.com/Reactive-Extensions/RxJS/blob/master/src/core/linq/observable/multicast.js" rel="nofollow">source</a> if you use a subjectSelector rather than just passing in the subject directly for each new subscription the subjectSelector will be invoked and a new ConnectableObservable will be created (Line 11).</p>
  
  <p>At this point I am unsure if sharing (via some cache) and disposing (when an error is detected) of the Subjects will actually give multicasting to subscribers?</p>
  
  <p>In getting to this point I have also written a RecoverableReplaySubject whereby I had taken out the error state when disposing, this was more for testing and would expect the RxJS team put this workflow in for a good reason. </p>
  
  <p>Any guidance and experience on this topic would be greatly appreciated.</p>
  
  <p>Thanks </p>