diff --git a/src/main/java/io/reactivex/rxjava4/core/StreamSink.java b/src/main/java/io/reactivex/rxjava4/core/StreamSink.java index 535243b04f..8d162f48c7 100644 --- a/src/main/java/io/reactivex/rxjava4/core/StreamSink.java +++ b/src/main/java/io/reactivex/rxjava4/core/StreamSink.java @@ -52,6 +52,27 @@ public interface StreamSink<@NonNull T> { @NonNull CompletionStage 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. diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamable.java b/src/main/java/io/reactivex/rxjava4/core/Streamable.java index d93ffcfacd..390523a502 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamable.java @@ -612,7 +612,7 @@ static Streamable> zip(Iterable> s // 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 @@ default T blockingFirst() { 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 @@ default Streamable hide() { 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 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. diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlocking.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlocking.java index 52bd4d9fd4..9b75fd840e 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlocking.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlocking.java @@ -64,4 +64,41 @@ public static T blockingFirst(Streamable source) { return result; } + /** + * Consumes all upstream items and returns the very last or throws + * a {@link NoSuchElementException}. + * @param 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 blockingLast(Streamable 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; + } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableIgnoreElements.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableIgnoreElements.java new file mode 100644 index 0000000000..502701e901 --- /dev/null +++ b/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(Streamable source) +implements Streamable { + + @Override + public @NonNull Streamer<@NonNull T> stream(@NonNull StreamerCancellation cancellation) { + return new IgnoreElementsStreamer<>(source.stream(cancellation)); + } + + static final class IgnoreElementsStreamer extends AtomicInteger + implements Streamer, BiConsumer { + + @Serial + private static final long serialVersionUID = 2265801211815192189L; + + final Streamer upstream; + + final CompletableFuture waiter; + + int stage; + + Throwable mainError; + + volatile boolean done; + + IgnoreElementsStreamer(Streamer upstream) { + this.upstream = upstream; + this.waiter = new CompletableFuture<>(); + } + + @Override + public @NonNull CompletionStage 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 finish() { + return FINISHED; + } + + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingFirstTest.java similarity index 97% rename from src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingTest.java rename to src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingFirstTest.java index bc39a89e25..004c1b46ee 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingTest.java +++ b/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 { diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingLastTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingLastTest.java new file mode 100644 index 0000000000..00980ace8d --- /dev/null +++ b/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]); + } + +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableIgnoreElementsTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableIgnoreElementsTest.java new file mode 100644 index 0000000000..d39a1a9d82 --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableIgnoreElementsTest.java @@ -0,0 +1,126 @@ +/* + * 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.util.NoSuchElementException; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.disposables.CompositeDisposable; +import io.reactivex.rxjava4.exceptions.TestException; +import io.reactivex.rxjava4.processors.DispatchStreamProcessor; + +public class StreamableIgnoreElementsTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + Streamable.range(1, 5) + .ignoreElements() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void empty() throws Throwable { + Streamable.empty() + .ignoreElements() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void just() throws Throwable { + Streamable.just(1) + .ignoreElements() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void error() throws Throwable { + Streamable.error(new TestException()) + .ignoreElements() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void finishCrash() throws Throwable { + StreamableFailingFinish.MAIN_COMPLETES + .ignoreElements() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void sourceAndFinishCrash() throws Throwable { + StreamableFailingFinish.MAIN_FAILS + .ignoreElements() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class) + .assertError(e -> e.getSuppressed()[0] instanceof TestException) + ; + } + + @Test + public void nextTwice() { + var streamer = Streamable.empty() + .ignoreElements() + .stream(new CompositeDisposable()); + + assertFalse(streamer.awaitNext(), "awaitNext-1"); + assertFalse(streamer.awaitNext(), "awaitNext-2"); + + streamer.awaitFinish(); + } + + @Test + public void dispatcher() throws Throwable { + var dsp = new DispatchStreamProcessor<>(); + + var ts = dsp.ignoreElements().test(); + + ts.awaitOnSubscribe(1, TimeUnit.SECONDS); + + awaitStreamers(dsp, 1000); + + dsp.awaitNext(1); + + dsp.awaitNext(2); + + dsp.awaitFinish(null); + + ts.awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void currentThrows() { + assertThrows(NoSuchElementException.class, () -> { + var dsp = new DispatchStreamProcessor<>(); + var streamer = dsp.ignoreElements().stream(new CompositeDisposable()); + streamer.current(); + }); + } +}