4.x: Streamable + doOnError, forEach(StreamerInput) (#8218)

* 4.x: Streamable + doOnError, forEach(StreamerInput)

* fix tests

* Improve API, fix bugs, add coverage

* fix style

* Fix flaky DispatchStreamProcessorTest > normalMulti()

* improve coverage of withCancellation, rename to StreamSink

* Remove ThrowableWrapper, fix attempt at flaky tests

* flaky forEachInputCancelUpfront fix?

Author: akarnokd

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<T> stage, @NonNull Dis
      * <p>
      * 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);
         }
     }
 

src/main/java/io/reactivex/rxjava4/core/StreamProcessor.java

@@ -20,20 +20,26 @@
 
 /**
  * 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 <In> the element type of the input side
  * @param <Out> the element type of the output side
  * @since 4.0.0
  */
-public interface StreamProcessor<@NonNull In, @NonNull Out> extends Streamable<Out>, StreamerInput<In> {
+public interface StreamProcessor<@NonNull In, @NonNull Out> extends Streamable<Out>, StreamSink<In> {
 
     /**
      * Returns {@code true} if this {@link StreamProcessor} has {@link Streamer}s.
      * @return {@code true} if this {@link StreamProcessor} has {@link Streamer}s.
      */
     boolean hasStreamers();
 
+    /**
+     * Returns the current number of {@link Streamer}s subscribed to this {@link StreamProcessor}
+     * @return the current number of {@link Streamer}s subscribed to this {@link StreamProcessor}
+     */
+    int streamerCount();
+
     /**
      * Returns {@code true} if this {@code StreamProcessor} was completed normally via {@link #finish(Throwable)}.
      * @return {@code true} if this {@code StreamProcessor} was completed normally via {@link #finish(Throwable)}.

src/main/java/io/reactivex/rxjava4/core/StreamSink.java

@@ -0,0 +1,101 @@
+/*
+ * 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.core;
+
+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
+ * said item or terminal event has completed, similar to how {@link Subscriber} can receive events.
+ * <p>
+ * The general contract is to call {@link #next(Object)} zero or more times, then
+ * call {@link #finish(Throwable)} at most once, all in a non-overlapping fashion and only if the
+ * returned {@link CompletionStage} has completed in some fashion.
+ * @param <T> the item type to be offered
+ * @since 4.0.0
+ */
+public interface StreamSink<@NonNull T> {
+
+    /**
+     * Offer the next item.
+     * @param item the item being offered
+     * @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<Boolean> next(T item);
+
+    /**
+     * Offer the final, terminal event.
+     * @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<Void> 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.
+     * <p>
+     * The default implementation returns a fresh {@link CompositeDisposable}.
+     * @return the {@code DisposableContainer}
+     */
+    @NonNull
+    default DisposableContainer cancellation() {
+        return new CompositeDisposable();
+    }
+
+    /**
+     * Returns a new {@link StreamSink} that returns the given {@link DisposableContainer}
+     * in its {@link #cancellation()}, allowing overriding the cancellation management
+     * of this {@code StreamSink}
+     * @param cancellation the {@link DisposableContainer} to use as cancellation management
+     * @return the new {@code StreamSink} instance
+     * @throws NullPointerException if {@code cancellation} is {@code null}
+     */
+    @NonNull
+    default StreamSink<T> withCancellation(DisposableContainer cancellation) {
+        Objects.requireNonNull(cancellation, "cancellation is null");
+        return new StreamSinkWithCancellation<>(this, cancellation);
+    }
+
+    /**
+     * Creates a {@link StreamSink} via lambda callbacks for {@link #next(Object)} and
+     * {@link #finish(Throwable)}.
+     * <p>
+     * Non-fatal exceptions thrown by the callbacks are turned into failed
+     * {@link CompletableFuture#failedFuture(Throwable)}s.
+     * @param <T> 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 StreamSink} instance
+     */
+    @NonNull
+    static <T> StreamSink<T> create(
+            @NonNull Function<? super T, ? extends CompletionStage<Boolean>> onNext,
+            @NonNull Function<? super Throwable, ? extends CompletionStage<Void>> onFinish
+    ) {
+        Objects.requireNonNull(onNext, "onNext is null");
+        Objects.requireNonNull(onFinish, "onFinish is null");
+        return new StreamSinkLambda<>(onNext, onFinish);
+    }
+}

src/main/java/io/reactivex/rxjava4/core/Streamable.java

@@ -589,6 +589,17 @@ default Streamable<T> 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<T> doOnError(Consumer<? super Throwable> 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,34 @@ default void subscribe(@NonNull Flow.Subscriber<? super T> subscriber) {
         subscribe(subscriber, Executors.newVirtualThreadPerTaskExecutor());
     }
 
+    /**
+     * 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
+     * @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 CompletionStage<Void> subscribe(@NonNull StreamSink<? super T> consumer) {
+        return subscribe(consumer, Executors.newVirtualThreadPerTaskExecutor());
+    }
+
+    /**
+     * 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
+     * @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 CompletionStage<Void> subscribe(@NonNull StreamSink<? super T> consumer, ExecutorService executor) {
+        Objects.requireNonNull(consumer, "consumer is null");
+        Objects.requireNonNull(executor, "executor is null");
+        return StreamableForEach.forEach(this, consumer, executor);
+    }
+
     /**
      * Creates a new {@link TestSubscriber} and subscribes it to this {@code Streamable}.
      * @return the created test subscriber

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<Boolean> 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<Void> stage) {
+        if (stage == FINISHED) {
             return;
         }
-        s.toCompletableFuture().join();
+        stage.toCompletableFuture().join();
     }
 
     /**

src/main/java/io/reactivex/rxjava4/core/StreamerInput.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