From 99e084982f40824b9a0044e31baf7ac53aa83a45 Mon Sep 17 00:00:00 2001 From: akarnokd Date: Thu, 9 Jul 2026 16:12:14 +0200 Subject: [PATCH] 4.x: Streamable +blockingFirst,last[OrError],skip ,S.flattenAsStreamable --- .../reactivex/rxjava4/core/Completable.java | 3 + .../java/io/reactivex/rxjava4/core/Maybe.java | 5 +- .../io/reactivex/rxjava4/core/Single.java | 38 ++++- .../io/reactivex/rxjava4/core/Streamable.java | 55 +++++++ .../internal/disposables/DisposableOnly.java | 30 ++++ .../operators/streamable/StageResumable.java | 18 +++ .../streamable/StreamableBlocking.java | 67 ++++++++ .../streamable/StreamableLastAsSingle.java | 136 ++++++++++++++++ .../streamable/StreamableSingleFlattenAs.java | 146 ++++++++++++++++++ .../operators/streamable/StreamableSkip.java | 93 +++++++++++ .../rxjava4/disposables/DisposableTest.java | 17 +- .../streamable/StreamableBaseTest.java | 24 ++- .../streamable/StreamableBlockingTest.java | 78 ++++++++++ .../streamable/StreamableDeferTest.java | 9 ++ .../streamable/StreamableFailingFinish.java | 4 +- .../StreamableLastAsSingleTest.java | 109 +++++++++++++ .../StreamableSingleFlattenAsTest.java | 87 +++++++++++ .../streamable/StreamableSkipTest.java | 88 +++++++++++ .../CheckBaseTypeAnnotationsTest.java | 3 +- .../validators/CheckParamValidationTest.java | 2 + 20 files changed, 1004 insertions(+), 8 deletions(-) create mode 100644 src/main/java/io/reactivex/rxjava4/internal/disposables/DisposableOnly.java create mode 100644 src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlocking.java create mode 100644 src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLastAsSingle.java create mode 100644 src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSingleFlattenAs.java create mode 100644 src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSkip.java create mode 100644 src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingTest.java create mode 100644 src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLastAsSingleTest.java create mode 100644 src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSingleFlattenAsTest.java create mode 100644 src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSkipTest.java diff --git a/src/main/java/io/reactivex/rxjava4/core/Completable.java b/src/main/java/io/reactivex/rxjava4/core/Completable.java index aa6c729bb3..4a5d205e1c 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Completable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Completable.java @@ -3228,6 +3228,8 @@ public final Future toFuture() { *

* *

+ *
Backpressure:
+ *
The returned {@code Streamable} honors the backpressure of the downstream consumer.
*
Scheduler:
*
{@code toStreamable} does not operate by default on a particular {@link Scheduler}.
*
@@ -3236,6 +3238,7 @@ public final Future toFuture() { * @since 4.0.0 */ @SchedulerSupport(SchedulerSupport.NONE) + @BackpressureSupport(BackpressureKind.FULL) @CheckReturnValue @NonNull public final <@NonNull T> Streamable toStreamable() { diff --git a/src/main/java/io/reactivex/rxjava4/core/Maybe.java b/src/main/java/io/reactivex/rxjava4/core/Maybe.java index 60c78eee3d..db11f98c9b 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Maybe.java +++ b/src/main/java/io/reactivex/rxjava4/core/Maybe.java @@ -3896,14 +3896,17 @@ public final Single toSingle() { *

* *

+ *
Backpressure:
+ *
The returned {@code Streamable} honors the backpressure of the downstream consumer.
*
Scheduler:
*
{@code toStreamable} does not operate by default on a particular {@link Scheduler}.
*
* @return the new {@code Streamable} instance * @since 4.0.0 */ - @CheckReturnValue @SchedulerSupport(SchedulerSupport.NONE) + @BackpressureSupport(BackpressureKind.FULL) + @CheckReturnValue @NonNull public final Streamable toStreamable() { return RxJavaPlugins.onAssembly(new StreamableFromMaybe<>(this)); diff --git a/src/main/java/io/reactivex/rxjava4/core/Single.java b/src/main/java/io/reactivex/rxjava4/core/Single.java index 851ff50e5d..b1ba92a65a 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Single.java +++ b/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 filter(@NonNull Predicate 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. + *

+ * + *

+ *
Backpressure:
+ *
The operator honors backpressure from downstream.
+ *
Scheduler:
+ *
{@code flattenAsStreamable} does not operate by default on a particular {@link Scheduler}.
+ *
+ * + * @param + * 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 ReactiveX operators documentation: FlatMap + * @since 4.0.0 + */ + @BackpressureSupport(BackpressureKind.FULL) + @CheckReturnValue + @NonNull + @SchedulerSupport(SchedulerSupport.NONE) + public final <@NonNull U> Streamable flattenAsStreamable(@NonNull Function> 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 toObservable() { *

* *

+ *
Backpressure:
+ *
The returned {@code Streamable} honors the backpressure of the downstream consumer.
*
Scheduler:
*
{@code toStreamable} does not operate by default on a particular {@link Scheduler}.
*
* @return the new {@code Streamable} instance * @since 4.0.0 */ - @CheckReturnValue @SchedulerSupport(SchedulerSupport.NONE) + @BackpressureSupport(BackpressureKind.FULL) + @CheckReturnValue @NonNull public final Streamable toStreamable() { return RxJavaPlugins.onAssembly(new StreamableFromSingle<>(this)); diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamable.java b/src/main/java/io/reactivex/rxjava4/core/Streamable.java index 706b0b94d8..d93ffcfacd 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamable.java @@ -611,6 +611,19 @@ static Streamable> zip(Iterable> 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 intercept(StreamableInterceptConfig 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 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 lastOrError() { + return RxJavaPlugins.onAssembly(new StreamableLastAsSingle<>(this, null)); + } + /** * This method requires advanced knowledge about building operators, please consider * other standard composition methods first; @@ -801,6 +841,21 @@ default Streamable onErrorResumeNext(@NonNull Function(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 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. diff --git a/src/main/java/io/reactivex/rxjava4/internal/disposables/DisposableOnly.java b/src/main/java/io/reactivex/rxjava4/internal/disposables/DisposableOnly.java new file mode 100644 index 0000000000..2f4e760ba8 --- /dev/null +++ b/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"); + } +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StageResumable.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StageResumable.java index eb78cf664b..31ecb43beb 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StageResumable.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StageResumable.java @@ -25,6 +25,7 @@ * one use/thread can signal {@link #ready()} to wake up another use/thread * on a {@link #await()} call. * @param the element type of the notification pass-around + * @since 4.0.0 */ public final class StageResumable extends AtomicReference> implements BiConsumer { @@ -32,6 +33,13 @@ public final class StageResumable extends AtomicReference ready() { @@ -49,6 +57,12 @@ public CompletableFuture 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 await() { @@ -66,6 +80,10 @@ public CompletableFuture 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); 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 new file mode 100644 index 0000000000..52bd4d9fd4 --- /dev/null +++ b/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 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) { + 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; + } + +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLastAsSingle.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLastAsSingle.java new file mode 100644 index 0000000000..be79e40f6d --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLastAsSingle.java @@ -0,0 +1,136 @@ +/* + * 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.atomic.AtomicInteger; +import java.util.function.BiConsumer; + +import io.reactivex.rxjava4.annotations.*; +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.internal.disposables.DisposableOnly; +import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; +import io.reactivex.rxjava4.internal.util.ExceptionHelper; + +public final class StreamableLastAsSingle +extends Single implements HasUpstreamStreamableSource { + + final Streamable source; + + final T defaultItem; + + public StreamableLastAsSingle(@Nullable Streamable source, @Nullable T defaultItem) { + this.source = source; + this.defaultItem = defaultItem; + } + + @Override + public @NonNull Streamable<@NonNull T> source() { + return source; + } + + @Override + protected void subscribeActual(@NonNull SingleObserver observer) { + var cancellation = new CompositeDisposable(); + var drainer = new LastStreamer<>(observer, defaultItem, cancellation); + observer.onSubscribe(drainer); + drainer.upstream = source.stream(cancellation); + drainer.drain(); + } + + static final class LastStreamer + extends AtomicInteger + implements BiConsumer, DisposableOnly { + + @Serial + private static final long serialVersionUID = 2423155860070143815L; + + final SingleObserver downstream; + + final DisposableStreamerCancellation cancellation; + + final T defaultItem; + + Streamer upstream; + + T current; + + Throwable nextFailure; + + volatile boolean done; + + LastStreamer( + SingleObserver downstream, + T defaultItem, + DisposableStreamerCancellation cancellation) { + this.downstream = downstream; + this.defaultItem = defaultItem; + this.cancellation = cancellation; + } + + 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 (u != null || nextFailure != null) { + downstream.onError(ExceptionHelper.unwrapAndCombine(nextFailure, u)); + } else { + if (current == null) { + if (defaultItem != null) { + downstream.onSuccess(defaultItem); + } else { + downstream.onError(new NoSuchElementException()); + } + } else { + downstream.onSuccess(current); + } + } + } else { + if (u != null) { + nextFailure = u; + done = true; + drain(); + } else { + if ((Boolean)t) { + current = upstream.current(); + } else { + done = true; + } + drain(); + } + } + } + + @Override + public void dispose() { + cancellation.dispose(); + } + } +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSingleFlattenAs.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSingleFlattenAs.java new file mode 100644 index 0000000000..2e525960a9 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSingleFlattenAs.java @@ -0,0 +1,146 @@ +/* + * 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.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; + +import io.reactivex.rxjava4.annotations.NonNull; +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.exceptions.Exceptions; +import io.reactivex.rxjava4.functions.Function; +import io.reactivex.rxjava4.internal.disposables.*; +import io.reactivex.rxjava4.internal.fuseable.HasUpstreamSingleSource; + +public record StreamableSingleFlattenAs( + SingleSource source, + Function> mapper +) implements Streamable, HasUpstreamSingleSource { + + @Override + public @NonNull Streamer<@NonNull U> stream(@NonNull StreamerCancellation cancellation) { + var observer = new FlattenAsSingleObserver<>(mapper, cancellation); + cancellation.add(observer); + source.subscribe(observer); + return observer; + } + + static final class FlattenAsSingleObserver + extends AtomicInteger + implements SingleObserver, Streamer, DisposableOnly, java.util.function.Function { + + @Serial + private static final long serialVersionUID = 796267562672678347L; + + final StreamerCancellation cancellation; + + final Function> mapper; + + final SequentialDisposable upstream; + + final CompletableFuture iteratorReady; + + volatile Iterator iteratorHandover; + + U current; + + Iterator currentIterator; + + FlattenAsSingleObserver( + Function> mapper, + StreamerCancellation cancellation) { + this.cancellation = cancellation; + this.mapper = mapper; + this.iteratorReady = new CompletableFuture<>(); + this.upstream = new SequentialDisposable(); + } + + @Override + public @NonNull CompletionStage next() { + var it = currentIterator; + if (it == null) { + return iteratorReady.thenApply(this); + } + if (it.hasNext()) { + current = it.next(); + return NEXT_TRUE; + } + return NEXT_FALSE; + } + + @Override + public Boolean apply(Boolean t) { + if (t) { + currentIterator = iteratorHandover; + iteratorHandover = null; + return true; + } else { + currentIterator = null; + iteratorHandover = null; + } + return false; + } + + @Override + public @NonNull U current() { + return current; + } + + @Override + public @NonNull CompletionStage finish() { + currentIterator = null; + DisposableHelper.dispose(upstream); + return FINISHED; + } + + @Override + public void dispose() { + DisposableHelper.dispose(upstream); + iteratorReady.completeExceptionally(new CancellationException()); + } + + @Override + public void onSubscribe(@NonNull Disposable d) { + DisposableHelper.setOnce(upstream, d); + } + + @Override + public void onSuccess(@NonNull T t) { + try { + var iterator = Objects.requireNonNull(mapper.apply(t), "The mapper returned a null Iterable").iterator(); + this.iteratorHandover = iterator; + if (iterator.hasNext()) { + current = iterator.next(); + iteratorReady.complete(true); + } else { + iteratorReady.complete(false); + } + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + upstream.lazySet(DisposableHelper.DISPOSED); + iteratorReady.completeExceptionally(ex); + return; + } + } + + @Override + public void onError(@NonNull Throwable e) { + upstream.lazySet(DisposableHelper.DISPOSED); + iteratorReady.completeExceptionally(e); + } + } +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSkip.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSkip.java new file mode 100644 index 0000000000..a2fd7f0cb0 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSkip.java @@ -0,0 +1,93 @@ +/* + * 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.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; + +public record StreamableSkip(Streamable source, long count) implements Streamable { + + @Override + public @NonNull Streamer<@NonNull T> stream(@NonNull StreamerCancellation cancellation) { + return new SkipStreamer<>(source.stream(cancellation), count); + } + + static final class SkipStreamer extends AtomicInteger implements Streamer, BiConsumer { + + @Serial + private static final long serialVersionUID = 1988154737845167665L; + + final Streamer upstream; + + long remaining; + + CompletableFuture waiter; + + SkipStreamer(Streamer upstream, long count) { + this.upstream = upstream; + this.remaining = count; + this.waiter = new CompletableFuture<>(); + } + + @Override + public @NonNull CompletionStage next() { + if (remaining <= 0) { + return upstream.next(); + } + drain(); + return waiter; + } + + void drain() { + if (getAndIncrement() != 0) { + return; + } + do { + upstream.next().whenComplete(this); + } while (decrementAndGet() != 0); + } + + @Override + public void accept(Boolean t, Throwable u) { + if (u != null) { + waiter.completeExceptionally(u); + } else + if (t) { + if (remaining-- > 0) { + drain(); + } else { + waiter.complete(true); + } + } else { + waiter.complete(false); + } + } + + @Override + public @NonNull T current() { + return upstream.current(); + } + + @Override + public @NonNull CompletionStage finish() { + return upstream.finish(); + } + } +} diff --git a/src/test/java/io/reactivex/rxjava4/disposables/DisposableTest.java b/src/test/java/io/reactivex/rxjava4/disposables/DisposableTest.java index 855191829c..865fc1fd08 100644 --- a/src/test/java/io/reactivex/rxjava4/disposables/DisposableTest.java +++ b/src/test/java/io/reactivex/rxjava4/disposables/DisposableTest.java @@ -27,7 +27,7 @@ import io.reactivex.rxjava4.core.RxJavaTest; import io.reactivex.rxjava4.exceptions.TestException; import io.reactivex.rxjava4.functions.Action; -import io.reactivex.rxjava4.internal.disposables.DisposableHelper; +import io.reactivex.rxjava4.internal.disposables.*; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.testsupport.TestHelper; @@ -250,4 +250,19 @@ public void autoCloseable() throws Throwable { assertTrue(d.isDisposed(), "d is not disposed"); } + + @Test + public void disposableOnlyThrows() { + assertThrows(UnsupportedOperationException.class, () -> { + enum Only implements DisposableOnly { + INSTANCE; + @Override + public void dispose() { + // not used here + } + } + + Only.INSTANCE.isDisposed(); + }); + } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBaseTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBaseTest.java index a3ee18d522..69553c9f99 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBaseTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBaseTest.java @@ -16,10 +16,11 @@ import java.lang.ref.Cleaner; import java.util.*; import java.util.concurrent.TimeoutException; -import java.util.function.BiConsumer; +import java.util.function.*; import org.junit.jupiter.api.*; +import io.reactivex.rxjava4.annotations.NonNull; import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.core.config.StreamableInterceptConfig; import io.reactivex.rxjava4.exceptions.CompositeException; @@ -167,4 +168,25 @@ public static void awaitNoStreamers(StreamProcessor sp, long timeoutMillis } } } + + /** + * Awaits a given {@link BooleanSupplier} to return the expected {@code value} + * within the given time period by sleeping in 1 microsecond increments until + * the timeout happens. + * @param value the expected value within the timeout period + * @param condition the condition to repeatedly call to assess the state + * @param timeoutMillis how long to wait for the condition to become as expected + * @throws InterruptedException if the sleep is interrupted + * @throws TimeoutException if the wait times out + */ + public static void awaitCondition(boolean value, @NonNull BooleanSupplier condition, long timeoutMillis) + throws InterruptedException, TimeoutException { + long timeout = timeoutMillis * 1_000_000L; + while (condition.getAsBoolean() != value) { + Thread.sleep(0, 1000); + if (--timeout <= 0L) { + throw new TimeoutException("condition still " + (!value)); + } + } + } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingTest.java new file mode 100644 index 0000000000..bc39a89e25 --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlockingTest.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 StreamableBlockingTest extends StreamableBaseTest { + + @Test + public void blockingFirstNormal() throws Throwable { + assertEquals(1, Streamable.just(1).blockingFirst()); + } + + @Test + public void blockingFirstMany() throws Throwable { + assertEquals(1, Streamable.range(1, 5).blockingFirst()); + } + + @Test + public void blockingFirstEmpty() throws Throwable { + assertThrows(NoSuchElementException.class, () -> { + Streamable.empty().blockingFirst(); + }); + } + + @Test + public void blockingFirstErrorUnchecked() throws Throwable { + assertThrows(TestException.class, () -> { + Streamable.error(new TestException()).blockingFirst(); + }); + } + + @Test + public void blockingFirstErrorChecked() throws Throwable { + var ex = assertThrows(CompletionException.class, () -> { + Streamable.error(new IOException()).blockingFirst(); + }); + + assertTrue(ex.getCause() instanceof IOException, "Wrong exception? " + ex.getCause()); + } + + @Test + public void blockingFirstFinishCrash() throws Throwable { + assertThrows(TestException.class, () -> { + StreamableFailingFinish.MAIN_COMPLETES.blockingFirst(); + }); + } + + @Test + public void blockingFirstNextAndFinishCrash() throws Throwable { + var ex = assertThrows(TestException.class, () -> { + StreamableFailingFinish.MAIN_FAILS.blockingFirst(); + }); + + assertTrue(ex.getSuppressed()[0] instanceof TestException, "Wrong exception? " + ex.getSuppressed()[0]); + } + +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDeferTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDeferTest.java index 9a4d545322..abc3285b7d 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDeferTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDeferTest.java @@ -17,6 +17,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; + import io.reactivex.rxjava4.core.Streamable; import io.reactivex.rxjava4.exceptions.TestException; @@ -45,4 +46,12 @@ public void crash() throws Throwable { .awaitDone(5, TimeUnit.SECONDS) .assertFailure(TestException.class); } + + @Test + public void error() throws Throwable { + Streamable.defer(() -> Streamable.error(new TestException())) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFailingFinish.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFailingFinish.java index 4ddb86d7bf..c0b7a8419b 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFailingFinish.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFailingFinish.java @@ -40,7 +40,7 @@ private StreamableFailingFinishStreamer(@NonNull StreamerCancellation dc) { var d = Disposable.fromFuture(cf, true); dc.add(d); if (mode == 1) { - cf.completeExceptionally(new TestException("StreamableFailingFinish(true)")); + cf.completeExceptionally(new TestException("StreamableFailingFinish.next()")); } else if (mode == 2) { cf.complete(false); @@ -55,7 +55,7 @@ private StreamableFailingFinishStreamer(@NonNull StreamerCancellation dc) { @Override public @NonNull CompletionStage finish() { - return CompletableFuture.failedFuture(new TestException()); + return CompletableFuture.failedFuture(new TestException("StreamableFailingFinish.finish()")); } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLastAsSingleTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLastAsSingleTest.java new file mode 100644 index 0000000000..f62bf4911f --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLastAsSingleTest.java @@ -0,0 +1,109 @@ +/* + * 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.assertTrue; + +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.exceptions.TestException; +import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; +import io.reactivex.rxjava4.processors.DispatchStreamProcessor; + +public class StreamableLastAsSingleTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + Streamable.range(1, 5) + .lastOrError() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(5); + } + + @Test + public void emptyDefault() throws Throwable { + Streamable.empty() + .last(6) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(6); + } + + @Test + public void error() throws Throwable { + Streamable.error(new TestException()) + .lastOrError() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void emptyNoDefault() throws Throwable { + Streamable.empty() + .lastOrError() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(NoSuchElementException.class); + } + + @Test + public void nonEmptyDefault() throws Throwable { + Streamable.range(1, 5) + .last(7) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(5); + } + + @Test + public void finallyCrash() throws Throwable { + StreamableFailingFinish.MAIN_COMPLETES + .last(7) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void cancelled() throws Throwable { + var dsp = new DispatchStreamProcessor<>(); + + var to = dsp.lastOrError() + .test(); + + to.awaitOnSubscribe(1, TimeUnit.SECONDS); + + awaitStreamers(dsp, 1000); + + to.dispose(); + + awaitNoStreamers(dsp, 1000); + } + + @Test + public void hasSource() { + var source = Streamable.range(1, 5); + + var operator = source.lastOrError(); + + assertTrue(operator instanceof HasUpstreamStreamableSource huss && source == huss.source(), + "HasUpstreamStreamableSource not supported or source() returns something unexpected: " + operator); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSingleFlattenAsTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSingleFlattenAsTest.java new file mode 100644 index 0000000000..3feae44e35 --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSingleFlattenAsTest.java @@ -0,0 +1,87 @@ +/* + * 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.List; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.Single; +import io.reactivex.rxjava4.exceptions.TestException; +import io.reactivex.rxjava4.subjects.SingleSubject; + +public class StreamableSingleFlattenAsTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + Single.just(1) + .flattenAsStreamable(v -> List.of(v * 10, v * 20, v * 30, v * 40, v * 50)) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(10, 20, 30, 40, 50); + } + + @Test + public void error() throws Throwable { + Single.error(new TestException()) + .flattenAsStreamable(v -> List.of(v * 10, v * 20, v * 30, v * 40, v * 50)) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void mapperNull() throws Throwable { + Single.just(1) + .flattenAsStreamable(_ -> null) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(NullPointerException.class); + } + + @Test + public void mapperCrash() throws Throwable { + Single.just(1) + .flattenAsStreamable(_ -> { throw new TestException(); }) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void emptyInner() throws Throwable { + Single.just(1) + .flattenAsStreamable(_ -> List.of()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void cancel() throws Throwable { + var ss = SingleSubject.create(); + + var to = ss.flattenAsStreamable(_ -> List.of()) + .test(); + + to.awaitOnSubscribe(1, TimeUnit.SECONDS); + + awaitCondition(true, () -> ss.hasObservers(), 1000); + + to.cancel(); + + awaitCondition(false, () -> ss.hasObservers(), 1000); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSkipTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSkipTest.java new file mode 100644 index 0000000000..c810d4d51d --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableSkipTest.java @@ -0,0 +1,88 @@ +/* + * 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.TimeUnit; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.exceptions.TestException; + +public class StreamableSkipTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + Streamable.range(1, 5) + .skip(2) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(3, 4, 5); + } + + @Test + public void doubleSkip() throws Throwable { + Streamable.range(1, 5) + .skip(2) + .skip(2) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(5); + } + + @Test + public void crash() throws Throwable { + Streamable.error(new TestException()) + .skip(2) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void zeroSkip() throws Throwable { + Streamable.range(1, 5) + .skip(0) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void empty() throws Throwable { + Streamable.empty() + .skip(2) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void skipAll() throws Throwable { + Streamable.range(1, 5) + .skip(5) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void skipMore() throws Throwable { + Streamable.range(1, 5) + .skip(6) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/validators/CheckBaseTypeAnnotationsTest.java b/src/test/java/io/reactivex/rxjava4/validators/CheckBaseTypeAnnotationsTest.java index 5449d83520..50e0177c03 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/CheckBaseTypeAnnotationsTest.java +++ b/src/test/java/io/reactivex/rxjava4/validators/CheckBaseTypeAnnotationsTest.java @@ -147,7 +147,8 @@ static void checkBackpressureSupport(Class clazz) { } } else { if (m.getReturnType() == Flowable.class - || m.getReturnType() == ParallelFlowable.class) { + || m.getReturnType() == ParallelFlowable.class + || m.getReturnType() == Streamable.class) { if (!m.isAnnotationPresent(BackpressureSupport.class)) { b.append("No @BackpressureSupport annotation (having ") .append(m.getReturnType().getSimpleName()) diff --git a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java index 9b72fd1501..1d7d0abce7 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java +++ b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java @@ -636,6 +636,8 @@ public void checkStreamable() { addIgnore(new ParamIgnore(Streamable.class, "intervalRange", Long.TYPE, Long.TYPE, Long.TYPE, Long.TYPE, TimeUnit.class, ExecutorService.class)); // zero take is allowed addOverride(new ParamOverride(Streamable.class, 0, ParamMode.NON_NEGATIVE, "take", Long.TYPE)); + // negative skip count is ignored + addOverride(new ParamOverride(Streamable.class, 0, ParamMode.NON_NEGATIVE, "skip", Long.TYPE)); // -----------------------------------------------------------------------------------