@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);
}