Upgrade components, add Streamable.create for the universal generator

Author: akarnokd

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

@@ -15,8 +15,30 @@
 
 import java.util.concurrent.CompletionStage;
 
-import io.reactivex.rxjava4.disposables.Disposable;
+import io.reactivex.rxjava4.annotations.NonNull;
+import io.reactivex.rxjava4.disposables.*;
 
-public record CompletionStageDisposable<T>(CompletionStage<T> stage, Disposable disposable) {
+/**
+ * Consist of a terminal stage and a disposable to be able to cancel a sequence.
+ * @param <T> the return and element type of the various stages
+ * @param stage the embedded stage to work with
+ * @param disposable the way to cancel the stage concurrently
+ * @since 4.0.0
+ */
+public record CompletionStageDisposable<T>(@NonNull CompletionStage<T> stage, @NonNull Disposable disposable) {
+
+    /**
+     * Await the completion of the current stage.
+     */
+    public void await() {
+        Streamer.await(stage);
+    }
 
+    /**
+     * Await the completion of the current stage.
+     * @param canceller the canceller link
+     */
+    public void await(DisposableContainer canceller) {
+        Streamer.await(stage, canceller);
+    }
 }

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

@@ -21062,16 +21062,44 @@ public final Stream<T> blockingStream(int prefetch) {
      *  relays the values one-by-one to the {@link Streamer } view of the sequence.
      *  </dd>
      *  <dt><b>Scheduler:</b></dt>
-     *  <dd>The operator by design does not run on any scheduler or executor.</dd>
+     *  <dd>The operator by design runs on the default virtual executor of the system.</dd>
      * </dl>
      * <p>
      * @return the new {@code Streamable} instance
+     * @since 4.0.0
      */
     @CheckReturnValue
     @BackpressureSupport(BackpressureKind.FULL)
-    @SchedulerSupport(SchedulerSupport.NONE)
+    @SchedulerSupport(SchedulerSupport.VIRTUAL)
     @NonNull
     public final Streamable<T> toStreamable() {
-        return new StreamableFromPublisher <>(this);
+        return toStreamable(Executors.newVirtualThreadPerTaskExecutor());
+    }
+
+    /**
+     * Converts this {@code Flowable} into a {@link Streamable} instance,
+     * transparently relaying signals between the two async representations of a sequence.
+     * <p>
+     * <dl>
+     *  <dt><b>Backpressure:</b></dt>
+     *  <dd>This operator requests from the upstream in a bounded manner and
+     *  relays the values one-by-one to the {@link Streamer } view of the sequence.
+     *  </dd>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>The operator runs on the provided {@link ExecutorService}.</dd>
+     * </dl>
+     * <p>
+     * @param executor where the coordination will happen
+     * @return the new {@code Streamable} instance
+     * @throws NullPointerException if {@code executor} is {@code null}
+     * @since 4.0.0
+     */
+    @CheckReturnValue
+    @BackpressureSupport(BackpressureKind.FULL)
+    @SchedulerSupport(SchedulerSupport.NONE)
+    @NonNull
+    public final Streamable<T> toStreamable(ExecutorService executor) {
+        Objects.requireNonNull(executor, "executor is null");
+        return new StreamableFromPublisher<>(this, executor);
     }
 }

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

@@ -14,7 +14,8 @@
 package io.reactivex.rxjava4.core;
 
 import java.util.Objects;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicReference;
 
 import io.reactivex.rxjava4.annotations.*;
 import io.reactivex.rxjava4.disposables.Disposable;
@@ -381,6 +382,29 @@ public <S extends Scheduler & Disposable> S when(@NonNull Function<Flowable<Flow
         return (S) new SchedulerWhen(combine, this);
     }
 
+    /**
+     * Turn this Scheduler into an ExecutorService implementation
+     * using its various *Direct() methods instead of workers.
+     * @return the ExecutorService view of this Scheduler
+     * @since 4.0.0
+     */
+    public ExecutorService toExecutorService() {
+        return new SchedulerToExecutorService(this, null);
+    }
+
+    /**
+     * Turn this Scheduler into an ExecutorService implementation
+     * using its various *Direct() methods or worker methods,
+     * depending on the parameter.
+     * @param useWorker if true, one of the workers is used as an executorservice,
+     *                  if false, the whole scheduler and its *Direct methods are used.
+     * @return the ExecutorService view of this Scheduler
+     * @since 4.0.0
+     */
+    public ExecutorService toExecutorService(boolean useWorker) {
+        return new SchedulerToExecutorService(this, new AtomicReference<>());
+    }
+
     /**
      * Represents an isolated, sequential worker of a parent Scheduler for executing {@code Runnable} tasks on
      * an underlying task-execution scheme (such as custom Threads, event loop, {@link java.util.concurrent.Executor Executor} or Actor system).
@@ -572,6 +596,35 @@ public Runnable getWrappedRunnable() {
                 return this.decoratedRun;
             }
         }
+
+        /**
+         * A stateless Worker that reports itself as shutdown and doesn't do anything.
+         * @since 4.0.0
+         */
+        @NonNull
+        public static final Worker SHUTDOWN = new ShutdownWorker();
+    }
+
+    /**
+     * Implementation of a stateless, shutdown worker. For cleanup and termination purposes.
+     * @since 4.0.0
+     */
+    static final class ShutdownWorker extends Worker {
+
+        @Override
+        public void dispose() {
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return true;
+        }
+
+        @Override
+        public @NonNull Disposable schedule(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
+            return Disposable.disposed();
+        }
+
     }
 
     static final class PeriodicDirectTask

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

