@Test public void failAfterItemsBackpressure() { TestSubscriber<Integer> ts = new TestSubscriber<>(); Multi.concat(Multi.range(1, 3), Multi.<Integer>error(new IOException())) .onErrorResumeWith(v -> Multi.range(4, 2)) .subscribe(ts); ts.assertEmpty() .request(3) .assertValuesOnly(1, 2, 3) .request(2) .assertResult(1, 2, 3, 4, 5); }
@Override public void subscribe(Flow.Subscriber<? super T> subscriber) { source.subscribe(new OnErrorResumeWithSubscriber<>(subscriber, fallbackFunction)); }
MultiOnErrorResumeWith implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { source.subscribe(new OnErrorResumeWithSubscriber<>(subscriber, fallbackFunction)); } }
MultiOnErrorResumeWith implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { source.subscribe(new OnErrorResumeWithSubscriber<>(subscriber, fallbackFunction)); } MultiOnErrorResumeWith(Multi<T> source,
           Function<? super Throwable, ? extends Flow.Publisher<? extends T>> fallbackFunction); }
MultiOnErrorResumeWith implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { source.subscribe(new OnErrorResumeWithSubscriber<>(subscriber, fallbackFunction)); } MultiOnErrorResumeWith(Multi<T> source,
           Function<? super Throwable, ? extends Flow.Publisher<? extends T>> fallbackFunction); @Override void subscribe(Flow.Subscriber<? super T> subscriber); }
MultiOnErrorResumeWith implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { source.subscribe(new OnErrorResumeWithSubscriber<>(subscriber, fallbackFunction)); } MultiOnErrorResumeWith(Multi<T> source,
           Function<? super Throwable, ? extends Flow.Publisher<? extends T>> fallbackFunction); @Override void subscribe(Flow.Subscriber<? super T> subscriber);  }