@Test public void concurrentRequestN() throws InterruptedException { final int expectedSubscribers = 50; Publisher<Integer> multicast = source.multicastToExactly(expectedSubscribers, expectedSubscribers); @SuppressWarnings("unchecked") TestPublisherSubscriber<Integer>[] subscribers = (TestPublisherSubscriber<Integer>[]) new TestPublisherSubscriber[expectedSubscribers]; final int expectedSubscribersMinus1 = expectedSubscribers - 1; for (int i = 0; i < expectedSubscribersMinus1; ++i) { subscribers[i] = new TestPublisherSubscriber<>(); toSource(multicast).subscribe(subscribers[i]); } subscribers[expectedSubscribersMinus1] = new TestPublisherSubscriber<>(); toSource(multicast).subscribe(subscribers[expectedSubscribersMinus1]); for (int i = 0; i < expectedSubscribersMinus1; ++i) { assertThat(subscribers[i].subscriptionReceived(), is(true)); } source.onSubscribe(subscription); ExecutorService executorService = new ThreadPoolExecutor(0, expectedSubscribers, 1, SECONDS, new SynchronousQueue<>()); try { CyclicBarrier barrier = new CyclicBarrier(expectedSubscribers); CountDownLatch doneLatch = new CountDownLatch(expectedSubscribers); AtomicReference<Throwable> throwableRef = new AtomicReference<>(); for (int i = 1; i <= expectedSubscribers; ++i) { executorService.execute(requestIRunnable(subscribers, i, barrier, throwableRef, doneLatch)); } doneLatch.await(); assertThat(throwableRef.get(), is(nullValue())); assertThat(subscription.requested(), is((long) expectedSubscribers)); assertThat(subscription.isCancelled(), is(false)); } finally { executorService.shutdown(); } }
@Override public void onSubscribe(Subscription s) { delayedSubscription.delayedSubscription(s); }
MulticastPublisher extends AbstractNoHandleSubscribePublisher<T> implements Subscriber<T> { @Override public void onSubscribe(Subscription s) { delayedSubscription.delayedSubscription(s); } }
MulticastPublisher extends AbstractNoHandleSubscribePublisher<T> implements Subscriber<T> { @Override public void onSubscribe(Subscription s) { delayedSubscription.delayedSubscription(s); } MulticastPublisher(Publisher<T> original, int expectedSubscribers, Executor executor);   MulticastPublisher(Publisher<T> original, int expectedSubscribers, int maxQueueSize, Executor executor); }
MulticastPublisher extends AbstractNoHandleSubscribePublisher<T> implements Subscriber<T> { @Override public void onSubscribe(Subscription s) { delayedSubscription.delayedSubscription(s); } MulticastPublisher(Publisher<T> original, int expectedSubscribers, Executor executor);   MulticastPublisher(Publisher<T> original, int expectedSubscribers, int maxQueueSize, Executor executor); @Override void onSubscribe(Subscription s); @Override void onNext(T t); @Override void onError(Throwable t); @Override void onComplete(); }
MulticastPublisher extends AbstractNoHandleSubscribePublisher<T> implements Subscriber<T> { @Override public void onSubscribe(Subscription s) { delayedSubscription.delayedSubscription(s); } MulticastPublisher(Publisher<T> original, int expectedSubscribers, Executor executor);   MulticastPublisher(Publisher<T> original, int expectedSubscribers, int maxQueueSize, Executor executor); @Override void onSubscribe(Subscription s); @Override void onNext(T t); @Override void onError(Throwable t); @Override void onComplete();  }