@Test public void duplicateOnSubscribeIsInvalid() { MulticastPublisher<Integer> source = new MulticastPublisher<>(new Publisher<Integer>(immediate()) { @Override protected void handleSubscribe(Subscriber<? super Integer> subscriber) { } }, 2, immediate()); source.forEach(t -> { }); source.forEach(t -> { }); Subscription sub = mock(Subscription.class); source.onSubscribe(sub); Subscription dup = mock(Subscription.class); source.onSubscribe(dup); verify(dup).cancel(); verify(sub, times(0)).cancel(); }
@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();  }