Create the "await" keyword, as static helper method.

10 foot pile of shit, 11 foot high ladder, Brian.

Author: akarnokd

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

@@ -23,6 +23,8 @@
 import io.reactivex.rxjava4.functions.Consumer;
 import io.reactivex.rxjava4.internal.operators.streamable.*;
 
+import static io.reactivex.rxjava4.core.Streamer.await;
+
 /**
  * The {@code IAsyncEnumerable} of the Java world.
  * Runs best with Virtual Threads.
@@ -119,7 +121,7 @@ public final CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T
         var future = executor.submit(() -> {
             try (var str = me.stream(canceller)) {
                 while (!canceller.isDisposed()) {
-                    if (str.next().toCompletableFuture().join()) {
+                    if (await(str.next(canceller), canceller)) {
                         consumer.accept(Objects.requireNonNull(str.current(), "The upstream Streamable " + me.getClass() + " produced a null element!"));
                     } else {
                         break;

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

@@ -16,7 +16,7 @@
 import java.util.NoSuchElementException;
 import java.util.concurrent.CompletionStage;
 
-import io.reactivex.rxjava4.annotations.NonNull;
+import io.reactivex.rxjava4.annotations.*;
 import io.reactivex.rxjava4.disposables.*;
 
 /**
@@ -69,4 +69,34 @@ default CompletionStage<Boolean> next() {
     default void close() {
         cancel().toCompletableFuture().join();
     }
+
+    /**
+     * The {@code await} keyword for async/await.
+     * @param <T> the type of the returned value if any.
+     * @param stage the stage to await virtual-blockingly
+     * @return the awaited value
+     */
+    @Nullable
+    static <T> T await(@NonNull CompletionStage<T> stage) {
+        return await(stage, null);
+    }
+
+    /**
+     * The cancellable {@code await} keyword for async/await.
+     * @param <T> the type of the returned value if any.
+     * @param stage the stage to await virtual-blockingly
+     * @param cancellation the container that can trigger a cancellation on demand
+     * @return the awaited value
+     */
+    @Nullable
+    static <T> T await(@NonNull CompletionStage<T> stage, @Nullable DisposableContainer cancellation) {
+        var f = stage.toCompletableFuture();
+        if (cancellation == null) {
+            return f.join();
+        }
+        var d = Disposable.fromFuture(f, true);
+        try (var _ = cancellation.subscribe(d)) {
+            return f.join();
+        }
+    }
 }