4.x: Streamable +blockingFirst,last[OrError],skip ,S.flattenAsStreamable (#8229)

Author: akarnokd

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

@@ -3228,6 +3228,8 @@ public final Future<Void> toFuture() {
      * <p>
      * <img width="640" height="293" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Completable.toStreamable.png" alt="">
      * <dl>
+     *  <dt><b>Backpressure:</b></dt>
+     *  <dd>The returned {@code Streamable} honors the backpressure of the downstream consumer.</dd>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code toStreamable} does not operate by default on a particular {@link Scheduler}.</dd>
      * </dl>
@@ -3236,6 +3238,7 @@ public final Future<Void> toFuture() {
      * @since 4.0.0
      */
     @SchedulerSupport(SchedulerSupport.NONE)
+    @BackpressureSupport(BackpressureKind.FULL)
     @CheckReturnValue
     @NonNull
     public final <@NonNull T> Streamable<T> toStreamable() {

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

@@ -3896,14 +3896,17 @@ public final Single<T> toSingle() {
      * <p>
      * <img width="640" height="346" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.toStreamable.png" alt="">
      * <dl>
+     *  <dt><b>Backpressure:</b></dt>
+     *  <dd>The returned {@code Streamable} honors the backpressure of the downstream consumer.</dd>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code toStreamable} does not operate by default on a particular {@link Scheduler}.</dd>
      * </dl>
      * @return the new {@code Streamable} instance
      * @since 4.0.0
      */
-    @CheckReturnValue
     @SchedulerSupport(SchedulerSupport.NONE)
+    @BackpressureSupport(BackpressureKind.FULL)
+    @CheckReturnValue
     @NonNull
     public final Streamable<T> toStreamable() {
         return RxJavaPlugins.onAssembly(new StreamableFromMaybe<>(this));

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

@@ -33,7 +33,7 @@
 import io.reactivex.rxjava4.internal.operators.mixed.*;
 import io.reactivex.rxjava4.internal.operators.observable.ObservableSingleSingle;
 import io.reactivex.rxjava4.internal.operators.single.*;
-import io.reactivex.rxjava4.internal.operators.streamable.StreamableFromSingle;
+import io.reactivex.rxjava4.internal.operators.streamable.*;
 import io.reactivex.rxjava4.observers.TestObserver;
 import io.reactivex.rxjava4.plugins.RxJavaPlugins;
 import io.reactivex.rxjava4.schedulers.*;
@@ -2976,6 +2976,37 @@ public final Maybe<T> filter(@NonNull Predicate<? super T> predicate) {
         return RxJavaPlugins.onAssembly(new SingleFlatMapIterableObservable<>(this, mapper));
     }
 
+    /**
+     * Maps the success value of the current {@code Single} into an {@link Iterable} and emits its items as a
+     * {@link Streamable} sequence.
+     * <p>
+     * <img width="640" height="373" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/flattenAsStreamable.png" alt="">
+     * <dl>
+     *  <dt><b>Backpressure:</b></dt>
+     *  <dd>The operator honors backpressure from downstream.</dd>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code flattenAsStreamable} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param <U>
+     *            the type of item emitted by the resulting {@code Iterable}
+     * @param mapper
+     *            a function that returns an {@code Iterable} sequence of values for when given an item emitted by the
+     *            current {@code Single}
+     * @return the new {@code Streamable} instance
+     * @throws NullPointerException if {@code mapper} is {@code null}
+     * @see <a href="http://reactivex.io/documentation/operators/flatmap.html">ReactiveX operators documentation: FlatMap</a>
+     * @since 4.0.0
+     */
+    @BackpressureSupport(BackpressureKind.FULL)
+    @CheckReturnValue
+    @NonNull
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final <@NonNull U> Streamable<U> flattenAsStreamable(@NonNull Function<? super T, @NonNull ? extends Iterable<? extends U>> mapper) {
+        Objects.requireNonNull(mapper, "mapper is null");
+        return RxJavaPlugins.onAssembly(new StreamableSingleFlattenAs<>(this, mapper));
+    }
+
     /**
      * Returns an {@link Observable} that is based on applying a specified function to the item emitted by the current {@code Single},
      * where that function returns an {@link ObservableSource}.
@@ -4894,14 +4925,17 @@ public final Observable<T> toObservable() {
      * <p>
      * <img width="640" height="305" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.toStreamable.v3.png" alt="">
      * <dl>
+     *  <dt><b>Backpressure:</b></dt>
+     *  <dd>The returned {@code Streamable} honors the backpressure of the downstream consumer.</dd>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>{@code toStreamable} does not operate by default on a particular {@link Scheduler}.</dd>
      * </dl>
      * @return the new {@code Streamable} instance
      * @since 4.0.0
      */
-    @CheckReturnValue
     @SchedulerSupport(SchedulerSupport.NONE)
+    @BackpressureSupport(BackpressureKind.FULL)
+    @CheckReturnValue
     @NonNull
     public final Streamable<T> toStreamable() {
         return RxJavaPlugins.onAssembly(new StreamableFromSingle<>(this));

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

@@ -611,6 +611,19 @@ static <T> Streamable<List<T>> zip(Iterable<? extends Streamable<? extends T>> s
     // Operators
     // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
 
+    /**
+     * Blocks the current thread until this {@code Streamable} produces one item, which is then returned
+     * @return the first item of this {@code Streamable}
+     * @throws NoSuchElementException if the this {@code Streamable} is empty
+     * @throws CancellationException if this {@code Streamable} failed with a checked exception
+     * @throws RuntimeException if this {@code Streamable} failed with an unchecked exception
+     */
+    @CheckReturnValue
+    @NonNull
+    default T blockingFirst() {
+        return StreamableBlocking.blockingFirst(this);
+    }
+
     /**
      * 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}.
@@ -732,6 +745,33 @@ default Streamable<T> intercept(StreamableInterceptConfig<T> config) {
                 config.onStream(), config.onNext(), config.onCurrent(), config.onFinish()));
     }
 
+    /**
+     * Signals the last item from the upstream {@code Streamable} or
+     * the provided {@code defaultItem} if the upstream is empty, as a
+     * {@link Single} instance.
+     * @param defaultItem the item to signal if the upstream turns out to be empty
+     * @return the new {@code Single} instance
+     * @throws NullPointerException if {@code defaultItem} is {@code null}
+     */
+    @CheckReturnValue
+    @NonNull
+    default Single<T> last(@NonNull T defaultItem) {
+        Objects.requireNonNull(defaultItem, "defaultItem is null");
+        return RxJavaPlugins.onAssembly(new StreamableLastAsSingle<>(this, defaultItem));
+    }
+
+    /**
+     * Signals the last item from the upstream {@code Streamable} or a
+     * {@link NoSuchElementException} if the upstream is empty, as a
+     * {@link Single} instance.
+     * @return the new {@code Single} instance
+     */
+    @CheckReturnValue
+    @NonNull
+    default Single<T> lastOrError() {
+        return RxJavaPlugins.onAssembly(new StreamableLastAsSingle<>(this, null));
+    }
+
     /**
      * <strong>This method requires advanced knowledge about building operators, please consider
      * other standard composition methods first;</strong>
@@ -801,6 +841,21 @@ default Streamable<T> onErrorResumeNext(@NonNull Function<? super Throwable, ? e
         return RxJavaPlugins.onAssembly(new StreamableOnErrorResumeNext<>(this, fallbackMapper));
     }
 
+    /**
+     * Skips the first {@code count} items and relays the rest to the downstream.
+     * @param count the number of items to skip
+     * @return the new {@Streamable} instance
+     * @throws IllegalArgumentException if {@code count} is negative
+     */
+    @CheckReturnValue
+    @NonNull
+    default Streamable<T> skip(long count) {
+        if (count < 0) {
+            throw new IllegalArgumentException("count >= 0 expected but it was " + count);
+        }
+        return RxJavaPlugins.onAssembly(new StreamableSkip<>(this, count));
+    }
+
     /**
      * Takes at most the given number of items from the upstream and relays it to the downstream,
      * then cancels the rest of the sequence.

src/main/java/io/reactivex/rxjava4/internal/disposables/DisposableOnly.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.internal.disposables;
+
+import io.reactivex.rxjava4.disposables.Disposable;
+
+/**
+ * An extension to {@link Disposable} that allows not
+ * implementing the {@link Disposable#isDisposed()} as it
+ * is practically never needed or cannot be observed anyways.
+ * @since 4.0.0
+ */
+public interface DisposableOnly extends Disposable {
+
+    @Override
+    default boolean isDisposed() {
+        throw new UnsupportedOperationException("The class " + this.getClass() + " does not support isDisposed");
+    }
+}

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

@@ -25,13 +25,21 @@
  * one use/thread can signal {@link #ready()} to wake up another use/thread
  * on a {@link #await()} call.
  * @param <T> the element type of the notification pass-around
+ * @since 4.0.0
  */
 public final class StageResumable<T> extends AtomicReference<CompletableFuture<T>>
 implements BiConsumer<T, Throwable> {
 
     @Serial
     private static final long serialVersionUID = -7518852864146380895L;
 
+    /**
+     * When the producer has arranged the item transfer via some field or queue,
+     * call this method and call {@link CompletableFuture#complete(Object)}
+     * or {@link CompletableFuture#completeExceptionally(Throwable)} to
+     * signal resumption for any current or upcoming {@link #await()} caller.
+     * @return the {@code CompletableFuture} to complete in some way
+     */
     @CheckReturnValue
     @NonNull
     public CompletableFuture<T> ready() {
@@ -49,6 +57,12 @@ public CompletableFuture<T> ready() {
         return cf;
     }
 
+    /**
+     * When the consumer is ready to receive an item, call this method
+     * and apply a continuation function, such as {@link CompletableFuture#whenComplete(BiConsumer)}
+     * to it to handle the signal and process any external data made ready.
+     * @return the {@code CompletableFuture} to observe a completion value or exception
+     */
     @CheckReturnValue
     @NonNull
     public CompletableFuture<T> await() {
@@ -66,6 +80,10 @@ public CompletableFuture<T> await() {
         return cf.whenComplete(this);
     }
 
+    /// Used to clear any waiting [CompletableFuture] when the await finishes
+    /// no concern to users and should not be called.
+    /// @param t the completion value if any, ignored
+    /// @param u the exception if any, ignored
     @Override
     public void accept(T t, Throwable u) {
         getAndSet(null);

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

@@ -0,0 +1,67 @@
+/*
+ * 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.NoSuchElementException;
+import java.util.concurrent.CompletionException;
+
+import io.reactivex.rxjava4.annotations.*;
+import io.reactivex.rxjava4.core.Streamable;
+import io.reactivex.rxjava4.disposables.CompositeDisposable;
+import io.reactivex.rxjava4.exceptions.Exceptions;
+import io.reactivex.rxjava4.internal.util.ExceptionHelper;
+
+public record StreamableBlocking() {
+
+    /**
+     * Consumes the first item and finishes the {@link Streamable},
+     * throwing {@link NoSuchElementException} if the source is empty.
+     * @param <T> the element type
+     * @param source the source {@code Streamable}
+     * @return the first item
+     * @throws RuntimeException if the source signals an unchecked exception
+     * @throws CompletionException if the source signals a checked exception
+     */
+    @CheckReturnValue
+    @NonNull
+    public static <T> T blockingFirst(Streamable<T> source) {
+        var streamer = source.stream(new CompositeDisposable());
+        Throwable nextException = null;
+        Throwable finishException = null;
+        T result = null;
+        try {
+            if (streamer.awaitNext()) {
+                result = streamer.current();
+            }
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            nextException = ex;
+        }
+        try {
+            streamer.awaitFinish();
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            finishException = ex;
+        }
+
+        if (nextException != null || finishException != null) {
+            throw ExceptionHelper.wrapOrThrow(ExceptionHelper.unwrapAndCombine(nextException, finishException));
+        }
+        if (result == null) {
+            throw new NoSuchElementException();
+        }
+        return result;
+    }
+
+}