### Question:
AsyncMulticastEventStreamPublisher implements EventStreamPublisher<K, V> { @Override public void subscribe(final Flow.Subscriber<? super KV<K, V>> subscriber) { Objects.requireNonNull(subscriber, "subscriber cannot be null"); if (closed.get()) { subscriber.onSubscribe(new Flow.Subscription() { @Override public void cancel() {} @Override public void request(long n) {} }); subscriber.onComplete(); return; } new InternalEventStreamSubscription(subscriber, idGenerator.generateNext()).init(); } AsyncMulticastEventStreamPublisher(final EventStream<K, V> stream); private  AsyncMulticastEventStreamPublisher(final EventStream<K, V> stream,
                                               final SubscriptionIdGenerator subscriptionIdGenerator); @Override String type(); @Override void subscribe(final Flow.Subscriber<? super KV<K, V>> subscriber);  }### Answer:
@Test public void test(){ AsyncMulticastEventStreamPublisher<String, Long> publisher = createPublisher(5); CaptureSubscriber<String, Long> subscriber = new CaptureSubscriber<>(); publisher.subscribe(subscriber); waitAndAssertReceived(subscriber, 0); subscriber.subscription.request(1); waitAndAssertReceived(subscriber, 1); subscriber.subscription.request(1); subscriber.subscription.request(2); waitAndAssertReceived(subscriber, 4); }