diff --git a/src/main/java/io/reactivex/rxjava4/processors/DispatchStreamProcessor.java b/src/main/java/io/reactivex/rxjava4/processors/DispatchStreamProcessor.java index a889fbfbf1..5ec17bfd05 100644 --- a/src/main/java/io/reactivex/rxjava4/processors/DispatchStreamProcessor.java +++ b/src/main/java/io/reactivex/rxjava4/processors/DispatchStreamProcessor.java @@ -46,8 +46,14 @@ public final class DispatchStreamProcessor implements StreamProcessor { @Override public @NonNull Streamer<@NonNull T> stream(@NonNull StreamerCancellation cancellation) { var result = new DispatchStreamer(this); - cancellation.add(result); if (add(result)) { + // Register with cancellation after a successful add. If cancellation + // already disposed the streamer (or dispose races with add), remove again + // because the first remove may have observed the pre-add array. + cancellation.add(result); + if (result.isDisposed()) { + remove(result); + } return result; } var t = terminalEvent; diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/DispatchStreamProcessorTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/DispatchStreamProcessorTest.java index bc6e857cb0..b8990549e0 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/DispatchStreamProcessorTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/DispatchStreamProcessorTest.java @@ -40,7 +40,7 @@ public void normal() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); - awaitStreamers(dsp, 1000); + assertHasStreamers(dsp, 1000); for (int i = 1; i < 6; i++) { dsp.next(i).toCompletableFuture().join(); @@ -50,7 +50,7 @@ public void normal() throws Throwable { ts.awaitDone(5, TimeUnit.SECONDS) .assertResult(1, 2, 3, 4, 5); - assertFalse(dsp.hasStreamers(), "dsp has streamers?"); + assertNoStreamers(dsp, 1000); assertTrue(dsp.hasComplete(), "dsp has completed?"); assertFalse(dsp.hasThrowable(), "dsp has throwable?"); assertNull(dsp.getThrowable(), "dsp has a non-null throwable?"); @@ -69,7 +69,7 @@ public void endsInError() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); - awaitStreamers(dsp, 1000); + assertHasStreamers(dsp, 1000); for (int i = 1; i < 6; i++) { dsp.next(i).toCompletableFuture().join(); @@ -80,7 +80,7 @@ public void endsInError() throws Throwable { ts.awaitDone(5, TimeUnit.SECONDS) .assertFailure(TestException.class, 1, 2, 3, 4, 5); - assertFalse(dsp.hasStreamers(), "dsp has streamers?"); + assertNoStreamers(dsp, 1000); assertFalse(dsp.hasComplete(), "dsp has completed?"); assertTrue(dsp.hasThrowable(), "dsp has no throwable?"); assertSame(te, dsp.getThrowable(), "dsp has the wrong throwable?"); @@ -100,9 +100,7 @@ public void normalDebug() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); - awaitStreamers(dsp, 1000); - - assertTrue(dsp.hasStreamers(), "dsp has no streamers?"); + assertHasStreamers(dsp, 1000); for (int i = 1; i < 6; i++) { dsp.next(i).toCompletableFuture().join(); @@ -112,7 +110,7 @@ public void normalDebug() throws Throwable { ts.awaitDone(5, TimeUnit.SECONDS) .assertResult(1, 2, 3, 4, 5); - assertFalse(dsp.hasStreamers(), "dsp has streamers?"); + assertNoStreamers(dsp, 1000); assertTrue(dsp.hasComplete(), "dsp has completed?"); assertFalse(dsp.hasThrowable(), "dsp has throwable?"); assertNull(dsp.getThrowable(), "dsp has a non-null throwable?"); @@ -178,16 +176,14 @@ public void normalTake3AltDebug() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); - awaitStreamers(dsp, 1000); + assertHasStreamers(dsp, 1000); for (int i = 1; i < 4; i++) { IO.println(i + " -> next"); dsp.next(i).toCompletableFuture().join(); } - awaitNoStreamers(dsp, 1000); - - assertFalse(dsp.hasStreamers(), "dsp has streamers?"); + assertNoStreamers(dsp, 1000); for (int i = 4; i < 6; i++) { IO.println(i + " -> next"); @@ -201,7 +197,7 @@ public void normalTake3AltDebug() throws Throwable { ts.awaitDone(5, TimeUnit.SECONDS) .assertResult(1, 2, 3); - assertFalse(dsp.hasStreamers(), "dsp has streamers?"); + assertNoStreamers(dsp, 1000); assertTrue(dsp.hasComplete(), "dsp has completed?"); assertFalse(dsp.hasThrowable(), "dsp has throwable?"); assertNull(dsp.getThrowable(), "dsp has a non-null throwable?"); @@ -222,7 +218,7 @@ public void normalTake3Debug() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); - awaitStreamers(dsp, 1000); + assertHasStreamers(dsp, 1000); for (int i = 1; i < 6; i++) { dsp.next(i).toCompletableFuture().join(); @@ -235,7 +231,7 @@ public void normalTake3Debug() throws Throwable { ts.awaitDone(5, TimeUnit.SECONDS) .assertResult(1, 2, 3); - assertFalse(dsp.hasStreamers(), "dsp has streamers?"); + assertNoStreamers(dsp, 1000); assertTrue(dsp.hasComplete(), "dsp has completed?"); assertFalse(dsp.hasThrowable(), "dsp has throwable?"); assertNull(dsp.getThrowable(), "dsp has a non-null throwable?"); @@ -257,7 +253,7 @@ public void normalMulti() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); ts2.awaitOnSubscribe(1, TimeUnit.SECONDS); - awaitStreamers(dsp, 1000, 2); + assertHasStreamers(dsp, 1000, 2); for (int i = 1; i < 6; i++) { dsp.next(i).toCompletableFuture().join(); @@ -270,7 +266,7 @@ public void normalMulti() throws Throwable { ts2.awaitDone(5, TimeUnit.SECONDS) .assertResult(1, 2, 3, 4, 5); - assertFalse(dsp.hasStreamers(), "dsp has streamers?"); + assertNoStreamers(dsp, 1000); assertTrue(dsp.hasComplete(), "dsp has completed?"); assertFalse(dsp.hasThrowable(), "dsp has throwable?"); assertNull(dsp.getThrowable(), "dsp has a non-null throwable?"); @@ -291,7 +287,7 @@ public void normalMultiOtherCancels() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); ts2.awaitOnSubscribe(1, TimeUnit.SECONDS); - awaitStreamers(dsp, 1000, 2); + assertHasStreamers(dsp, 1000, 2); ts2.cancel(); @@ -305,7 +301,7 @@ public void normalMultiOtherCancels() throws Throwable { ts2.assertEmpty(); - assertFalse(dsp.hasStreamers(), "dsp has streamers?"); + assertNoStreamers(dsp, 1000); assertTrue(dsp.hasComplete(), "dsp has completed?"); assertFalse(dsp.hasThrowable(), "dsp has throwable?"); assertNull(dsp.getThrowable(), "dsp has a non-null throwable?"); @@ -326,7 +322,7 @@ public void normalMultiFirstCancels() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); ts2.awaitOnSubscribe(1, TimeUnit.SECONDS); - awaitStreamers(dsp, 1000, 2); + assertHasStreamers(dsp, 1000, 2); ts.cancel(); @@ -340,7 +336,7 @@ public void normalMultiFirstCancels() throws Throwable { ts.assertEmpty(); - assertFalse(dsp.hasStreamers(), "dsp has streamers?"); + assertNoStreamers(dsp, 1000); assertTrue(dsp.hasComplete(), "dsp has completed?"); assertFalse(dsp.hasThrowable(), "dsp has throwable?"); assertNull(dsp.getThrowable(), "dsp has a non-null throwable?"); @@ -367,14 +363,14 @@ public void raceToStream() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); ts2.awaitOnSubscribe(1, TimeUnit.SECONDS); - awaitStreamers(dsp, 1000); + // Both subscribers must be fully attached before cancellation, + // otherwise a late stream() can re-add a streamer after the wait. + assertHasStreamers(dsp, 1000, 2); ts.cancel(); ts2.cancel(); - awaitNoStreamers(dsp, 1000); - - assertFalse(dsp.hasStreamers(), "dsp has streamers?"); + assertNoStreamers(dsp, 1000); assertFalse(dsp.hasComplete(), "dsp has completed?"); assertFalse(dsp.hasThrowable(), "dsp has throwable?"); assertNull(dsp.getThrowable(), "dsp has a non-null throwable?"); @@ -394,13 +390,11 @@ public void comeAndGo() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); - awaitStreamers(dsp, 1000); + assertHasStreamers(dsp, 1000); ts.cancel(); - awaitNoStreamers(dsp, 1000); - - assertFalse(dsp.hasStreamers(), "dsp has streamers?"); + assertNoStreamers(dsp, 1000); assertFalse(dsp.hasComplete(), "dsp has completed?"); assertFalse(dsp.hasThrowable(), "dsp has throwable?"); assertNull(dsp.getThrowable(), "dsp has a non-null throwable?"); diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBaseTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBaseTest.java index 69553c9f99..b5c67a9bb3 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBaseTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBaseTest.java @@ -15,7 +15,7 @@ import java.lang.ref.Cleaner; import java.util.*; -import java.util.concurrent.TimeoutException; +import java.util.concurrent.*; import java.util.function.*; import org.junit.jupiter.api.*; @@ -119,13 +119,7 @@ public static StreamableInterceptConfig debugIntercept() { public static void awaitStreamers(StreamProcessor sp, long timeoutMillis) throws InterruptedException, TimeoutException { - long timeout = timeoutMillis * 1_000_000L; - while (!sp.hasStreamers()) { - Thread.sleep(0, 1000); - if (--timeout <= 0L) { - throw new TimeoutException("hasStreamers still false"); - } - } + awaitCondition(true, sp::hasStreamers, timeoutMillis, "hasStreamers still false"); } /** @@ -140,13 +134,8 @@ public static void awaitStreamers(StreamProcessor sp, long timeoutMillis) public static void awaitStreamers(StreamProcessor sp, long timeoutMillis, int atLeast) throws InterruptedException, TimeoutException { - long timeout = timeoutMillis * 1_000_000L; - while (sp.streamerCount() < atLeast) { - Thread.sleep(0, 1000); - if (--timeout <= 0L) { - throw new TimeoutException("hasStreamers still false"); - } - } + awaitCondition(true, () -> sp.streamerCount() >= atLeast, timeoutMillis, + "streamerCount still below " + atLeast + " (was " + sp.streamerCount() + ")"); } /** @@ -160,12 +149,55 @@ public static void awaitStreamers(StreamProcessor sp, long timeoutMillis, public static void awaitNoStreamers(StreamProcessor sp, long timeoutMillis) throws InterruptedException, TimeoutException { - long timeout = timeoutMillis * 1_000_000L; - while (sp.hasStreamers()) { - Thread.sleep(0, 1000); - if (--timeout <= 0L) { - throw new TimeoutException("hasStreamers still false"); - } + awaitCondition(false, sp::hasStreamers, timeoutMillis, "hasStreamers still true"); + } + + /** + * Awaits until the processor has no streamers and asserts that state. + * @param sp the processor + * @param timeoutMillis how long to wait for the streamer(s) to leave + * @throws InterruptedException if the sleep is interrupted + * @throws TimeoutException if the wait times out + */ + public static void assertNoStreamers(StreamProcessor sp, long timeoutMillis) + throws InterruptedException, TimeoutException + { + awaitNoStreamers(sp, timeoutMillis); + if (sp.hasStreamers()) { + throw new AssertionError("Processor still has streamers: " + sp.streamerCount()); + } + } + + /** + * Awaits until the processor has at least one streamer and asserts that state. + * @param sp the processor + * @param timeoutMillis how long to wait for the streamer(s) to arrive + * @throws InterruptedException if the sleep is interrupted + * @throws TimeoutException if the wait times out + */ + public static void assertHasStreamers(StreamProcessor sp, long timeoutMillis) + throws InterruptedException, TimeoutException + { + awaitStreamers(sp, timeoutMillis); + if (!sp.hasStreamers()) { + throw new AssertionError("Processor has no streamers"); + } + } + + /** + * Awaits until the processor has at least {@code atLeast} streamers and asserts that state. + * @param sp the processor + * @param timeoutMillis how long to wait for the streamer(s) to arrive + * @param atLeast the minimum number of streamers expected + * @throws InterruptedException if the sleep is interrupted + * @throws TimeoutException if the wait times out + */ + public static void assertHasStreamers(StreamProcessor sp, long timeoutMillis, int atLeast) + throws InterruptedException, TimeoutException + { + awaitStreamers(sp, timeoutMillis, atLeast); + if (sp.streamerCount() < atLeast) { + throw new AssertionError("Processor streamerCount below " + atLeast + ": " + sp.streamerCount()); } } @@ -181,12 +213,31 @@ public static void awaitNoStreamers(StreamProcessor sp, long timeoutMillis */ public static void awaitCondition(boolean value, @NonNull BooleanSupplier condition, long timeoutMillis) throws InterruptedException, TimeoutException { - long timeout = timeoutMillis * 1_000_000L; - while (condition.getAsBoolean() != value) { - Thread.sleep(0, 1000); - if (--timeout <= 0L) { - throw new TimeoutException("condition still " + (!value)); + awaitCondition(value, condition, timeoutMillis, "condition still " + (!value)); + } + + /** + * Awaits a given {@link BooleanSupplier} to return the expected {@code value} + * within the given wall-clock time period. + * @param value the expected value within the timeout period + * @param condition the condition to repeatedly call to assess the state + * @param timeoutMillis how long to wait for the condition to become as expected + * @param timeoutMessage the message used when the wait times out + * @throws InterruptedException if the sleep is interrupted + * @throws TimeoutException if the wait times out + */ + public static void awaitCondition(boolean value, @NonNull BooleanSupplier condition, long timeoutMillis, + @NonNull String timeoutMessage) + throws InterruptedException, TimeoutException { + long end = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMillis); + for (;;) { + if (condition.getAsBoolean() == value) { + return; + } + if (System.nanoTime() >= end) { + throw new TimeoutException(timeoutMessage); } + Thread.sleep(0, 1000); } } }