@Test
public void creation() throws InterruptedException {

	final var lock = new ReentrantLock();
	final var finished = lock.newCondition();
	final var running = new AtomicBoolean(true);
	final var generation = new AtomicLong();
	final Stream<Long> stream = _engine.stream()
		.limit(33)
		.map(EvolutionResult::generation);
	try (var publisher = new StreamPublisher<Long>()) {
		publisher.subscribe(new Subscriber<>() {
			private Subscription _subscription;
			@Override
			public void onSubscribe(final Subscription subscription) {
				_subscription = subscription;
				_subscription.request(1);
			}
			@Override
			public void onNext(final Long g) {
				generation.set(g);
				_subscription.request(1);
			}
			@Override
			public void onError(final Throwable throwable) {
			}
			@Override
			public void onComplete() {
				lock.lock();
				try {
					running.set(false);
					finished.signal();
				} finally {
					lock.unlock();
				}
			}
		});
		publisher.attach(stream);
		lock.lock();
		try {
			while (running.get()) {
				finished.await();
			}
		} finally {
			lock.unlock();
		}
	}
	Assert.assertEquals(generation.get(), 33);
}