title: RxJava: calling onError without finishing / unsubscribing
body: <p>I have the following code(*) that implements polling using a scheduler that recursively calls the supplied observable. </p>
  
  <p>(*) inspired from <a href="https://github.com/ReactiveX/RxJava/issues/448" rel="nofollow">https://github.com/ReactiveX/RxJava/issues/448</a></p>
  
  <p>This is working correctly when I only pass the <code>onNext</code> event to the subscriber. But when I pass the <code>onError</code> event to the subscriber, the unsubscribe event is called and this in turn kills the scheduler.</p>
  
  <p>I'd like to also pass the errors to the subscriber. Any ideas how to achieve that?</p>
  
  <pre class="lang-java prettyprint-override"><code>public Observable&lt;Status&gt; observe() {
      return Observable.create(new PollingSubscriberAction&lt;&gt;(service.getStatusObservable(), 5, TimeUnit.SECONDS));
  }
  
  private class PollingSubscriberAction&lt;T&gt; implements Observable.OnSubscribe&lt;T&gt; {
      private Subscription subscription;
      private Subscription innerSubscription;
      private Scheduler.Worker worker = Schedulers.newThread().createWorker();
  
      private Observable&lt;T&gt; observable;
      private long delayTime;
      private TimeUnit unit;
  
      public PollingSubscriberAction(final Observable&lt;T&gt; observable, long delayTime, TimeUnit unit) {
          this.observable = observable;
          this.delayTime = delayTime;
          this.unit = unit;
      }
  
      @Override
      public void call(final Subscriber&lt;? super T&gt; subscriber) {
          subscription = worker.schedule(new Action0() {
              @Override
              public void call() {
                  schedule(subscriber, true);
              }
          });
  
          subscriber.add(Subscriptions.create(new Action0() {
              @Override
              public void call() {
                  subscription.unsubscribe();
                  if (innerSubscription != null) {
                      innerSubscription.unsubscribe();
                  }
              }
          }));
      }
  
      private void schedule(final Subscriber&lt;? super T&gt; subscriber, boolean immediately) {
          long delayTime = immediately ? 0 : this.delayTime;
          subscription = worker.schedule(createInnerAction(subscriber), delayTime, unit);
      }
  
      private Action0 createInnerAction(final Subscriber&lt;? super T&gt; subscriber) {
          return new Action0() {
              @Override
              public void call() {
                  innerSubscription = observable.subscribe(new Observer&lt;T&gt;() {
                      @Override
                      public void onCompleted() {
                          schedule(subscriber, false);
                      }
  
                      @Override
                      public void onError(Throwable e) {
                          // Doesn't work.
                          // subscriber.onError(e);
                          schedule(subscriber, false);
                      }
  
                      @Override
                      public void onNext(T t) {
                          subscriber.onNext(t);
                      }
                  });
              }
          };
      }
  }
  </code></pre>