4.x: Add virtualCreate and virtualTransform to Flowable

Author: akarnokd

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

@@ -15,10 +15,9 @@
 
 import java.util.*;
 import java.util.concurrent.*;
+import java.util.concurrent.Flow.*;
 import java.util.stream.*;
 
-import static java.util.concurrent.Flow.*;
-
 import io.reactivex.rxjava4.annotations.*;
 import io.reactivex.rxjava4.disposables.*;
 import io.reactivex.rxjava4.exceptions.*;
@@ -34,6 +33,7 @@
 import io.reactivex.rxjava4.internal.schedulers.ImmediateThinScheduler;
 import io.reactivex.rxjava4.internal.subscribers.*;
 import io.reactivex.rxjava4.internal.util.*;
+import io.reactivex.rxjava4.internal.virtual.*;
 import io.reactivex.rxjava4.operators.ScalarSupplier;
 import io.reactivex.rxjava4.parallel.ParallelFlowable;
 import io.reactivex.rxjava4.plugins.RxJavaPlugins;
@@ -20896,4 +20896,122 @@ public final Stream<T> blockingStream(int prefetch) {
         ObjectHelper.verifyPositive(prefetch, "prefetch");
         return RxJavaPlugins.onAssembly(new FlowableFlatMapStream<>(this, mapper, prefetch));
     }
+    
+    /**
+     * Construct a {@code Flowable} and use the given {@code generator}
+     * to generate items on demand while running on the given {@link ExecutorService}.
+     * <p>
+     * <dl>
+     *  <dt><b>Backpressure:</b></dt>
+     *  <dd>This operator honors backpressure from downstream and blocks the emitter if
+     *  the downstream is not ready.
+     *  </dd>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>You specify which {@link Scheduler} this operator will use.</dd>
+     * </dl>
+     * <p>
+     * Note that backpressure is handled via blocking so it is recommended the provided
+     * {@code ExecutorService} uses virtual threads, such as the one returned by
+     * {@link Executors#newVirtualThreadPerTaskExecutor()}.
+     * <p>
+     * Examples:
+     * <pre><code>
+     * try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
+     *     Flowable.&lt;Integer&gt;virtualCreate(emitter -> {
+     *         for (int i = 0; i &lt; 10; i++) {
+     *             Thread.sleep(1000);
+     *             emitter.emit(i);
+     *         }
+     *     }, executor)
+     *     .subscribe(
+     *         System.out::println,
+     *         Throwable::printStackTrace,
+     *         () -&gt; System.out.println("Done")
+     *     );
+     * }
+     * </code></pre>
+     * @param <T> the element type to emit
+     * @param generator the callback used to generate items on demand by the downstream
+     * @param executor the target {@code ExecutorService} to use for running the callback
+     * @return the new {@code Flowable} instance
+     * @throws NullPointerException if {@code generator} or {@code executor} is {@code null}
+     * @since 4.0.0
+     */
+    @CheckReturnValue
+    @BackpressureSupport(BackpressureKind.FULL)
+    @SchedulerSupport(SchedulerSupport.NONE)
+    @NonNull
+    public static <@NonNull T> Flowable<T> virtualCreate(@NonNull VirtualGenerator<T> generator, @NonNull ExecutorService executor) {
+        Objects.requireNonNull(generator, "generator is null");
+        Objects.requireNonNull(executor, "executor is null");
+        return RxJavaPlugins.onAssembly(new FlowableVirtualCreateExecutor<>(generator, executor));
+    }
+
+    /**
+     * Returns a {@code Flowable} that turns an upstream item an upstream item into 
+     * zero or more downstream values by running on the given {@link ExecutorService}.
+     * <p>
+     * <dl>
+     *  <dt><b>Backpressure:</b></dt>
+     *  <dd>This operator honors backpressure from downstream and blocks the emitter if
+     *  the downstream is not ready.
+     *  </dd>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>You specify which {@link Scheduler} this operator will use.</dd>
+     * </dl>
+     * <p>
+     * Note that backpressure is handled via blocking so it is recommended the provided
+     * {@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)} is invoked for each upstream item
+     * @param executor the target {@code ExecutorService} to use for running the callback
+     * @return the new {@code Flowable} instance
+     * @throws NullPointerException if {@code transformer} or {@code executor} is {@code null}
+     * @since 4.0.0
+     */
+    @CheckReturnValue
+    @BackpressureSupport(BackpressureKind.FULL)
+    @SchedulerSupport(SchedulerSupport.NONE)
+    @NonNull
+    public final <@NonNull R> Flowable<R> virtualTransform(@NonNull VirtualTransformer<T, R> transformer, @NonNull ExecutorService executor) {
+        return virtualTransform(transformer, executor, Flowable.bufferSize());
+    }
+
+    /**
+     * Returns a {@code Flowable} that turns an upstream item into zero or more downstream
+     * values by running on the given {@link ExecutorService}.
+     * <p>
+     * <dl>
+     *  <dt><b>Backpressure:</b></dt>
+     *  <dd>This operator honors backpressure from downstream and blocks the emitter if
+     *  the downstream is not ready.
+     *  </dd>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>You specify which {@link Scheduler} this operator will use.</dd>
+     * </dl>
+     * <p>
+     * Note that backpressure is handled via blocking so it is recommended the provided
+     * {@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)} 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.
+     * @return the new {@code Flowable} instance
+     * @throws NullPointerException if {@code transformer} or {@code executor} is {@code null}
+     * @throws IllegalArgumentException if {@code prefetch} is non-positive
+     * @since 4.0.0
+     */
+    @CheckReturnValue
+    @BackpressureSupport(BackpressureKind.FULL)
+    @SchedulerSupport(SchedulerSupport.NONE)
+    @NonNull
+    public final <@NonNull R> Flowable<R> virtualTransform(@NonNull VirtualTransformer<T, R> transformer, @NonNull ExecutorService executor, int prefetch) {
+        Objects.requireNonNull(transformer, "transformer is null");
+        Objects.requireNonNull(executor, "executor is null");
+        ObjectHelper.verifyPositive(prefetch, "prefetch");
+        return new FlowableVirtualTransformExecutor<>(this, transformer, executor, prefetch);
+    }
+
 }

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

