avoid constructor escape

reactive-streams · May 13, 2023 · 2b7af33 · 2b7af33
1 parent 11b1a04
commit 2b7af33
Showing 1 changed file with 15 additions and 9 deletions.
diff --git a/examples/src/test/java/org/reactivestreams/example/unicast/AsyncRangePublisherTest.java b/examples/src/test/java/org/reactivestreams/example/unicast/AsyncRangePublisherTest.java
@@ -51,37 +51,39 @@ private AsyncPublisher(Publisher<T> original, Executor executor) {
 
         @Override
         public void subscribe(Subscriber<? super T> s) {
-            original.subscribe(new AsyncSubscriber<T>(requireNonNull(s), executor));
+            AsyncSubscriber.wrapAndSubscribe(original, requireNonNull(s), executor);
         }
 
         private static final class AsyncSubscriber<T> implements Subscriber<T> {
             private final BlockingQueue<Object> signalQueue = new LinkedBlockingQueue<Object>();
 
-            private AsyncSubscriber(final Subscriber<? super T> original, final Executor executor) {
+            static <T> void wrapAndSubscribe(final Publisher<T> publisher,
+                    final Subscriber<? super T> targetSubscriber, final Executor executor) {
+                final AsyncSubscriber<T> asyncSubscriber = new AsyncSubscriber<T>();
                 try {
                     executor.execute(new Runnable() {
                         @Override
                         public void run() {
                             for (; ; ) {
                                 try {
-                                    final Object signal = signalQueue.take();
+                                    final Object signal = asyncSubscriber.signalQueue.take();
                                     if (signal instanceof Cancelled) {
                                         return;
                                     } else if (signal instanceof TerminalSignal) {
                                         Thread.sleep(TERMINAL_DELAY_MS);
                                         TerminalSignal terminalSignal = (TerminalSignal) signal;
                                         if (terminalSignal.cause == null) {
-                                            original.onComplete();
+                                            targetSubscriber.onComplete();
                                         } else {
-                                            original.onError(terminalSignal.cause);
+                                            targetSubscriber.onError(terminalSignal.cause);
                                         }
                                         return;
                                     } else if (signal instanceof OnSubscribeSignal) {
-                                        original.onSubscribe(((OnSubscribeSignal) signal).subscription);
+                                        targetSubscriber.onSubscribe(((OnSubscribeSignal) signal).subscription);
                                     } else {
                                         @SuppressWarnings("unchecked")
                                         final T onNextSignal = ((OnNextSignal<T>) signal).onNext;
-                                        original.onNext(onNextSignal);
+                                        targetSubscriber.onNext(onNextSignal);
                                     }
                                 } catch (InterruptedException ex) {
                                     throw new RuntimeException(ex);
@@ -90,7 +92,7 @@ public void run() {
                         }
                     });
                 } catch (Throwable cause) {
-                    original.onSubscribe(new Subscription() {
+                    targetSubscriber.onSubscribe(new Subscription() {
                         @Override
                         public void request(long n) {
                         }
@@ -99,8 +101,12 @@ public void request(long n) {
                         public void cancel() {
                         }
                     });
-                    original.onError(new IllegalStateException("Executor rejected", cause));
+                    targetSubscriber.onError(new IllegalStateException("Executor rejected", cause));
+                    // Publisher rejected the target subscriber and terminated it, don't continue to subscribe to avoid
+                    // duplicate termination.
+                    return;
                 }
+                publisher.subscribe(asyncSubscriber);
             }
 
             @Override