From c9891bf209c4fe5e39c820a93703e27da43207b0 Mon Sep 17 00:00:00 2001 From: akarnokd Date: Fri, 3 Jul 2026 21:47:44 +0200 Subject: [PATCH] 4.x: Streamable + takeWhile, + fromXXX, + toObservable + ~.toStreamble --- .../reactivex/rxjava4/core/Completable.java | 21 +++ .../java/io/reactivex/rxjava4/core/Maybe.java | 20 ++ .../io/reactivex/rxjava4/core/Single.java | 20 ++ .../io/reactivex/rxjava4/core/Streamable.java | 172 ++++++++++-------- .../streamable/StreamableForEach.java | 100 ++++++++++ .../streamable/StreamableFromCompletable.java | 88 +++++++++ .../streamable/StreamableFromMaybe.java | 102 +++++++++++ .../streamable/StreamableFromSingle.java | 97 ++++++++++ .../streamable/StreamableTakeWhile.java | 76 ++++++++ .../streamable/StreamableToObservable.java | 107 +++++++++++ .../internal/util/ExceptionHelper.java | 28 ++- .../streamable/StreamableBaseTest.java | 1 - .../StreamableFromCompletableTest.java | 72 ++++++++ .../streamable/StreamableFromMaybeTest.java | 81 +++++++++ .../streamable/StreamableFromSingleTest.java | 72 ++++++++ .../StreamableLifecycleVerifier.java | 81 +++++++++ .../StreamableLifecycleVerifierTest.java | 45 +++++ .../streamable/StreamableTakeWhileTest.java | 94 ++++++++++ .../StreamableToObservableTest.java | 131 +++++++++++++ .../internal/util/ExceptionHelperTest.java | 52 +++++- .../CheckParamValidationNamingTest.java | 16 +- 21 files changed, 1398 insertions(+), 78 deletions(-) create mode 100644 src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java create mode 100644 src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromCompletable.java create mode 100644 src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromMaybe.java create mode 100644 src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromSingle.java create mode 100644 src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeWhile.java create mode 100644 src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableToObservable.java create mode 100644 src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromCompletableTest.java create mode 100644 src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromMaybeTest.java create mode 100644 src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromSingleTest.java create mode 100644 src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLifecycleVerifier.java create mode 100644 src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLifecycleVerifierTest.java create mode 100644 src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeWhileTest.java create mode 100644 src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableToObservableTest.java diff --git a/src/main/java/io/reactivex/rxjava4/core/Completable.java b/src/main/java/io/reactivex/rxjava4/core/Completable.java index a702fa7e7e1..aa6c729bb3e 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Completable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Completable.java @@ -30,6 +30,7 @@ import io.reactivex.rxjava4.internal.operators.maybe.*; import io.reactivex.rxjava4.internal.operators.mixed.*; import io.reactivex.rxjava4.internal.operators.single.SingleDelayWithCompletable; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableFromCompletable; import io.reactivex.rxjava4.observers.TestObserver; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.schedulers.Schedulers; @@ -3221,6 +3222,26 @@ public final Future toFuture() { return RxJavaPlugins.onAssembly(new CompletableToSingle<>(this, null, completionValue)); } + /** + * Returns an {@link Streamable} which when subscribed to subscribes to this {@code Completable} and + * relays the terminal events to the downstream {@link Streamer}. + *

+ * + *

+ *
Scheduler:
+ *
{@code toStreamable} does not operate by default on a particular {@link Scheduler}.
+ *
+ * @param the value type + * @return the new {@code Streamable} instance + * @since 4.0.0 + */ + @SchedulerSupport(SchedulerSupport.NONE) + @CheckReturnValue + @NonNull + public final <@NonNull T> Streamable toStreamable() { + return RxJavaPlugins.onAssembly(new StreamableFromCompletable<>(this)); + } + /** * Returns a {@code Completable} which makes sure when an observer disposes the subscription, the * {@code dispose()} method is called on the specified {@link Scheduler}. diff --git a/src/main/java/io/reactivex/rxjava4/core/Maybe.java b/src/main/java/io/reactivex/rxjava4/core/Maybe.java index ae2cbb58516..60c78eee3df 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Maybe.java +++ b/src/main/java/io/reactivex/rxjava4/core/Maybe.java @@ -31,6 +31,7 @@ import io.reactivex.rxjava4.internal.operators.maybe.*; import io.reactivex.rxjava4.internal.operators.mixed.*; import io.reactivex.rxjava4.internal.operators.observable.ObservableElementAtMaybe; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableFromMaybe; import io.reactivex.rxjava4.observers.TestObserver; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.schedulers.*; @@ -3889,6 +3890,25 @@ public final Single toSingle() { return RxJavaPlugins.onAssembly(new MaybeToSingle<>(this, null)); } + /** + * Returns an {@link Streamable} which when subscribed to subscribes to this {@code Maybe} and + * relays the terminal events to the downstream {@link Streamer}. + *

+ * + *

+ *
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) + @NonNull + public final Streamable toStreamable() { + return RxJavaPlugins.onAssembly(new StreamableFromMaybe<>(this)); + } + /** * Returns a {@code Maybe} instance that if this {@code Maybe} emits an error, it will emit an {@code onComplete} * and swallow the throwable. diff --git a/src/main/java/io/reactivex/rxjava4/core/Single.java b/src/main/java/io/reactivex/rxjava4/core/Single.java index 554ed0a7e42..851ff50e5db 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Single.java +++ b/src/main/java/io/reactivex/rxjava4/core/Single.java @@ -33,6 +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.observers.TestObserver; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.schedulers.*; @@ -4887,6 +4888,25 @@ public final Observable toObservable() { return RxJavaPlugins.onAssembly(new SingleToObservable<>(this)); } + /** + * Returns an {@link Streamable} which when subscribed to subscribes to this {@code Single} and + * relays the terminal events to the downstream {@link Streamer}. + *

+ * + *

+ *
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) + @NonNull + public final Streamable toStreamable() { + return RxJavaPlugins.onAssembly(new StreamableFromSingle<>(this)); + } + /** * Returns a {@code Single} which makes sure when a {@link SingleObserver} disposes the {@link Disposable}, * that call is propagated up on the specified {@link Scheduler}. diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamable.java b/src/main/java/io/reactivex/rxjava4/core/Streamable.java index 7d226232c66..2480799b92d 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamable.java @@ -21,11 +21,9 @@ import io.reactivex.rxjava4.annotations.*; import io.reactivex.rxjava4.core.config.*; import io.reactivex.rxjava4.disposables.*; -import io.reactivex.rxjava4.exceptions.Exceptions; import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.functions.ObjectHelper; import io.reactivex.rxjava4.internal.operators.streamable.*; -import io.reactivex.rxjava4.internal.util.ExceptionHelper; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.schedulers.Schedulers; import io.reactivex.rxjava4.subscribers.TestSubscriber; @@ -95,7 +93,7 @@ public interface Streamable<@NonNull T> { * @param sources a streamable of inner streamables * @param executor the executorservice where to run the virtual wait * @return the new {@code Streamable} instance. - * @throws NullPointerException if {@code sources} or {@code exec} is {@code null} + * @throws NullPointerException if {@code sources} or {@code executor} is {@code null} */ @CheckReturnValue @NonNull @@ -160,7 +158,8 @@ public interface Streamable<@NonNull T> { * @param the element type * @param generator the generator to use * @param scheduler the scheduler to run the virtual generator on - * @return the streamable instance + * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code generator} or {@code scheduler} is {@code null} */ @CheckReturnValue @NonNull @@ -231,11 +230,11 @@ public interface Streamable<@NonNull T> { } /** - * Filters out the upstream items that do not pass the given predicate + * Filters out the upstream items that do not pass the given predicate. * @param predicate the callback that should return {@code true} to let the upstream value pass * or {@code false} to ignore it and continue with the next upstream item * @return the new {@code Streamable} instance - * @throw NullPointerException if {@code predicate} is {@code null} + * @throws NullPointerException if {@code predicate} is {@code null} */ @CheckReturnValue @NonNull @@ -259,6 +258,21 @@ default Streamable filter(@NonNull Predicate predicate) { return RxJavaPlugins.onAssembly(new StreamableFromArray<>(items)); } + /** + * Convert a {@link CompletableSource} into a {@code Streamable} and + * relay its terminal events. + *

+ * The resulting {@code Streamable} will never produce any items. + * @param the target type of the sequence + * @param source the source {@code CompletableSource} to convert + * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code source} is {@code null} + */ + static <@NonNull T> Streamable fromCompletable(@NonNull CompletableSource source) { + Objects.requireNonNull(source, "source is null"); + return RxJavaPlugins.onAssembly(new StreamableFromCompletable(source)); + } + /** * Streams all elements of the given {@link Iterable} sequence. * @param the element type of the items @@ -273,11 +287,27 @@ default Streamable filter(@NonNull Predicate predicate) { return RxJavaPlugins.onAssembly(new StreamableFromIterable<>(items)); } + /** + * Convert a {@link MaybeSource} into a {@code Streamable} and + * relay its terminal events. + *

+ * The resulting {@code Streamable} will never produce any items. + * @param the target type of the sequence + * @param source the source {@code MaybeSource} to convert + * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code source} is {@code null} + */ + static <@NonNull T> Streamable fromMaybe(@NonNull MaybeSource source) { + Objects.requireNonNull(source, "source is null"); + return RxJavaPlugins.onAssembly(new StreamableFromMaybe(source)); + } + /** * Convert any {@link java.util.concurrent.Flow.Publisher} into a {@code Streamable} sequence. * @param the element type * @param source Flow.Publisher to convert * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code source} is {@code null} */ @CheckReturnValue @NonNull @@ -292,6 +322,7 @@ static Streamable fromPublisher(@NonNull Flow.Publisher source) { * @param source Flow.Publisher to convert * @param executor where the conversion will run * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code source} or {@code executor} is {@code null} */ @CheckReturnValue @NonNull @@ -301,6 +332,21 @@ static Streamable fromPublisher(@NonNull Flow.Publisher source, @NonNu return RxJavaPlugins.onAssembly(new StreamableFromPublisher<>(source, executor)); } + /** + * Convert a {@link SingleSource} into a {@code Streamable} and + * relay its terminal events. + *

+ * The resulting {@code Streamable} will never produce any items. + * @param the target type of the sequence + * @param source the source {@code SingleSource} to convert + * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code source} is {@code null} + */ + static <@NonNull T> Streamable fromSingle(@NonNull SingleSource source) { + Objects.requireNonNull(source, "source is null"); + return RxJavaPlugins.onAssembly(new StreamableFromSingle(source)); + } + /** * Streams all elements of the given {@link Stream} sequence. * @param the element type of the items @@ -327,6 +373,8 @@ static Streamable fromPublisher(@NonNull Flow.Publisher source, @NonNu * @param unit the time unit for both {@code initialDelay} and {@code period} * @param scheduler the scheduler to use for the timed waiting * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} + * @throws IllegalArgumentException if {@code count} is negative */ static Streamable intervalRange(long start, long count, long initialDelay, long period, TimeUnit unit, Scheduler scheduler) { @@ -360,6 +408,8 @@ static Streamable intervalRange(long start, long count, * @param unit the time unit for both {@code initialDelay} and {@code period} * @param executor the executor to use * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code unit} or {@code executor} is {@code null} + * @throws IllegalArgumentException if {@code count} is negative */ static Streamable intervalRange(long start, long count, long initialDelay, long period, TimeUnit unit, ExecutorService executor) { @@ -381,6 +431,7 @@ static Streamable intervalRange(long start, long count, * @param the element type * @param item the constant item to produce * @return the {@code Streamable} instance + * @throws NullPointerException if {@code item} is {@code null} */ @CheckReturnValue @NonNull @@ -406,6 +457,7 @@ static Streamable intervalRange(long start, long count, * @param start the start element * @param count the number of elements to emit * @return the new {@code Streamable} instance + * @throws IllegalArgumentException if {@code count} is negative */ @CheckReturnValue @NonNull @@ -430,6 +482,7 @@ static Streamable range(int start, int count) { * @param start the start element * @param count the number of elements to emit * @return the new {@code Streamable} instance + * @throws IllegalArgumentException if {@code count} is negative */ @CheckReturnValue @NonNull @@ -536,7 +589,7 @@ default Streamable hide() { * and allows the modification of them via Function callbacks. * @param config the configuration record for this operator * @return the new {@code Streamable} instance - * @thros NullPointerException if {@code config} is {@code null} + * @throws NullPointerException if {@code config} is {@code null} */ @CheckReturnValue @NonNull @@ -576,7 +629,7 @@ default Streamable intercept(StreamableInterceptConfig config) { * @param mapper the function that takes an upstream item and returns an item to be emitted * to the downstream * @return the new {@code Streamable} instance - * @throw NullPointerException if {@code mapper} is {@code null} + * @throws NullPointerException if {@code mapper} is {@code null} */ @CheckReturnValue @NonNull @@ -591,7 +644,7 @@ default Streamable intercept(StreamableInterceptConfig config) { * @param mapper the function that takes an upstream item and returns an optional item to be emitted / skipped * to the downstream * @return the new {@code Streamable} instance - * @throw NullPointerException if {@code mapper} is {@code null} + * @throws NullPointerException if {@code mapper} is {@code null} */ @CheckReturnValue @NonNull @@ -605,6 +658,7 @@ default Streamable intercept(StreamableInterceptConfig config) { * then cancels the rest of the sequence. * @param n the maximum number of items to relay * @return the new {@code Streamable} instance + * @throws IllegalArgumentException if {@code n} is non-positive */ @CheckReturnValue @NonNull @@ -621,6 +675,19 @@ default Streamable take(long n) { }); } + /** + * Relays items from this {@code Streamable} while the predicate returns {@code true} + * @param predicate the predicate to test if the sequence should keep going + * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code predicate} is {@code null} + */ + @CheckReturnValue + @NonNull + default Streamable takeWhile(@NonNull Predicate predicate) { + Objects.requireNonNull(predicate, "predicate is null"); + return RxJavaPlugins.onAssembly(new StreamableTakeWhile<>(this, predicate)); + } + /** * Calls the specified converter function during assembly time and returns its resulting value. *

@@ -648,19 +715,33 @@ default Flowable toFlowable() { } /** - * Converts the streamable into a Flowable representation, running + * Converts this {@code Streamable} into a {@link Flowable} representation, running * on the provided executor service. * @param executor the executor to use - * @return the new Flowable instance + * @return the new {@code Flowable} instance + * @throws NullPointerException if {@code executor} is {@code null} */ @CheckReturnValue @NonNull default Flowable toFlowable(@NonNull ExecutorService executor) { - Objects.requireNonNull(executor, "executir is null"); + Objects.requireNonNull(executor, "executor is null"); var me = this; return Flowable.virtualCreate(emitter -> me.forEach(emitter::emit).await(), executor); } + /** + * Converts this {@code Streamable} into a {@link Observable} representation, + * emitting items on whatever thread produces items in the current {@code Streamable}. + *

+ * Unlike {@link #toFlowable(ExecutorService)}, the lack of backpressure doesn't require + * any blocking thus any need to run the conversion on any particular {@link Scheduler} + * or {@link ExecutorService} on its own. + * @return the new {@code Observable} instance + */ + default Observable toObservable() { + return RxJavaPlugins.onAssembly(new StreamableToObservable<>(this)); + } + /** * Transforms the upstream sequence into zero or more elements for the downstream. * @param the result element type @@ -678,7 +759,8 @@ default Flowable toFlowable(@NonNull ExecutorService executor) { * @param the result element type * @param transformer the interface to implement the transforming logic * @param executor where to run the transform and blocking operations - * @return the new Streamable instance + * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code transformer} or {@code executor} is {@code null} */ @CheckReturnValue @NonNull @@ -740,7 +822,8 @@ default CompletionStageDisposable forEach(@NonNull Consumer con * @param consumer the callback that gets the elements until completion * @param canceller the container to trigger cancellation of the sequence * @param executor the service that hosts the blocking waits. - * @return the {@code CompletionStage} that gets notified when the sequence ends + * @return the new {@code CompletionStageDisposable} that gets notified when the sequence ends + * @throws NullPointerException if {@code consumer} or {@code canceller} or {@code executor} is {@code null} */ @CheckReturnValue @NonNull @@ -748,34 +831,7 @@ default CompletionStageDisposable forEach(@NonNull Consumer con Objects.requireNonNull(consumer, "consumer is null"); Objects.requireNonNull(canceller, "canceller is null"); Objects.requireNonNull(executor, "executor is null"); - final Streamable me = this; - var future = CompletableFuture.supplyAsync(() -> { - var str = me.stream(canceller); - try { - try { - while (!canceller.isDisposed()) { - if (str.awaitNext()) { - // System.out.println("Received " + str.current()); - consumer.accept(Objects.requireNonNull(str.current(), "The upstream Streamable " + me.getClass() + " produced a null element!")); - } else { - // System.out.println("EOF "); - break; - } - } - } finally { - str.awaitFinish(); - } - } catch (final Throwable crash) { - Exceptions.throwIfFatal(crash); - if (crash instanceof CompletionException ce) { - throw ExceptionHelper.wrapOrThrow(ce.getCause()); - } - throw ExceptionHelper.wrapOrThrow(crash); - } - return null; - }, executor); - canceller.add(Disposable.fromFuture(future)); - return new CompletionStageDisposable<>(future, canceller); + return StreamableForEach.forEach(this, consumer, canceller, executor); } /** @@ -784,6 +840,7 @@ default CompletionStageDisposable forEach(@NonNull Consumer con * @param canceller the container to trigger cancellation of the sequence * @param executor the service that hosts the blocking waits. * @return the {@code CompletionStage} that gets notified when the sequence ends + * @throws NullPointerException if {@code consumer} or {@code canceller} or {@code executor} is {@code null} */ @CheckReturnValue @NonNull @@ -794,36 +851,7 @@ default CompletionStageDisposable forEach( Objects.requireNonNull(consumer, "consumer is null"); Objects.requireNonNull(canceller, "canceller is null"); Objects.requireNonNull(executor, "executor is null"); - final Streamable me = this; - var future = CompletableFuture.supplyAsync(() -> { - var str = me.stream(canceller); - try { - try { - var stopper = Disposable.empty(); - while (!canceller.isDisposed() && !stopper.isDisposed()) { - if (str.awaitNext()) { - // System.out.println("Received " + str.current()); - var v = Objects.requireNonNull(str.current(), "The upstream Streamable " + me.getClass() + " produced a null element!"); - consumer.accept(v, stopper); - } else { - // System.out.println("EOF "); - break; - } - } - } finally { - str.awaitFinish(); - } - } catch (final Throwable crash) { - Exceptions.throwIfFatal(crash); - if (crash instanceof CompletionException ce) { - throw ExceptionHelper.wrapOrThrow(ce.getCause()); - } - throw ExceptionHelper.wrapOrThrow(crash); - } - return null; - }); - canceller.add(Disposable.fromFuture(future)); - return new CompletionStageDisposable<>(future, canceller); + return StreamableForEach.forEach(this, consumer, canceller, executor); } /** diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java new file mode 100644 index 00000000000..c905a0b4e4f --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java @@ -0,0 +1,100 @@ +/* + * 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 java.util.concurrent.*; + +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.*; +import io.reactivex.rxjava4.internal.util.ExceptionHelper; + +/** + * ForEach implementation to unclutter the {@link Streamable} type. + */ +public record StreamableForEach() { + + public static CompletionStageDisposable forEach( + @NonNull Streamable me, + @NonNull Consumer consumer, + @NonNull DisposableContainer canceller, + @NonNull ExecutorService executor) { + var future = CompletableFuture.supplyAsync(() -> { + var str = me.stream(canceller); + try { + try { + while (!canceller.isDisposed()) { + if (str.awaitNext()) { + // System.out.println("Received " + str.current()); + consumer.accept(Objects.requireNonNull(str.current(), "The upstream Streamable " + me.getClass() + " produced a null element!")); + } else { + // System.out.println("EOF "); + break; + } + } + } finally { + str.awaitFinish(); + } + } catch (final Throwable crash) { + Exceptions.throwIfFatal(crash); + if (crash instanceof CompletionException ce) { + throw ExceptionHelper.wrapOrThrow(ce.getCause()); + } + throw ExceptionHelper.wrapOrThrow(crash); + } + return null; + }, executor); + canceller.add(Disposable.fromFuture(future)); + return new CompletionStageDisposable<>(future, canceller); + } + + public static CompletionStageDisposable forEach( + @NonNull Streamable me, + @NonNull BiConsumer consumer, + @NonNull DisposableContainer canceller, + @NonNull ExecutorService executor) { + var future = CompletableFuture.supplyAsync(() -> { + var str = me.stream(canceller); + try { + try { + var stopper = Disposable.empty(); + while (!canceller.isDisposed() && !stopper.isDisposed()) { + if (str.awaitNext()) { + // System.out.println("Received " + str.current()); + var v = Objects.requireNonNull(str.current(), "The upstream Streamable " + me.getClass() + " produced a null element!"); + consumer.accept(v, stopper); + } else { + // System.out.println("EOF "); + break; + } + } + } finally { + str.awaitFinish(); + } + } catch (final Throwable crash) { + Exceptions.throwIfFatal(crash); + if (crash instanceof CompletionException ce) { + throw ExceptionHelper.wrapOrThrow(ce.getCause()); + } + throw ExceptionHelper.wrapOrThrow(crash); + } + return null; + }); + canceller.add(Disposable.fromFuture(future)); + return new CompletionStageDisposable<>(future, canceller); + } +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromCompletable.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromCompletable.java new file mode 100644 index 00000000000..4f2d8424ebf --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromCompletable.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.io.Serial; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicReference; + +import io.reactivex.rxjava4.annotations.NonNull; +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.internal.disposables.DisposableHelper; +import io.reactivex.rxjava4.internal.fuseable.HasUpstreamCompletableSource; + +public record StreamableFromCompletable(CompletableSource source) +implements Streamable, HasUpstreamCompletableSource { + + @Override + public @NonNull Streamer stream(@NonNull DisposableContainer cancellation) { + var streamer = new CompletableStreamer(); + cancellation.add(streamer); + source.subscribe(streamer); + return streamer; + } + + static final class CompletableStreamer + extends AtomicReference + implements Streamer, CompletableObserver, Disposable { + + @Serial + private static final long serialVersionUID = -4580514428263096178L; + + final CompletableFuture waiter = new CompletableFuture<>(); + + @Override + public void onSubscribe(@NonNull Disposable d) { + DisposableHelper.setOnce(this, d); + } + + @Override + public void onComplete() { + waiter.complete(false); + } + + @Override + public void onError(@NonNull Throwable e) { + waiter.completeExceptionally(e); + } + + @Override + public @NonNull CompletionStage next() { + return waiter; + } + + @Override + public @NonNull T current() { + return null; // never has any items + } + + @Override + public @NonNull CompletionStage finish() { + DisposableHelper.dispose(this); + return FINISHED; + } + + @Override + public void dispose() { + DisposableHelper.dispose(this); + waiter.completeExceptionally(new CancellationException()); + } + + @Override + public boolean isDisposed() { + return get() == DisposableHelper.DISPOSED; + } + } +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromMaybe.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromMaybe.java new file mode 100644 index 00000000000..49fbf0345ce --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromMaybe.java @@ -0,0 +1,102 @@ +/* + * 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.AtomicReference; + +import io.reactivex.rxjava4.annotations.NonNull; +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.internal.disposables.DisposableHelper; +import io.reactivex.rxjava4.internal.fuseable.HasUpstreamMaybeSource; + +public record StreamableFromMaybe(MaybeSource source) +implements Streamable, HasUpstreamMaybeSource { + + @Override + public @NonNull Streamer stream(@NonNull DisposableContainer cancellation) { + var streamer = new MaybeStreamer(); + cancellation.add(streamer); + source.subscribe(streamer); + return streamer; + } + + static final class MaybeStreamer + extends AtomicReference + implements Streamer, MaybeObserver, Disposable { + + @Serial + private static final long serialVersionUID = -4580514428263096178L; + + final CompletableFuture waiter = new CompletableFuture<>(); + + int stage; + + volatile T current; + + @Override + public void onSubscribe(@NonNull Disposable d) { + DisposableHelper.setOnce(this, d); + } + + @Override + public void onSuccess(@NonNull T t) { + current = t; + waiter.complete(true); + } + + @Override + public void onComplete() { + waiter.complete(false); + } + + @Override + public void onError(@NonNull Throwable e) { + waiter.completeExceptionally(e); + } + + @Override + public @NonNull CompletionStage next() { + if (stage++ == 0) { + return waiter; + } + return NEXT_FALSE; + } + + @Override + public @NonNull T current() { + return current; + } + + @Override + public @NonNull CompletionStage finish() { + current = null; + DisposableHelper.dispose(this); + return FINISHED; + } + + @Override + public void dispose() { + DisposableHelper.dispose(this); + waiter.completeExceptionally(new CancellationException()); + } + + @Override + public boolean isDisposed() { + return get() == DisposableHelper.DISPOSED; + } + } +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromSingle.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromSingle.java new file mode 100644 index 00000000000..970d4ef7cc4 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromSingle.java @@ -0,0 +1,97 @@ +/* + * 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.AtomicReference; + +import io.reactivex.rxjava4.annotations.NonNull; +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.internal.disposables.DisposableHelper; +import io.reactivex.rxjava4.internal.fuseable.HasUpstreamSingleSource; + +public record StreamableFromSingle(SingleSource source) +implements Streamable, HasUpstreamSingleSource { + + @Override + public @NonNull Streamer stream(@NonNull DisposableContainer cancellation) { + var streamer = new MaybeStreamer(); + cancellation.add(streamer); + source.subscribe(streamer); + return streamer; + } + + static final class MaybeStreamer + extends AtomicReference + implements Streamer, SingleObserver, Disposable { + + @Serial + private static final long serialVersionUID = -4580514428263096178L; + + final CompletableFuture waiter = new CompletableFuture<>(); + + int stage; + + volatile T current; + + @Override + public void onSubscribe(@NonNull Disposable d) { + DisposableHelper.setOnce(this, d); + } + + @Override + public void onSuccess(@NonNull T t) { + current = t; + waiter.complete(true); + } + + @Override + public void onError(@NonNull Throwable e) { + waiter.completeExceptionally(e); + } + + @Override + public @NonNull CompletionStage next() { + if (stage++ == 0) { + return waiter; + } + return NEXT_FALSE; + } + + @Override + public @NonNull T current() { + return current; + } + + @Override + public @NonNull CompletionStage finish() { + current = null; + DisposableHelper.dispose(this); + return FINISHED; + } + + @Override + public void dispose() { + DisposableHelper.dispose(this); + waiter.completeExceptionally(new CancellationException()); + } + + @Override + public boolean isDisposed() { + return get() == DisposableHelper.DISPOSED; + } + } +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeWhile.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeWhile.java new file mode 100644 index 00000000000..0586dafe425 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeWhile.java @@ -0,0 +1,76 @@ +/* + * 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.*; + +import io.reactivex.rxjava4.annotations.NonNull; +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.DisposableContainer; +import io.reactivex.rxjava4.exceptions.Exceptions; +import io.reactivex.rxjava4.functions.*; +import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; + +public record StreamableTakeWhile(Streamable source, Predicate predicate) +implements Streamable, HasUpstreamStreamableSource { + + @Override + public @NonNull Streamer<@NonNull T> stream(@NonNull DisposableContainer cancellation) { + return new TakeWhileStreamer<>(source.stream(cancellation), predicate); + } + + static final class TakeWhileStreamer + implements Streamer, java.util.function.Function> { + + final Streamer upstream; + + final Predicate predicate; + + TakeWhileStreamer(Streamer upstream, Predicate predicate) { + this.upstream = upstream; + this.predicate = predicate; + } + + @Override + public @NonNull CompletionStage next() { + return upstream.next().thenCompose(this); + } + + @Override + public @NonNull T current() { + return upstream.current(); + } + + @Override + public @NonNull CompletionStage finish() { + return upstream.finish(); + } + + @Override + public @NonNull CompletionStage apply(@NonNull Boolean t) { + if (t) { + try { + if (predicate.test(upstream.current())) { + return NEXT_TRUE; + } + return NEXT_FALSE; + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + return CompletableFuture.failedFuture(ex); + } + } + return NEXT_FALSE; + } + } +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableToObservable.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableToObservable.java new file mode 100644 index 00000000000..af0d2c9b4b2 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableToObservable.java @@ -0,0 +1,107 @@ +/* + * 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.atomic.*; +import java.util.function.BiConsumer; + +import io.reactivex.rxjava4.annotations.NonNull; +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; +import io.reactivex.rxjava4.internal.util.ExceptionHelper; + +public final class StreamableToObservable extends Observable +implements HasUpstreamStreamableSource { + + final Streamable source; + + public StreamableToObservable(Streamable source) { + this.source = source; + } + + @Override + public @NonNull Streamable<@NonNull T> source() { + return source; + } + + @Override + protected void subscribeActual(@NonNull Observer observer) { + var cs = new CompositeDisposable(); + observer.onSubscribe(cs); + var sto = new StreamToObserver<>(source.stream(cs), observer, cs, + new AtomicInteger(), new AtomicBoolean(), new AtomicReference<>(), new AtomicBoolean()); + cs.add(sto); + sto.drain(); + } + + record StreamToObserver(Streamer streamer, Observer observer, DisposableContainer cancellation, + AtomicInteger wip, AtomicBoolean done, AtomicReference mainError, AtomicBoolean disposed) + implements BiConsumer, Disposable { + + void drain() { + if (wip.getAndIncrement() != 0) { + return; + } + + do { + if (done.get()) { + streamer.finish().whenComplete(this); + break; + } else { + streamer.next().whenComplete(this); + } + } while (wip.decrementAndGet() != 0); + } + + @Override + public void accept(Object t, Throwable u) { + if (disposed.get()) { + return; + } + if (done.get()) { + var w = ExceptionHelper.unwrapAndCombine(mainError.get(), u); + if (w != null) { + observer.onError(w); + } else { + observer.onComplete(); + } + } else { + if (u != null) { + mainError.lazySet(u); + done.lazySet(true); + } else { + if ((Boolean)t) { + observer.onNext(streamer.current()); + } else { + done.lazySet(true); + } + } + drain(); + } + } + + @Override + public void dispose() { + disposed.lazySet(true); + done.lazySet(true); + drain(); + } + + @Override + public boolean isDisposed() { + return disposed.get(); + } + } +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/util/ExceptionHelper.java b/src/main/java/io/reactivex/rxjava4/internal/util/ExceptionHelper.java index d15d87275f5..733e72f0855 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/util/ExceptionHelper.java +++ b/src/main/java/io/reactivex/rxjava4/internal/util/ExceptionHelper.java @@ -15,9 +15,10 @@ import java.io.Serial; import java.util.*; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicReference; +import io.reactivex.rxjava4.annotations.Nullable; import io.reactivex.rxjava4.exceptions.*; /** @@ -181,4 +182,29 @@ public static T nullCheck(T value, String prefix) { } return value; } + + /** + * Unwraps both throwables if they are wrapped into a {@link CompletionException} or + * {@link ThrowableWrapper}, then if both are present, add {@code b} as suppressed to {@code a} + * and return a; return b otherwise + * @param main the first throwable + * @param secondary the second throwable + * @return the unwrapped and combined throwable or null if both where + */ + @Nullable + public static Throwable unwrapAndCombine(@Nullable Throwable main, @Nullable Throwable secondary) { + if (main instanceof CompletionException || main instanceof ThrowableWrapper) { + main = main.getCause(); + } + if (secondary instanceof CompletionException || secondary instanceof ThrowableWrapper) { + secondary = secondary.getCause(); + } + if (main != null && secondary != null) { + main.addSuppressed(secondary); + } + if (main != null) { + return main; + } + return secondary; + } } 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 816e7ef6367..a0c0caa90c4 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 @@ -15,7 +15,6 @@ import java.lang.ref.Cleaner; import java.util.*; -import java.util.concurrent.*; import org.junit.jupiter.api.*; diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromCompletableTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromCompletableTest.java new file mode 100644 index 00000000000..7e1c6c04012 --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromCompletableTest.java @@ -0,0 +1,72 @@ +/* + * 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.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.exceptions.TestException; +import io.reactivex.rxjava4.subjects.CompletableSubject; + +public class StreamableFromCompletableTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + Streamable.fromCompletable(Completable.complete()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void normalViaCompletable() throws Throwable { + Completable.complete() + .toStreamable() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void crash() throws Throwable { + Completable.error(new TestException()) + .toStreamable() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void dispose() { + var cs = CompletableSubject.create(); + + var cd = new CompositeDisposable(); + var ts = cs.toStreamable().stream(cd); + + assertFalse((ts instanceof Disposable d) && d.isDisposed(), "Disposed?"); + assertTrue(cs.hasObservers(), "has no observers?"); + + assertNull(ts.current()); + + cd.dispose(); + + assertFalse(cs.hasObservers(), "has observers?"); + assertTrue((ts instanceof Disposable d) && d.isDisposed(), "Not Disposed?"); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromMaybeTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromMaybeTest.java new file mode 100644 index 00000000000..16125fe7de3 --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromMaybeTest.java @@ -0,0 +1,81 @@ +/* + * 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.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.exceptions.TestException; +import io.reactivex.rxjava4.subjects.MaybeSubject; + +public class StreamableFromMaybeTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + Streamable.fromMaybe(Maybe.just(1)) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1); + } + + @Test + public void normalViaMaybe() throws Throwable { + Maybe.just(1) + .toStreamable() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1); + } + + @Test + public void crash() throws Throwable { + Maybe.error(new TestException()) + .toStreamable() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void empty() throws Throwable { + Maybe.empty() + .toStreamable() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void dispose() { + var cs = MaybeSubject.create(); + + var cd = new CompositeDisposable(); + var ts = cs.toStreamable().stream(cd); + + assertFalse((ts instanceof Disposable d) && d.isDisposed(), "Disposed?"); + assertTrue(cs.hasObservers(), "has no observers?"); + + assertNull(ts.current()); + + cd.dispose(); + + assertFalse(cs.hasObservers(), "has observers?"); + assertTrue((ts instanceof Disposable d) && d.isDisposed(), "Not Disposed?"); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromSingleTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromSingleTest.java new file mode 100644 index 00000000000..c7ffb65506a --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromSingleTest.java @@ -0,0 +1,72 @@ +/* + * 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.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.exceptions.TestException; +import io.reactivex.rxjava4.subjects.SingleSubject; + +public class StreamableFromSingleTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + Streamable.fromSingle(Single.just(1)) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1); + } + + @Test + public void normalViaSingle() throws Throwable { + Single.just(1) + .toStreamable() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1); + } + + @Test + public void crash() throws Throwable { + Single.error(new TestException()) + .toStreamable() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void dispose() { + var cs = SingleSubject.create(); + + var cd = new CompositeDisposable(); + var ts = cs.toStreamable().stream(cd); + + assertFalse((ts instanceof Disposable d) && d.isDisposed(), "Disposed?"); + assertTrue(cs.hasObservers(), "has no observers?"); + + assertNull(ts.current()); + + cd.dispose(); + + assertFalse(cs.hasObservers(), "has observers?"); + assertTrue((ts instanceof Disposable d) && d.isDisposed(), "Not Disposed?"); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLifecycleVerifier.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLifecycleVerifier.java new file mode 100644 index 00000000000..53af03f1a2d --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLifecycleVerifier.java @@ -0,0 +1,81 @@ +/* + * 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.concurrent.atomic.AtomicInteger; + +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.core.config.StreamableInterceptConfig; +import io.reactivex.rxjava4.disposables.DisposableContainer; + +/** + * Tracks the calls to the various lifecycle events and allows verifying the call patterns. + * @param the element type of the sequence + * @since 4.0.0 + */ +public final class StreamableLifecycleVerifier { + + final StreamableInterceptConfig config; + + final AtomicInteger onStreamCount; + + final AtomicInteger onNextCount; + + final AtomicInteger onCurrentCount; + + final AtomicInteger onFinishCount; + + public StreamableLifecycleVerifier() { + onStreamCount = new AtomicInteger(); + onNextCount = new AtomicInteger(); + onCurrentCount = new AtomicInteger(); + onFinishCount = new AtomicInteger(); + config = new StreamableInterceptConfig<>( + (_, v) -> { onStreamCount.getAndIncrement(); return v; }, + (_, v) -> { onNextCount.getAndIncrement(); return v; }, + (v) -> { onCurrentCount.getAndIncrement(); return v; }, + (_, v) -> { onFinishCount.getAndIncrement(); return v; } + ); + } + + public StreamableInterceptConfig config() { + return config; + } + + /** + * Verify the intercept registered only one {@link Streamable#stream(DisposableContainer)} + * call and only one {@link Streamer#finish()} call. + */ + public void verify() { + assertAll( + () -> assertEquals(1, onStreamCount.get(), "onStreamCount"), + () -> assertEquals(1, onFinishCount.get(), "onFinishCount") + ); + } + + /** + * Verify the intercept registered only one {@link Streamable#stream(DisposableContainer)} + * call and only one {@link Streamer#finish()} call and one more + * {@link Streamer#next()} calls than {@link Streamer#current()} calls + */ + public void verifyStrict() { + assertAll( + () -> assertEquals(1, onStreamCount.get(), "onStreamCount"), + () -> assertEquals(onNextCount.get() - 1, onCurrentCount.get(), "onNextCount > onCurrentCount"), + () -> assertEquals(1, onFinishCount.get(), "onFinishCount") + ); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLifecycleVerifierTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLifecycleVerifierTest.java new file mode 100644 index 00000000000..94fdbece6c7 --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableLifecycleVerifierTest.java @@ -0,0 +1,45 @@ +/* + * 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.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.Streamable; + +public class StreamableLifecycleVerifierTest { + + @Test + public void checkVerify() { + var v = new StreamableLifecycleVerifier(); + + assertThrows(AssertionError.class, () -> v.verify()); + + assertThrows(AssertionError.class, () -> v.verifyStrict()); + + Streamable.range(1, 5) + .intercept(v.config()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + + v.verify(); + + v.verifyStrict(); + } + +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeWhileTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeWhileTest.java new file mode 100644 index 00000000000..dfa52692cc0 --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeWhileTest.java @@ -0,0 +1,94 @@ +/* + * 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 StreamableTakeWhileTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + var verifier = new StreamableLifecycleVerifier(); + + Streamable.range(1, 5) + .intercept(verifier.config()) + .takeWhile(v -> v < 4) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3); + + verifier.verify(); + } + + @Test + public void crash() throws Throwable { + var verifier = new StreamableLifecycleVerifier(); + + Streamable.error(new TestException()) + .intercept(verifier.config()) + .takeWhile(v -> v < 4) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + + verifier.verify(); + } + + @Test + public void allPass() throws Throwable { + var verifier = new StreamableLifecycleVerifier(); + + Streamable.range(1, 5) + .intercept(verifier.config()) + .takeWhile(_ -> true) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + + verifier.verify(); + } + + @Test + public void nonePass() throws Throwable { + var verifier = new StreamableLifecycleVerifier(); + + Streamable.range(1, 5) + .intercept(verifier.config()) + .takeWhile(_ -> false) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + + verifier.verify(); + } + + @Test + public void predicateCrash() throws Throwable { + var verifier = new StreamableLifecycleVerifier(); + + Streamable.range(1, 5) + .intercept(verifier.config()) + .takeWhile(_ -> { throw new TestException(); }) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + + verifier.verify(); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableToObservableTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableToObservableTest.java new file mode 100644 index 00000000000..0d6a0dd078b --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableToObservableTest.java @@ -0,0 +1,131 @@ +/* + * 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.concurrent.TimeUnit; +import java.util.concurrent.atomic.*; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.core.config.StreamableInterceptConfig; +import io.reactivex.rxjava4.disposables.CompositeDisposable; +import io.reactivex.rxjava4.exceptions.TestException; +import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; +import io.reactivex.rxjava4.internal.operators.streamable.StreamableToObservable.StreamToObserver; +import io.reactivex.rxjava4.observers.TestObserver; +import io.reactivex.rxjava4.processors.PublishProcessor; + +public class StreamableToObservableTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + var onFinishCounter = new AtomicInteger(); + + Streamable.range(1, 5) + .intercept(new StreamableInterceptConfig<>( + (_, v) -> v, (_, v) -> v, v -> v, (_, v) -> { onFinishCounter.incrementAndGet(); return v; })) + .toObservable() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + + assertEquals(1, onFinishCounter.get(), "onFinishCounter"); + } + + @Test + public void crash() throws Throwable { + AtomicInteger onFinishCounter = new AtomicInteger(); + + Streamable.error(new TestException()) + .intercept(new StreamableInterceptConfig<>( + (_, v) -> v, (_, v) -> v, v -> v, (_, v) -> { onFinishCounter.incrementAndGet(); return v; })) + .toObservable() + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + + assertEquals(1, onFinishCounter.get(), "onFinishCounter"); + } + + @Test + public void take() throws Throwable { + var onFinishCounter = new AtomicInteger(); + + Streamable.range(1, 5) + .intercept(new StreamableInterceptConfig<>( + (_, v) -> v, (_, v) -> v, v -> v, (_, v) -> { onFinishCounter.incrementAndGet(); return v; })) + .toObservable() + .take(3) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3); + + assertEquals(1, onFinishCounter.get(), "onFinishCounter"); + } + + @Test + public void subject() throws Throwable { + var pp = PublishProcessor.create(); + var ts = pp.toStreamable() + .toObservable() + .test(); + + while (!pp.hasSubscribers()) { + Thread.sleep(1); + } + + while (!ts.hasSubscription()) { + Thread.sleep(1); + } + + ts.dispose(); + + while (pp.hasSubscribers()) { + Thread.sleep(1); + } + } + + @Test + public void upstream() { + var s = Streamable.range(1, 5); + + var o = s.toObservable(); + + if (o instanceof HasUpstreamStreamableSource uso) { + assertSame(s, uso.source()); + } else { + fail(o.getClass() + " doesn't implement HasUpstreamStreamable or it is hidden."); + } + } + + @Test + public void disposable() { + var observer = new TestObserver<>(); + var cs = new CompositeDisposable(); + observer.onSubscribe(cs); + var sto = new StreamToObserver<>(Streamable.never().stream(cs), observer, cs, + new AtomicInteger(), new AtomicBoolean(), new AtomicReference<>(), new AtomicBoolean()); + cs.add(sto); + + assertFalse(sto.isDisposed(), "sto disposed?"); + + observer.dispose(); + + assertTrue(cs.isDisposed(), "cs not disposed?"); + assertTrue(sto.isDisposed(), "sto not disposed?"); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/util/ExceptionHelperTest.java b/src/test/java/io/reactivex/rxjava4/internal/util/ExceptionHelperTest.java index 9b487ebefac..bfa88cd969f 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/util/ExceptionHelperTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/util/ExceptionHelperTest.java @@ -15,12 +15,13 @@ import static org.junit.jupiter.api.Assertions.*; +import java.util.concurrent.CompletionException; import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Test; import io.reactivex.rxjava4.core.RxJavaTest; -import io.reactivex.rxjava4.exceptions.TestException; +import io.reactivex.rxjava4.exceptions.*; import io.reactivex.rxjava4.testsupport.TestHelper; public class ExceptionHelperTest extends RxJavaTest { @@ -49,4 +50,53 @@ public void throwIfThrowable() throws Exception { ExceptionHelper.throwIfThrowable(new InternalError()); }); } + + @Test + public void unwrapAndCombine1() { + assertNull(ExceptionHelper.unwrapAndCombine(null, null)); + } + + @Test + public void unwrapAndCombine2() { + var te = new TestException(); + assertSame(te, ExceptionHelper.unwrapAndCombine(te, null)); + } + + @Test + public void unwrapAndCombine3() { + var te = new TestException(); + assertSame(te, ExceptionHelper.unwrapAndCombine(null, te)); + } + + @Test + public void unwrapAndCombine4() { + var te = new TestException(); + var te2 = new TestException(); + assertSame(te, ExceptionHelper.unwrapAndCombine(te, te2)); + assertSame(te2, te.getSuppressed()[0]); + } + + @Test + public void unwrapAndCombine5() { + var te = new TestException(); + assertSame(te, ExceptionHelper.unwrapAndCombine(new CompletionException(te), null)); + } + + @Test + public void unwrapAndCombine6() { + var te = new TestException(); + assertSame(te, ExceptionHelper.unwrapAndCombine(null, new CompletionException(te))); + } + + @Test + public void unwrapAndCombine7() { + var te = new TestException(); + assertSame(te, ExceptionHelper.unwrapAndCombine(new ThrowableWrapper(te), null)); + } + + @Test + public void unwrapAndCombine8() { + var te = new TestException(); + assertSame(te, ExceptionHelper.unwrapAndCombine(null, new ThrowableWrapper(te))); + } } diff --git a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationNamingTest.java b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationNamingTest.java index 05f8c5183fd..3c9e7ca7414 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationNamingTest.java +++ b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationNamingTest.java @@ -177,6 +177,11 @@ public void checkCompositeDisposable() throws Exception { processFile(CompositeDisposable.class); } + @Test + public void checkStreamable() throws Exception { + processFile(Streamable.class); + } + static void processFile(Class clazz) throws Exception { String baseClassName = clazz.getSimpleName(); File f = TestHelper.findSource(baseClassName, clazz.getPackage().getName()); @@ -233,9 +238,11 @@ static void processFile(Class clazz) throws Exception { // find the method declaration for (; midx >= 0; midx--) { String linek = lines.get(midx).trim(); - if (linek.startsWith("public") || linek.startsWith("private") + if (linek.startsWith("public") + || linek.startsWith("private") || linek.startsWith("protected") || linek.startsWith("static") + || linek.startsWith("default") || linek.startsWith(baseClassName)) { break; } @@ -354,8 +361,11 @@ static void processFile(Class clazz) throws Exception { } } - if (line.startsWith("public") || line.startsWith("protected") || line.startsWith("final") || line.startsWith("private") - || line.startsWith("static")) { + if (line.startsWith("public") || line.startsWith("protected") + || line.startsWith("final") + || line.startsWith("private") + || line.startsWith("static") + || line.startsWith("default")) { for (ValidatorStrings validatorStr : TYPICAL_ARGUMENT_STRINGS) { // find the method declaration ending { for (int i = j; i < lines.size(); i++) {