title: rxjava : avoid cancelling subscription on error in downstream operations
body: <p>I am rather new to RxJava, and have a problem for which I cannot find a decent solution, even after hours of research.</p>
  
  <p>Here it is : I am working on a banking application that connects to some server to receive pricing updates, process these updates, and saves them in some internal database. Using RxJava 2, it was pretty easy to get something working when no error happens, but things tend to get ugly if there are errors in the processing stage : the subscription is cancelled, and no more messages are received.</p>
  
  <p>I used to work with an ad-hoc streaming library that simply catches exceptions, logs the errors and messages that caused them, and keeps the stream open. This is a very simple and pragmatic way of handling errors, but it ensured a rather good level of robustness for applications.</p>
  
  <p>Using RxJava2, the only approach that I can think of that would really work is to surround all of my mapping operators by a wrapper that catches these exceptions. It does not feel very robust, as anyone could forget to catch errors, which would lead to the termination of the application.</p>
  
  <p>To hopefully make my case clearer, here is a test that currently fails, and that I would like to fix :</p>
  
  <pre><code>@Test
  public void shouldKeepProducingWithErrorsInMap() {
  
      List&lt;Integer&gt; output = new ArrayList&lt;&gt;();
      List&lt;Throwable&gt; errors = new ArrayList&lt;&gt;();
      AtomicInteger startCount = new AtomicInteger(0);
  
      Observable&lt;Integer&gt; source = Observable.defer(() -&gt; {
          startCount.incrementAndGet();
          return Observable.range(1, 10);
      });
  
      source
              .map(i -&gt; {
                  if (i % 3 == 0) {
                      throw new IllegalStateException("I cannot accept multiples of 3, this is " + i);
                  }
                  return i;
              })
              .subscribe(output::add, errors::add, () -&gt; log.info("done"));
  
      Assertions.assertThat(startCount.get()).isEqualTo(1);
      Assertions.assertThat(output).containsExactly(1, 2, 4, 5, 7, 8, 10);
      Assertions.assertThat(errors).hasSize(0);
  
  }
  </code></pre>
  
  <p>To achieve the right level of robustness, I would like the following:</p>
  
  <ol>
  <li>encapsulate the logic of connecting to some arbitrary source, by returning an Observable from some factory method</li>
  <li>avoid reconnecting to the source in case of errors that happened at some point downstream</li>
  <li>in case of downstream error, just log the error and move to the next message</li>
  <li>implement this "source protection" feature in he factory method itself, to avoid interrupting the feed when unexpected errors happen.</li>
  </ol>
  
  <p>Is there a way to achieve this in RxJava2 using available operators, or should I implement a custom operator for this?</p>