@Test public void fallbackBackpressure() throws Exception { TestSubscriber<Long> ts = new TestSubscriber<>(1L); SubmissionPublisher<Long> sp = new SubmissionPublisher<>(Runnable::run, 128); SubmissionPublisher<Long> sp2 = new SubmissionPublisher<>(Runnable::run, 128); Multi.create(sp) .timeout(100, TimeUnit.MILLISECONDS, executor, sp2) .subscribe(ts); sp.submit(1L); for (int i = 0; i < 5000; i++) { if (!sp.hasSubscribers()) { break; } Thread.sleep(10); } assertFalse(sp.hasSubscribers()); ts.assertValuesOnly(1L); assertTrue(sp2.hasSubscribers()); sp2.submit(2L); sp2.close(); ts.assertValuesOnly(1L); ts.request(1) .assertResult(1L, 2L); }
@Override public void subscribe(Flow.Subscriber<? super T> subscriber) { TimeoutSubscriber<T> parent = new TimeoutSubscriber<>(subscriber, timeout, unit, executor, fallback); subscriber.onSubscribe(parent); parent.schedule(0L); source.subscribe(parent); }
MultiTimeout implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { TimeoutSubscriber<T> parent = new TimeoutSubscriber<>(subscriber, timeout, unit, executor, fallback); subscriber.onSubscribe(parent); parent.schedule(0L); source.subscribe(parent); } }
MultiTimeout implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { TimeoutSubscriber<T> parent = new TimeoutSubscriber<>(subscriber, timeout, unit, executor, fallback); subscriber.onSubscribe(parent); parent.schedule(0L); source.subscribe(parent); } MultiTimeout(Multi<T> source, long timeout, TimeUnit unit,
                 ScheduledExecutorService executor, Flow.Publisher<T> fallback); }
MultiTimeout implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { TimeoutSubscriber<T> parent = new TimeoutSubscriber<>(subscriber, timeout, unit, executor, fallback); subscriber.onSubscribe(parent); parent.schedule(0L); source.subscribe(parent); } MultiTimeout(Multi<T> source, long timeout, TimeUnit unit,
                 ScheduledExecutorService executor, Flow.Publisher<T> fallback); @Override void subscribe(Flow.Subscriber<? super T> subscriber); }
MultiTimeout implements Multi<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { TimeoutSubscriber<T> parent = new TimeoutSubscriber<>(subscriber, timeout, unit, executor, fallback); subscriber.onSubscribe(parent); parent.schedule(0L); source.subscribe(parent); } MultiTimeout(Multi<T> source, long timeout, TimeUnit unit,
                 ScheduledExecutorService executor, Flow.Publisher<T> fallback); @Override void subscribe(Flow.Subscriber<? super T> subscriber);  }