4.x: Streamable Ops: doOnNext, intercept, lift, to; test cleanup (#8209)

Author: akarnokd

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

@@ -19,7 +19,7 @@
 import java.util.stream.Stream;
 
 import io.reactivex.rxjava4.annotations.*;
-import io.reactivex.rxjava4.core.config.StandardConcurrentConfig;
+import io.reactivex.rxjava4.core.config.*;
 import io.reactivex.rxjava4.disposables.*;
 import io.reactivex.rxjava4.exceptions.Exceptions;
 import io.reactivex.rxjava4.functions.*;
@@ -37,6 +37,7 @@
  * @param <T> the element type of the stream.
  * @since 4.0.0
  */
+@FunctionalInterface
 public interface Streamable<@NonNull T> {
 
     // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
@@ -394,52 +395,83 @@ static Streamable<Long> rangeLong(long start, long count) {
     // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
 
     /**
-     * Converts the streamable into a Flowable representation, running
-     * on the default Executors.newVirtualThreadPerTaskExecutor() virtual thread.
-     * @return the new Flowable instance
+     * Calls the given consumer whenever an upstream item becomes available.
+     * @param consumer the callback to invoke with the next item from upstream
+     * @return the new {@code Streamable} instance
+     * @throws NullPointerException if {@code consumer} is {@code null}
+     */
+    default Streamable<T> doOnNext(Consumer<? super T> consumer) {
+        Objects.requireNonNull(consumer, "consumer is null");
+        return intercept(new StreamableInterceptConfig<>(v -> { consumer.accept(v); return v; } ));
+    }
+
+    /**
+     * Maps each upstream item onto a {@code Streamable} and runs them concurrently while
+     * relaying inner items as first-come-first-served manner.
+     * @param <R> the element type of the output sequence
+     * @param mapper the function that turns an upstream item into a {@code Streamable} inner sequence
+     * @param config the configuration record for this operator
+     * @return the new {@code Streamable} instance
+     * @throws NullPointerException if {@code mapper} or {@code config} is {@code null}
      */
     @CheckReturnValue
     @NonNull
-    default Flowable<T> toFlowable() {
-        return toFlowable(Executors.newVirtualThreadPerTaskExecutor());
+    default <R> Streamable<R> flatMap(@NonNull Function<? super T, ? extends Streamable<? extends R>> mapper,
+            @NonNull StandardConcurrentConfig config) {
+        Objects.requireNonNull(mapper, "mapper is null");
+        Objects.requireNonNull(config, "config is null");
+        return RxJavaPlugins.onAssembly(new StreamableFlatMap<>(this, mapper, config.maxConcurrency()));
     }
 
     /**
-     * Converts the streamable into a Flowable representation, running
-     * on the provided executor service.
-     * @param executor the executor to use
-     * @return the new Flowable instance
+     * Hides the identity of this {@code Streamable} and its {@link Streamer}.
+     * <p>
+     * Use it to break optimizations or hide concrete implementations.
+     * @return the new {@code Streamable} instance
      */
     @CheckReturnValue
     @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).await(emitter.canceller()), executor);
+    default Streamable<T> hide() {
+        return RxJavaPlugins.onAssembly(new StreamableHide<>(this));
     }
 
     /**
-     * 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
+     * Intercepts the lifecycle method calls of {@code Streamable} and {@link Streamer}
+     * and allows the modification of them via Function callbacks.
+     * @param config the configuration record for this operator
      * @return the new {@code Streamable} instance
+     * @thros NullPointerException if {@code config} is {@code null}
      */
     @CheckReturnValue
     @NonNull
-    default <@NonNull R> Streamable<R> transform(@NonNull VirtualTransformer<T, R> transformer) {
-        return transform(transformer, Executors.newVirtualThreadPerTaskExecutor());
+    default Streamable<T> intercept(StreamableInterceptConfig<T> config) {
+        Objects.requireNonNull(config, "config is null");
+        return RxJavaPlugins.onAssembly(new StreamableIntercept<>(this,
+                config.onStream(), config.onNext(), config.onCurrent(), config.onFinish()));
     }
 
     /**
-     * Hides the identity of this {@code Streamable} and its {@link Streamer}.
+     * <strong>This method requires advanced knowledge about building operators, please consider
+     * other standard composition methods first;</strong>
+     * Returns a {@code Streamable} instance which when its {@link #stream(DisposableContainer)} is invoked,
+     * applies the specified operator callback to the upstream {@link Streamer} to produce
+     * an actual {@code Streamer} instance to be handed downstream.
      * <p>
-     * Use it to break optimizations or hide concrete implementations.
+     * Use it to implement operators without creating the surrounding {@code Streamable} class.
+     * <p>
+     * If the {@code lifter} returns {@code null} or throws, the downstream will receive a
+     * standard error streamer.
+     * @param <R> the downstream type of the sequence
+     * @param lifter the callback that will be invoked with the upstream {@code Streamer} and is expected
+     *               to produce a {@code Streamer} for the downstream.
      * @return the new {@code Streamable} instance
+     * @throws NullPointerException if {@code lifter} is {@code null}
      */
     @CheckReturnValue
     @NonNull
-    default Streamable<T> hide() {
-        return RxJavaPlugins.onAssembly(new StreamableHide<>(this));
+    default <@NonNull R> Streamable<R> lift(@NonNull StreamableOperator<? super T, ? extends R> lifter) {
+        Objects.requireNonNull(lifter, "lifter is null");
+        return RxJavaPlugins.onAssembly(new StreamableLift<T, R>(this, lifter));
     }
 
     /**
@@ -472,27 +504,6 @@ default Streamable<T> hide() {
         return RxJavaPlugins.onAssembly(new StreamableMapOptional<>(this, mapper));
     }
 
-    /**
-     * 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
-     */
-    @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, stopper) -> {
-            // System.out.println("item " + item);
-            transformer.transform(item, emitter, stopper);
-        }, emitter.canceller(), executor)
-        .await(emitter.canceller()), executor);
-    }
-
     /**
      * Takes at most the given number of items from the upstream and relays it to the downstream,
      * then cancels the rest of the sequence.
@@ -515,21 +526,76 @@ default Streamable<T> take(long n) {
     }
 
     /**
-     * Maps each upstream item onto a {@code Streamable} and runs them concurrently while
-     * relaying inner items as first-come-first-served manner.
-     * @param <R> the element type of the output sequence
-     * @param mapper the function that turns an upstream item into a {@code Streamable} inner sequence
-     * @param config the configuration record for this operator
+     * Calls the specified converter function during assembly time and returns its resulting value.
+     * <p>
+     * This allows fluent conversion to any other type.
+     * @param <R> the resulting object type
+     * @param converter the function that receives the current {@code Observable} instance and returns a value
+     * @return the converted value
+     * @throws NullPointerException if {@code converter} is {@code null}
+     */
+    @CheckReturnValue
+    @NonNull
+    default <@NonNull R> R to(@NonNull StreamableConverter<T, ? extends R> converter) {
+        return Objects.requireNonNull(converter, "converter is null").apply(this);
+    }
+
+    /**
+     * Converts the streamable into a Flowable representation, running
+     * on the default Executors.newVirtualThreadPerTaskExecutor() virtual thread.
+     * @return the new Flowable instance
+     */
+    @CheckReturnValue
+    @NonNull
+    default Flowable<T> toFlowable() {
+        return toFlowable(Executors.newVirtualThreadPerTaskExecutor());
+    }
+
+    /**
+     * Converts the streamable into a Flowable representation, running
+     * on the provided executor service.
+     * @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");
+        var me = this;
+        return Flowable.virtualCreate(emitter -> 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 {@code Streamable} instance
-     * @throws NullPointerException if {@code mapper} or {@code config} is {@code null}
      */
     @CheckReturnValue
     @NonNull
