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);  }
@Test public void testSimpleValueGeneration() throws Exception { final int limit = 100; AtomicInteger valueSource = new AtomicInteger(); AtomicInteger emitCounter = new AtomicInteger(); CountDownLatch latch = new CountDownLatch(1); Subscription subscription = ObservableExt.generatorFrom(valueSource::getAndIncrement).subscribe(new Subscriber<Integer>() { private Producer producer; @Override public void setProducer(Producer p) { this.producer = p; p.request(1); } @Override public void onCompleted() { latch.countDown(); } @Override public void onError(Throwable e) { latch.countDown(); e.printStackTrace(); } @Override public void onNext(Integer integer) { if (emitCounter.incrementAndGet() < limit) { producer.request(1); } else { unsubscribe(); latch.countDown(); } } }); latch.await(); assertThat(emitCounter.get()).isEqualTo(limit); Awaitility.await().timeout(5, TimeUnit.SECONDS).until(subscription::isUnsubscribed); }
@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"); }