diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamable.java b/src/main/java/io/reactivex/rxjava4/core/Streamable.java index 89955ae678..acb4a85b0a 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamable.java @@ -238,6 +238,21 @@ public interface Streamable<@NonNull T> { return RxJavaPlugins.onAssembly(new StreamableError<>(throwable)); } + /** + * Creates a {@code Streamable} that invokes the given {@link Supplier} to produce a {@link Throwable} + * that will be immediately signaled when the streaming begins. + * @param the element type of the sequence + * @param throwableSupplier the {@code Supplier} to produce a {@code Throwable} for each of the streamers. + * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code throwableSupplier} is {@code null} + */ + @CheckReturnValue + @NonNull + static <@NonNull T> Streamable error(Supplier throwableSupplier) { + Objects.requireNonNull(throwableSupplier, "throwableSupplier is null"); + return defer(() -> error(throwableSupplier.get())); + } + /** * Returns an empty {@code Streamable} that never produces an item and just completes. * @param the element type @@ -629,8 +644,25 @@ static Streamable> zip(Iterable> s @CheckReturnValue @NonNull default T blockingFirst() { - return StreamableBlocking.blockingFirst(this); + return StreamableBlocking.blockingFirst(this, null, new CompositeDisposable()); } + + /** + * Blocks the current thread until this {@code Streamable} produces one item, which is then returned immediately, + * or the {@code defaultItem} is returned if the upstream is empty. + * @param defaultItem if the upstream is empty + * @return the first item of this {@code Streamable} + * @throws CancellationException if this {@code Streamable} failed with a checked exception + * @throws RuntimeException if this {@code Streamable} failed with an unchecked exception + * @throws NullPointerException if {@code defaultItem} is {@code null} + */ + @CheckReturnValue + @NonNull + default T blockingFirst(@NonNull T defaultItem) { + Objects.requireNonNull(defaultItem, "defaultItem is null"); + return StreamableBlocking.blockingFirst(this, defaultItem, new CompositeDisposable()); + } + /** * Blocks the current thread until this {@code Streamable} produces one item, which is then returned. * @param cancellation the external cancellation resource to pass into the chain @@ -642,9 +674,27 @@ default T blockingFirst() { */ @CheckReturnValue @NonNull - default T blockingFirst(StreamerCancellation cancellation) { + default T blockingFirst(@NonNull StreamerCancellation cancellation) { + Objects.requireNonNull(cancellation, "cancellation is null"); + return StreamableBlocking.blockingFirst(this, null, cancellation); + } + + /** + * Blocks the current thread until this {@code Streamable} produces one item, which is then returned immediately, + * or the {@code defaultItem} is returned if the upstream is empty. + * @param defaultItem if the upstream is empty + * @param cancellation the external cancellation resource to pass into the chain + * @return the first item of this {@code Streamable} + * @throws CancellationException if this {@code Streamable} failed with a checked exception + * @throws RuntimeException if this {@code Streamable} failed with an unchecked exception + * @throws NullPointerException if {@code defaultItem} or {@code cancellation} is {@code null} + */ + @CheckReturnValue + @NonNull + default T blockingFirst(@NonNull T defaultItem, @NonNull StreamerCancellation cancellation) { + Objects.requireNonNull(defaultItem, "defaultItem is null"); Objects.requireNonNull(cancellation, "cancellation is null"); - return StreamableBlocking.blockingFirst(this, cancellation); + return StreamableBlocking.blockingFirst(this, defaultItem, cancellation); } /** @@ -658,7 +708,25 @@ default T blockingFirst(StreamerCancellation cancellation) { @CheckReturnValue @NonNull default T blockingLast() { - return StreamableBlocking.blockingLast(this); + return StreamableBlocking.blockingLast(this, null, new CompositeDisposable()); + } + + /** + * Blocks the current thread until this {@code Streamable} produces all of its items + * and the very last is then returned, or returns the {@code defaultItem} if the + * upstream is empty. + * @param defaultItem if the upstream is empty + * @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 + * @throws NullPointerException if {@code defaultItem} is {@code null} + */ + @CheckReturnValue + @NonNull + default T blockingLast(@NonNull T defaultItem) { + Objects.requireNonNull(defaultItem, "defaultItem is null"); + return StreamableBlocking.blockingLast(this, defaultItem, new CompositeDisposable()); } /** @@ -673,9 +741,29 @@ default T blockingLast() { */ @CheckReturnValue @NonNull - default T blockingLast(StreamerCancellation cancellation) { + default T blockingLast(@NonNull StreamerCancellation cancellation) { Objects.requireNonNull(cancellation, "cancellation is null"); - return StreamableBlocking.blockingLast(this, cancellation); + return StreamableBlocking.blockingLast(this, null, cancellation); + } + + /** + * Blocks the current thread until this {@code Streamable} produces all of its items + * and the very last is then returned, or returns the {@code defaultItem} if the + * upstream is empty. + * @param defaultItem if the upstream is empty + * @param cancellation the external cancellation resource to pass into the chain + * @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 + * @throws NullPointerException if {@code defaultItem} or {@code cancellation} is {@code null} + */ + @CheckReturnValue + @NonNull + default T blockingLast(@NonNull T defaultItem, @NonNull StreamerCancellation cancellation) { + Objects.requireNonNull(defaultItem, "defaultItem is null"); + Objects.requireNonNull(cancellation, "cancellation is null"); + return StreamableBlocking.blockingLast(this, defaultItem, cancellation); } /** @@ -696,6 +784,24 @@ default Streamable collect(@NonNull Collector collector) { return RxJavaPlugins.onAssembly(new StreamableCollector<>(this, collector)); } + /** + * Collects all upstream values into a data structure, provided for each streamer, via + * a custom collector callback and emits this single item as the result of the returned + * {@code Streamable}. + * @param the element type of the data structure and returned value + * @param initialItemSupplier provides an individual item for each streamer to collect upstream items into + * @param collector the callback that receives the individual data structure item and the current upstream value + * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code initialItemSupplier} or {@code collector} is {@code null} + */ + @CheckReturnValue + @NonNull + default Streamable collect(Supplier initialItemSupplier, BiConsumer collector) { + Objects.requireNonNull(initialItemSupplier, "initialItemSupplier is null"); + Objects.requireNonNull(collector, "collector is null"); + return RxJavaPlugins.onAssembly(new StreamableCollect<>(this, initialItemSupplier, collector)); + } + /** * Delays the delivery of each upstream item by the given time amount. * @param time the delay time @@ -1191,6 +1297,36 @@ default Observable toObservable() { .await(), executor); } + /** + * Replaces the {@link StreamerCancellation} instance of the downstream travelling through + * this operator with the provided instance to allow external cancellation. + * @param cancellation the new {@code StreamerCancellation} to use + * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code cancellation} is {@code null} + */ + @CheckReturnValue + @NonNull + default Streamable withCancellation(StreamerCancellation cancellation) { + Objects.requireNonNull(cancellation, "cancellation is null"); + return withCancellation(_ -> cancellation); + } + + /** + * Replaces the {@link StreamerCancellation} instance of the downstream travelling through + * this operator with the instance returned by the {@code onStream} callback to allow external, + * possibly dependent cancellation. + * @param onStream the function called with the downstream {@code StreamerCancellation} and expected + * to return a {@code StreamerCancellation} to pass it to the upstream source + * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code onStream} is {@code null} + */ + @CheckReturnValue + @NonNull + default Streamable withCancellation(Function onStream) { + Objects.requireNonNull(onStream, "onStream is null"); + return RxJavaPlugins.onAssembly(new StreamableWithCancellation<>(this, onStream)); + } + // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo // Consumption methods and outgoing converters // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo 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 cf2e1583fe..e15ea2c52f 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 @@ -26,24 +26,10 @@ public record StreamableBlocking() { /** * Consumes the first item and finishes the {@link Streamable}, - * throwing {@link NoSuchElementException} if the source is empty. - * @param 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 blockingFirst(Streamable source) { - return blockingFirst(source, new CompositeDisposable()); - } - - /** - * Consumes the first item and finishes the {@link Streamable}, - * throwing {@link NoSuchElementException} if the source is empty. + * throwing {@link NoSuchElementException} if the source is empty and {@code defaultItem} is {@code null}. * @param the element type * @param source the source {@code Streamable} + * @param defaultItem if not null and the source is empty, this value will be returned * @param cancellation the external cancellation manager * @return the first item * @throws RuntimeException if the source signals an unchecked exception @@ -51,7 +37,8 @@ public static T blockingFirst(Streamable source) { */ @CheckReturnValue @NonNull - public static T blockingFirst(Streamable source, StreamerCancellation cancellation) { + public static T blockingFirst(@NonNull Streamable source, + @Nullable T defaultItem, @NonNull StreamerCancellation cancellation) { var streamer = source.stream(cancellation); Throwable nextException = null; Throwable finishException = null; @@ -75,6 +62,9 @@ public static T blockingFirst(Streamable source, StreamerCancellation can throw ExceptionHelper.wrapOrThrow(ExceptionHelper.unwrapAndCombine(nextException, finishException)); } if (result == null) { + if (defaultItem != null) { + return defaultItem; + } throw new NoSuchElementException(); } return result; @@ -82,28 +72,16 @@ public static T blockingFirst(Streamable source, StreamerCancellation can /** * 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) { - return blockingLast(source, new CompositeDisposable()); - } - - /** - * Consumes all upstream items and returns the very last or throws - * a {@link NoSuchElementException}. + * a {@link NoSuchElementException} if {@code defaultItem} is also {@code null}. * @param the element type * @param source the source sequence + * @param defaultItem if not null, item to return if the source is empty * @param cancellation the external cancellation manager * @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, StreamerCancellation cancellation) { + public static T blockingLast(Streamable source, T defaultItem, StreamerCancellation cancellation) { var streamer = source.stream(cancellation); Throwable nextException = null; Throwable finishException = null; @@ -127,6 +105,9 @@ public static T blockingLast(Streamable source, StreamerCancellation canc throw ExceptionHelper.wrapOrThrow(ExceptionHelper.unwrapAndCombine(nextException, finishException)); } if (result == null) { + if (defaultItem != null) { + return defaultItem; + } throw new NoSuchElementException(); } return result; diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollect.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollect.java new file mode 100644 index 0000000000..3b92a1c895 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollect.java @@ -0,0 +1,247 @@ +/* + * 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.Objects; +import java.util.concurrent.*; +import java.util.concurrent.Future.State; +import java.util.concurrent.atomic.AtomicInteger; + +import io.reactivex.rxjava4.annotations.NonNull; +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.StreamerCancellation; +import io.reactivex.rxjava4.exceptions.Exceptions; +import io.reactivex.rxjava4.functions.*; +import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; +import io.reactivex.rxjava4.operators.*; + +public record StreamableCollect( + Streamable source, + Supplier initialItemSupplier, + BiConsumer collector +) implements Streamable, HasUpstreamStreamableSource { + + @Override + public @NonNull Streamer<@NonNull U> stream(@NonNull StreamerCancellation cancellation) { + U storage; + try { + storage = Objects.requireNonNull(initialItemSupplier.get()); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + return StreamableError.createFailed(ex); + } + return new CollectStreamable<>( + source.stream(cancellation), + storage, + collector, + cancellation); + } + + static final class CollectStreamable + extends AtomicInteger + implements Streamer, java.util.function.BiConsumer { + + @Serial + private static final long serialVersionUID = -2893736592415047972L; + + final Streamer upstream; + + final U storage; + + final BiConsumer collector; + + final CompletableFuture nextReady; + + final CompletableFuture finishReady; + + final StreamerCancellation cancellation; + + U current; + + boolean once; + + boolean deferredOnce; + + volatile boolean done; + + CollectStreamable(Streamer upstream, U storage, + BiConsumer collector, + StreamerCancellation cancellation) { + this.upstream = upstream; + this.storage = storage; + this.collector = collector; + this.nextReady = new CompletableFuture<>(); + this.finishReady = new CompletableFuture<>(); + this.cancellation = cancellation; + } + + @SuppressWarnings("unchecked") + @Override + public @NonNull CompletionStage next() { + if (!once) { + once = true; + + if (upstream instanceof IndexableSource isrc) { + long max = isrc.limit(); + for (long index = 0; index < max; index++) { + if (cancellation.isDisposed()) { + return CompletableFuture.failedFuture(new CancellationException()); + } + + try { + var value = (T)isrc.elementAt(index); + collector.accept(storage, value); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + return CompletableFuture.failedFuture(ex); + } + } + current = storage; + return NEXT_TRUE; + } else + if (upstream instanceof DeferredEnumerableSource dsrc) { + StreamableHelper.whenComplete(dsrc.enumerableReady(), this::deferredEnumerate); + return nextReady; + } else + if (upstream instanceof EnumerableSource esrc) { + try { + while (esrc.nextSync()) { + if (cancellation.isDisposed()) { + return CompletableFuture.failedFuture(new CancellationException()); + } + collector.accept(storage, (T)esrc.current()); + } + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + return CompletableFuture.failedFuture(ex); + } + current = storage; + return NEXT_TRUE; + } + drain(); + return nextReady; + } + return NEXT_FALSE; + } + + void deferredEnumerate(Boolean hasInitialValue, Throwable error) { + if (error != null) { + nextReady.completeExceptionally(error); + return; + } + if (hasInitialValue) { + @SuppressWarnings("unchecked") + var upstreamCast = (DeferredEnumerableSource)upstream; + + try { + while (upstreamCast.nextSync()) { + if (cancellation.isDisposed()) { + nextReady.completeExceptionally(new CancellationException()); + return; + } + collector.accept(storage, upstreamCast.current()); + } + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + nextReady.completeExceptionally(ex); + return; + } + } + current = storage; + nextReady.complete(true); + } + + @Override + public @NonNull U current() { + return current; + } + + @Override + public @NonNull CompletionStage finish() { + done = true; + drain(); + return finishReady; + } + + void drain() { + for (;;) { + if (done) { + StreamableHelper.whenComplete(upstream.finish(), this); + return; + } else { + var upstreamNext = upstream.next().toCompletableFuture(); + var state = upstreamNext.state(); + if (state == State.RUNNING) { + set(1); + upstreamNext.whenComplete(this); + if (compareAndSet(1, 0)) { + return; + } + state = upstreamNext.state(); + } + + if (state == State.SUCCESS) { + if (upstreamNext.getNow((false))) { + try { + collector.accept(storage, upstream.current()); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + nextReady.completeExceptionally(ex); + return; + } + } else { + current = storage; + nextReady.complete(true); + return; + } + } else { + nextReady.completeExceptionally(upstreamNext.exceptionNow()); + return; + } + } + } + } + + @Override + public void accept(Object t, Throwable u) { + if (done) { + if (u != null) { + finishReady.completeExceptionally(u); + } else { + finishReady.complete(null); + } + } else { + if (!compareAndSet(1, 2)) { + if (u != null) { + nextReady.completeExceptionally(u); + } else + if ((Boolean)t) { + try { + collector.accept(storage, upstream.current()); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + nextReady.completeExceptionally(ex); + return; + } + drain(); + } else { + current = storage; + nextReady.complete(true); + } + } + } + } + } +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollector.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollector.java index a1b3c9a7e0..4f6a617303 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollector.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollector.java @@ -133,7 +133,7 @@ static final class CollectorStreamable return NEXT_FALSE; } - void deferredEnumerate(boolean hasInitialValue, Throwable error) { + void deferredEnumerate(Boolean hasInitialValue, Throwable error) { if (error != null) { nextReady.completeExceptionally(error); return; diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableWithCancellation.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableWithCancellation.java new file mode 100644 index 0000000000..5413f89786 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableWithCancellation.java @@ -0,0 +1,46 @@ +/* + * 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.Objects; + +import io.reactivex.rxjava4.annotations.NonNull; +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.StreamerCancellation; +import io.reactivex.rxjava4.exceptions.Exceptions; +import io.reactivex.rxjava4.functions.Function; +import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; + +/// Allows replacing the [StreamerCancellation] the [#stream(StreamerCancellation)] method receives +/// via a callback function. +/// @param the element type of the `Streamable` +/// @param source the upstream source `Streamable` +/// @param onStream the function called with the downstream `StreamerCancellation` and expected +/// to return a `StreamerCancellation` to pass it to the upstream `source` +public record StreamableWithCancellation( + Streamable source, + Function onStream) +implements Streamable, HasUpstreamStreamableSource { + + @Override + public @NonNull Streamer<@NonNull T> stream(@NonNull StreamerCancellation cancellation) { + try { + var altCancellation = Objects.requireNonNull(onStream.apply(cancellation), "The onStream returned a null StreamerCancellation"); + return source.stream(altCancellation); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + return StreamableError.createFailed(ex); + } + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/maybe/MaybeAmbTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/maybe/MaybeAmbTest.java index 1cd69f1cca..456a208aa2 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/maybe/MaybeAmbTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/maybe/MaybeAmbTest.java @@ -29,7 +29,6 @@ import io.reactivex.rxjava4.observers.TestObserver; import io.reactivex.rxjava4.processors.PublishProcessor; import io.reactivex.rxjava4.schedulers.Schedulers; -import io.reactivex.rxjava4.subjects.*; import io.reactivex.rxjava4.testsupport.TestHelper; public class MaybeAmbTest extends RxJavaTest { diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingFirstTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingFirstTest.java index 004c1b46ee..c3bb4dbab5 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingFirstTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingFirstTest.java @@ -22,6 +22,7 @@ import org.junit.jupiter.api.Test; import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.disposables.CompositeDisposable; import io.reactivex.rxjava4.exceptions.TestException; public class StreamableBlockingFirstTest extends StreamableBaseTest { @@ -75,4 +76,13 @@ public void blockingFirstNextAndFinishCrash() throws Throwable { assertTrue(ex.getSuppressed()[0] instanceof TestException, "Wrong exception? " + ex.getSuppressed()[0]); } + @Test + public void blockingFirstDefaultEmpty() throws Throwable { + assertEquals(1, Streamable.empty().blockingFirst(1)); + } + + @Test + public void blockingFirstDefaultCancelEmpty() throws Throwable { + assertEquals(1, Streamable.empty().blockingFirst(1, new CompositeDisposable())); + } } 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 index 63349a08b2..14159d3e8b 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingLastTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingLastTest.java @@ -22,7 +22,7 @@ import org.junit.jupiter.api.Test; import io.reactivex.rxjava4.core.Streamable; -import io.reactivex.rxjava4.disposables.DisposableStreamerCancellation; +import io.reactivex.rxjava4.disposables.*; import io.reactivex.rxjava4.exceptions.TestException; public class StreamableBlockingLastTest extends StreamableBaseTest { @@ -81,4 +81,14 @@ public void normalCancellation() throws Throwable { assertEquals(1, Streamable.just(1).blockingLast(DisposableStreamerCancellation.never())); } + @Test + public void blockingLastDefaultEmpty() throws Throwable { + assertEquals(1, Streamable.empty().blockingLast(1)); + } + + @Test + public void blockingLastDefaultCancelEmpty() throws Throwable { + assertEquals(1, Streamable.empty().blockingLast(1, new CompositeDisposable())); + } + } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollectTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollectTest.java new file mode 100644 index 0000000000..46be424874 --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollectTest.java @@ -0,0 +1,303 @@ +/* + * 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.assertThrows; + +import java.io.IOException; +import java.util.*; +import java.util.concurrent.*; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.CompositeDisposable; +import io.reactivex.rxjava4.exceptions.TestException; +import io.reactivex.rxjava4.processors.DispatchStreamProcessor; +import io.reactivex.rxjava4.schedulers.Schedulers; + +public class StreamableCollectTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + Streamable.range(1, 5) + .hide() + .collect(ArrayList::new, ArrayList::add) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(new ArrayList<>(List.of(1, 2, 3, 4, 5))); + } + + @Test + public void normalOptimized() throws Throwable { + Streamable.range(1, 5) + .collect(ArrayList::new, ArrayList::add) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(new ArrayList<>(List.of(1, 2, 3, 4, 5))); + } + + @Test + public void empty() throws Throwable { + Streamable.empty() + .collect(ArrayList::new, ArrayList::add) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(new ArrayList<>(List.of())); + } + + @Test + public void crash() throws Throwable { + Streamable.error(new TestException()) + .collect(ArrayList::new, ArrayList::add) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void finishCrash() throws Throwable { + StreamableFailingFinish.MAIN_COMPLETES + .collect(ArrayList::new, ArrayList::add) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class, new ArrayList<>(List.of())); + } + + @Test + public void indexerDisposed() throws Throwable { + assertThrows(CancellationException.class, () -> { + var cd = new CompositeDisposable(); + cd.dispose(); + + Streamable.range(1, 5) + .collect(ArrayList::new, ArrayList::add) + .stream(cd) + .awaitNext(); + }); + } + + @Test + public void intelvalRange() throws Throwable { + Streamable.intervalRange(1, 5, 1, 1, TimeUnit.MILLISECONDS, Schedulers.single()) + .hide() + .collect(ArrayList::new, ArrayList::add) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(new ArrayList<>(List.of(1L, 2L, 3L, 4L, 5L))); + } + + @Test + public void viaDispatch() throws Throwable { + var dsp = new DispatchStreamProcessor<>(); + + var ts = dsp + .collect(ArrayList::new, ArrayList::add) + .test(); + + ts.awaitOnSubscribe(1, TimeUnit.SECONDS); + + awaitStreamers(dsp, 1000); + + dsp.next(1).toCompletableFuture().join(); + dsp.next(2).toCompletableFuture().join(); + dsp.next(3).toCompletableFuture().join(); + dsp.finish(null).toCompletableFuture().join(); + + ts.awaitDone(5, TimeUnit.SECONDS) + .assertResult(new ArrayList<>(List.of(1, 2, 3))); + } + + @Test + public void viaDispatchError() throws Throwable { + var dsp = new DispatchStreamProcessor<>(); + + var ts = dsp + .collect(ArrayList::new, ArrayList::add) + .test(); + + ts.awaitOnSubscribe(1, TimeUnit.SECONDS); + + awaitStreamers(dsp, 1000); + + dsp.next(1).toCompletableFuture().join(); + dsp.next(2).toCompletableFuture().join(); + dsp.next(3).toCompletableFuture().join(); + dsp.finish(new TestException()).toCompletableFuture().join(); + + ts.awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void initialItemSupplierNull() { + Streamable.range(1, 5) + .collect(() -> null, (_, _) -> { }) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(NullPointerException.class); + } + + @Test + public void initialItemSupplierCrash() { + Streamable.range(1, 5) + .collect(() -> { throw new IOException(); }, (_, _) -> { }) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(IOException.class); + } + + @Test + public void indexedSourceCollectorCrash() { + Streamable.range(1, 5) + .collect(ArrayList::new, (_, _) -> { throw new TestException(); }) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void enumerableSourceCollectorNormal() { + Streamable.fromIterable(List.of(1, 2, 3, 4, 5)) + .collect(ArrayList::new, ArrayList::add) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(new ArrayList<>(List.of(1, 2, 3, 4, 5))); + } + + @Test + public void enumerableSourceCollectorCrash() { + Streamable.fromIterable(List.of(1, 2, 3, 4, 5)) + .collect(ArrayList::new, (_, _) -> { throw new TestException(); }) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void genericSourceCollectorCrash() { + Streamable.range(1, 5) + .hide() + .collect(ArrayList::new, (_, _) -> { throw new TestException(); }) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void deferredEnumerableSourceCollectorNormal() { + Single.just(1) + .flattenAsStreamable(_ -> List.of(1, 2, 3, 4, 5)) + .collect(ArrayList::new, ArrayList::add) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(new ArrayList<>(List.of(1, 2, 3, 4, 5))); + } + + @Test + public void deferredEnumerableSourceCollectorEmpty() { + Single.just(1) + .flattenAsStreamable(_ -> List.of()) + .collect(ArrayList::new, ArrayList::add) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(new ArrayList<>(List.of())); + } + + @Test + public void enumerableSourceCancel() { + var cd = new CompositeDisposable(); + Streamable.fromIterable(List.of(1, 2, 3, 4, 5)) + .map(v -> { + if (v == 3) { + cd.dispose(); + } + return v; + }) + .collect(ArrayList::new, ArrayList::add) + .withCancellation(cd) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(CancellationException.class) + ; + } + + @Test + public void deferredEnumerableSourceCrash() { + Single.error(new TestException()) + .flattenAsStreamable(_ -> List.of(1, 2, 3, 4, 5)) + .collect(ArrayList::new, ArrayList::add) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void deferredEnumerableSourceCrashDebug() throws Throwable { + withCachedExecutor(exec -> { + Single.error(new TestException()) + .flattenAsStreamable(_ -> List.of(1, 2, 3, 4, 5)) + .collect(ArrayList::new, ArrayList::add) + .test(exec) + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + }); + } + + @Test + public void deferredEnumerableCollectorCrash() { + Single.just(1) + .flattenAsStreamable(_ -> List.of(1, 2, 3, 4, 5)) + .collect(ArrayList::new, (_, _) -> { + throw new TestException(); + }) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void deferredEnumerableCollectorCancels() { + var cd = new CompositeDisposable(); + Single.just(1) + .flattenAsStreamable(_ -> List.of(1, 2, 3, 4, 5)) + .collect(ArrayList::new, (_, _) -> { + cd.dispose(); + }) + .withCancellation(cd) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(CancellationException.class); + } + + @Test + public void viaDispatchCrash() throws Throwable { + var dsp = new DispatchStreamProcessor<>(); + + var ts = dsp + .collect(ArrayList::new, (_, _) -> { + throw new TestException(); + }) + .test(); + + ts.awaitOnSubscribe(1, TimeUnit.SECONDS); + + awaitStreamers(dsp, 1000); + + dsp.next(1).toCompletableFuture().join(); + + ts.awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollectorTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollectorTest.java index 1d811efcca..9d22035ba5 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollectorTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollectorTest.java @@ -21,7 +21,7 @@ import org.junit.jupiter.api.Test; -import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.CompositeDisposable; import io.reactivex.rxjava4.exceptions.TestException; import io.reactivex.rxjava4.processors.DispatchStreamProcessor; @@ -139,4 +139,16 @@ public void viaDispatchError() throws Throwable { ts.awaitDone(5, TimeUnit.SECONDS) .assertFailure(TestException.class); } + + @Test + public void deferredEnumerableSourceCrashDebug() throws Throwable { + withCachedExecutor(exec -> { + Single.error(new TestException()) + .flattenAsStreamable(_ -> List.of(1, 2, 3, 4, 5)) + .collect(Collectors.toList()) + .test(exec) + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + }); + } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableErrorTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableErrorTest.java index 0117157819..75ba8fd582 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableErrorTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableErrorTest.java @@ -15,9 +15,12 @@ import static org.junit.jupiter.api.Assertions.assertThrows; +import java.io.IOException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; + import io.reactivex.rxjava4.core.Streamable; import io.reactivex.rxjava4.exceptions.TestException; import io.reactivex.rxjava4.internal.operators.streamable.StreamableError.ErrorStreamer; @@ -48,4 +51,33 @@ public void currentThrows() { new ErrorStreamer<>(new TestException()).current(); }); } + + @Test + public void supplierNormal() { + var index = new AtomicInteger(); + var source = Streamable.error(() -> new TestException(index.getAndIncrement() + "")); + for (int i = 0; i < 10; i++) { + var j = i; + source.test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class) + .assertError(t -> t.getMessage().equals("" + j)); + } + } + + @Test + public void supplierNull() { + Streamable.error(() -> null) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(NullPointerException.class); + } + + @Test + public void supplierCrash() { + Streamable.error(() -> { throw new IOException(); }) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(IOException.class); + } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableWithCancellationTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableWithCancellationTest.java new file mode 100644 index 0000000000..4cdb66aa88 --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableWithCancellationTest.java @@ -0,0 +1,59 @@ +/* + * 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.IOException; +import java.util.concurrent.*; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.disposables.CompositeDisposable; +import io.reactivex.rxjava4.schedulers.Schedulers; + +public class StreamableWithCancellationTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + var cd = new CompositeDisposable(); + + var ts = Streamable.timer(1, TimeUnit.MINUTES, Schedulers.computation()) + .withCancellation(cd) + .test() + .awaitOnSubscribe(5, TimeUnit.SECONDS); + + cd.dispose(); + + ts.awaitDone(5, TimeUnit.SECONDS) + .assertFailure(CancellationException.class); + } + + @Test + public void crash() throws Throwable { + Streamable.never() + .withCancellation(_ -> { throw new IOException(); }) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(IOException.class); + } + + @Test + public void nullReturn() throws Throwable { + Streamable.never() + .withCancellation(_ -> null) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(NullPointerException.class); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/schedulers/SchedulerToExecutorServiceTest.java b/src/test/java/io/reactivex/rxjava4/internal/schedulers/SchedulerToExecutorServiceTest.java index f42d89e452..465884c839 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/schedulers/SchedulerToExecutorServiceTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/schedulers/SchedulerToExecutorServiceTest.java @@ -221,6 +221,7 @@ public void invokeAllTimeoutDoesTimeout() throws Throwable { public void awaitTerminationUsesRequestedTimeUnit() throws Exception { Scheduler scheduler = Schedulers.computation(); Scheduler.Worker worker = scheduler.createWorker(); + @SuppressWarnings("resource") SchedulerToExecutorService executor = new SchedulerToExecutorService( scheduler, new AtomicReference<>(worker)); @@ -235,6 +236,7 @@ public void awaitTerminationUsesRequestedTimeUnit() throws Exception { @Test public void awaitTerminationReturnsTrueIfAlreadyTerminated() throws Exception { + @SuppressWarnings("resource") SchedulerToExecutorService executor = new SchedulerToExecutorService( Schedulers.computation(), new AtomicReference<>(null));