@@ -0,0 +1,30 @@
+/*
+ * 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;
+
+/**
+ * Interface handed to user code in {@link Flowable#virtualCreate(VirtualGenerator, java.util.concurrent.ExecutorService)} callback.
+ * @param <T> the element type to emit
+ * @since 4.0.0
+ */
+@FunctionalInterface
+public interface VirtualEmitter<T> {
+
+    /**
+     * Signal the next item
+     * @param item the item to signal
+     * @throws Throwable an arbitrary exception if the downstream cancelled
+     */
+    void emit(T item) throws Throwable;
+}

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

@@ -0,0 +1,34 @@
+/*
+ * 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;
+
+/**
+ * Interface to implement to produce elements when asked by
+ * {@link Flowable#virtaulCreate(VirtualGenerator, java.util.concurrent.ExecutorService)}.
+ * <p>
+ * To signal {@code onComplete}, return normally from {@link #generate(VirtualEmitter)}.
+ * To signal {@code onError}, throw any exception from {@link #generate(VirtualEmitter)}.
+ * @param <T> the element type generated
+ * @since 4.0.0
+ */
+@FunctionalInterface
+public interface VirtualGenerator<T> {
+
+    /**
+     * The method to implement and start emitting items.
+     * @param emitter use {@link VirtualEmitter#emit(Object)} to generate values
+     * @throws Throwable if the generator wishes to signal {@code onError}.
+     */
+    void generate(VirtualEmitter<T> emitter) throws Throwable;
+}

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