@@ -22,6 +22,7 @@
 import io.reactivex.rxjava4.exceptions.Exceptions;
 import io.reactivex.rxjava4.functions.Consumer;
 import io.reactivex.rxjava4.internal.operators.streamable.*;
+import io.reactivex.rxjava4.schedulers.Schedulers;
 import io.reactivex.rxjava4.subscribers.TestSubscriber;
 
 /**
@@ -79,6 +80,7 @@ default Streamer<T> stream() {
      * @param item the constant item to produce
      * @return the {@code Streamable} instance
      */
+    @NonNull
     static <@NonNull T> Streamable<T> just(@NonNull T item) {
         Objects.requireNonNull(item, "item is null");
         return new StreamableJust<>(item);
@@ -90,10 +92,75 @@ default Streamer<T> stream() {
      * @param source Flow.Publisher to convert
      * @return the new Streamable instance
      */
-    static <T> Streamable<T> fromPublisher(Flow.Publisher<T> source) {
+    @NonNull
+    static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source) {
         Objects.requireNonNull(source, "source is null");
-        return new StreamableFromPublisher<T>(source);
+        return fromPublisher(source, Executors.newVirtualThreadPerTaskExecutor());
+    }
+
+    /**
+     * Convert any Flow.Publisher into a Streamable sequence.
+     * @param <T> the element type
+     * @param source Flow.Publisher to convert
+     * @param executor where the conversion will run
+     * @return the new Streamable instance
+     */
+    @NonNull
+    static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source, @NonNull ExecutorService executor) {
+        Objects.requireNonNull(source, "source is null");
+        return new StreamableFromPublisher<T>(source, executor);
+    }
+
+    /**
+     * Generate a sequence of values via a virtual generator callback (yielder)
+     * which is free to block and is natively backpressured.
+     * <p>
+     * Runs on the {@link Schedulers#virtual()} scheduler.
+     * @param <T> the element type
+     * @param generator the generator to use
+     * @return the streamable instance
+     */
+    @NonNull
+    static <@NonNull T> Streamable<T> create(@NonNull VirtualGenerator<T> generator) {
+        // FIXME native implementation
+        return Flowable.virtualCreate(generator)
+                .toStreamable();
+    }
+
+    /**
+     * Generate a sequence of values via a virtual generator callback (yielder)
+     * which is free to block and is natively backpressured.
+     * <p>
+     * Runs on the given scheduler.
+     * @param <T> the element type
+     * @param generator the generator to use
+     * @param scheduler the scheduler to run the virtual generator on
+     * @return the streamable instance
+     */
+    @NonNull
+    static <@NonNull T> Streamable<T> create(@NonNull VirtualGenerator<T> generator, @NonNull Scheduler scheduler) {
+        // FIXME native implementation
+        return Flowable.virtualCreate(generator, scheduler)
+                .toStreamable();
+    }
+
+    /**
+     * Generate a sequence of values via a virtual generator callback (yielder)
+     * which is free to block and is natively backpressured.
+     * <p>
+     * Runs on the given executor service.
+     * @param <T> the element type
+     * @param generator the generator to use
+     * @param executor the executor to run the virtual generator on
+     * @return the streamable instance
+     */
+    @NonNull
+    static <@NonNull T> Streamable<T> create(@NonNull VirtualGenerator<T> generator, @NonNull ExecutorService executor) {
+        // FIXME native implementation
+        return Flowable.virtualCreate(generator, executor)
+                .toStreamable();
     }
+
     // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
     // Operators
     // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
@@ -103,6 +170,7 @@ static <T> Streamable<T> fromPublisher(Flow.Publisher<T> source) {
      * on the default Executors.newVirtualThreadPerTaskExecutor() virtual thread.
      * @return the new Flowable instance
      */
+    @NonNull
     default Flowable<T> toFlowable() {
         return toFlowable(Executors.newVirtualThreadPerTaskExecutor());
     }
@@ -113,10 +181,44 @@ default Flowable<T> toFlowable() {
      * @param executor the executor to use
      * @return the new Flowable instance
      */
-    default Flowable<T> toFlowable(ExecutorService executor) {
+    @NonNull
+    default Flowable<T> toFlowable(@NonNull ExecutorService executor) {
+        Objects.requireNonNull(executor, "executir is null");
         var me = this;
         return Flowable.virtualCreate(emitter -> {
-            me.forEach(emitter::emit);
+            me.forEach(emitter::emit).await(emitter.canceller());
+        }, executor);
+    }
+
+    /**
+     * Transforms the upstream sequence into zero or more elements for the downstream.
+     * @param <R> the result element type
+     * @param transformer the interface to implement the transforming logic
+     * @return the new Streamable instance
+     */
+    @NonNull
+    default <@NonNull R> Streamable<R> transform(@NonNull VirtualTransformer<T, R> transformer) {
+        return transform(transformer, Executors.newVirtualThreadPerTaskExecutor());
+    }
+
+    /**
+     * Transforms the upstream sequence into zero or more elements for the downstream.
+     * @param <R> the result element type
+     * @param transformer the interface to implement the transforming logic
+     * @param executor where to run the transform and blocking operations
+     * @return the new Streamable instance
+     */
+    @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());
         }, executor);
     }
 
