@Test public void multipleAppendAfterItemsBackpressure() { TestSubscriber<Integer> ts = new TestSubscriber<>(); Multi.concat(Multi.<Integer>empty(), Multi.just(1, 2, 3)) .onCompleteResumeWith(Multi.just(4, 5)) .onCompleteResumeWith(Multi.just(6, 7)) .subscribe(ts); ts.assertEmpty() .request(3) .assertValuesOnly(1, 2, 3) .request(2) .assertValuesOnly(1, 2, 3, 4, 5) .request(1) .assertValuesOnly(1, 2, 3, 4, 5, 6) .request(1) .assertResult(1, 2, 3, 4, 5, 6, 7); }
@Override public void subscribe(Flow.Subscriber<? super T> subscriber) { source.subscribe(new OnCompleteResumeWithSubscriber<>(subscriber, fallbackPublisher)); }
MultiOnCompleteResumeWith implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { source.subscribe(new OnCompleteResumeWithSubscriber<>(subscriber, fallbackPublisher)); } }
MultiOnCompleteResumeWith implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { source.subscribe(new OnCompleteResumeWithSubscriber<>(subscriber, fallbackPublisher)); } MultiOnCompleteResumeWith(Multi<T> source, Flow.Publisher<? extends T> fallbackPublisher); }
MultiOnCompleteResumeWith implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { source.subscribe(new OnCompleteResumeWithSubscriber<>(subscriber, fallbackPublisher)); } MultiOnCompleteResumeWith(Multi<T> source, Flow.Publisher<? extends T> fallbackPublisher); @Override void subscribe(Flow.Subscriber<? super T> subscriber); }
MultiOnCompleteResumeWith implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { source.subscribe(new OnCompleteResumeWithSubscriber<>(subscriber, fallbackPublisher)); } MultiOnCompleteResumeWith(Multi<T> source, Flow.Publisher<? extends T> fallbackPublisher); @Override void subscribe(Flow.Subscriber<? super T> subscriber);  }