@Test public void onRequestCrashAndThenOnErrorPass() { TestSubscriber<Integer> ts = new TestSubscriber<>(); AtomicInteger calls = new AtomicInteger(); new MultiTappedPublisher<>( Multi.singleton(1), null, null, e -> { calls.getAndIncrement(); }, null, r -> { calls.getAndIncrement(); throw new IllegalArgumentException(); }, null ) .subscribe(ts); ts.requestMax(); assertEquals(ts.getItems(), Collections.singletonList(1)); assertThat(ts.getLastError(), is(nullValue())); assertThat(ts.isComplete(), is(true)); 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)); }
MultiTappedPublisher implements Multi<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)); } }
MultiTappedPublisher implements Multi<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)); } MultiTappedPublisher(Multi<T> source,
                         Consumer<? super Flow.Subscription> onSubscribeCallback,
                         Consumer<? super T> onNextCallback,
                         Consumer<? super Throwable> onErrorCallback,
                         Runnable onCompleteCallback,
                         LongConsumer onRequestCallback,
                         Runnable onCancelCallback); private  MultiTappedPublisher(Builder<T> builder); }
MultiTappedPublisher implements Multi<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)); } MultiTappedPublisher(Multi<T> source,
                         Consumer<? super Flow.Subscription> onSubscribeCallback,
                         Consumer<? super T> onNextCallback,
                         Consumer<? super Throwable> onErrorCallback,
                         Runnable onCompleteCallback,
                         LongConsumer onRequestCallback,
                         Runnable onCancelCallback); private  MultiTappedPublisher(Builder<T> builder); static Builder<T> builder(Multi<T> source); @Override void subscribe(Flow.Subscriber<? super T> subscriber); @Override Multi<T> onComplete(Runnable onTerminate); @Override Multi<T> onError(Consumer<? super Throwable> onErrorConsumer); @Override Multi<T> onTerminate(Runnable onTerminate); @Override Multi<T> peek(Consumer<? super T> consumer); @Override Multi<T> onCancel(Runnable onCancel); }
MultiTappedPublisher implements Multi<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)); } MultiTappedPublisher(Multi<T> source,
                         Consumer<? super Flow.Subscription> onSubscribeCallback,
                         Consumer<? super T> onNextCallback,
                         Consumer<? super Throwable> onErrorCallback,
                         Runnable onCompleteCallback,
                         LongConsumer onRequestCallback,
                         Runnable onCancelCallback); private  MultiTappedPublisher(Builder<T> builder); static Builder<T> builder(Multi<T> source); @Override void subscribe(Flow.Subscriber<? super T> subscriber); @Override Multi<T> onComplete(Runnable onTerminate); @Override Multi<T> onError(Consumer<? super Throwable> onErrorConsumer); @Override Multi<T> onTerminate(Runnable onTerminate); @Override Multi<T> peek(Consumer<? super T> consumer); @Override Multi<T> onCancel(Runnable onCancel);  }