-    default <R> Streamable<R> flatMap(@NonNull Function<? super T, ? extends Streamable<? extends R>> mapper,
-            @NonNull StandardConcurrentConfig config) {
-        Objects.requireNonNull(mapper, "mapper is null");
-        Objects.requireNonNull(config, "config is null");
-        return RxJavaPlugins.onAssembly(new StreamableFlatMap<>(this, mapper, config.maxConcurrency()));
+    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
+     */
+    @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, stopper) -> {
+            // System.out.println("item " + item);
+            transformer.transform(item, emitter, stopper);
+        }, emitter.canceller(), executor)
+        .await(emitter.canceller()), executor);
     }
 
     // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo

src/main/java/io/reactivex/rxjava4/core/StreamableConverter.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;
+
+import io.reactivex.rxjava4.annotations.NonNull;
+
+/**
+ * Convenience interface and callback used by the {@link Streamable#to} operator to turn an {@link Streamable} into another
+ * value fluently.
+ * @param <T> the upstream type
+ * @param <R> the output type
+ * @since 4.0.0
+ */
+@FunctionalInterface
+public interface StreamableConverter<@NonNull T, @NonNull R> {
+    /**
+     * Applies a function to the upstream {@link Streamable} and returns a converted value of type {@code R}.
+     *
+     * @param upstream the upstream {@code Streamable} instance
+     * @return the converted value
+     */
+    R apply(@NonNull Streamable<T> upstream);
+}

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

