@Test public void mainCanceled() throws Exception { TestSubscriber<Long> ts = new TestSubscriber<>(Long.MAX_VALUE); SubmissionPublisher<Long> sp = new SubmissionPublisher<>(Runnable::run, 128); Single.create(sp) .timeout(1, TimeUnit.MILLISECONDS, executor) .subscribe(ts); for (int i = 0; i < 5000; i++) { if (!sp.hasSubscribers()) { return; } Thread.sleep(10); } throw new AssertionError("Did not cancel the main source!"); }
@Override public void subscribe(Flow.Subscriber<? super T> subscriber) { TimeoutSubscriber<T> parent = new TimeoutSubscriber<>(subscriber, fallback); subscriber.onSubscribe(parent); parent.setFuture(executor.schedule(parent, timeout, unit)); source.subscribe(parent); }
SingleTimeout extends CompletionSingle<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { TimeoutSubscriber<T> parent = new TimeoutSubscriber<>(subscriber, fallback); subscriber.onSubscribe(parent); parent.setFuture(executor.schedule(parent, timeout, unit)); source.subscribe(parent); } }
SingleTimeout extends CompletionSingle<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { TimeoutSubscriber<T> parent = new TimeoutSubscriber<>(subscriber, fallback); subscriber.onSubscribe(parent); parent.setFuture(executor.schedule(parent, timeout, unit)); source.subscribe(parent); } SingleTimeout(Single<T> source, long timeout, TimeUnit unit,
                  ScheduledExecutorService executor, Single<T> fallback); }
SingleTimeout extends CompletionSingle<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { TimeoutSubscriber<T> parent = new TimeoutSubscriber<>(subscriber, fallback); subscriber.onSubscribe(parent); parent.setFuture(executor.schedule(parent, timeout, unit)); source.subscribe(parent); } SingleTimeout(Single<T> source, long timeout, TimeUnit unit,
                  ScheduledExecutorService executor, Single<T> fallback); @Override void subscribe(Flow.Subscriber<? super T> subscriber); }
SingleTimeout extends CompletionSingle<T> { @Override public void subscribe(Flow.Subscriber<? super T> subscriber) { TimeoutSubscriber<T> parent = new TimeoutSubscriber<>(subscriber, fallback); subscriber.onSubscribe(parent); parent.setFuture(executor.schedule(parent, timeout, unit)); source.subscribe(parent); } SingleTimeout(Single<T> source, long timeout, TimeUnit unit,
                  ScheduledExecutorService executor, Single<T> fallback); @Override void subscribe(Flow.Subscriber<? super T> subscriber);  }