@Test public void onCancelCrashThenOnErrorPass() { TestSubscriber<Integer> ts = new TestSubscriber<>(); AtomicInteger calls = new AtomicInteger(); new SingleTappedPublisher<>( Single.just(1), null, null, e -> { calls.getAndIncrement(); }, null, null, () -> { calls.getAndIncrement(); throw new IllegalArgumentException(); } ) .subscribe(ts); ts.getSubcription().cancel(); assertThat(ts.getItems().isEmpty(), is(true)); assertThat(ts.getLastError(), is(nullValue())); assertThat(ts.isComplete(), is(false)); assertThat(calls.get(), is(2)); }
@Override public void subscribe(Flow.Subscriber<? super T> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); source.subscribe(new MultiTappedSubscriber<>(subscriber, onSubscribeCallback, onNextCallback, onErrorCallback, onCompleteCallback, onRequestCallback, onCancelCallback)); }
SingleTappedPublisher extends CompletionSingle<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); source.subscribe(new MultiTappedSubscriber<>(subscriber, onSubscribeCallback, onNextCallback, onErrorCallback, onCompleteCallback, onRequestCallback, onCancelCallback)); } }
SingleTappedPublisher extends CompletionSingle<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); source.subscribe(new MultiTappedSubscriber<>(subscriber, onSubscribeCallback, onNextCallback, onErrorCallback, onCompleteCallback, onRequestCallback, onCancelCallback)); } SingleTappedPublisher(Single<T> source,
                          Consumer<? super Flow.Subscription> onSubscribeCallback,
                          Consumer<? super T> onNextCallback,
                          Consumer<? super Throwable> onErrorCallback,
                          Runnable onCompleteCallback,
                          LongConsumer onRequestCallback,
                          Runnable onCancelCallback); }
SingleTappedPublisher extends CompletionSingle<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); source.subscribe(new MultiTappedSubscriber<>(subscriber, onSubscribeCallback, onNextCallback, onErrorCallback, onCompleteCallback, onRequestCallback, onCancelCallback)); } SingleTappedPublisher(Single<T> source,
                          Consumer<? super Flow.Subscription> onSubscribeCallback,
                          Consumer<? super T> onNextCallback,
                          Consumer<? super Throwable> onErrorCallback,
                          Runnable onCompleteCallback,
                          LongConsumer onRequestCallback,
                          Runnable onCancelCallback); @Override void subscribe(Flow.Subscriber<? super T> subscriber); @Override Single<T> onComplete(Runnable onTerminate); @Override Single<T> onError(Consumer<? super Throwable> onErrorConsumer); @Override Single<T> onTerminate(Runnable onTerminate); @Override Single<T> peek(Consumer<? super T> consumer); @Override Single<T> onCancel(Runnable onCancel); }
SingleTappedPublisher extends CompletionSingle<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); source.subscribe(new MultiTappedSubscriber<>(subscriber, onSubscribeCallback, onNextCallback, onErrorCallback, onCompleteCallback, onRequestCallback, onCancelCallback)); } SingleTappedPublisher(Single<T> source,
                          Consumer<? super Flow.Subscription> onSubscribeCallback,
                          Consumer<? super T> onNextCallback,
                          Consumer<? super Throwable> onErrorCallback,
                          Runnable onCompleteCallback,
                          LongConsumer onRequestCallback,
                          Runnable onCancelCallback); @Override void subscribe(Flow.Subscriber<? super T> subscriber); @Override Single<T> onComplete(Runnable onTerminate); @Override Single<T> onError(Consumer<? super Throwable> onErrorConsumer); @Override Single<T> onTerminate(Runnable onTerminate); @Override Single<T> peek(Consumer<? super T> consumer); @Override Single<T> onCancel(Runnable onCancel);  }