@@ -0,0 +1,35 @@
+/*
+ * 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;
+/**
+ * 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.
+ *
+ * @param <T> the source value type
+ * @param <R> the result value type
+ * @since 4.0.0
+ */
+@FunctionalInterface
+public interface VirtualTransformer<T, R> {
+
+    /**
+     * Implement this method to generate any number of items via
+     * {@link VirtualEmitter#emit(Object)}.
+     * 
+     * @param value the upstream value
+     * @param emitter the emitter to use to generate result value(s)
+     * @throws Throwable signaled as {@code onError} for the downstream.
+     */
+    void transform(T value, VirtualEmitter<R> emitter) throws Throwable;
+}

src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualCreateExecutor.java

@@ -0,0 +1,123 @@
+/*
+ * 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.internal.virtual;
+
+import java.util.Objects;
+import java.util.concurrent.*;
+import java.util.concurrent.Flow.*;
+import java.util.concurrent.atomic.AtomicLong;
+
+import io.reactivex.rxjava4.core.*;
+import io.reactivex.rxjava4.exceptions.Exceptions;
+import io.reactivex.rxjava4.internal.util.BackpressureHelper;
+
+/**
+ * Runs a generator callback on a virtual thread backed by a Worker of the given scheduler
+ * and signals events emitted by the generator considering any downstream backpressure.
+ *
+ * @param <T> the element type of the flow
+ * @since 4.0.0
+ */
+public final class FlowableVirtualCreateExecutor<T> extends Flowable<T> {
+
+    final VirtualGenerator<T> generator;
+
+    final ExecutorService executor;
+
+    public FlowableVirtualCreateExecutor(VirtualGenerator<T> generator, ExecutorService executor) {
+        this.generator = generator;
+        this.executor = executor;
+    }
+
+    @Override
+    protected void subscribeActual(Subscriber<? super T> s) {
+        var parent = new ExecutorVirtualCreateSubscription<>(s, generator);
+        s.onSubscribe(parent);
+
+        executor.submit(parent);
+    }
+
+    static final class ExecutorVirtualCreateSubscription<T> extends AtomicLong 
+    implements Subscription, Callable<Void>, VirtualEmitter<T> {
+
+        private static final long serialVersionUID = -6959205135542203083L;
+
+        Subscriber<? super T> downstream;
+
+        final VirtualGenerator<T> generator;
+
+        final VirtualResumable consumerReady;
+
+        volatile boolean cancelled;
+
+        static final Throwable STOP = new Throwable("Downstream cancelled");
+
+        long produced;
+
+        ExecutorVirtualCreateSubscription(Subscriber<? super T> downstream, VirtualGenerator<T> generator) {
+            this.downstream = downstream;
+            this.generator = generator;
+            this.consumerReady = new VirtualResumable();
+        }
+
+        @Override
+        public Void call() {
+            try {
+                try {
+                    generator.generate(this);
+                } catch (Throwable ex) {
+                    Exceptions.throwIfFatal(ex);
+                    if (ex != STOP && !cancelled) {
+                        downstream.onError(ex);
+                    }
+                    return null;
+                }
+                if (!cancelled) {
+                    downstream.onComplete();
+                }
+            } finally {
+                downstream = null;
+            }
+            return null;
+        }
+
+        @Override
+        public void request(long n) {
+            BackpressureHelper.add(this, n);
+            consumerReady.resume();
+        }
+
+        @Override
+        public void cancel() {
+            cancelled = true;
+            request(1);
+        }
+
+        @Override
+        public void emit(T item) throws Throwable {
+            Objects.requireNonNull(item, "item is null");
+            var p = produced;
+            while (get() == p && !cancelled) {
+                consumerReady.await();
+            }
+
+            if (cancelled) {
+                throw STOP;
+            }
+
+            downstream.onNext(item);
+            produced = p + 1;
+        }
+    }
+}