@Test public void appendChain() { TestSubscriber<Integer> ts = new TestSubscriber<>(); Multi.<Integer>just(1, 2, 3) .onCompleteResume(4) .onCompleteResume(5) .onCompleteResumeWith(Multi.just(6, 7)) .onCompleteResume(8) .subscribe(ts); ts.assertEmpty() .requestMax() .assertComplete() .assertResult(1, 2, 3, 4, 5, 6, 7, 8); }
@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);  }