@Test public void testFailureInValueProviderIsPropagatedToSubscriber() throws Exception { Supplier<Integer> failingSupplier = () -> { throw new RuntimeException("simulated error"); }; AtomicReference<Throwable> errorRef = new AtomicReference<>(); CountDownLatch latch = new CountDownLatch(1); Subscription subscription = ObservableExt.generatorFrom(failingSupplier).subscribe(new Subscriber<Integer>() { @Override public void setProducer(Producer p) { p.request(1); } @Override public void onCompleted() { latch.countDown(); } @Override public void onError(Throwable e) { errorRef.set(e); latch.countDown(); } @Override public void onNext(Integer integer) { } }); latch.await(); Awaitility.await().timeout(5, TimeUnit.SECONDS).until(subscription::isUnsubscribed); assertThat(errorRef.get()).hasMessageContaining("simulated error"); }
@Override public void request(long n) { if (n > 0) { BackpressureUtils.getAndAddRequest(requested, n); worker.schedule(this::drain); } }
ValueGenerator implements Producer { @Override public void request(long n) { if (n > 0) { BackpressureUtils.getAndAddRequest(requested, n); worker.schedule(this::drain); } } }
ValueGenerator implements Producer { @Override public void request(long n) { if (n > 0) { BackpressureUtils.getAndAddRequest(requested, n); worker.schedule(this::drain); } } ValueGenerator(Function<Long, T> source, Subscriber<? super T> subscriber, Scheduler scheduler); }
ValueGenerator implements Producer { @Override public void request(long n) { if (n > 0) { BackpressureUtils.getAndAddRequest(requested, n); worker.schedule(this::drain); } } ValueGenerator(Function<Long, T> source, Subscriber<? super T> subscriber, Scheduler scheduler); @Override void request(long n); }
ValueGenerator implements Producer { @Override public void request(long n) { if (n > 0) { BackpressureUtils.getAndAddRequest(requested, n); worker.schedule(this::drain); } } ValueGenerator(Function<Long, T> source, Subscriber<? super T> subscriber, Scheduler scheduler); @Override void request(long n);  }