4.x: Streamable/Streamer API rewamp, dropping next/finish args (#8212)

Author: akarnokd

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

@@ -91,21 +91,9 @@ public CompletionStageDisposable(@NonNull CompletionStage<T> stage, @NonNull Dis
      * @throws ThrowableWrapper if the original exception was a checked exception
      */
     public void await() {
-        await(null);
-    }
-
-    /**
-     * Await the completion of the current stage.
-     * <p>
-     * Rethrows any original unchecked exceptions as is.
-     * @param canceller the canceller link
-     * @throws CancellationException if the computation was cancelled
-     * @throws ThrowableWrapper if the original exception was a checked exception
-     */
-    public void await(DisposableContainer canceller) {
         state.lazySet(true);
         try {
-            AwaitCoordinatorStatic.await(stage, canceller);
+            stage.toCompletableFuture().join();
         } catch (CompletionException ce) {
             throw ExceptionHelper.wrapOrThrow(ce.getCause());
         }

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

@@ -25,18 +25,46 @@
 import io.reactivex.rxjava4.functions.*;
 import io.reactivex.rxjava4.internal.functions.ObjectHelper;
 import io.reactivex.rxjava4.internal.operators.streamable.*;
-import io.reactivex.rxjava4.internal.util.*;
+import io.reactivex.rxjava4.internal.util.ExceptionHelper;
 import io.reactivex.rxjava4.plugins.RxJavaPlugins;
 import io.reactivex.rxjava4.schedulers.Schedulers;
 import io.reactivex.rxjava4.subscribers.TestSubscriber;
 
-/**
- * The holographically emergent {@code IAsyncEnumerable} of the Java world.
- * Runs best with Virtual Threads.
- * TODO proper docs
- * @param <T> the element type of the stream.
- * @since 4.0.0
- */
+/// Represents a virtual-thread capable, multi-valued (a)synchronous sequence of values that
+/// builds upon the Java [CompletionStage]-based concurrency-coordination model.
+///
+/// The lifecycle of the sequence is as follows:
+///
+/// - consumer calls {@link #stream(DisposableContainer)}
+/// - consumer calls {@link Streamer#next()} in a loop and consumer checks if what
+///   the {@link CompletionStage} outcome is:
+///   - if it succeeded with a boolean {@code true}, it is safe to call {@link Streamer#current()}.
+///   - if it succeeded with a boolean {@code false}, no further values are coming.
+///   - if it failed with a {@code Throwable}, no further values are coming and the error can be propagated further
+/// - consumer calls {@link Streamer#finish()}
+///
+/// It is always necessary to have the consumer call {@code finish} because that is responsible for cleaning up
+/// resources of the upstream.
+///
+/// Downstream cancellations are signaled via the [DisposableContainer], where operators can register their own
+/// [Disposable]s that get disposed. Because dispose can happen at any time and asynchronously to the consumption loop,
+/// the sensitive sources must complete their waiting `CompletionStage` returned by `next` exceptionally via a
+/// [CancellationException]. This will unblock the loops and invoke the `finish` method of the lifecycle at
+/// the consumer thread. Depending on the operator, the `CancellationException` may not be propagated further.
+///
+/// If a source wishes to fail, it must signal the [Throwable] via the returned {@code CompletionStage} of {@code next}.
+/// If the `finish` also throws, its `Throwable` should be added as suppressed exception to the original `Throwable`.
+///
+/// The `Streamer` methods must be invoked sequentially and non-overlappingly, similar to the
+/// <a href='https://github.com/reactive-streams/reactive-streams-jvm#1.3'>Reactive Streams rule §1.3</a>.
+///
+/// This reactive type was modeled after the C# `IAsyncEnumerable` and `IAsyncEnumerator` interfaces. Unfortunately,
+/// Java never added any `async`/`await` infrastructure, plus `CompletionStage` doesn't even have any native way to blockingly
+/// join to it. Therefore, when running in a (virtual) blocking fashion, one may use the {@link Streamer#awaitNext()}
+/// or {@link Streamer#awaitFinish()} helper methods.
+///
+/// @param <T> the element type of the {@code Streamable} sequence.
+/// @since 4.0.0
 @FunctionalInterface
 public interface Streamable<@NonNull T> {
 
@@ -258,32 +286,6 @@ static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source, @NonNu
                 .toStreamable();
     }
 
-    /**
-     * Generates a sequence in order which the stages complete in any form.
-     * @param <T> the common element type
-     * @param stages the iterable of stages to be relayed in the order they complete
-     * @param executor the executor to run the blocking operator
-     * @return the new Streamable instance
-     */
-    @SuppressWarnings("unchecked")
-    @CheckReturnValue
-    @NonNull
-    static <@NonNull T> Streamable<CompletionStage<T>> fromStages(
-            @NonNull Iterable<? extends CompletionStage<? extends T>> stages, ExecutorService executor) {
-        Objects.requireNonNull(stages, "stages is null");
-        Objects.requireNonNull(executor, "executor is null");
-        return create(emitter -> {
-            var list = new ArrayList<CompletionStage<? extends T>>();
-            for(var stage : stages) {
-                list.add(stage);
-            }
-            while (!list.isEmpty()) {
-                var winner = AwaitCoordinatorStatic.awaitFirstIndex(list, emitter.canceller());
-                emitter.emit((CompletionStage<T>)list.remove(winner));
-            }
-        }, executor);
-    }
-
     /**
      * Defers the creation of the actual {@code Streamable}, allowing a per streamer
      * state to be created along with it.
@@ -330,10 +332,10 @@ static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source, @NonNu
         return create(emitter -> {
             try (var mainSource = sources.forEach(item -> {
                 try (var innerSource = item.forEach(emitter::emit, emitter.canceller().derive(), executor)) {
-                    innerSource.await(emitter.canceller());
+                    innerSource.await();
                 }
             }, emitter.canceller(), executor)) {
-                mainSource.await(emitter.canceller());
+                mainSource.await();
             }
         }, executor);
     }
@@ -562,7 +564,7 @@ default Flowable<T> toFlowable() {
     default Flowable<T> toFlowable(@NonNull ExecutorService executor) {
         Objects.requireNonNull(executor, "executir is null");
         var me = this;
-        return Flowable.virtualCreate(emitter -> me.forEach(emitter::emit).await(emitter.canceller()), executor);
+        return Flowable.virtualCreate(emitter -> me.forEach(emitter::emit).await(), executor);
     }
 
     /**
@@ -595,7 +597,7 @@ default Flowable<T> toFlowable(@NonNull ExecutorService executor) {
             // System.out.println("item " + item);
             transformer.transform(item, emitter, stopper);
         }, emitter.canceller(), executor)
-        .await(emitter.canceller()), executor);
+        .await(), executor);
     }
 
     // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
@@ -658,7 +660,7 @@ default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> con
             try {
                 try {
                     while (!canceller.isDisposed()) {
-                        if (str.awaitNext(canceller)) {
+                        if (str.awaitNext()) {
                             // System.out.println("Received " + str.current());
                             consumer.accept(Objects.requireNonNull(str.current(), "The upstream Streamable " + me.getClass() + " produced a null element!"));
                         } else {
@@ -667,7 +669,7 @@ default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> con
                         }
                     }
                 } finally {
-                    str.awaitFinish(canceller);
+                    str.awaitFinish();
                 }
             } catch (final Throwable crash) {
                 Exceptions.throwIfFatal(crash);
@@ -705,7 +707,7 @@ default CompletionStageDisposable<Void> forEach(
                 try {
                 var stopper = Disposable.empty();
                     while (!canceller.isDisposed() && !stopper.isDisposed()) {
-                        if (str.awaitNext(canceller)) {
+                        if (str.awaitNext()) {
                             // System.out.println("Received " + str.current());
                             var v = Objects.requireNonNull(str.current(), "The upstream Streamable " + me.getClass() + " produced a null element!");
                             consumer.accept(v, stopper);
@@ -715,7 +717,7 @@ default CompletionStageDisposable<Void> forEach(
                         }
                     }
                 } finally {
-                    str.awaitFinish(canceller);
+                    str.awaitFinish();
                 }
             } catch (final Throwable crash) {
                 Exceptions.throwIfFatal(crash);
@@ -739,7 +741,7 @@ default void subscribe(@NonNull Flow.Subscriber<? super T> subscriber, @NonNull
         final Streamable<T> me = this;
         Flowable.<T>virtualCreate(emitter -> {
             me.forEach(emitter::emit, emitter.canceller().derive(), executor)
-            .await(emitter.canceller());
+            .await();
         }, executor)
         .subscribe(subscriber);
     }
@@ -751,7 +753,7 @@ default void subscribe(@NonNull Flow.Subscriber<? super T> subscriber, @NonNull
     default void subscribe(@NonNull Flow.Subscriber<? super T> subscriber) {
         final Streamable<T> me = this;
         Flowable.<T>virtualCreate(emitter -> {
-            me.forEach(emitter::emit).await(emitter.canceller());
+            me.forEach(emitter::emit).await();
         })
         .subscribe(subscriber);
     }

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

@@ -13,94 +13,104 @@
 
 package io.reactivex.rxjava4.core;
 
-import java.util.*;
 import java.util.concurrent.*;
 
 import io.reactivex.rxjava4.annotations.NonNull;
-import io.reactivex.rxjava4.disposables.*;
-import io.reactivex.rxjava4.internal.util.AwaitCoordinator;
 
-/**
- * A realized stream which can then be consumed asynchronously in steps.
- * Think of it as the {@IAsyncEnumerator} of the Java world. Runs best on Virtual Threads.
- * <p>
- * To make sure you can run finish, use {@link DisposableContainer#clear()} or {@link DisposableContainer#reset()}
- * to get rid of all previous registered disposables. finish() will create its own, and if that
- * gets stuck, just call clear()/dispose() on the container to get rid of this sequence for good.
- * @param <T> the element type.
- * TODO proper docs
- * @since 4.0.0
- */
-public interface Streamer<@NonNull T> extends AwaitCoordinator {
+/// A realized stream which can then be consumed asynchronously in steps.
+/// Think of it as the `IAsyncEnumerator` from C# ported to the clumsy Java world.
+///
+/// The `Streamer` methods must be invoked sequentially and non-overlappingly, similar to the
+/// <a href='https://github.com/reactive-streams/reactive-streams-jvm#1.3'>Reactive Streams rule §1.3</a>.
+///
+/// For an optimized synchronous operation, please consider using the {@link #NEXT_TRUE}, {@link #NEXT_FALSE}
+/// and {@link #FINISHED} constant CompletionStages.
+/// @param <T> the element type.
+/// @since 4.0.0
+public interface Streamer<@NonNull T> {
 
     // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
     // API
     // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
 
     /**
      * Determine if there are more elements available from the source.
-     * @param cancellation ability to perform cancellation on a per-virtual-pull request.
-     * @return eventually true or false, indicating availability or termination
+     * @return a `CompletionStage` with 3 outcomes
+     * <ul>
+     * <li>`true` indicates there is an item available for consumption via {@link #current()}
+     * <li>`false` indicates there are no more items available
+     *< li>`Throwable` indicates there was an upstream error
+     * </ul>
      */
     @NonNull
-    CompletionStage<Boolean> next(@NonNull DisposableContainer cancellation);
+    CompletionStage<Boolean> next();
 
     /**
-     * Returns the current element if {@link #next(DisposableContainer)} yielded {@code true}.
-     * Can be called multiple times between {@link #next(DisposableContainer)} calls.
-     * @return the current element
-     * @throws NoSuchElementException before the very first {@link #next(DisposableContainer)}
-     *  or after {@link #next(DisposableContainer)} returned {@code false}
+     * Returns the currently available item synchronously if the previous call to [#next()] yielded `true`.
+     * 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
      */
     @NonNull
     T current();
 
     /**
-     * Called when the stream ends or gets cancelled. Should be always invoked.
-     * TODO, this is inherited from {@code IAsyncDisposable} in C#...
-     * @param cancellation to cancel a stuck finish operation, just in case.
-     * @return the stage you can await to cleanups to happen
+     * Finish the sequence once all processing has been done to it either via exhaustion or via cancellation.
+     * <p>
+     * Usually involves resource cleanup, so this method must be always called.
+     * <p>
+     * If the cleanup crashes and the [#next()] crashed too, the cleanup `Throwable` will be added as suppressed
+     * to the main crash `Throwable` from `next`.
+     *
+     * @return a `CompletionStage` that completes when the resource cleanup completes normally or exceptionally
      */
     @NonNull
-    CompletionStage<Void> finish(@NonNull DisposableContainer cancellation);
+    CompletionStage<Void> finish();
 
     // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
     // HELPERS
     // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
 
     /**
-     * Moves and awaits the sequence's next element, returns false if there are no more
-     * data.
-     * @param cancellation to efficiently cancel this await if necessary
-     * @return true if the next element via {@link #current()} can be read, or false if
-     * the stream ended.
+     * Convenience method to blockingly await the CompletionStage returned by the {@link #next()} method.
+     * @return true if there are more items, false if no more items are coming, or crashes
      */
-    default boolean awaitNext(@NonNull DisposableContainer cancellation) {
-        return await(next(cancellation), cancellation);
+    default boolean awaitNext() {
+        var s = next();
+        if (s == NEXT_TRUE) {
+            return true;
+        } else
+        if (s == NEXT_FALSE) {
+            return false;
+        }
+        return s.toCompletableFuture().join();
     }
 
     /**
-     * Who cancels the cancellation attempt? Another cancellation attempt!
-     * @param cancellation the token to cancel and ongoing cancel attempt
+     * Convenience method to blockingly await the CompletionStage returned by the {@link #finish()} method.
      */
-    default void awaitFinish(@NonNull DisposableContainer cancellation) {
-        await(finish(cancellation), cancellation);
+    default void awaitFinish() {
+        var s = finish();
+        if (s == FINISHED) {
+            return;
+        }
+        s.toCompletableFuture().join();
     }
 
     /**
-     * Use this constant in {@link #next(DisposableContainer)} to indicate
+     * Use this constant in {@link #next()} to indicate
      * the next value is available, synchronously.
      */
     CompletionStage<Boolean> NEXT_TRUE = CompletableFuture.completedStage(true);
 
     /**
-     * Use this constant in {@link #next(DisposableContainer)} to indicate
+     * Use this constant in {@link #next()} to indicate
      * no more values will be available, synchronously.
      */
     CompletionStage<Boolean> NEXT_FALSE = CompletableFuture.completedStage(false);
 
     /**
-     * Use this constant in {@link #finish(DisposableContainer)} to indicate
+     * Use this constant in {@link #finish()} to indicate
      * the cleanup was done synchronously.
      */
     CompletionStage<Void> FINISHED = CompletableFuture.completedStage(null);

src/main/java/io/reactivex/rxjava4/core/config/StreamableInterceptConfig.java

@@ -25,9 +25,9 @@
  * Configuration record the intercept() operator with various lifecylce-stage transforming callbacks
  * @param <T> the element type of the sequence
  * @param onStream called when the {@link Streamable#stream(io.reactivex.rxjava4.disposables.DisposableContainer)} is invoked
- * @param onNext called when the {@link Streamer#next(io.reactivex.rxjava4.disposables.DisposableContainer)} is invoked
+ * @param onNext called when the {@link Streamer#next()} is invoked
  * @param onCurrent called when the {@link Streamer#current()} is invoked
- * @param onFinish called when the {@link Streamer#finish(io.reactivex.rxjava4.disposables.DisposableContainer)} is invoked
+ * @param onFinish called when the {@link Streamer#finish()} is invoked
  * @since 4.0.0
  */
 public record StreamableInterceptConfig<T>(
@@ -56,9 +56,9 @@ public StreamableInterceptConfig(@NonNull Function<? super T, ? extends T> onCur
     /**
      * Constructs a fully configured record.
      * @param onStream called when the {@link Streamable#stream(io.reactivex.rxjava4.disposables.DisposableContainer)} is invoked
-     * @param onNext called when the {@link Streamer#next(io.reactivex.rxjava4.disposables.DisposableContainer)} is invoked
+     * @param onNext called when the {@link Streamer#next()} is invoked
      * @param onCurrent called when the {@link Streamer#current()} is invoked
-     * @param onFinish called when the {@link Streamer#finish(io.reactivex.rxjava4.disposables.DisposableContainer)} is invoked
+     * @param onFinish called when the {@link Streamer#finish()} is invoked
      */
     public StreamableInterceptConfig {
         Objects.requireNonNull(onStream, "onStream is null");

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableEmpty.java

@@ -34,7 +34,7 @@ enum EmptyStreamer implements Streamer<Object> {
         INSTANCE;
 
         @Override
-        public @NonNull CompletionStage<Boolean> next(DisposableContainer cancellation) {
+        public @NonNull CompletionStage<Boolean> next() {
             return NEXT_FALSE;
         }
 
@@ -44,7 +44,7 @@ enum EmptyStreamer implements Streamer<Object> {
         }
 
         @Override
-        public @NonNull CompletionStage<Void> finish(DisposableContainer canceller) {
+        public @NonNull CompletionStage<Void> finish() {
             return FINISHED;
         }
     }