Fix flaky tests, Fix a missing await() call. F java

Author: akarnokd

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

@@ -16024,7 +16024,7 @@ public final void subscribe(@NonNull FlowableSubscriber<? super T> subscriber) {
             // can't call onSubscribe because the call might have set a Subscription already
             RxJavaPlugins.onError(e);
 
-            NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
+            var npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
             npe.initCause(e);
             throw npe;
         }
@@ -20929,7 +20929,7 @@ public final Stream<T> blockingStream(int prefetch) {
      * {@code ExecutorService} uses virtual threads, such as the one returned by
      * {@link Executors#newVirtualThreadPerTaskExecutor()}.
      * @param <R> the downstream element type
-     * @param transformer the callback whose {@link VirtualTransformer#transform(Object, VirtualEmitter)}
+     * @param transformer the callback whose {@link VirtualTransformer#transform(Object, VirtualEmitter, Disposable)}
      *  is invoked for each upstream item
      * @param executor the target {@code ExecutorService} to use for running the callback
      * @return the new {@code Flowable} instance
@@ -20962,7 +20962,7 @@ public final Stream<T> blockingStream(int prefetch) {
      * {@link Scheduler} uses virtual threads, such as the one returned by
      * {@link Executors#newVirtualThreadPerTaskExecutor()}.
      * @param <R> the downstream element type
-     * @param transformer the callback whose {@link VirtualTransformer#transform(Object, VirtualEmitter)}
+     * @param transformer the callback whose {@link VirtualTransformer#transform(Object, VirtualEmitter, Disposable)}
      *  is invoked for each upstream item
      * @return the new {@code Flowable} instance
      * @throws NullPointerException if {@code transformer} is {@code null}
@@ -20993,7 +20993,7 @@ public final Stream<T> blockingStream(int prefetch) {
      * {@code Scheduler} uses virtual threads, such as the one returned by
      * {@link Schedulers#virtual()}.
      * @param <R> the downstream element type
-     * @param transformer the callback whose {@link VirtualTransformer#transform(Object, VirtualEmitter)}
+     * @param transformer the callback whose {@link VirtualTransformer#transform(Object, VirtualEmitter, Disposable)}
      * is invoked for each upstream item
      * @param scheduler the target {@code Scheduler} to use for running the callback
      * @param prefetch the number of items to fetch from the upstream.
@@ -21031,7 +21031,7 @@ public final Stream<T> blockingStream(int prefetch) {
      * {@code ExecutorService} uses virtual threads, such as the one returned by
      * {@link Executors#newVirtualThreadPerTaskExecutor()}.
      * @param <R> the downstream element type
-     * @param transformer the callback whose {@link VirtualTransformer#transform(Object, VirtualEmitter)}
+     * @param transformer the callback whose {@link VirtualTransformer#transform(Object, VirtualEmitter, Disposable)}
      *  is invoked for each upstream item
      * @param executor the target {@code ExecutorService} to use for running the callback
      * @param prefetch the number of items to fetch from the upstream.

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

@@ -17,10 +17,10 @@
 import java.util.Objects;
 import java.util.concurrent.*;
 
-import io.reactivex.rxjava4.annotations.NonNull;
+import io.reactivex.rxjava4.annotations.*;
 import io.reactivex.rxjava4.disposables.*;
 import io.reactivex.rxjava4.exceptions.Exceptions;
-import io.reactivex.rxjava4.functions.Consumer;
+import io.reactivex.rxjava4.functions.*;
 import io.reactivex.rxjava4.internal.operators.streamable.*;
 import io.reactivex.rxjava4.schedulers.Schedulers;
 import io.reactivex.rxjava4.subscribers.TestSubscriber;
@@ -43,6 +43,7 @@ public interface Streamable<@NonNull T> {
      * @param cancellation where to register and listen for cancellation calls.
      * @return the Streamer instance to consume.
      */
+    @CheckReturnValue
     @NonNull
     Streamer<T> stream(@NonNull DisposableContainer cancellation);
 
@@ -55,6 +56,7 @@ public interface Streamable<@NonNull T> {
      * Realizes the stream and returns an interface that let's one consume it.
      * @return the Streamer instance to consume.
      */
+    @CheckReturnValue
     @NonNull
     default Streamer<T> stream() {
         return stream(new CompositeDisposable()); // FIXME, use a practically no-op disposable container instead
@@ -69,6 +71,7 @@ default Streamer<T> stream() {
      * @param <T> the element type
      * @return the {@code Streamable} instance
      */
+    @CheckReturnValue
     @NonNull
     static <@NonNull T> Streamable<T> empty() {
         return new StreamableEmpty<>();
@@ -80,6 +83,7 @@ default Streamer<T> stream() {
      * @param item the constant item to produce
      * @return the {@code Streamable} instance
      */
+    @CheckReturnValue
     @NonNull
     static <@NonNull T> Streamable<T> just(@NonNull T item) {
         Objects.requireNonNull(item, "item is null");
@@ -92,6 +96,7 @@ default Streamer<T> stream() {
      * @param source Flow.Publisher to convert
      * @return the new Streamable instance
      */
+    @CheckReturnValue
     @NonNull
     static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source) {
         Objects.requireNonNull(source, "source is null");
@@ -105,6 +110,7 @@ static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source) {
      * @param executor where the conversion will run
      * @return the new Streamable instance
      */
+    @CheckReturnValue
     @NonNull
     static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source, @NonNull ExecutorService executor) {
         Objects.requireNonNull(source, "source is null");
@@ -120,6 +126,7 @@ static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source, @NonNu
      * @param generator the generator to use
      * @return the streamable instance
      */
+    @CheckReturnValue
     @NonNull
     static <@NonNull T> Streamable<T> create(@NonNull VirtualGenerator<T> generator) {
         // FIXME native implementation
@@ -137,6 +144,7 @@ static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source, @NonNu
      * @param scheduler the scheduler to run the virtual generator on
      * @return the streamable instance
      */
+    @CheckReturnValue
     @NonNull
     static <@NonNull T> Streamable<T> create(@NonNull VirtualGenerator<T> generator, @NonNull Scheduler scheduler) {
         // FIXME native implementation
@@ -154,6 +162,7 @@ static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source, @NonNu
      * @param executor the executor to run the virtual generator on
      * @return the streamable instance
      */
+    @CheckReturnValue
     @NonNull
     static <@NonNull T> Streamable<T> create(@NonNull VirtualGenerator<T> generator, @NonNull ExecutorService executor) {
         // FIXME native implementation
@@ -170,6 +179,7 @@ static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source, @NonNu
      * on the default Executors.newVirtualThreadPerTaskExecutor() virtual thread.
      * @return the new Flowable instance
      */
+    @CheckReturnValue
     @NonNull
     default Flowable<T> toFlowable() {
         return toFlowable(Executors.newVirtualThreadPerTaskExecutor());
@@ -181,6 +191,7 @@ default Flowable<T> toFlowable() {
      * @param executor the executor to use
      * @return the new Flowable instance
      */
+    @CheckReturnValue
     @NonNull
     default Flowable<T> toFlowable(@NonNull ExecutorService executor) {
         Objects.requireNonNull(executor, "executir is null");
@@ -196,6 +207,7 @@ default Flowable<T> toFlowable(@NonNull ExecutorService executor) {
      * @param transformer the interface to implement the transforming logic
      * @return the new Streamable instance
      */
+    @CheckReturnValue
     @NonNull
     default <@NonNull R> Streamable<R> transform(@NonNull VirtualTransformer<T, R> transformer) {
         return transform(transformer, Executors.newVirtualThreadPerTaskExecutor());
@@ -208,17 +220,19 @@ default Flowable<T> toFlowable(@NonNull ExecutorService executor) {
      * @param executor where to run the transform and blocking operations
      * @return the new Streamable instance
      */
+    @CheckReturnValue
     @NonNull
     default <@NonNull R> Streamable<R> transform(@NonNull VirtualTransformer<T, R> transformer,
             @NonNull ExecutorService executor) {
         Objects.requireNonNull(transformer, "transformer is null");
         Objects.requireNonNull(executor, "executor is null");
         var me = this;
         return create(emitter -> {
-            me.forEach(item -> {
-                System.out.println("item " + item);
-                transformer.transform(item, emitter);
-            }, executor).await(emitter.canceller());
+            me.forEach((item, stopper) -> {
+                // System.out.println("item " + item);
+                transformer.transform(item, emitter, stopper);
+            }, emitter.canceller(), executor)
+            .await(emitter.canceller());
         }, executor);
     }
 
@@ -231,6 +245,7 @@ default Flowable<T> toFlowable(@NonNull ExecutorService executor) {
      * @param consumer the callback that gets the elements until completion
      * @return a Disposable that let's one cancel the sequence asynchronously.
      */
+    @CheckReturnValue
     @NonNull
     default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> consumer) {
         CompositeDisposable canceller = new CompositeDisposable();
@@ -243,6 +258,8 @@ default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> con
      * @param canceller the container to trigger cancellation of the sequence
      * @return the {@code CompletionStage} that gets notified when the sequence ends
      */
+    @CheckReturnValue
+    @NonNull
     default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> consumer, @NonNull DisposableContainer canceller) {
         return forEach(consumer, canceller, Executors.newVirtualThreadPerTaskExecutor());
     }
@@ -253,6 +270,7 @@ default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> con
      * @param executor the service that hosts the blocking waits.
      * @return a Disposable that let's one cancel the sequence asynchronously.
      */
+    @CheckReturnValue
     @NonNull
     default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> consumer, @NonNull ExecutorService executor) {
         CompositeDisposable canceller = new CompositeDisposable();
@@ -266,6 +284,8 @@ default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> con
      * @param executor the service that hosts the blocking waits.
      * @return the {@code CompletionStage} that gets notified when the sequence ends
      */
+    @CheckReturnValue
+    @NonNull
     @SuppressWarnings("unchecked")
     default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> consumer, @NonNull DisposableContainer canceller, @NonNull ExecutorService executor) {
         Objects.requireNonNull(consumer, "consumer is null");
@@ -276,14 +296,18 @@ default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> con
             try (var str = me.stream(canceller)) {
                 while (!canceller.isDisposed()) {
                     if (str.awaitNext(canceller)) {
-                        System.out.println("Received " + str.current());
+                        // System.out.println("Received " + str.current());
                         consumer.accept(Objects.requireNonNull(str.current(), "The upstream Streamable " + me.getClass() + " produced a null element!"));
                     } else {
+                        // System.out.println("EOF ");
                         break;
                     }
                 }
+                // System.out.println("Canceller status after loop: " + canceller.isDisposed());
             } catch (final Throwable crash) {
                 Exceptions.throwIfFatal(crash);
+                // System.out.println("Canceller status in error: " + canceller.isDisposed());
+                crash.printStackTrace();
                 if (crash instanceof RuntimeException ex) {
                     throw ex;
                 }
@@ -298,6 +322,57 @@ default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> con
         return new CompletionStageDisposable<Void>(StreamableHelper.toCompletionStage((Future<Void>)(Future<?>)future), canceller);
     }
 
+    /**
+     * Consumes elements from this {@code Streamable} via the provided executor service.
+     * @param consumer the callback that gets the elements until completion
+     * @param canceller the container to trigger cancellation of the sequence
+     * @param executor the service that hosts the blocking waits.
+     * @return the {@code CompletionStage} that gets notified when the sequence ends
+     */
+    @CheckReturnValue
+    @NonNull
+    @SuppressWarnings("unchecked")
+    default CompletionStageDisposable<Void> forEach(
+            @NonNull BiConsumer<? super T, ? super Disposable> consumer,
+            @NonNull DisposableContainer canceller,
+            @NonNull ExecutorService executor) {
+        Objects.requireNonNull(consumer, "consumer is null");
+        Objects.requireNonNull(canceller, "canceller is null");
+        Objects.requireNonNull(executor, "executor is null");
+        final Streamable<T> me = this;
+        var future = executor.submit(() -> {
+            try (var str = me.stream(canceller)) {
+                var stopper = Disposable.empty();
+                while (!canceller.isDisposed() && !stopper.isDisposed()) {
+                    if (str.awaitNext(canceller)) {
+                        // 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);
+                    } else {
+                        // System.out.println("EOF ");
+                        break;
+                    }
+                }
+                // System.out.println("Canceller status after loop: " + canceller.isDisposed());
+            } catch (final Throwable crash) {
+                Exceptions.throwIfFatal(crash);
+                // System.out.println("Canceller status in error: " + canceller.isDisposed());
+                // crash.printStackTrace();
+                if (crash instanceof RuntimeException ex) {
+                    throw ex;
+                }
+                if (crash instanceof Exception ex) {
+                    throw ex;
+                }
+                throw new InvocationTargetException(crash);
+            }
+            return null;
+        });
+        canceller.add(Disposable.fromFuture(future));
+        return new CompletionStageDisposable<Void>(
+                StreamableHelper.toCompletionStage((Future<Void>)(Future<?>)future), canceller);
+    }
+
     /**
      * Consume this {@code Streamable} via the given flow-reactive-streams subscriber.
      * @param subscriber the subscriber to consume with.
@@ -310,7 +385,7 @@ default void subscribe(@NonNull Flow.Subscriber<? super T> subscriber, @NonNull
             me.forEach(v -> {
                 // System.out.println("subscribe::virtualCreate::forEach::emit");
                 emitter.emit(v);
-            }).await();
+            }).await(emitter.canceller());
         }, executor)
         .subscribe(subscriber);
     }
@@ -325,7 +400,7 @@ default void subscribe(@NonNull Flow.Subscriber<? super T> subscriber) {
             me.forEach(v -> {
                 // System.out.println("Emitting " + v);
                 emitter.emit(v);
-            });
+            }).await(emitter.canceller());
         })
         .subscribe(subscriber);
     }
@@ -334,6 +409,8 @@ default void subscribe(@NonNull Flow.Subscriber<? super T> subscriber) {
      * Creates a new {@link TestSubscriber} and subscribes it to this {@code Streamable}.
      * @return the created test subscriber
      */
+    @CheckReturnValue
+    @NonNull
     default TestSubscriber<T> test() {
         var ts = new TestSubscriber<T>();
         subscribe(ts);
@@ -345,7 +422,9 @@ default TestSubscriber<T> test() {
      * @param executor the executor to use
      * @return the created test subscriber
      */
-    default TestSubscriber<T> test(ExecutorService executor) {
+    @CheckReturnValue
+    @NonNull
+    default TestSubscriber<T> test(@NonNull ExecutorService executor) {
         var ts = new TestSubscriber<T>();
         subscribe(ts, executor);
         return ts;

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

@@ -12,6 +12,9 @@
  */
 
 package io.reactivex.rxjava4.core;
+
+import io.reactivex.rxjava4.disposables.Disposable;
+
 /**
  * Interface called by the {@link Flowable#virtualTransform(VirtualTransformer, java.util.concurrent.ExecutorService)}
  * operator to generate any number of output values based of the current input of the upstream.
@@ -29,7 +32,8 @@ public interface VirtualTransformer<T, R> {
      * 
      * @param value the upstream value
      * @param emitter the emitter to use to generate result value(s)
+     * @param stopper call to stop the upstream
      * @throws Throwable signaled as {@code onError} for the downstream.
      */
-    void transform(T value, VirtualEmitter<R> emitter) throws Throwable;
+    void transform(T value, VirtualEmitter<R> emitter, Disposable stopper) throws Throwable;
 }

src/main/java/io/reactivex/rxjava4/functions/Consumer3.java

@@ -0,0 +1,36 @@
+/*
+ * 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.functions;
+
+import io.reactivex.rxjava4.annotations.NonNull;
+
+/**
+ * A functional interface (callback) that accepts two values (of possibly different types).
+ * @param <T1> the first value type
+ * @param <T2> the second value type
+ * @param <T3> the third value type
+ * @since 4.0.0
+ */
+@FunctionalInterface
+public interface Consumer3<@NonNull T1, @NonNull T2, @NonNull T3> {
+
+    /**
+     * Performs an operation on the given values.
+     * @param t1 the first value
+     * @param t2 the second value
+     * @param t3 the third value
+     * @throws Throwable if the implementation wishes to throw any type of exception
+     */
+    void accept(T1 t1, T2 t2, T3 t3) throws Throwable;
+}