@Test public void duplicateSubscriber() { toSource(processor).subscribe(subscriber); TestPublisherSubscriber<Integer> subscriber2 = new TestPublisherSubscriber<>(); toSource(processor).subscribe(subscriber2); verifyTerminal(instanceOf(DuplicateSubscribeException.class), subscriber2); assertThat("Unexpected terminal.", subscriber.takeTerminal(), is(nullValue())); }
@Override public void subscribe(final Subscriber<? super T> subscriber) { subscribeInternal(subscriber); }
PublisherProcessor extends Publisher<T> implements Processor<T, T>, Subscription { @Override public void subscribe(final Subscriber<? super T> subscriber) { subscribeInternal(subscriber); } }
PublisherProcessor extends Publisher<T> implements Processor<T, T>, Subscription { @Override public void subscribe(final Subscriber<? super T> subscriber) { subscribeInternal(subscriber); } PublisherProcessor(final PublisherProcessorSignalsHolder<T> buffer); }
PublisherProcessor extends Publisher<T> implements Processor<T, T>, Subscription { @Override public void subscribe(final Subscriber<? super T> subscriber) { subscribeInternal(subscriber); } PublisherProcessor(final PublisherProcessorSignalsHolder<T> buffer); @Override void onSubscribe(final Subscription subscription); @Override void onNext(@Nullable final T t); @Override void onError(final Throwable t); @Override void onComplete(); @Override void subscribe(final Subscriber<? super T> subscriber); @Override void request(final long n); @Override void cancel(); }
PublisherProcessor extends Publisher<T> implements Processor<T, T>, Subscription { @Override public void subscribe(final Subscriber<? super T> subscriber) { subscribeInternal(subscriber); } PublisherProcessor(final PublisherProcessorSignalsHolder<T> buffer); @Override void onSubscribe(final Subscription subscription); @Override void onNext(@Nullable final T t); @Override void onError(final Throwable t); @Override void onComplete(); @Override void subscribe(final Subscriber<? super T> subscriber); @Override void request(final long n); @Override void cancel();  }