@@ -174,6 +276,7 @@ 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());
                         consumer.accept(Objects.requireNonNull(str.current(), "The upstream Streamable " + me.getClass() + " produced a null element!"));
                     } else {
                         break;
@@ -203,9 +306,11 @@ default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> con
     default void subscribe(@NonNull Flow.Subscriber<? super T> subscriber, @NonNull ExecutorService executor) {
         final Streamable<T> me = this;
         Flowable.<T>virtualCreate(emitter -> {
+            // System.out.println("subscribe::virtualCreate");
             me.forEach(v -> {
+                // System.out.println("subscribe::virtualCreate::forEach::emit");
                 emitter.emit(v);
-            });
+            }).await();
         }, executor)
         .subscribe(subscriber);
     }
@@ -218,6 +323,7 @@ default void subscribe(@NonNull Flow.Subscriber<? super T> subscriber) {
         final Streamable<T> me = this;
         Flowable.<T>virtualCreate(emitter -> {
             me.forEach(v -> {
+                // System.out.println("Emitting " + v);
                 emitter.emit(v);
             });
         })
@@ -233,4 +339,15 @@ default TestSubscriber<T> test() {
         subscribe(ts);
         return ts;
     }
+
+    /**
+     * Creates a new {@link TestSubscriber} and subscribes it to this {@code Streamable}.
+     * @param executor the executor to use
+     * @return the created test subscriber
+     */
+    default TestSubscriber<T> test(ExecutorService executor) {
+        var ts = new TestSubscriber<T>();
+        subscribe(ts, executor);
+        return ts;
+    }
 }