WrappedSubscriber implements Subscriber<T> { @Override public void onSubscribe(Subscription subscription) { Objects.requireNonNull(subscription); if (subscribed.compareAndSet(false, true)) { source.onSubscribe( new WrappedSubscription(subscription, () -> future.completeExceptionally(new CancellationException()))); } else { subscription.cancel(); } } WrappedSubscriber(Subscriber<T> delegate); CompletionStage<Void> future(); @Override void onSubscribe(Subscription subscription); @Override void onNext(T item); @Override void onError(Throwable throwable); @Override void onComplete();  }
@Test public void checkThatOnSubscribeCanOnlyBeCallOnce() { AtomicReference<Subscription> subscriptionReference = new AtomicReference<>(); WrappedSubscriber<Integer> subscriber = new WrappedSubscriber<>(new Subscriber<Integer>() { @Override public void onSubscribe(Subscription s) { subscriptionReference.set(s); } @Override public void onNext(Integer integer) { } @Override public void onError(Throwable t) { } @Override public void onComplete() { } }); AtomicBoolean cancellationReference1 = new AtomicBoolean(); subscriber.onSubscribe(new Subscription() { @Override public void request(long n) { } @Override public void cancel() { cancellationReference1.set(true); } }); assertThat(cancellationReference1.get()).isFalse(); assertThat(subscriptionReference.get()).isNotNull(); AtomicBoolean cancellationReference2 = new AtomicBoolean(); subscriber.onSubscribe(new Subscription() { @Override public void request(long n) { } @Override public void cancel() { cancellationReference2.set(true); } }); assertThat(cancellationReference1.get()).isFalse(); assertThat(cancellationReference2.get()).isTrue(); assertThat(subscriptionReference.get()).isNotNull(); }