4.x: Streamable: +delay +timeout +zip +onErrorResumeNext, cleanup (#8216)

* 4.x: Streamable: +delay +timeout +zip +onErrorResumeNext, cleanup

* Try fixing StreamableGroupByTest.groupDisposeTest

* Disable flaky test

Author: akarnokd

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

@@ -0,0 +1,31 @@
+/*
+ * 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 java.util.concurrent.CompletionStage;
+import java.util.concurrent.Flow.Processor;
+
+import io.reactivex.rxjava4.annotations.NonNull;
+
+/**
+ * A {@link Processor}-like interface combining the {@code Streamable} interface and the
+ * {@link StreamerInput} interface to establish a push-pull bridge based on {@link CompletionStage}-based
+ * asynchronous processing and dispatching of values and errors.
+ * @param <In> the element type of the input side
+ * @param <Out> the element type of the output side
+ * @since 4.0.0
+ */
+public interface StreamProcessor<@NonNull In, @NonNull Out> extends Streamable<Out>, StreamerInput<In> {
+
+}

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

@@ -16,7 +16,7 @@
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.stream.Stream;
+import java.util.stream.*;
 
 import io.reactivex.rxjava4.annotations.*;
 import io.reactivex.rxjava4.core.config.*;
@@ -539,10 +539,56 @@ static Streamable<Long> timer(long delay, TimeUnit unit, ExecutorService executo
         return RxJavaPlugins.onAssembly(new StreamableTimer(delay, unit, null, executor));
     }
 
+    /**
+     * Takes the next element from each source {@code Streamable} and emits them a a single
+     * row of {@link List}.
+     * <p>
+     * If any of the sources is shorter than the rest or any of them fails, the
+     * sequence is completed early normally or with an exception, respectively.
+     * @param <T> the common element type of the sequences
+     * @param sources the iterable sequence of the source {@code Streamable}s
+     * @return the new {@code Streamable} instance
+     * @throws NullPointerException if {@code sources} is {@&ode null}
+     */
+    static <T> Streamable<List<T>> zip(Iterable<? extends Streamable<? extends T>> sources) {
+        Objects.requireNonNull(sources, "sources is null");
+        return RxJavaPlugins.onAssembly(new StreamableZip<>(sources));
+    }
+
     // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
     // Operators
     // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
 
+    /**
+     * Collects all upstream values via the use of a {@link Collector} configuration
+     * and emits its resulting value as a single item of the returned {@code Streamable}.
+     * <p>
+     * See {@link Collectors} for the most typical collector standard implementations.
+     * @param <A> the accumulator type of the collector
+     * @param <R> the result type of the collector and the returned {@code Streamamble}
+     * @param collector the Java collector instance to use
+     * @return the new {@code Streamable} instance
+     * @throws NullPointerException if {@code collector} is {@code null}
+     */
+    default <A, R> Streamable<R> collect(Collector<T, A, R> collector) {
+        Objects.requireNonNull(collector, "collector is null");
+        return RxJavaPlugins.onAssembly(new StreamableCollector<>(this, collector));
+    }
+
+    /**
+     * Delays the delivery of each upstream item by the given time amount.
+     * @param time the delay time
+     * @param unit the time unit
+     * @param scheduler the scheduler where the timed wait is happening
+     * @return the new {@code Streamable} instance
+     * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null}
+     */
+    default Streamable<T> delay(long time, TimeUnit unit, Scheduler scheduler) {
+        Objects.requireNonNull(unit, "unit is null");
+        Objects.requireNonNull(scheduler, "scheduler is null");
+        return RxJavaPlugins.onAssembly(new StreamableDelay<>(this, time, unit, scheduler));
+    }
+
     /**
      * Calls the given consumer whenever an upstream item becomes available.
      * @param consumer the callback to invoke with the next item from upstream
@@ -667,6 +713,19 @@ default Streamable<T> intercept(StreamableInterceptConfig<T> config) {
         return RxJavaPlugins.onAssembly(new StreamableMapOptional<>(this, mapper));
     }
 
+    /**
+     * When the upstream fails, the sequence is resumed by the {@code Streamable} that is returned for the
+     * failure {@link Throwable}.
+     * @param fallbackMapper the function that receives the upstream error and should return a
+     *                       {@code Streamable} to resume the sequence with
+     * @return the new {@code Streamable} instance
+     * @throws NullPointerException if {@code fallbackMapper} is {@code null}
+     */
+    default Streamable<T> onErrorResumeNext(Function<? super Throwable, ? extends Streamable<? extends T>> fallbackMapper) {
+        Objects.requireNonNull(fallbackMapper, "fallbackMapper is null");
+        return RxJavaPlugins.onAssembly(new StreamableOnErrorResumeNext<>(this, fallbackMapper));
+    }
+
     /**
      * Takes at most the given number of items from the upstream and relays it to the downstream,
      * then cancels the rest of the sequence.
@@ -689,6 +748,21 @@ default Streamable<T> take(long n) {
         });
     }
 
+    /**
+     * Relays items from this {@code Streamable} until the other {@code Streamable} signals
+     * an item or completes.
+     * @param <U> the element type of the other {@code Streamable}
+     * @param other the {@code Streamable} expected to signal when to stop taking items from this {@code Streamable}
+     * @return the new {@code Streamable} instance
+     * @throws NullPointerException if {@code other} is {@code null}
+     */
+    @CheckReturnValue
+    @NonNull
+    default <U> Streamable<T> takeUntil(@NonNull Streamable<U> other) {
+        Objects.requireNonNull(other, "other is null");
+        return RxJavaPlugins.onAssembly(new StreamableTakeUntil<>(this, other));
+    }
+
     /**
      * Relays items from this {@code Streamable} while the predicate returns {@code true}
      * @param predicate the predicate to test if the sequence should keep going
@@ -703,18 +777,21 @@ default Streamable<T> takeWhile(@NonNull Predicate<? super T> predicate) {
     }
 
     /**
-     * Relays items from this {@code Streamable} until the other {@code Streamable} signals
-     * an item or completes.
-     * @param <U> the element type of the other {@code Streamable}
-     * @param other the {@code Streamable} expected to signal when to stop taking items from this {@code Streamable}
+     * Applies a timeout to each upstream item and switches to the fallback {@code Streamable}
+     * if the upstream doesn't produce an item within the given timeout period.
+     * @param timeout the time to wait for each upstream item
+     * @param unit the time unit
+     * @param scheduler the scheduler where to wait for the next upstream item
+     * @param fallback the {@code Streamable} to switch to if the upstream doesn't produce an item
+     *                 within the time limit
      * @return the new {@code Streamable} instance
-     * @throws NullPointerException if {@code other} is {@code null}
+     * @throws NullPointerException if {@code unit} or {@code scheduler} or {@code fallback} is {@code null}
      */
-    @CheckReturnValue
-    @NonNull
-    default <U> Streamable<T> takeUntil(@NonNull Streamable<U> other) {
-        Objects.requireNonNull(other, "other is null");
-        return RxJavaPlugins.onAssembly(new StreamableTakeUntil<>(this, other));
+    default Streamable<T> timeout(long timeout, TimeUnit unit, Scheduler scheduler, Streamable<T> fallback) {
+        Objects.requireNonNull(unit, "unit is null");
+        Objects.requireNonNull(scheduler, "scheduler is null");
+        Objects.requireNonNull(fallback, "fallback is null");
+        return RxJavaPlugins.onAssembly(new StreamableTimeout<>(this, timeout, unit, scheduler, fallback));
     }
 
     /**
@@ -891,8 +968,8 @@ default CompletionStageDisposable<Void> forEach(
     default void subscribe(@NonNull Flow.Subscriber<? super T> subscriber, @NonNull ExecutorService executor) {
         final Streamable<T> me = this;
         Flowable.<T>virtualCreate(emitter -> {
-            me.forEach(emitter::emit, emitter.canceller().derive(), executor)
-            .await();
+            var cf = me.forEach(emitter::emit, emitter.canceller().derive(), executor);
+            cf.await();
         }, executor)
         .subscribe(subscriber);
     }

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

@@ -0,0 +1,49 @@
+/*
+ * 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 java.util.concurrent.CompletionStage;
+import java.util.concurrent.Flow.Subscriber;
+
+import io.reactivex.rxjava4.annotations.*;
+
+/**
+ * An interface to submit items and terminal events to a consumer that indacates when the processing of
+ * said item or terminal event has completed, similar to how {@link Subscriber} can receive events.
+ * <p>
+ * The general contract is to call {@link #next(Object)} zero or more times, then
+ * call {@link #finish(Throwable)} at most once, all in a non-overlapping fashion and only if the
+ * returned {@link CompletionStage} has completed in some fashion.
+ * @param <T> the item type to be offered
+ * @since 4.0.0
+ */
+public interface StreamerInput<@NonNull T> {
+
+    /**
+     * Offer the next item.
+     * @param item the item being offered
+     * @return a {@link CompletionStage} that should complete with {@code true} if the
+     *         item was accepted, or {@code false} if the item could not be accepted at this time,
+     *         or via a {@code Throwable} indicating some error.
+     */
+    CompletionStage<Boolean> next(T item);
+
+    /**
+     * Offer the final, terminal event.
+     * @param throwable the optional throwable to signal error, null to signal normal completion
+     * @return the {@link CompletionStage} that should complete normally if the terminal event was accepted,
+     *         or via a {@code Throwable} indicating an error
+     */
+    CompletionStage<Void> finish(@Nullable Throwable throwable);
+}

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollector.java

@@ -0,0 +1,131 @@
+/*
+ * 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.operators.streamable;
+
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.*;
+import java.util.stream.Collector;
+
+import io.reactivex.rxjava4.annotations.NonNull;
+import io.reactivex.rxjava4.core.*;
+import io.reactivex.rxjava4.disposables.DisposableContainer;
+import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource;
+
+public record StreamableCollector<T, A, R>(
+        Streamable<T> source,
+        Collector<T, A, R> collector
+) implements Streamable<R>, HasUpstreamStreamableSource<T> {
+
+    @Override
+    public @NonNull Streamer<@NonNull R> stream(@NonNull DisposableContainer cancellation) {
+        return new CollectorStreamable<>(
+                source.stream(cancellation),
+                collector.supplier().get(),
+                collector.accumulator(),
+                collector.finisher());
+    }
+
+    static final class CollectorStreamable<T, A, R>
+    implements Streamer<R>, BiConsumer<Object, Throwable> {
+
+        final AtomicInteger wip;
+
+        final Streamer<T> upstream;
+
+        final A storage;
+
+        final BiConsumer<A, T> accumulator;
+
+        final Function<A, R> finisher;
+
+        final CompletableFuture<Boolean> nextReady;
+
+        final CompletableFuture<Void> finishReady;
+
+        R current;
+
+        int stage;
+
+        volatile boolean done;
+
+        CollectorStreamable(Streamer<T> upstream, A storage, BiConsumer<A, T> accumulator, Function<A, R> finisher) {
+            this.upstream = upstream;
+            this.wip = new AtomicInteger();
+            this.storage = storage;
+            this.accumulator = accumulator;
+            this.finisher = finisher;
+            this.nextReady = new CompletableFuture<>();
+            this.finishReady = new CompletableFuture<>();
+        }
+
+        @Override
+        public @NonNull CompletionStage<Boolean> next() {
+            if (stage++ == 0) {
+                drain();
+                return nextReady;
+            }
+            return NEXT_FALSE;
+        }
+
+        @Override
+        public @NonNull R current() {
+            return current;
+        }
+
+        @Override
+        public @NonNull CompletionStage<Void> finish() {
+            done = true;
+            drain();
+            return finishReady;
+        }
+
+        void drain() {
+            if (wip.getAndIncrement() != 0) {
+                return;
+            }
+
+            do {
+                if (done) {
+                    upstream.finish().whenComplete(this);
+                    break;
+                } else {
+                    upstream.next().whenComplete(this);
+                }
+            } while (wip.decrementAndGet() != 0);
+        }
+
+        @Override
+        public void accept(Object t, Throwable u) {
+            if (done) {
+                if (u != null) {
+                    finishReady.completeExceptionally(u);
+                } else {
+                    finishReady.complete(null);
+                }
+            } else {
+                if (u != null) {
+                    nextReady.completeExceptionally(u);
+                } else
+                if ((Boolean)t) {
+                    accumulator.accept(storage, upstream.current());
+                    drain();
+                } else {
+                    current = finisher.apply(storage);
+                    nextReady.complete(true);
+                }
+            }
+        }
+    }
+}