@@ -0,0 +1,38 @@
+/*
+ * 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;
+
+import io.reactivex.rxjava4.annotations.NonNull;
+import io.reactivex.rxjava4.disposables.DisposableContainer;
+
+/**
+ * Interface to map/wrap an upstream {@link Streamer} to an downstream {@code Streamer}.
+ *
+ * @param <T> the value type of the upstream
+ * @param <R> the value type of the downstream
+ * @since 4.0.0
+ */
+@FunctionalInterface
+public interface StreamableOperator<@NonNull T, @NonNull R> {
+    /**
+     * Applies a function to the upstream {@link Streamer} and returns a new downstream {@code Streamer}.
+     * @param container the {@link DisposableContainer} handling the cancellation propagation for the downstream
+     * @param streamer the upstream {@code Streamer} instance
+     * @return the downstream {@code Streamer} instance
+     * @throws Throwable on failure
+     */
+    @NonNull
+    Streamer<? extends R> apply(@NonNull DisposableContainer container,
+            @NonNull Streamer<? extends T> streamer) throws Throwable;
+}

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

@@ -0,0 +1,69 @@
+/*
+ * 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.config;
+
+import java.util.Objects;
+import java.util.concurrent.CompletionStage;
+
+import io.reactivex.rxjava4.annotations.NonNull;
+import io.reactivex.rxjava4.core.*;
+import io.reactivex.rxjava4.disposables.DisposableContainer;
+import io.reactivex.rxjava4.functions.*;
+
+/**
+ * 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 onCurrent called when the {@link Streamer#current()} is invoked
+ * @param onFinish called when the {@link Streamer#finish(io.reactivex.rxjava4.disposables.DisposableContainer)} is invoked
+ * @since 4.0.0
+ */
+public record StreamableInterceptConfig<T>(
+        @NonNull BiFunction<? super DisposableContainer, ? super Streamer<? extends T>, ? extends Streamer<? extends T>> onStream,
+        @NonNull BiFunction<? super DisposableContainer, ? super CompletionStage<Boolean>, ? extends CompletionStage<Boolean>> onNext,
+        @NonNull Function<? super T, ? extends T> onCurrent,
+        @NonNull BiFunction<? super DisposableContainer, ? super CompletionStage<Void>, ? extends CompletionStage<Void>> onFinish
+) {
+
+    /**
+     * Constructs a configuration with a custom {@link #onNext} intercept and everything else is pass-through.
+     * @param onNext the callback for intercepting the {@code next()} calls
+     */
+    public StreamableInterceptConfig(@NonNull BiFunction<? super DisposableContainer, ? super CompletionStage<Boolean>, ? extends CompletionStage<Boolean>> onNext) {
+        this((_, v) -> v, onNext, v -> v, (_, v) -> v);
+    }
+
+    /**
+     * Constructs a configuration with a custom {@link #onCurrent} intercept and everything else is pass-through.
+     * @param onCurrent the callback for when an item is ready
+     */
+    public StreamableInterceptConfig(@NonNull Function<? super T, ? extends T> onCurrent) {
+        this((_, v) -> v, (_, v) -> v, onCurrent, (_, v) -> v);
+    }
+
+    /**
+     * 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 onCurrent called when the {@link Streamer#current()} is invoked
+     * @param onFinish called when the {@link Streamer#finish(io.reactivex.rxjava4.disposables.DisposableContainer)} is invoked
+     */
+    public StreamableInterceptConfig {
+        Objects.requireNonNull(onStream, "onStream is null");
+        Objects.requireNonNull(onNext, "onNext is null");
+        Objects.requireNonNull(onCurrent, "onCurrent is null");
+        Objects.requireNonNull(onFinish, "onFinish is null");
+    }
+}