4.x: Streamable + blockingLast + ignoreElements

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

@@ -52,6 +52,27 @@ public interface StreamSink<@NonNull T> {
     @NonNull
     CompletionStage<Void> finish(@Nullable Throwable throwable);
 
+    /**
+     * Offers the given item and then awaits its consumption in a blocking fashion.
+     * @param item the item being offered
+     * @return true if the item was accepted, false if not
+     * @throws CancellationException if there was a cancellation issued
+     * @throws CompletionException if the upstream failed
+     */
+    default boolean awaitNext(T item) {
+        return next(item).toCompletableFuture().join();
+    }
+
+    /**
+     * Offer the final, terminal event and then awaits its consumption in a blocking fashion.
+     * @param throwable the optional throwable to signal error, {@code null} to signal normal completion
+     * @throws CancellationException if there was a cancellation issued
+     * @throws CompletionException if the upstream failed
+     */
+    default void awaitFinish(Throwable throwable) {
+        finish(throwable).toCompletableFuture().join();
+    }
+
     /**
      * Returns the {@link DisposableContainer} to use to detect if the consumer has indicated no more
      * items it is willing to accept.

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

@@ -612,7 +612,7 @@
     // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
 
     /**
-     * Blocks the current thread until this {@code Streamable} produces one item, which is then returned
+     * 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
@@ -624,6 +624,20 @@
         return StreamableBlocking.blockingFirst(this);
     }
 
+    /**
+     * Blocks the current thread until this {@code Streamable} produces all of its items
+     * and the very last is then returned.
+     * @return the very last 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 blockingLast() {
+        return StreamableBlocking.blockingLast(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}.
@@ -730,6 +744,16 @@
         return RxJavaPlugins.onAssembly(new StreamableHide<>(this));
     }
 
+    /**
+     * Ignores all elements from the current {@link Streamable} and completes.
+     * @return the new {@code Streamable} instance
+     */
+    @CheckReturnValue
+    @NonNull
+    default Streamable<T> ignoreElements() {
+        return RxJavaPlugins.onAssembly(new StreamableIgnoreElements<>(this));
+    }
+
     /**
      * Intercepts the lifecycle method calls of {@code Streamable} and {@link Streamer}
      * and allows the modification of them via Function callbacks.
@@ -844,7 +868,7 @@
     /**
      * 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

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

@@ -64,4 +64,41 @@ public static <T> T blockingFirst(Streamable<T> source) {
         return result;
     }
 
+    /**
+     * Consumes all upstream items and returns the very last or throws
+     * a {@link NoSuchElementException}.
+     * @param <T> the element type
+     * @param source the source sequence
+     * @return the very last value
+     * @throws RuntimeException if the source signals an unchecked exception
+     * @throws CompletionException if the source signals a checked exception
+     */
+    public static <T> T blockingLast(Streamable<T> source) {
+        var streamer = source.stream(new CompositeDisposable());
+        Throwable nextException = null;
+        Throwable finishException = null;
+        T result = null;
+        try {
+            while (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;
+    }
 }

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

@@ -0,0 +1,112 @@
+/*
+ * 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.io.Serial;
+import java.util.NoSuchElementException;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiConsumer;
+
+import io.reactivex.rxjava4.annotations.NonNull;
+import io.reactivex.rxjava4.core.*;
+import io.reactivex.rxjava4.disposables.StreamerCancellation;
+import io.reactivex.rxjava4.internal.util.ExceptionHelper;
+
+public record StreamableIgnoreElements<T>(Streamable<T> source)
+implements Streamable<T> {
+
+    @Override
+    public @NonNull Streamer<@NonNull T> stream(@NonNull StreamerCancellation cancellation) {
+        return new IgnoreElementsStreamer<>(source.stream(cancellation));
+    }
+
+    static final class IgnoreElementsStreamer<T> extends AtomicInteger
+    implements Streamer<T>, BiConsumer<Object, Throwable> {
+
+        @Serial
+        private static final long serialVersionUID = 2265801211815192189L;
+
+        final Streamer<T> upstream;
+
+        final CompletableFuture<Boolean> waiter;
+
+        int stage;
+
+        Throwable mainError;
+
+        volatile boolean done;
+
+        IgnoreElementsStreamer(Streamer<T> upstream)  {
+            this.upstream = upstream;
+            this.waiter = new CompletableFuture<>();
+        }
+
+        @Override
+        public @NonNull CompletionStage<Boolean> next() {
+            if (stage == 0) {
+                stage = 1;
+                drain();
+                return waiter;
+            }
+            return NEXT_FALSE;
+        }
+
+        void drain() {
+            if (getAndIncrement() != 0) {
+                return;
+            }
+
+            do {
+                if (done) {
+                    upstream.finish().whenComplete(this);
+                    break;
+                } else {
+                    upstream.next().whenComplete(this);
+                }
+            } while (decrementAndGet() != 0);
+        }
+
+        @Override
+        public void accept(Object t, Throwable u) {
+            if (done) {
+                if (mainError != null || u != null) {
+                    waiter.completeExceptionally(ExceptionHelper.unwrapAndCombine(mainError, u));
+                } else {
+                    waiter.complete(false);
+                }
+            } else {
+                if (u != null) {
+                    mainError = u;
+                    done = true;
+                } else
+                if (!(Boolean)t) {
+                    done = true;
+                }
+                drain();
+            }
+        }
+
+        @Override
+        public @NonNull T current() {
+            throw new NoSuchElementException();
+        }
+
+        @Override
+        public @NonNull CompletionStage<Void> finish() {
+            return FINISHED;
+        }
+
+    }
+}

src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingTest.java → src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingFirstTest.java

@@ -24,7 +24,7 @@
 import io.reactivex.rxjava4.core.Streamable;
 import io.reactivex.rxjava4.exceptions.TestException;
 
-public class StreamableBlockingTest extends StreamableBaseTest {
+public class StreamableBlockingFirstTest extends StreamableBaseTest {
 
     @Test
     public void blockingFirstNormal() throws Throwable {

src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingLastTest.java

@@ -0,0 +1,78 @@
+/*
+ * 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 static org.junit.jupiter.api.Assertions.*;
+
+import java.io.IOException;
+import java.util.NoSuchElementException;
+import java.util.concurrent.CompletionException;
+
+import org.junit.jupiter.api.Test;
+
+import io.reactivex.rxjava4.core.Streamable;
+import io.reactivex.rxjava4.exceptions.TestException;
+
+public class StreamableBlockingLastTest extends StreamableBaseTest {
+
+    @Test
+    public void normal() throws Throwable {
+        assertEquals(1, Streamable.just(1).blockingLast());
+    }
+
+    @Test
+    public void many() throws Throwable {
+        assertEquals(5, Streamable.range(1, 5).blockingLast());
+    }
+
+    @Test
+    public void empty() throws Throwable {
+        assertThrows(NoSuchElementException.class, () -> {
+            Streamable.empty().blockingLast();
+        });
+    }
+
+    @Test
+    public void errorUnchecked() throws Throwable {
+        assertThrows(TestException.class, () -> {
+            Streamable.error(new TestException()).blockingLast();
+        });
+    }
+
+    @Test
+    public void errorChecked() throws Throwable {
+        var ex = assertThrows(CompletionException.class, () -> {
+            Streamable.error(new IOException()).blockingLast();
+        });
+
+        assertTrue(ex.getCause() instanceof IOException, "Wrong exception? " + ex.getCause());
+    }
+
+    @Test
+    public void finishCrash() throws Throwable {
+        assertThrows(TestException.class, () -> {
+            StreamableFailingFinish.MAIN_COMPLETES.blockingLast();
+        });
+    }
+
+    @Test
+    public void nextAndFinishCrash() throws Throwable {
+        var ex = assertThrows(TestException.class, () -> {
+            StreamableFailingFinish.MAIN_FAILS.blockingLast();
+        });
+
+        assertTrue(ex.getSuppressed()[0] instanceof TestException, "Wrong exception? " + ex.getSuppressed()[0]);
+    }
+
+}