From 682143a422f088ead4398cec8b58334a00af243b Mon Sep 17 00:00:00 2001 From: akarnokd Date: Mon, 6 Jul 2026 22:56:04 +0200 Subject: [PATCH 1/8] 4.x: Streamable + doOnError, forEach(StreamerInput) --- .../{StreamableDoOnXTest.java => StreamableDoOnNextTest.java} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename src/test/java/io/reactivex/rxjava4/internal/operators/streamable/{StreamableDoOnXTest.java => StreamableDoOnNextTest.java} (100%) diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDoOnXTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDoOnNextTest.java similarity index 100% rename from src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDoOnXTest.java rename to src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDoOnNextTest.java From 07ec596ecd4c0cddaf5496a89d8bb68f7d24f6bc Mon Sep 17 00:00:00 2001 From: akarnokd Date: Mon, 6 Jul 2026 23:00:31 +0200 Subject: [PATCH 2/8] fix tests --- .../io/reactivex/rxjava4/core/Streamable.java | 32 +++++++++ .../io/reactivex/rxjava4/core/Streamer.java | 37 +++++++--- .../reactivex/rxjava4/core/StreamerInput.java | 10 +++ .../streamable/StreamableForEach.java | 40 +++++++++++ .../streamable/StreamableHelper.java | 32 ++++++++- .../streamable/StreamableDoOnErrorTest.java | 67 +++++++++++++++++++ .../streamable/StreamableDoOnNextTest.java | 2 +- .../streamable/StreamableForEachTest.java | 62 ++++++++++++++++- .../operators/streamable/StreamableTest.java | 7 +- .../validators/CheckParamValidationTest.java | 3 + 10 files changed, 275 insertions(+), 17 deletions(-) create mode 100644 src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDoOnErrorTest.java diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamable.java b/src/main/java/io/reactivex/rxjava4/core/Streamable.java index b067279ab3..cfe01c408f 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamable.java @@ -589,6 +589,17 @@ default Streamable delay(long time, TimeUnit unit, Scheduler scheduler) { return RxJavaPlugins.onAssembly(new StreamableDelay<>(this, time, unit, scheduler)); } + /** + * Calls the specific {@link Consumer} if there is an error from the upstream. + * @param consumer the consumer to call with the Throwable + * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code consumer} is {@code null} + */ + default Streamable doOnError(Consumer consumer) { + Objects.requireNonNull(consumer, "consumer is null"); + return intercept(StreamableHelper.createOnError(consumer)); + } + /** * Calls the given consumer whenever an upstream item becomes available. * @param consumer the callback to invoke with the next item from upstream @@ -983,6 +994,27 @@ default void subscribe(@NonNull Flow.Subscriber subscriber) { subscribe(subscriber, Executors.newVirtualThreadPerTaskExecutor()); } + /** + * Relays the events of the upstream into a {@link StreamerInput} consumer. + * @param consumer the consumer to relay events into + * @throws NullPointerException if {@code consumer} or {@code executor} is {@code null} + */ + default void subscribe(@NonNull StreamerInput consumer) { + subscribe(consumer, Executors.newVirtualThreadPerTaskExecutor()); + } + + /** + * Relays the events of the upstream into a {@link StreamerInput} consumer. + * @param consumer the consumer to relay events into + * @param executor the {@link ExecutorService} to run the blocking consume and emissions + * @throws NullPointerException if {@code consumer} or {@code executor} is {@code null} + */ + default void subscribe(@NonNull StreamerInput consumer, ExecutorService executor) { + Objects.requireNonNull(consumer, "consumer is null"); + Objects.requireNonNull(executor, "executor is null"); + StreamableForEach.forEach(this, consumer, executor); + } + /** * Creates a new {@link TestSubscriber} and subscribes it to this {@code Streamable}. * @return the created test subscriber diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamer.java b/src/main/java/io/reactivex/rxjava4/core/Streamer.java index 8aef3ec26b..36d6731529 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamer.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamer.java @@ -13,6 +13,7 @@ package io.reactivex.rxjava4.core; +import java.util.NoSuchElementException; import java.util.concurrent.*; import io.reactivex.rxjava4.annotations.NonNull; @@ -50,6 +51,7 @@ public interface Streamer<@NonNull T> { * Calling it during an ongoing [#next()] or [#finish()] call, or beyond the lifecycle of the `Streamer` * is an undefined behavior. It may yield `null` or throw. * @return the current item + * @throws NoSuchElementException if there are no items to return */ @NonNull T current(); @@ -76,25 +78,42 @@ public interface Streamer<@NonNull T> { * @return true if there are more items, false if no more items are coming, or crashes */ default boolean awaitNext() { - var s = next(); - if (s == NEXT_TRUE) { + return awaitBoolean(next()); + } + + /** + * Convenience method to blockingly await the CompletionStage returned by the {@link #finish()} method. + */ + default void awaitFinish() { + awaitVoid(finish()); + } + + /** + * Convenience method to await the completion of a boolean stage, optimized + * for handling {@value #NEXT_TRUE} and {@value #NEXT_FALSE} directly. + * @param stage the stage to await + * @return the result of the stage + */ + static boolean awaitBoolean(CompletionStage stage) { + if (stage == NEXT_TRUE) { return true; } else - if (s == NEXT_FALSE) { + if (stage == NEXT_FALSE) { return false; } - return s.toCompletableFuture().join(); + return stage.toCompletableFuture().join(); } /** - * Convenience method to blockingly await the CompletionStage returned by the {@link #finish()} method. + * Convenience method to await the completion of a stage, optimized + * for handling {@value #FINISHED} directly. + * @param stage the stage to await */ - default void awaitFinish() { - var s = finish(); - if (s == FINISHED) { + static void awaitVoid(CompletionStage stage) { + if (stage == FINISHED) { return; } - s.toCompletableFuture().join(); + stage.toCompletableFuture().join(); } /** diff --git a/src/main/java/io/reactivex/rxjava4/core/StreamerInput.java b/src/main/java/io/reactivex/rxjava4/core/StreamerInput.java index 2325f60061..43bf0699d4 100644 --- a/src/main/java/io/reactivex/rxjava4/core/StreamerInput.java +++ b/src/main/java/io/reactivex/rxjava4/core/StreamerInput.java @@ -17,6 +17,7 @@ import java.util.concurrent.Flow.Subscriber; import io.reactivex.rxjava4.annotations.*; +import io.reactivex.rxjava4.disposables.*; /** * An interface to submit items and terminal events to a consumer that indacates when the processing of @@ -45,4 +46,13 @@ public interface StreamerInput<@NonNull T> { * or exceptionally on error */ CompletionStage finish(@Nullable Throwable throwable); + + /** + * Returns the {@link DisposableContainer} to use to detect if the consumer has indicated no more + * items it is willing to accept. + * @return the {@code DisposableContainer} + */ + default DisposableContainer cancellation() { + return new CompositeDisposable(); + } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java index 00fc5f52e0..438aeb9045 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java @@ -22,6 +22,7 @@ import io.reactivex.rxjava4.exceptions.Exceptions; import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.util.ExceptionHelper; +import io.reactivex.rxjava4.plugins.RxJavaPlugins; /** * ForEach implementation to unclutter the {@link Streamable} type. @@ -109,4 +110,43 @@ public static CompletionStageDisposable forEach( canceller.add(Disposable.fromFuture(future)); return new CompletionStageDisposable<>(future, canceller); } + + public static void forEach(Streamable me, StreamerInput consumer, ExecutorService executor) { + CompletableFuture.runAsync(() -> { + Throwable error = null; + var cancellation = consumer.cancellation(); + var streamer = me.stream(cancellation); + try { + try { + while (!cancellation.isDisposed()) { + if (streamer.awaitNext()) { + Streamer.awaitBoolean(consumer.next(streamer.current())); + } else { + break; + } + } + } finally { + try { + streamer.awaitFinish(); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + error = ExceptionHelper.unwrap(ex); + } + } + } catch (Throwable crash) { + Exceptions.throwIfFatal(crash); + crash = ExceptionHelper.unwrap(crash); + if (error != null) { + crash.addSuppressed(error); + } + error = crash; + } + try { + Streamer.awaitVoid(consumer.finish(error)); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + RxJavaPlugins.onError(ex); + } + }, executor); + } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelper.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelper.java index d2e6af862c..26ec63eef4 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelper.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelper.java @@ -21,8 +21,10 @@ import io.reactivex.rxjava4.annotations.*; import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.core.config.StreamableInterceptConfig; import io.reactivex.rxjava4.disposables.Disposable; -import io.reactivex.rxjava4.exceptions.CompositeException; +import io.reactivex.rxjava4.exceptions.*; +import io.reactivex.rxjava4.functions.Consumer; import io.reactivex.rxjava4.internal.util.*; /** @@ -373,4 +375,30 @@ public void accept(Boolean t, Throwable u) { } } } -} + + /** + * Create a {@link StreamableInterceptConfig} that can consume the {@link Streamer#next()} errors. + * @param the element type of the {@link Streamable} + * @param consumer the consumer to be called with the error + * @return the new {@code StreamableInterceptConfig} instance + */ + public static StreamableInterceptConfig createOnError(Consumer consumer) { + return new StreamableInterceptConfig<>((_, v) -> v, (_, v) -> { + var cf = new CompletableFuture(); + v.whenComplete((u, e) -> { + if (e != null) { + try { + consumer.accept(e); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + ex.addSuppressed(e); + e = ex; + } + cf.completeExceptionally(e); + } else { + cf.complete(u); + } + }); + return cf; + }, v -> v, (_, v) -> v); + }} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDoOnErrorTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDoOnErrorTest.java new file mode 100644 index 0000000000..bc4b889088 --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDoOnErrorTest.java @@ -0,0 +1,67 @@ +/* + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.rxjava4.internal.operators.streamable; + +import static org.junit.jupiter.api.Assertions.*; + +import java.io.IOException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.exceptions.TestException; + +public class StreamableDoOnErrorTest extends StreamableBaseTest { + + @Test + public void normal() { + AtomicReference error = new AtomicReference<>(); + Streamable.range(1, 5) + .doOnError(error::set) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5) + ; + + assertNull(error.get(), "error is not empty?"); + } + + @Test + public void hasError() { + AtomicReference error = new AtomicReference<>(); + var te = new TestException(); + Streamable.error(te) + .doOnError(error::set) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class) + ; + + assertSame(te, error.get(), "doOnError differs from TestSubscriber.onError?"); + } + + @Test + public void consumerCrash() { + var te = new TestException(); + Streamable.error(te) + .doOnError(_ -> { throw new IOException(); }) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(IOException.class) + .assertError(e -> e.getSuppressed()[0] == te) + ; + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDoOnNextTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDoOnNextTest.java index cecb1109d5..5ab75dfbc8 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDoOnNextTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDoOnNextTest.java @@ -23,7 +23,7 @@ import io.reactivex.rxjava4.core.Streamable; import io.reactivex.rxjava4.exceptions.TestException; -public class StreamableDoOnXTest extends StreamableBaseTest { +public class StreamableDoOnNextTest extends StreamableBaseTest { @Test public void passthrough() { diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java index 0aef535c25..2b788c852a 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java @@ -15,13 +15,15 @@ import static org.junit.jupiter.api.Assertions.*; -import java.util.concurrent.CancellationException; +import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; -import org.junit.jupiter.api.*; +import org.junit.jupiter.api.Test; + import io.reactivex.rxjava4.core.Streamable; import io.reactivex.rxjava4.disposables.CompositeDisposable; import io.reactivex.rxjava4.exceptions.*; +import io.reactivex.rxjava4.processors.DispatchStreamProcessor; public class StreamableForEachTest extends StreamableBaseTest { @@ -189,4 +191,60 @@ public void forEachBiInsideCancel() throws Throwable { assertEquals(1, counter.get()); }); } + + @Test + public void forEachInput() throws Throwable { + var dsp = new DispatchStreamProcessor<>(); + var ts = dsp.test(); + + ts.awaitOnSubscribe(1, TimeUnit.SECONDS); + + while (!dsp.hasStreamers()) { + Thread.sleep(0, 1000); + } + + Streamable.range(1, 5) + .subscribe(dsp); + + ts.awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void forEachInputDebug() throws Throwable { + withCachedExecutor(exec -> { + var dsp = new DispatchStreamProcessor<>(); + var ts = dsp.test(exec); + + ts.awaitOnSubscribe(1, TimeUnit.SECONDS); + + while (!dsp.hasStreamers()) { + Thread.sleep(0, 1000); + } + + Streamable.range(1, 5) + .subscribe(dsp); + + ts.awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + }); + } + + @Test + public void forEachInputError() throws Throwable { + var dsp = new DispatchStreamProcessor<>(); + var ts = dsp.test(); + + ts.awaitOnSubscribe(1, TimeUnit.SECONDS); + + while (!dsp.hasStreamers()) { + Thread.sleep(0, 1000); + } + + Streamable.error(new TestException()) + .subscribe(dsp); + + ts.awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTest.java index 813fadad4e..cf68e65f41 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTest.java @@ -15,11 +15,12 @@ import static org.junit.jupiter.api.Assertions.*; -import java.util.*; -import java.util.concurrent.*; -import java.util.concurrent.atomic.AtomicInteger; +import java.util.NoSuchElementException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.*; import org.junit.jupiter.api.*; + import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.exceptions.TestException; import io.reactivex.rxjava4.internal.subscriptions.EmptySubscription; diff --git a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java index 4f338b4905..206dffd806 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java +++ b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java @@ -34,6 +34,7 @@ import io.reactivex.rxjava4.internal.operators.streamable.StreamableNever; import io.reactivex.rxjava4.parallel.*; import io.reactivex.rxjava4.plugins.RxJavaPlugins; +import io.reactivex.rxjava4.processors.DispatchStreamProcessor; import io.reactivex.rxjava4.schedulers.Schedulers; import io.reactivex.rxjava4.testsupport.TestHelper; @@ -620,6 +621,8 @@ public void checkStreamable() { defaultValues.put(StreamableInterceptConfig.class, new StreamableInterceptConfig((_, v) -> v)); + defaultValues.put(StreamerInput.class, new DispatchStreamProcessor<>()); + // TODO insert new config record types here @SuppressWarnings("rawtypes") From c1efe46a038d7fe76275adad78669f32fb87ad88 Mon Sep 17 00:00:00 2001 From: akarnokd Date: Tue, 7 Jul 2026 08:28:22 +0200 Subject: [PATCH 3/8] Improve API, fix bugs, add coverage --- .../io/reactivex/rxjava4/core/Streamable.java | 21 +- .../reactivex/rxjava4/core/StreamerInput.java | 47 +++- .../streamable/StreamableForEach.java | 16 +- .../streamable/StreamerInputLambda.java | 59 +++++ .../StreamerInputWithCancellation.java | 50 +++++ .../DispatchStreamProcessorTest.java | 105 ++------- .../streamable/StreamableBaseTest.java | 40 ++++ .../streamable/StreamableForEachTest.java | 210 +++++++++++++++++- 8 files changed, 432 insertions(+), 116 deletions(-) create mode 100644 src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamerInputLambda.java create mode 100644 src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamerInputWithCancellation.java diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamable.java b/src/main/java/io/reactivex/rxjava4/core/Streamable.java index cfe01c408f..6f7f83921c 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamable.java @@ -995,24 +995,31 @@ default void subscribe(@NonNull Flow.Subscriber subscriber) { } /** - * Relays the events of the upstream into a {@link StreamerInput} consumer. + * Relays the events of the upstream into a {@link StreamerInput} consumer + * via the help of the standard {@link Executors#newVirtualThreadPerTaskExecutor()} + * as a mediator for pull-to-push. * @param consumer the consumer to relay events into - * @throws NullPointerException if {@code consumer} or {@code executor} is {@code null} + * @return the stage that gets completed normally or with an exception when this + * {@code Streamable} terminates + * @throws NullPointerException if {@code consumer} is {@code null} */ - default void subscribe(@NonNull StreamerInput consumer) { - subscribe(consumer, Executors.newVirtualThreadPerTaskExecutor()); + default CompletionStage subscribe(@NonNull StreamerInput consumer) { + return subscribe(consumer, Executors.newVirtualThreadPerTaskExecutor()); } /** - * Relays the events of the upstream into a {@link StreamerInput} consumer. + * Relays the events of the upstream into a {@link StreamerInput} consumer + * via the help of the given {@link ExecutorService} as a mediator for pull-to-push. * @param consumer the consumer to relay events into * @param executor the {@link ExecutorService} to run the blocking consume and emissions + * @return the stage that gets completed normally or with an exception when this + * {@code Streamable} terminates * @throws NullPointerException if {@code consumer} or {@code executor} is {@code null} */ - default void subscribe(@NonNull StreamerInput consumer, ExecutorService executor) { + default CompletionStage subscribe(@NonNull StreamerInput consumer, ExecutorService executor) { Objects.requireNonNull(consumer, "consumer is null"); Objects.requireNonNull(executor, "executor is null"); - StreamableForEach.forEach(this, consumer, executor); + return StreamableForEach.forEach(this, consumer, executor); } /** diff --git a/src/main/java/io/reactivex/rxjava4/core/StreamerInput.java b/src/main/java/io/reactivex/rxjava4/core/StreamerInput.java index 43bf0699d4..a746ce8b59 100644 --- a/src/main/java/io/reactivex/rxjava4/core/StreamerInput.java +++ b/src/main/java/io/reactivex/rxjava4/core/StreamerInput.java @@ -13,11 +13,14 @@ package io.reactivex.rxjava4.core; -import java.util.concurrent.CompletionStage; +import java.util.Objects; +import java.util.concurrent.*; import java.util.concurrent.Flow.Subscriber; import io.reactivex.rxjava4.annotations.*; import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.functions.Function; +import io.reactivex.rxjava4.internal.operators.streamable.*; /** * An interface to submit items and terminal events to a consumer that indacates when the processing of @@ -37,22 +40,62 @@ public interface StreamerInput<@NonNull T> { * @return a {@link CompletionStage} that completes with {@code true} if the value was successfully consumed, * {@code false} if the value was rejected or exceptionally on error */ + @NonNull CompletionStage next(T item); /** * Offer the final, terminal event. - * @param throwable the optional throwable to signal error, null to signal normal completion + * @param throwable the optional throwable to signal error, {@code null} to signal normal completion * @return a {@link CompletionStage} that completes with {@code null} if the call succeeded * or exceptionally on error */ + @NonNull CompletionStage finish(@Nullable Throwable throwable); /** * Returns the {@link DisposableContainer} to use to detect if the consumer has indicated no more * items it is willing to accept. + *

+ * The default implementation returns a fresh {@link CompositeDisposable}. * @return the {@code DisposableContainer} */ + @NonNull default DisposableContainer cancellation() { return new CompositeDisposable(); } + + /** + * Returns a new {@link StreamerInput} that returns the given {@link DisposableContainer} + * in its {@link #cancellation()}, allowing overriding the cancellation management + * of this {@code StreamerInput} + * @param cancellation the {@link DisposableContainer} to use as cancellation management + * @return the new {@code StreamerInput} instance + * @throws NullPointerException if {@code cancellation} is {@code null} + */ + @NonNull + default StreamerInput withCancellation(DisposableContainer cancellation) { + Objects.requireNonNull(cancellation, "cancellation is null"); + return new StreamerInputWithCancellation<>(this, cancellation); + } + + /** + * Creates a {@link StreamerInput} via lambda callbacks for {@link #next(Object)} and + * {@link #finish(Throwable)}. + *

+ * Non-fatal exceptions thrown by the callbacks are turned into failed + * {@link CompletableFuture#failedFuture(Throwable)}s. + * @param the element type of the stream + * @param onNext the callback for the {@code next} method + * @param onFinish the callback for the {@code finish} method + * @return the new {@link StreamerInput} instance + */ + @NonNull + static StreamerInput create( + @NonNull Function> onNext, + @NonNull Function> onFinish + ) { + Objects.requireNonNull(onNext, "onNext is null"); + Objects.requireNonNull(onFinish, "onFinish is null"); + return new StreamerInputLambda<>(onNext, onFinish); + } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java index 438aeb9045..e293ce39f9 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java @@ -22,7 +22,6 @@ import io.reactivex.rxjava4.exceptions.Exceptions; import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.util.ExceptionHelper; -import io.reactivex.rxjava4.plugins.RxJavaPlugins; /** * ForEach implementation to unclutter the {@link Streamable} type. @@ -100,10 +99,10 @@ public static CompletionStageDisposable forEach( } } catch (final Throwable crash) { Exceptions.throwIfFatal(crash); - throw ExceptionHelper.wrapOrThrow(ExceptionHelper.unwrapAndCombine(crash, finallyCrash)); + finallyCrash = ExceptionHelper.unwrapAndCombine(crash, finallyCrash); } if (finallyCrash != null) { - throw ExceptionHelper.wrapOrThrow(ExceptionHelper.unwrap(finallyCrash)); + throw ExceptionHelper.wrapOrThrow(finallyCrash); } return null; }); @@ -111,7 +110,8 @@ public static CompletionStageDisposable forEach( return new CompletionStageDisposable<>(future, canceller); } - public static void forEach(Streamable me, StreamerInput consumer, ExecutorService executor) { + public static CompletionStage forEach(Streamable me, StreamerInput consumer, ExecutorService executor) { + var cf = new CompletableFuture(); CompletableFuture.runAsync(() -> { Throwable error = null; var cancellation = consumer.cancellation(); @@ -145,8 +145,14 @@ public static void forEach(Streamable me, StreamerInput consum Streamer.awaitVoid(consumer.finish(error)); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); - RxJavaPlugins.onError(ex); + if (error != null) { + ex.addSuppressed(error); + } + cf.completeExceptionally(ex); + return; } + cf.complete(null); }, executor); + return cf; } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamerInputLambda.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamerInputLambda.java new file mode 100644 index 0000000000..4ddf89697f --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamerInputLambda.java @@ -0,0 +1,59 @@ +/* + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.rxjava4.internal.operators.streamable; + +import java.util.Objects; +import java.util.concurrent.*; + +import io.reactivex.rxjava4.annotations.*; +import io.reactivex.rxjava4.core.StreamerInput; +import io.reactivex.rxjava4.exceptions.Exceptions; +import io.reactivex.rxjava4.functions.Function; + +/** + * Creates a {@link StreamerInput} via lambda callbacks for {@link #next(Object)} and + * {@link #finish(Throwable)}. + * @param the element type of the stream + * @param onNext the callback for the {@code next} method + * @param onFinish the callback for the {@code finish} method + * @since 4.0.0 + */ +public record StreamerInputLambda<@NonNull T>( + @NonNull Function> onNext, + @NonNull Function> onFinish +) implements StreamerInput { + + @Override + public @NonNull CompletionStage next(@NonNull T item) { + try { + return Objects.requireNonNull(onNext.apply(item), "onNext returned a null CompletionStage"); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + return CompletableFuture.failedStage(ex); + } + } + + @Override + public @NonNull CompletionStage finish(@Nullable Throwable throwable) { + try { + return Objects.requireNonNull(onFinish.apply(throwable), "onFinish returned a null CompletionStage"); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + if (throwable != null) { + ex.addSuppressed(throwable); + } + return CompletableFuture.failedStage(ex); + } + } +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamerInputWithCancellation.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamerInputWithCancellation.java new file mode 100644 index 0000000000..43e50f3be1 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamerInputWithCancellation.java @@ -0,0 +1,50 @@ +/* + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.rxjava4.internal.operators.streamable; + +import java.util.Objects; +import java.util.concurrent.CompletionStage; + +import io.reactivex.rxjava4.annotations.*; +import io.reactivex.rxjava4.core.StreamerInput; +import io.reactivex.rxjava4.disposables.DisposableContainer; + +/** + * Wraps a {@link StreamerInput} and uses the given {@link DisposableContainer} to be + * returned via {@link #cancellation()}. + * @param the element type of the stream + * @param downstream the {@code StreamerInput} to relay events to + * @param cancellation the {@code DisposableContainer} to be used for indicating cancellation + * @since 4.0.0 + */ +public record StreamerInputWithCancellation<@NonNull T>( + @NonNull StreamerInput downstream, @NonNull DisposableContainer cancellation) +implements StreamerInput { + + public StreamerInputWithCancellation { + Objects.requireNonNull(downstream, "downstream is null"); + } + + @Override + @NonNull + public CompletionStage next(@NonNull T item) { + return downstream.next(item); + } + + @Override + @NonNull + public CompletionStage finish(@Nullable Throwable throwable) { + return downstream.finish(throwable); + } +} 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 3e6f097382..b9ad929805 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,13 +40,7 @@ public void normal() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); - int timeout = 1000; - while (!dsp.hasStreamers()) { - Thread.sleep(1); - if (timeout-- < 0) { - throw new TimeoutException("hasStreamers = " + dsp.hasStreamers()); - } - } + awaitStreamers(dsp, 1000); for (int i = 1; i < 6; i++) { dsp.next(i).toCompletableFuture().join(); @@ -75,13 +69,7 @@ public void endsInError() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); - int timeout = 1000; - while (!dsp.hasStreamers()) { - Thread.sleep(1); - if (timeout-- < 0) { - throw new TimeoutException("hasStreamers = " + dsp.hasStreamers()); - } - } + awaitStreamers(dsp, 1000); for (int i = 1; i < 6; i++) { dsp.next(i).toCompletableFuture().join(); @@ -112,13 +100,7 @@ public void normalDebug() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); - int timeout = 1000; - while (!dsp.hasStreamers()) { - Thread.sleep(1); - if (timeout-- < 0) { - throw new TimeoutException("hasStreamers = " + dsp.hasStreamers()); - } - } + awaitStreamers(dsp, 1000); assertTrue(dsp.hasStreamers(), "dsp has no streamers?"); @@ -196,26 +178,15 @@ public void normalTake3AltDebug() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); - int timeout = 1000; - while (!dsp.hasStreamers()) { - Thread.sleep(1); - if (timeout-- < 0) { - throw new TimeoutException("hasStreamers = " + dsp.hasStreamers()); - } - } + awaitStreamers(dsp, 1000); for (int i = 1; i < 4; i++) { IO.println(i + " -> next"); dsp.next(i).toCompletableFuture().join(); } - timeout = 1000; - while (dsp.hasStreamers()) { - Thread.sleep(1); - if (timeout-- < 0) { - throw new TimeoutException("hasStreamers = " + dsp.hasStreamers()); - } - } + awaitNoStreamers(dsp, 1000); + assertFalse(dsp.hasStreamers(), "dsp has streamers?"); for (int i = 4; i < 6; i++) { @@ -251,13 +222,7 @@ public void normalTake3Debug() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); - int timeout = 1000; - while (!dsp.hasStreamers()) { - Thread.sleep(1); - if (timeout-- < 0) { - throw new TimeoutException("hasStreamers = " + dsp.hasStreamers()); - } - } + awaitStreamers(dsp, 1000); for (int i = 1; i < 6; i++) { dsp.next(i).toCompletableFuture().join(); @@ -292,13 +257,7 @@ public void normalMulti() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); ts2.awaitOnSubscribe(1, TimeUnit.SECONDS); - int timeout = 1000; - while (!dsp.hasStreamers()) { - Thread.sleep(1); - if (timeout-- < 0) { - throw new TimeoutException("hasStreamers = " + dsp.hasStreamers()); - } - } + awaitStreamers(dsp, 1000); for (int i = 1; i < 6; i++) { dsp.next(i).toCompletableFuture().join(); @@ -332,13 +291,7 @@ public void normalMultiOtherCancels() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); ts2.awaitOnSubscribe(1, TimeUnit.SECONDS); - int timeout = 1000; - while (!dsp.hasStreamers()) { - Thread.sleep(1); - if (timeout-- < 0) { - throw new TimeoutException("hasStreamers = " + dsp.hasStreamers()); - } - } + awaitStreamers(dsp, 1000); ts2.cancel(); @@ -373,13 +326,7 @@ public void normalMultiFirstCancels() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); ts2.awaitOnSubscribe(1, TimeUnit.SECONDS); - int timeout = 1000; - while (!dsp.hasStreamers()) { - Thread.sleep(1); - if (timeout-- < 0) { - throw new TimeoutException("hasStreamers = " + dsp.hasStreamers()); - } - } + awaitStreamers(dsp, 1000); ts.cancel(); @@ -420,24 +367,12 @@ public void raceToStream() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); ts2.awaitOnSubscribe(1, TimeUnit.SECONDS); - int timeout = 1000; - while (!dsp.hasStreamers()) { - if (timeout-- < 0) { - throw new TimeoutException("hasStreamers = " + dsp.hasStreamers()); - } - Thread.sleep(0, 1000); - } + awaitStreamers(dsp, 1000); ts.cancel(); ts2.cancel(); - timeout = 1000; - while (dsp.hasStreamers()) { - if (timeout-- < 0) { - throw new TimeoutException("hasStreamers = " + dsp.hasStreamers()); - } - Thread.sleep(0, 1000); - } + awaitNoStreamers(dsp, 1000); assertFalse(dsp.hasStreamers(), "dsp has streamers?"); assertFalse(dsp.hasComplete(), "dsp has completed?"); @@ -459,23 +394,11 @@ public void comeAndGo() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); - int timeout = 1000; - while (!dsp.hasStreamers()) { - if (timeout-- < 0) { - throw new TimeoutException("hasStreamers = " + dsp.hasStreamers()); - } - Thread.sleep(1); - } + awaitStreamers(dsp, 1000); ts.cancel(); - timeout = 1000; - while (dsp.hasStreamers()) { - if (timeout-- < 0) { - throw new TimeoutException("hasStreamers = " + dsp.hasStreamers()); - } - Thread.sleep(1); - } + awaitNoStreamers(dsp, 1000); assertFalse(dsp.hasStreamers(), "dsp has streamers?"); assertFalse(dsp.hasComplete(), "dsp has completed?"); 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 dff5ecf8cf..289ec8e85d 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,6 +15,7 @@ import java.lang.ref.Cleaner; import java.util.*; +import java.util.concurrent.TimeoutException; import java.util.function.BiConsumer; import org.junit.jupiter.api.*; @@ -105,4 +106,43 @@ protected final void setUndeliverablesExpected(boolean isExpected) { public static StreamableInterceptConfig debugIntercept() { return (StreamableInterceptConfig)DEBUG_INTERCEPT; } + + /** + * Awaits the given {@link StreamProcessor#hasStreamers()} to register an + * incoming consumer. + * @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 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"); + } + } + } + /** + * Awaits the given {@link StreamProcessor#hasStreamers()} to lose + * all of its streamers. + * @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 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"); + } + } + } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java index 2b788c852a..dffc6246bf 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java @@ -16,11 +16,11 @@ import static org.junit.jupiter.api.Assertions.*; import java.util.concurrent.*; -import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.*; import org.junit.jupiter.api.Test; -import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.CompositeDisposable; import io.reactivex.rxjava4.exceptions.*; import io.reactivex.rxjava4.processors.DispatchStreamProcessor; @@ -199,9 +199,7 @@ public void forEachInput() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); - while (!dsp.hasStreamers()) { - Thread.sleep(0, 1000); - } + awaitStreamers(dsp, 1000); Streamable.range(1, 5) .subscribe(dsp); @@ -218,9 +216,7 @@ public void forEachInputDebug() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); - while (!dsp.hasStreamers()) { - Thread.sleep(0, 1000); - } + awaitStreamers(dsp, 1000); Streamable.range(1, 5) .subscribe(dsp); @@ -237,9 +233,7 @@ public void forEachInputError() throws Throwable { ts.awaitOnSubscribe(1, TimeUnit.SECONDS); - while (!dsp.hasStreamers()) { - Thread.sleep(0, 1000); - } + awaitStreamers(dsp, 1000); Streamable.error(new TestException()) .subscribe(dsp); @@ -247,4 +241,198 @@ public void forEachInputError() throws Throwable { ts.awaitDone(5, TimeUnit.SECONDS) .assertFailure(TestException.class); } + + @Test + public void forEachInputCancelUpfront() throws Throwable { + var dsp0 = new DispatchStreamProcessor<>(); + + var dsp = new DispatchStreamProcessor<>(); + var ts = dsp.test(); + + ts.awaitOnSubscribe(1, TimeUnit.SECONDS); + + awaitStreamers(dsp, 1000); + + var cd = new CompositeDisposable(); + cd.dispose(); + dsp0.subscribe(dsp.withCancellation(cd)); + + awaitNoStreamers(dsp0, 1000); + + assertFalse(dsp.hasComplete(), "dsp completes: error = " + dsp.hasThrowable()); + } + + @Test + public void forEachInputSendNull() throws Throwable { + IO.println("forEachInputSendNull"); + var error = new AtomicReference(); + var dsp = new DispatchStreamProcessor<>(); + var si = StreamerInput.create(_ -> null, e -> { error.set(e); return Streamer.FINISHED; }); + var f = dsp.subscribe(si); + + IO.println(" hasStreamers()"); + + awaitStreamers(dsp, 1000); + + IO.println(" next(1)"); + + dsp.next(1).toCompletableFuture().join(); + + IO.println(" f.toCompletableFuture.join"); + + f.toCompletableFuture().join(); + + assertTrue(error.get() instanceof NullPointerException, "" + error.get()); + + IO.println(" ."); + } + + @Test + public void forEachInputSendCrash() throws Throwable { + IO.println("forEachInputSendCrash"); + var error = new AtomicReference(); + var dsp = new DispatchStreamProcessor<>(); + var si = StreamerInput.create(_ -> { throw new TestException(); }, e -> { error.set(e); return Streamer.FINISHED; }); + var f = dsp.subscribe(si); + + IO.println(" hasStreamers()"); + + awaitStreamers(dsp, 1000); + + IO.println(" next(1)"); + dsp.next(1).toCompletableFuture().join(); + + IO.println(" f.toCompletableFuture.join"); + + f.toCompletableFuture().join(); + + assertTrue(error.get() instanceof TestException, "" + error.get()); + + IO.println(" ."); + } + + @Test + public void forEachInputTerminateNull() throws Throwable { + IO.println("forEachInputTerminateNull"); + var dsp = new DispatchStreamProcessor<>(); + var si = StreamerInput.create(_ -> Streamer.NEXT_TRUE, _ -> { return null; }); + var f = dsp.subscribe(si); + + IO.println(" hasStreamers()"); + + awaitStreamers(dsp, 1000); + + IO.println(" next(1)"); + dsp.next(1).toCompletableFuture().join(); + + IO.println(" finish()"); + dsp.finish(null).toCompletableFuture().join(); + + IO.println(" f.toCompletableFuture.join"); + + var ex = assertThrows(CompletionException.class, () -> { + f.toCompletableFuture().join(); + }); + + assertTrue(ex.getCause() instanceof NullPointerException, ex.getCause().toString()); + + IO.println(" ."); + } + + @Test + public void forEachInputTerminateCrash() throws Throwable { + IO.println("forEachInputTerminateNull"); + var dsp = new DispatchStreamProcessor<>(); + var si = StreamerInput.create(_ -> Streamer.NEXT_TRUE, _ -> { throw new TestException(); }); + var f = dsp.subscribe(si); + + IO.println(" hasStreamers()"); + + awaitStreamers(dsp, 1000); + + IO.println(" next(1)"); + dsp.next(1).toCompletableFuture().join(); + + IO.println(" finish()"); + dsp.finish(null).toCompletableFuture().join(); + + IO.println(" f.toCompletableFuture.join"); + + var ex = assertThrows(CompletionException.class, () -> { + f.toCompletableFuture().join(); + }); + + assertTrue(ex.getCause() instanceof TestException, ex.getCause().toString()); + + IO.println(" ."); + } + + @Test + public void forEachInputTerminateBothCrash() throws Throwable { + IO.println("forEachInputTerminateNull"); + var dsp = new DispatchStreamProcessor<>(); + var si = StreamerInput.create(_ -> null, _ -> { throw new TestException(); }); + var f = dsp.subscribe(si); + + IO.println(" hasStreamers()"); + + awaitStreamers(dsp, 1000); + + IO.println(" next(1)"); + dsp.next(1).toCompletableFuture().join(); + + IO.println(" finish()"); + dsp.finish(null).toCompletableFuture().join(); + + IO.println(" f.toCompletableFuture.join"); + + var ex = assertThrows(CompletionException.class, () -> { + f.toCompletableFuture().join(); + }); + + assertTrue(ex.getCause() instanceof TestException, ex.getCause().toString()); + assertTrue(ex.getCause().getSuppressed()[0] instanceof NullPointerException, + ex.getCause().getSuppressed()[0].toString()); + + IO.println(" ."); + } + + @Test + public void forEachUpstreamFinishCrash() throws Throwable { + var dsp = new DispatchStreamProcessor<>(); + var ts = dsp.test(); + + ts.awaitOnSubscribe(1, TimeUnit.SECONDS); + awaitStreamers(dsp, 1000); + + StreamableFailingFinish.MAIN_COMPLETES + .subscribe(dsp); + + ts.awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void forEachBiUpstreamFinishCrash() throws Throwable { + var fs = StreamableFailingFinish.MAIN_COMPLETES + .forEach((_, _) -> { + }, new CompositeDisposable(), Executors.newVirtualThreadPerTaskExecutor()); + + assertThrows(TestException.class, () -> { + fs.await(); + }); + } + + @Test + public void forEachBiUpstreamFinishCrashDebug() throws Throwable { + withCachedExecutor(exec -> { + var fs = StreamableFailingFinish.MAIN_COMPLETES + .forEach((_, _) -> { + }, new CompositeDisposable(), exec); + + assertThrows(TestException.class, () -> { + fs.await(); + }); + }); + } } From be8e2015fa5f9db8c1c9a168baa47030434dc163 Mon Sep 17 00:00:00 2001 From: akarnokd Date: Tue, 7 Jul 2026 08:30:34 +0200 Subject: [PATCH 4/8] fix style --- src/main/java/io/reactivex/rxjava4/core/Streamable.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamable.java b/src/main/java/io/reactivex/rxjava4/core/Streamable.java index 6f7f83921c..0059f3a90b 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamable.java @@ -1003,7 +1003,7 @@ default void subscribe(@NonNull Flow.Subscriber subscriber) { * {@code Streamable} terminates * @throws NullPointerException if {@code consumer} is {@code null} */ - default CompletionStage subscribe(@NonNull StreamerInput consumer) { + default CompletionStage subscribe(@NonNull StreamerInput consumer) { return subscribe(consumer, Executors.newVirtualThreadPerTaskExecutor()); } @@ -1016,7 +1016,7 @@ default CompletionStage subscribe(@NonNull StreamerInput consu * {@code Streamable} terminates * @throws NullPointerException if {@code consumer} or {@code executor} is {@code null} */ - default CompletionStage subscribe(@NonNull StreamerInput consumer, ExecutorService executor) { + default CompletionStage subscribe(@NonNull StreamerInput consumer, ExecutorService executor) { Objects.requireNonNull(consumer, "consumer is null"); Objects.requireNonNull(executor, "executor is null"); return StreamableForEach.forEach(this, consumer, executor); From ee5956ea46320ad43286a9875306db473a604f62 Mon Sep 17 00:00:00 2001 From: akarnokd Date: Tue, 7 Jul 2026 08:46:30 +0200 Subject: [PATCH 5/8] Fix flaky DispatchStreamProcessorTest > normalMulti() --- .../rxjava4/core/StreamProcessor.java | 6 +++++ .../processors/DispatchStreamProcessor.java | 5 +++++ .../DispatchStreamProcessorTest.java | 10 ++++----- .../streamable/StreamableBaseTest.java | 22 +++++++++++++++++++ 4 files changed, 38 insertions(+), 5 deletions(-) diff --git a/src/main/java/io/reactivex/rxjava4/core/StreamProcessor.java b/src/main/java/io/reactivex/rxjava4/core/StreamProcessor.java index e9606431d7..4d73f3e8b7 100644 --- a/src/main/java/io/reactivex/rxjava4/core/StreamProcessor.java +++ b/src/main/java/io/reactivex/rxjava4/core/StreamProcessor.java @@ -34,6 +34,12 @@ public interface StreamProcessor<@NonNull In, @NonNull Out> extends Streamable sp, long timeoutMillis) } } } + + /** + * Awaits the given {@link StreamProcessor#hasStreamers()} to register an + * incoming consumer. + * @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 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"); + } + } + } + /** * Awaits the given {@link StreamProcessor#hasStreamers()} to lose * all of its streamers. From f74b39c192d72f497b53c51b292dfb8c1a34913b Mon Sep 17 00:00:00 2001 From: akarnokd Date: Tue, 7 Jul 2026 08:57:49 +0200 Subject: [PATCH 6/8] improve coverage of withCancellation, rename to StreamSink --- .../rxjava4/core/StreamProcessor.java | 4 +-- .../{StreamerInput.java => StreamSink.java} | 20 +++++------ .../io/reactivex/rxjava4/core/Streamable.java | 8 ++--- ...InputLambda.java => StreamSinkLambda.java} | 8 ++--- ...n.java => StreamSinkWithCancellation.java} | 14 ++++---- .../streamable/StreamableForEach.java | 2 +- .../streamable/StreamableGroupBy.java | 2 +- .../streamable/StreamableForEachTest.java | 35 ++++++++++++++++--- .../validators/CheckParamValidationTest.java | 2 +- 9 files changed, 60 insertions(+), 35 deletions(-) rename src/main/java/io/reactivex/rxjava4/core/{StreamerInput.java => StreamSink.java} (85%) rename src/main/java/io/reactivex/rxjava4/internal/operators/streamable/{StreamerInputLambda.java => StreamSinkLambda.java} (90%) rename src/main/java/io/reactivex/rxjava4/internal/operators/streamable/{StreamerInputWithCancellation.java => StreamSinkWithCancellation.java} (77%) diff --git a/src/main/java/io/reactivex/rxjava4/core/StreamProcessor.java b/src/main/java/io/reactivex/rxjava4/core/StreamProcessor.java index 4d73f3e8b7..2757882935 100644 --- a/src/main/java/io/reactivex/rxjava4/core/StreamProcessor.java +++ b/src/main/java/io/reactivex/rxjava4/core/StreamProcessor.java @@ -20,13 +20,13 @@ /** * A {@link Processor}-like interface combining the {@code Streamable} interface and the - * {@link StreamerInput} interface to establish a push-pull bridge based on {@link CompletionStage}-based + * {@link StreamSink} interface to establish a push-pull bridge based on {@link CompletionStage}-based * asynchronous processing and dispatching of values and errors. * @param the element type of the input side * @param the element type of the output side * @since 4.0.0 */ -public interface StreamProcessor<@NonNull In, @NonNull Out> extends Streamable, StreamerInput { +public interface StreamProcessor<@NonNull In, @NonNull Out> extends Streamable, StreamSink { /** * Returns {@code true} if this {@link StreamProcessor} has {@link Streamer}s. diff --git a/src/main/java/io/reactivex/rxjava4/core/StreamerInput.java b/src/main/java/io/reactivex/rxjava4/core/StreamSink.java similarity index 85% rename from src/main/java/io/reactivex/rxjava4/core/StreamerInput.java rename to src/main/java/io/reactivex/rxjava4/core/StreamSink.java index a746ce8b59..ffcb40f5b5 100644 --- a/src/main/java/io/reactivex/rxjava4/core/StreamerInput.java +++ b/src/main/java/io/reactivex/rxjava4/core/StreamSink.java @@ -32,7 +32,7 @@ * @param the item type to be offered * @since 4.0.0 */ -public interface StreamerInput<@NonNull T> { +public interface StreamSink<@NonNull T> { /** * Offer the next item. @@ -65,21 +65,21 @@ default DisposableContainer cancellation() { } /** - * Returns a new {@link StreamerInput} that returns the given {@link DisposableContainer} + * Returns a new {@link StreamSink} that returns the given {@link DisposableContainer} * in its {@link #cancellation()}, allowing overriding the cancellation management - * of this {@code StreamerInput} + * of this {@code StreamSink} * @param cancellation the {@link DisposableContainer} to use as cancellation management - * @return the new {@code StreamerInput} instance + * @return the new {@code StreamSink} instance * @throws NullPointerException if {@code cancellation} is {@code null} */ @NonNull - default StreamerInput withCancellation(DisposableContainer cancellation) { + default StreamSink withCancellation(DisposableContainer cancellation) { Objects.requireNonNull(cancellation, "cancellation is null"); - return new StreamerInputWithCancellation<>(this, cancellation); + return new StreamSinkWithCancellation<>(this, cancellation); } /** - * Creates a {@link StreamerInput} via lambda callbacks for {@link #next(Object)} and + * Creates a {@link StreamSink} via lambda callbacks for {@link #next(Object)} and * {@link #finish(Throwable)}. *

* Non-fatal exceptions thrown by the callbacks are turned into failed @@ -87,15 +87,15 @@ default StreamerInput withCancellation(DisposableContainer cancellation) { * @param the element type of the stream * @param onNext the callback for the {@code next} method * @param onFinish the callback for the {@code finish} method - * @return the new {@link StreamerInput} instance + * @return the new {@link StreamSink} instance */ @NonNull - static StreamerInput create( + static StreamSink create( @NonNull Function> onNext, @NonNull Function> onFinish ) { Objects.requireNonNull(onNext, "onNext is null"); Objects.requireNonNull(onFinish, "onFinish is null"); - return new StreamerInputLambda<>(onNext, onFinish); + return new StreamSinkLambda<>(onNext, onFinish); } } diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamable.java b/src/main/java/io/reactivex/rxjava4/core/Streamable.java index 0059f3a90b..218dcc3c28 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamable.java @@ -995,7 +995,7 @@ default void subscribe(@NonNull Flow.Subscriber subscriber) { } /** - * Relays the events of the upstream into a {@link StreamerInput} consumer + * Relays the events of the upstream into a {@link StreamSink} consumer * via the help of the standard {@link Executors#newVirtualThreadPerTaskExecutor()} * as a mediator for pull-to-push. * @param consumer the consumer to relay events into @@ -1003,12 +1003,12 @@ default void subscribe(@NonNull Flow.Subscriber subscriber) { * {@code Streamable} terminates * @throws NullPointerException if {@code consumer} is {@code null} */ - default CompletionStage subscribe(@NonNull StreamerInput consumer) { + default CompletionStage subscribe(@NonNull StreamSink consumer) { return subscribe(consumer, Executors.newVirtualThreadPerTaskExecutor()); } /** - * Relays the events of the upstream into a {@link StreamerInput} consumer + * Relays the events of the upstream into a {@link StreamSink} consumer * via the help of the given {@link ExecutorService} as a mediator for pull-to-push. * @param consumer the consumer to relay events into * @param executor the {@link ExecutorService} to run the blocking consume and emissions @@ -1016,7 +1016,7 @@ default CompletionStage subscribe(@NonNull StreamerInput consum * {@code Streamable} terminates * @throws NullPointerException if {@code consumer} or {@code executor} is {@code null} */ - default CompletionStage subscribe(@NonNull StreamerInput consumer, ExecutorService executor) { + default CompletionStage subscribe(@NonNull StreamSink consumer, ExecutorService executor) { Objects.requireNonNull(consumer, "consumer is null"); Objects.requireNonNull(executor, "executor is null"); return StreamableForEach.forEach(this, consumer, executor); diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamerInputLambda.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamSinkLambda.java similarity index 90% rename from src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamerInputLambda.java rename to src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamSinkLambda.java index 4ddf89697f..f485622140 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamerInputLambda.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamSinkLambda.java @@ -17,22 +17,22 @@ import java.util.concurrent.*; import io.reactivex.rxjava4.annotations.*; -import io.reactivex.rxjava4.core.StreamerInput; +import io.reactivex.rxjava4.core.StreamSink; import io.reactivex.rxjava4.exceptions.Exceptions; import io.reactivex.rxjava4.functions.Function; /** - * Creates a {@link StreamerInput} via lambda callbacks for {@link #next(Object)} and + * Creates a {@link StreamSink} via lambda callbacks for {@link #next(Object)} and * {@link #finish(Throwable)}. * @param the element type of the stream * @param onNext the callback for the {@code next} method * @param onFinish the callback for the {@code finish} method * @since 4.0.0 */ -public record StreamerInputLambda<@NonNull T>( +public record StreamSinkLambda<@NonNull T>( @NonNull Function> onNext, @NonNull Function> onFinish -) implements StreamerInput { +) implements StreamSink { @Override public @NonNull CompletionStage next(@NonNull T item) { diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamerInputWithCancellation.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamSinkWithCancellation.java similarity index 77% rename from src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamerInputWithCancellation.java rename to src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamSinkWithCancellation.java index 43e50f3be1..4185786525 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamerInputWithCancellation.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamSinkWithCancellation.java @@ -17,22 +17,22 @@ import java.util.concurrent.CompletionStage; import io.reactivex.rxjava4.annotations.*; -import io.reactivex.rxjava4.core.StreamerInput; +import io.reactivex.rxjava4.core.StreamSink; import io.reactivex.rxjava4.disposables.DisposableContainer; /** - * Wraps a {@link StreamerInput} and uses the given {@link DisposableContainer} to be + * Wraps a {@link StreamSink} and uses the given {@link DisposableContainer} to be * returned via {@link #cancellation()}. * @param the element type of the stream - * @param downstream the {@code StreamerInput} to relay events to + * @param downstream the {@code StreamSink} to relay events to * @param cancellation the {@code DisposableContainer} to be used for indicating cancellation * @since 4.0.0 */ -public record StreamerInputWithCancellation<@NonNull T>( - @NonNull StreamerInput downstream, @NonNull DisposableContainer cancellation) -implements StreamerInput { +public record StreamSinkWithCancellation<@NonNull T>( + @NonNull StreamSink downstream, @NonNull DisposableContainer cancellation) +implements StreamSink { - public StreamerInputWithCancellation { + public StreamSinkWithCancellation { Objects.requireNonNull(downstream, "downstream is null"); } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java index e293ce39f9..6a1d437f86 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java @@ -110,7 +110,7 @@ public static CompletionStageDisposable forEach( return new CompletionStageDisposable<>(future, canceller); } - public static CompletionStage forEach(Streamable me, StreamerInput consumer, ExecutorService executor) { + public static CompletionStage forEach(Streamable me, StreamSink consumer, ExecutorService executor) { var cf = new CompletableFuture(); CompletableFuture.runAsync(() -> { Throwable error = null; diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableGroupBy.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableGroupBy.java index a2c612ca6d..e61718c35e 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableGroupBy.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableGroupBy.java @@ -176,7 +176,7 @@ boolean isDeleted(K key) { } static abstract class BasicGroupedStreamable extends GroupedStreamable - implements StreamerInput { + implements StreamSink { BasicGroupedStreamable(K key) { super(key); } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java index dffc6246bf..aaf55af3bd 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java @@ -262,12 +262,37 @@ public void forEachInputCancelUpfront() throws Throwable { assertFalse(dsp.hasComplete(), "dsp completes: error = " + dsp.hasThrowable()); } + @Test + public void forEachInputWithCancellationOverride() throws Throwable { + var dspMain = new DispatchStreamProcessor<>(); + + var dspSecondary = new DispatchStreamProcessor<>(); + var ts = dspSecondary.test(); + + ts.awaitOnSubscribe(1, TimeUnit.SECONDS); + + awaitStreamers(dspSecondary, 1000); + + var cd = new CompositeDisposable(); + dspMain.subscribe(dspSecondary.withCancellation(cd)); + + awaitStreamers(dspMain, 1000); + + dspMain.next(1).toCompletableFuture().join(); + dspMain.finish(null).toCompletableFuture().join(); + + ts.awaitDone(5, TimeUnit.SECONDS) + .assertResult(1); + + assertTrue(dspSecondary.hasComplete(), "dsp completes: error = " + dspSecondary.hasThrowable()); + } + @Test public void forEachInputSendNull() throws Throwable { IO.println("forEachInputSendNull"); var error = new AtomicReference(); var dsp = new DispatchStreamProcessor<>(); - var si = StreamerInput.create(_ -> null, e -> { error.set(e); return Streamer.FINISHED; }); + var si = StreamSink.create(_ -> null, e -> { error.set(e); return Streamer.FINISHED; }); var f = dsp.subscribe(si); IO.println(" hasStreamers()"); @@ -292,7 +317,7 @@ public void forEachInputSendCrash() throws Throwable { IO.println("forEachInputSendCrash"); var error = new AtomicReference(); var dsp = new DispatchStreamProcessor<>(); - var si = StreamerInput.create(_ -> { throw new TestException(); }, e -> { error.set(e); return Streamer.FINISHED; }); + var si = StreamSink.create(_ -> { throw new TestException(); }, e -> { error.set(e); return Streamer.FINISHED; }); var f = dsp.subscribe(si); IO.println(" hasStreamers()"); @@ -315,7 +340,7 @@ public void forEachInputSendCrash() throws Throwable { public void forEachInputTerminateNull() throws Throwable { IO.println("forEachInputTerminateNull"); var dsp = new DispatchStreamProcessor<>(); - var si = StreamerInput.create(_ -> Streamer.NEXT_TRUE, _ -> { return null; }); + var si = StreamSink.create(_ -> Streamer.NEXT_TRUE, _ -> { return null; }); var f = dsp.subscribe(si); IO.println(" hasStreamers()"); @@ -343,7 +368,7 @@ public void forEachInputTerminateNull() throws Throwable { public void forEachInputTerminateCrash() throws Throwable { IO.println("forEachInputTerminateNull"); var dsp = new DispatchStreamProcessor<>(); - var si = StreamerInput.create(_ -> Streamer.NEXT_TRUE, _ -> { throw new TestException(); }); + var si = StreamSink.create(_ -> Streamer.NEXT_TRUE, _ -> { throw new TestException(); }); var f = dsp.subscribe(si); IO.println(" hasStreamers()"); @@ -371,7 +396,7 @@ public void forEachInputTerminateCrash() throws Throwable { public void forEachInputTerminateBothCrash() throws Throwable { IO.println("forEachInputTerminateNull"); var dsp = new DispatchStreamProcessor<>(); - var si = StreamerInput.create(_ -> null, _ -> { throw new TestException(); }); + var si = StreamSink.create(_ -> null, _ -> { throw new TestException(); }); var f = dsp.subscribe(si); IO.println(" hasStreamers()"); diff --git a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java index 206dffd806..5a1dc68c9e 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java +++ b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java @@ -621,7 +621,7 @@ public void checkStreamable() { defaultValues.put(StreamableInterceptConfig.class, new StreamableInterceptConfig((_, v) -> v)); - defaultValues.put(StreamerInput.class, new DispatchStreamProcessor<>()); + defaultValues.put(StreamSink.class, new DispatchStreamProcessor<>()); // TODO insert new config record types here From ca2568c21e6d80b2faf06ada8a2a673d24108b1a Mon Sep 17 00:00:00 2001 From: akarnokd Date: Tue, 7 Jul 2026 09:28:10 +0200 Subject: [PATCH 7/8] Remove ThrowableWrapper, fix attempt at flaky tests --- .../core/CompletionStageDisposable.java | 9 ++- .../rxjava4/exceptions/Exceptions.java | 4 +- .../rxjava4/exceptions/ThrowableWrapper.java | 48 ------------- .../operators/streamable/StreamableTimer.java | 4 +- .../internal/util/ExceptionHelper.java | 42 ++++++++--- .../FlowableVirtualCreateExecutor.java | 4 +- .../FlowableVirtualTransformExecutor.java | 8 +-- .../exceptions/ThrowableWrapperTest.java | 70 ------------------- .../streamable/StreamableForEachTest.java | 6 +- .../internal/util/ExceptionHelperTest.java | 4 +- 10 files changed, 52 insertions(+), 147 deletions(-) delete mode 100644 src/main/java/io/reactivex/rxjava4/exceptions/ThrowableWrapper.java delete mode 100644 src/test/java/io/reactivex/rxjava4/exceptions/ThrowableWrapperTest.java diff --git a/src/main/java/io/reactivex/rxjava4/core/CompletionStageDisposable.java b/src/main/java/io/reactivex/rxjava4/core/CompletionStageDisposable.java index 8824e66006..b697126c6b 100644 --- a/src/main/java/io/reactivex/rxjava4/core/CompletionStageDisposable.java +++ b/src/main/java/io/reactivex/rxjava4/core/CompletionStageDisposable.java @@ -21,9 +21,8 @@ import java.util.function.Consumer; import io.reactivex.rxjava4.annotations.NonNull; -import io.reactivex.rxjava4.disposables.*; -import io.reactivex.rxjava4.exceptions.ThrowableWrapper; -import io.reactivex.rxjava4.internal.util.*; +import io.reactivex.rxjava4.disposables.Disposable; +import io.reactivex.rxjava4.internal.util.ExceptionHelper; import io.reactivex.rxjava4.plugins.RxJavaPlugins; /** @@ -88,14 +87,14 @@ public CompletionStageDisposable(@NonNull CompletionStage stage, @NonNull Dis *

* Rethrows any original unchecked exceptions as is. * @throws CancellationException if the computation was cancelled - * @throws ThrowableWrapper if the original exception was a checked exception + * @throws CompletionException if the original exception was a checked exception */ public void await() { state.lazySet(true); try { stage.toCompletableFuture().join(); } catch (CompletionException ce) { - throw ExceptionHelper.wrapOrThrow(ce.getCause()); + throw ExceptionHelper.unwrapOrThrow(ce); } } diff --git a/src/main/java/io/reactivex/rxjava4/exceptions/Exceptions.java b/src/main/java/io/reactivex/rxjava4/exceptions/Exceptions.java index a745d93450..4bab9a369b 100644 --- a/src/main/java/io/reactivex/rxjava4/exceptions/Exceptions.java +++ b/src/main/java/io/reactivex/rxjava4/exceptions/Exceptions.java @@ -13,6 +13,8 @@ package io.reactivex.rxjava4.exceptions; +import java.util.concurrent.CompletionException; + import io.reactivex.rxjava4.annotations.NonNull; import io.reactivex.rxjava4.internal.util.ExceptionHelper; @@ -28,7 +30,7 @@ private Exceptions() { } /** * Convenience method to throw a {@code RuntimeException} and {@code Error} directly - * or wrap any other exception type into a {@link ThrowableWrapper}. + * or wrap any other exception type into a {@link CompletionException}. * @param t the exception to throw directly or wrapped * @return because {@code propagate} itself throws an exception or error, this is a sort of phantom return * value; {@code propagate} does not actually return anything diff --git a/src/main/java/io/reactivex/rxjava4/exceptions/ThrowableWrapper.java b/src/main/java/io/reactivex/rxjava4/exceptions/ThrowableWrapper.java deleted file mode 100644 index ad0ed95953..0000000000 --- a/src/main/java/io/reactivex/rxjava4/exceptions/ThrowableWrapper.java +++ /dev/null @@ -1,48 +0,0 @@ -/* - * Copyright (c) 2016-present, RxJava Contributors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in - * compliance with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software distributed under the License is - * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See - * the License for the specific language governing permissions and limitations under the License. - */ - -package io.reactivex.rxjava4.exceptions; - -import java.io.Serial; -import java.util.concurrent.CompletionException; - -/** - * A runtime exception to sneak around checked exceptions. - *

- * If you encounter me, it means some operator forgot to unwrap the inner throwable - * at the right place. - * @since 4.0.0 - */ -public final class ThrowableWrapper extends CompletionException { - - @Serial - private static final long serialVersionUID = -5280780582536857320L; - - /** - * Constructs an instance with the given non-null original Throwable. - * @param original the original Throwable - */ - public ThrowableWrapper(Throwable original) { - super("You forgot to unwrap me!", original != null ? original : new NullPointerException("original is null")); - } - - /** - * Checks if the given {@link Throwable} is of type {@code ThrowableWrapper} and - * unwraps it, or returns the provided throwable as is. - * @param t the throwable to unwrap - * @return the possibly unwrapped throwable - */ - public static Throwable unwrap(Throwable t) { - return t instanceof ThrowableWrapper ? t.getCause() : t; - } -} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTimer.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTimer.java index e3e8034f1d..ad19b70bd9 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTimer.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTimer.java @@ -74,7 +74,9 @@ public void run() { } void interrupedSleep(InterruptedException ex) { - waiter.completeExceptionally(ex); + if (!isDisposed()) { + waiter.completeExceptionally(ex); + } } @Override diff --git a/src/main/java/io/reactivex/rxjava4/internal/util/ExceptionHelper.java b/src/main/java/io/reactivex/rxjava4/internal/util/ExceptionHelper.java index 17e1138d12..2995df6b8f 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/util/ExceptionHelper.java +++ b/src/main/java/io/reactivex/rxjava4/internal/util/ExceptionHelper.java @@ -18,8 +18,8 @@ import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicReference; -import io.reactivex.rxjava4.annotations.Nullable; -import io.reactivex.rxjava4.exceptions.*; +import io.reactivex.rxjava4.annotations.*; +import io.reactivex.rxjava4.exceptions.CompositeException; /** * Terminal atomics for Throwable containers. @@ -33,19 +33,39 @@ private ExceptionHelper() { /** * If the provided Throwable is an Error this method - * throws it, otherwise returns a RuntimeException wrapping the error + * throws it, otherwise returns a CompletionException wrapping the error * if that error is a checked exception. * @param error the error to wrap or throw * @return the (wrapped) error */ - public static RuntimeException wrapOrThrow(Throwable error) { + @NonNull + public static RuntimeException wrapOrThrow(@NonNull Throwable error) { if (error instanceof Error err) { throw err; } if (error instanceof RuntimeException rte) { return rte; } - return new ThrowableWrapper(error); + return new CompletionException("You forgot to unwrap me!", error); + } + /** + * Unwraps a {@link CompletionException} and rethrows its {@link Error} + * or {@link RuntimeException} inside it, or returns it as is if + * the {@code CompletionException} holds a checked exception. + * @param error the error to unwrap and rethrow its cause if possible + * @return the {@code error} if it has a checked exception cause + * @since 4.0.0 + */ + @NonNull + public static RuntimeException unwrapOrThrow(@NonNull CompletionException error) { + var cause = error.getCause(); + if (cause instanceof Error err) { + throw err; + } + if (cause instanceof RuntimeException rte) { + return rte; + } + return error; } /** @@ -184,8 +204,8 @@ public static T nullCheck(T value, String prefix) { } /** - * Unwraps both throwables if they are wrapped into a {@link CompletionException} or - * {@link ThrowableWrapper}, then if both are present, add {@code b} as suppressed to {@code a} + * Unwraps both throwables if they are wrapped into a {@link CompletionException}, + * then if both are present, add {@code b} as suppressed to {@code a} * and return a; return b otherwise * @param main the first throwable * @param secondary the second throwable @@ -193,10 +213,10 @@ public static T nullCheck(T value, String prefix) { */ @Nullable public static Throwable unwrapAndCombine(@Nullable Throwable main, @Nullable Throwable secondary) { - if (main instanceof CompletionException || main instanceof ThrowableWrapper) { + if (main instanceof CompletionException) { main = main.getCause(); } - if (secondary instanceof CompletionException || secondary instanceof ThrowableWrapper) { + if (secondary instanceof CompletionException) { secondary = secondary.getCause(); } if (main != null && secondary != null && main != secondary) { @@ -209,12 +229,12 @@ public static Throwable unwrapAndCombine(@Nullable Throwable main, @Nullable Thr } /** - * Unwraps the given {@link CompletionException} or {@link ThrowableWrapper} + * Unwraps the given {@link CompletionException}. * @param t the possible throwable to unwrap * @return the unwrapped Throwable */ public static Throwable unwrap(@Nullable Throwable t) { - if (t instanceof CompletionException || t instanceof ThrowableWrapper) { + if (t instanceof CompletionException) { t = t.getCause(); } return t; diff --git a/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualCreateExecutor.java b/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualCreateExecutor.java index 1fa122ad0a..2fe02df5c0 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualCreateExecutor.java +++ b/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualCreateExecutor.java @@ -22,7 +22,7 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.*; import io.reactivex.rxjava4.exceptions.*; -import io.reactivex.rxjava4.internal.util.BackpressureHelper; +import io.reactivex.rxjava4.internal.util.*; /** * Runs a generator callback on a virtual thread backed by a Worker of the given scheduler @@ -105,7 +105,7 @@ public Void call() { } catch (Throwable ex) { Exceptions.throwIfFatal(ex); if (ex != STOP && !cancelled) { - downstream.onError(ThrowableWrapper.unwrap(ex)); + downstream.onError(ExceptionHelper.unwrap(ex)); } return null; } diff --git a/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualTransformExecutor.java b/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualTransformExecutor.java index e047516cd8..262399f1ea 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualTransformExecutor.java +++ b/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualTransformExecutor.java @@ -22,8 +22,8 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.core.Scheduler.Worker; import io.reactivex.rxjava4.disposables.*; -import io.reactivex.rxjava4.exceptions.*; -import io.reactivex.rxjava4.internal.util.BackpressureHelper; +import io.reactivex.rxjava4.exceptions.Exceptions; +import io.reactivex.rxjava4.internal.util.*; import io.reactivex.rxjava4.operators.SpscArrayQueue; public final class FlowableVirtualTransformExecutor extends Flowable { @@ -206,7 +206,7 @@ public Void call() { if (d && empty) { var ex = error; if (ex != null) { - downstream.onError(ThrowableWrapper.unwrap(ex)); + downstream.onError(ExceptionHelper.unwrap(ex)); } else { downstream.onComplete(); } @@ -234,7 +234,7 @@ public Void call() { Exceptions.throwIfFatal(ex); if (ex != STOP && !cancelled) { upstream.cancel(); - downstream.onError(ThrowableWrapper.unwrap(ex)); + downstream.onError(ExceptionHelper.unwrap(ex)); } return null; } diff --git a/src/test/java/io/reactivex/rxjava4/exceptions/ThrowableWrapperTest.java b/src/test/java/io/reactivex/rxjava4/exceptions/ThrowableWrapperTest.java deleted file mode 100644 index c895f44477..0000000000 --- a/src/test/java/io/reactivex/rxjava4/exceptions/ThrowableWrapperTest.java +++ /dev/null @@ -1,70 +0,0 @@ -/* - * Copyright (c) 2016-present, RxJava Contributors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in - * compliance with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software distributed under the License is - * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See - * the License for the specific language governing permissions and limitations under the License. - */ - -package io.reactivex.rxjava4.exceptions; - -import static org.junit.jupiter.api.Assertions.*; - -import java.util.concurrent.TimeUnit; - -import org.junit.jupiter.api.Test; - -import io.reactivex.rxjava4.core.*; - -public class ThrowableWrapperTest extends RxJavaTest { - - @Test - public void basic() { - var original = new Throwable("original"); - - try { - throw new ThrowableWrapper(original); - } catch (RuntimeException ex) { - assertSame(original, ex.getCause()); - assertEquals("You forgot to unwrap me!", ex.getMessage()); - assertEquals("original", ex.getCause().getMessage()); - } - } - - @Test - public void basicNull() { - try { - throw new ThrowableWrapper(null); - } catch (RuntimeException ex) { - assertEquals("original is null", ex.getCause().getMessage()); - assertEquals("You forgot to unwrap me!", ex.getMessage()); - assertTrue(ex.getCause() instanceof NullPointerException, ex.getCause().toString()); - } - } - - @Test - public void virtualCreateUnwraps() { - Flowable.virtualCreate(_ -> { - throw new ThrowableWrapper(new TestException()); - }) - .test() - .awaitDone(5, TimeUnit.SECONDS) - .assertFailure(TestException.class); - } - - @Test - public void virtualTransformUnwraps() { - Flowable.just(1) - .virtualTransform((_, _, _) -> { - throw new ThrowableWrapper(new TestException()); - }) - .test() - .awaitDone(5, TimeUnit.SECONDS) - .assertFailure(TestException.class); - } -} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java index aaf55af3bd..0ebc7b14b6 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java @@ -29,7 +29,7 @@ public class StreamableForEachTest extends StreamableBaseTest { @Test public void forEachCheckedCrash() { - var ex = assertThrows(ThrowableWrapper.class, () -> { + var ex = assertThrows(CompletionException.class, () -> { Streamable.just(1) .forEach(_ -> { throw new Exception("test"); @@ -58,7 +58,7 @@ public void forEachUncheckedCrash() { @Test public void forEachExecCheckedCrash() throws Throwable { withCachedExecutor(exec -> { - var ex = assertThrows(ThrowableWrapper.class, () -> { + var ex = assertThrows(CompletionException.class, () -> { Streamable.just(1) .forEach(_ -> { throw new Exception("test"); @@ -90,7 +90,7 @@ public void forEachExecUncheckedCrash() throws Throwable { @Test public void forEachBiCheckedCrash() throws Throwable { withVirtual(exec -> { - var ex = assertThrows(ThrowableWrapper.class, () -> { + var ex = assertThrows(CompletionException.class, () -> { Streamable.just(1) .forEach((_, _) -> { throw new Exception("test"); diff --git a/src/test/java/io/reactivex/rxjava4/internal/util/ExceptionHelperTest.java b/src/test/java/io/reactivex/rxjava4/internal/util/ExceptionHelperTest.java index bfa88cd969..c2b396d26a 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/util/ExceptionHelperTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/util/ExceptionHelperTest.java @@ -91,12 +91,12 @@ public void unwrapAndCombine6() { @Test public void unwrapAndCombine7() { var te = new TestException(); - assertSame(te, ExceptionHelper.unwrapAndCombine(new ThrowableWrapper(te), null)); + assertSame(te, ExceptionHelper.unwrapAndCombine(new CompletionException(te), null)); } @Test public void unwrapAndCombine8() { var te = new TestException(); - assertSame(te, ExceptionHelper.unwrapAndCombine(null, new ThrowableWrapper(te))); + assertSame(te, ExceptionHelper.unwrapAndCombine(null, new CompletionException(te))); } } From 48a5a6270fde77d828d3e0fcc26737cec6c38f9c Mon Sep 17 00:00:00 2001 From: akarnokd Date: Tue, 7 Jul 2026 09:45:35 +0200 Subject: [PATCH 8/8] flaky forEachInputCancelUpfront fix? --- .../operators/streamable/StreamableForEachTest.java | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java index 0ebc7b14b6..5bf9b87026 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEachTest.java @@ -18,11 +18,11 @@ import java.util.concurrent.*; import java.util.concurrent.atomic.*; -import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.*; import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.CompositeDisposable; -import io.reactivex.rxjava4.exceptions.*; +import io.reactivex.rxjava4.exceptions.TestException; import io.reactivex.rxjava4.processors.DispatchStreamProcessor; public class StreamableForEachTest extends StreamableBaseTest { @@ -259,7 +259,9 @@ public void forEachInputCancelUpfront() throws Throwable { awaitNoStreamers(dsp0, 1000); - assertFalse(dsp.hasComplete(), "dsp completes: error = " + dsp.hasThrowable()); + awaitNoStreamers(dsp, 1000); + + assertTrue(dsp.hasComplete(), "dsp completes: error = " + dsp.hasThrowable()); } @Test