From d210cfa4dab38d795939b28f8a1cb89645331c0f Mon Sep 17 00:00:00 2001 From: akarnokd Date: Tue, 23 Jun 2026 19:40:24 +0200 Subject: [PATCH 1/2] 4.x: Single API reduction: merge --- .../io/reactivex/rxjava4/core/Single.java | 481 ++---------------- .../core/config/SingleMergeConfig.java | 62 +++ .../flowable/FlowableCombineLatestTest.java | 8 +- .../single/SingleMergeArrayTest.java | 5 +- .../operators/single/SingleMergeTest.java | 29 +- .../reactivex/rxjava4/single/SingleTest.java | 2 +- .../ParamValidationCheckerTest.java | 1 + 7 files changed, 129 insertions(+), 459 deletions(-) create mode 100644 src/main/java/io/reactivex/rxjava4/core/config/SingleMergeConfig.java diff --git a/src/main/java/io/reactivex/rxjava4/core/Single.java b/src/main/java/io/reactivex/rxjava4/core/Single.java index ae281aac006..d207192b2ee 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Single.java +++ b/src/main/java/io/reactivex/rxjava4/core/Single.java @@ -13,144 +13,30 @@ package io.reactivex.rxjava4.core; -import java.util.NoSuchElementException; -import java.util.Objects; -import java.util.Optional; -import java.util.concurrent.Callable; -import java.util.concurrent.CancellationException; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CompletionStage; -import java.util.concurrent.Flow.Publisher; -import java.util.concurrent.Flow.Subscriber; -import java.util.concurrent.Future; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; -import java.util.stream.IntStream; -import java.util.stream.Stream; - -import io.reactivex.rxjava4.annotations.BackpressureKind; -import io.reactivex.rxjava4.annotations.BackpressureSupport; -import io.reactivex.rxjava4.annotations.CheckReturnValue; -import io.reactivex.rxjava4.annotations.NonNull; -import io.reactivex.rxjava4.annotations.Nullable; -import io.reactivex.rxjava4.annotations.SchedulerSupport; -import io.reactivex.rxjava4.core.config.SingleConcatConfig; -import io.reactivex.rxjava4.core.config.SingleConcatEagerConfig; -import io.reactivex.rxjava4.disposables.CompositeDisposable; -import io.reactivex.rxjava4.disposables.Disposable; -import io.reactivex.rxjava4.disposables.DisposableContainer; -import io.reactivex.rxjava4.exceptions.CompositeException; -import io.reactivex.rxjava4.exceptions.Exceptions; -import io.reactivex.rxjava4.exceptions.UndeliverableException; -import io.reactivex.rxjava4.functions.Action; -import io.reactivex.rxjava4.functions.BiConsumer; -import io.reactivex.rxjava4.functions.BiFunction; -import io.reactivex.rxjava4.functions.BiPredicate; -import io.reactivex.rxjava4.functions.BooleanSupplier; -import io.reactivex.rxjava4.functions.Cancellable; -import io.reactivex.rxjava4.functions.Consumer; -import io.reactivex.rxjava4.functions.Function; -import io.reactivex.rxjava4.functions.Function3; -import io.reactivex.rxjava4.functions.Function4; -import io.reactivex.rxjava4.functions.Function5; -import io.reactivex.rxjava4.functions.Function6; -import io.reactivex.rxjava4.functions.Function7; -import io.reactivex.rxjava4.functions.Function8; -import io.reactivex.rxjava4.functions.Function9; -import io.reactivex.rxjava4.functions.Predicate; -import io.reactivex.rxjava4.functions.Supplier; -import io.reactivex.rxjava4.internal.functions.Functions; -import io.reactivex.rxjava4.internal.functions.ObjectHelper; -import io.reactivex.rxjava4.internal.fuseable.FuseToFlowable; -import io.reactivex.rxjava4.internal.fuseable.FuseToMaybe; -import io.reactivex.rxjava4.internal.fuseable.FuseToObservable; -import io.reactivex.rxjava4.internal.jdk8.CompletionStageConsumer; -import io.reactivex.rxjava4.internal.jdk8.SingleFlattenStreamAsFlowable; -import io.reactivex.rxjava4.internal.jdk8.SingleFlattenStreamAsObservable; -import io.reactivex.rxjava4.internal.jdk8.SingleFromCompletionStage; -import io.reactivex.rxjava4.internal.jdk8.SingleMapOptional; -import io.reactivex.rxjava4.internal.observers.BiConsumerSingleObserver; -import io.reactivex.rxjava4.internal.observers.BlockingDisposableMultiObserver; -import io.reactivex.rxjava4.internal.observers.BlockingMultiObserver; -import io.reactivex.rxjava4.internal.observers.ConsumerSingleObserver; -import io.reactivex.rxjava4.internal.observers.DisposableAutoReleaseMultiObserver; -import io.reactivex.rxjava4.internal.observers.FutureMultiObserver; -import io.reactivex.rxjava4.internal.observers.SafeSingleObserver; -import io.reactivex.rxjava4.internal.operators.completable.CompletableFromSingle; -import io.reactivex.rxjava4.internal.operators.completable.CompletableToFlowable; -import io.reactivex.rxjava4.internal.operators.flowable.FlowableFlatMapSinglePublisher; -import io.reactivex.rxjava4.internal.operators.flowable.FlowableSingleSingle; -import io.reactivex.rxjava4.internal.operators.maybe.MaybeFilterSingle; -import io.reactivex.rxjava4.internal.operators.maybe.MaybeFromSingle; -import io.reactivex.rxjava4.internal.operators.maybe.MaybeToSingle; -import io.reactivex.rxjava4.internal.operators.mixed.FlowableConcatMapSinglePublisher; -import io.reactivex.rxjava4.internal.operators.mixed.FlowableSwitchMapSinglePublisher; -import io.reactivex.rxjava4.internal.operators.mixed.ObservableConcatMapSingle; -import io.reactivex.rxjava4.internal.operators.mixed.SingleFlatMapObservable; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.Flow.*; +import java.util.stream.*; + +import io.reactivex.rxjava4.annotations.*; +import io.reactivex.rxjava4.core.config.*; +import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.exceptions.*; +import io.reactivex.rxjava4.functions.*; +import io.reactivex.rxjava4.internal.functions.*; +import io.reactivex.rxjava4.internal.fuseable.*; +import io.reactivex.rxjava4.internal.jdk8.*; +import io.reactivex.rxjava4.internal.observers.*; +import io.reactivex.rxjava4.internal.operators.completable.*; +import io.reactivex.rxjava4.internal.operators.flowable.*; +import io.reactivex.rxjava4.internal.operators.maybe.*; +import io.reactivex.rxjava4.internal.operators.mixed.*; import io.reactivex.rxjava4.internal.operators.observable.ObservableSingleSingle; -import io.reactivex.rxjava4.internal.operators.single.SingleAmb; -import io.reactivex.rxjava4.internal.operators.single.SingleCache; -import io.reactivex.rxjava4.internal.operators.single.SingleContains; -import io.reactivex.rxjava4.internal.operators.single.SingleCreate; -import io.reactivex.rxjava4.internal.operators.single.SingleDefer; -import io.reactivex.rxjava4.internal.operators.single.SingleDelay; -import io.reactivex.rxjava4.internal.operators.single.SingleDelayWithCompletable; -import io.reactivex.rxjava4.internal.operators.single.SingleDelayWithObservable; -import io.reactivex.rxjava4.internal.operators.single.SingleDelayWithPublisher; -import io.reactivex.rxjava4.internal.operators.single.SingleDelayWithSingle; -import io.reactivex.rxjava4.internal.operators.single.SingleDematerialize; -import io.reactivex.rxjava4.internal.operators.single.SingleDetach; -import io.reactivex.rxjava4.internal.operators.single.SingleDoAfterSuccess; -import io.reactivex.rxjava4.internal.operators.single.SingleDoAfterTerminate; -import io.reactivex.rxjava4.internal.operators.single.SingleDoFinally; -import io.reactivex.rxjava4.internal.operators.single.SingleDoOnDispose; -import io.reactivex.rxjava4.internal.operators.single.SingleDoOnError; -import io.reactivex.rxjava4.internal.operators.single.SingleDoOnEvent; -import io.reactivex.rxjava4.internal.operators.single.SingleDoOnLifecycle; -import io.reactivex.rxjava4.internal.operators.single.SingleDoOnSubscribe; -import io.reactivex.rxjava4.internal.operators.single.SingleDoOnSuccess; -import io.reactivex.rxjava4.internal.operators.single.SingleDoOnTerminate; -import io.reactivex.rxjava4.internal.operators.single.SingleEquals; -import io.reactivex.rxjava4.internal.operators.single.SingleError; -import io.reactivex.rxjava4.internal.operators.single.SingleFlatMap; -import io.reactivex.rxjava4.internal.operators.single.SingleFlatMapBiSelector; -import io.reactivex.rxjava4.internal.operators.single.SingleFlatMapCompletable; -import io.reactivex.rxjava4.internal.operators.single.SingleFlatMapIterableFlowable; -import io.reactivex.rxjava4.internal.operators.single.SingleFlatMapIterableObservable; -import io.reactivex.rxjava4.internal.operators.single.SingleFlatMapMaybe; -import io.reactivex.rxjava4.internal.operators.single.SingleFlatMapNotification; -import io.reactivex.rxjava4.internal.operators.single.SingleFlatMapPublisher; -import io.reactivex.rxjava4.internal.operators.single.SingleFromCallable; -import io.reactivex.rxjava4.internal.operators.single.SingleFromPublisher; -import io.reactivex.rxjava4.internal.operators.single.SingleFromSupplier; -import io.reactivex.rxjava4.internal.operators.single.SingleFromUnsafeSource; -import io.reactivex.rxjava4.internal.operators.single.SingleHide; -import io.reactivex.rxjava4.internal.operators.single.SingleInternalHelper; -import io.reactivex.rxjava4.internal.operators.single.SingleJust; -import io.reactivex.rxjava4.internal.operators.single.SingleLift; -import io.reactivex.rxjava4.internal.operators.single.SingleMap; -import io.reactivex.rxjava4.internal.operators.single.SingleMaterialize; -import io.reactivex.rxjava4.internal.operators.single.SingleNever; -import io.reactivex.rxjava4.internal.operators.single.SingleObserveOn; -import io.reactivex.rxjava4.internal.operators.single.SingleOnErrorComplete; -import io.reactivex.rxjava4.internal.operators.single.SingleOnErrorReturn; -import io.reactivex.rxjava4.internal.operators.single.SingleResumeNext; -import io.reactivex.rxjava4.internal.operators.single.SingleSubscribeOn; -import io.reactivex.rxjava4.internal.operators.single.SingleTakeUntil; -import io.reactivex.rxjava4.internal.operators.single.SingleTimeInterval; -import io.reactivex.rxjava4.internal.operators.single.SingleTimeout; -import io.reactivex.rxjava4.internal.operators.single.SingleTimer; -import io.reactivex.rxjava4.internal.operators.single.SingleToFlowable; -import io.reactivex.rxjava4.internal.operators.single.SingleToObservable; -import io.reactivex.rxjava4.internal.operators.single.SingleUnsubscribeOn; -import io.reactivex.rxjava4.internal.operators.single.SingleUsing; -import io.reactivex.rxjava4.internal.operators.single.SingleZipArray; -import io.reactivex.rxjava4.internal.operators.single.SingleZipIterable; +import io.reactivex.rxjava4.internal.operators.single.*; import io.reactivex.rxjava4.internal.util.ErrorMode; import io.reactivex.rxjava4.observers.TestObserver; import io.reactivex.rxjava4.plugins.RxJavaPlugins; -import io.reactivex.rxjava4.schedulers.Schedulers; -import io.reactivex.rxjava4.schedulers.Timed; +import io.reactivex.rxjava4.schedulers.*; /** * The {@code Single} class implements the Reactive Pattern for a single value response. @@ -1210,8 +1096,6 @@ public abstract class Single<@NonNull T> implements SingleSource { * {@link RxJavaPlugins#onError(Throwable)} method as {@link UndeliverableException} errors. Similarly, {@code Throwable}s * signaled by source(s) after the returned {@code Flowable} has been cancelled or terminated with a * (composite) error will be sent to the same global error handler. - * Use {@link #mergeDelayError(Iterable)} to merge sources and terminate only when all source {@code SingleSource}s - * have completed or failed with an error. * * * @param the common and resulting value type @@ -1219,7 +1103,6 @@ public abstract class Single<@NonNull T> implements SingleSource { * @return the new {@code Flowable} instance * @throws NullPointerException if {@code sources} is {@code null} * @since 2.0 - * @see #mergeDelayError(Iterable) */ @CheckReturnValue @NonNull @@ -1249,15 +1132,12 @@ public abstract class Single<@NonNull T> implements SingleSource { * {@link RxJavaPlugins#onError(Throwable)} method as {@link UndeliverableException} errors. Similarly, {@code Throwable}s * signaled by source(s) after the returned {@code Flowable} has been cancelled or terminated with a * (composite) error will be sent to the same global error handler. - * Use {@link #mergeDelayError(Publisher)} to merge sources and terminate only when all source {@code SingleSource}s - * have completed or failed with an error. * * * @param the common and resulting value type * @param sources the {@code Publisher} emitting a sequence of {@code SingleSource}s * @return the new {@code Flowable} instance * @throws NullPointerException if {@code sources} is {@code null} - * @see #mergeDelayError(Publisher) * @since 2.0 */ @CheckReturnValue @@ -1300,164 +1180,6 @@ public abstract class Single<@NonNull T> implements SingleSource { return RxJavaPlugins.onAssembly(new SingleFlatMap, T>(source, Functions.identity())); } - /** - * Flattens two {@link SingleSource}s into one {@link Flowable} sequence, without any transformation. - *

- * - *

- * You can combine items emitted by multiple {@code SingleSource}s so that they appear as a single {@code Flowable}, by - * using the {@code merge} method. - *

- *
Backpressure:
- *
The returned {@code Flowable} honors the backpressure of the downstream consumer.
- *
Scheduler:
- *
{@code merge} does not operate by default on a particular {@link Scheduler}.
- *
Error handling:
- *
If any of the source {@code SingleSource}s signal a {@link Throwable} via {@code onError}, the resulting - * {@code Flowable} terminates with that {@code Throwable} and all other source {@code SingleSource}s are disposed. - * If more than one {@code SingleSource} signals an error, the resulting {@code Flowable} may terminate with the - * first one's error or, depending on the concurrency of the sources, may terminate with a - * {@link CompositeException} containing two or more of the various error signals. - * {@code Throwable}s that didn't make into the composite will be sent (individually) to the global error handler via - * {@link RxJavaPlugins#onError(Throwable)} method as {@link UndeliverableException} errors. Similarly, {@code Throwable}s - * signaled by source(s) after the returned {@code Flowable} has been cancelled or terminated with a - * (composite) error will be sent to the same global error handler. - * Use {@link #mergeDelayError(SingleSource, SingleSource)} to merge sources and terminate only when all source {@code SingleSource}s - * have completed or failed with an error. - *
- *
- * - * @param the common value type - * @param source1 - * a {@code SingleSource} to be merged - * @param source2 - * a {@code SingleSource} to be merged - * @return the new {@code Flowable} that emits all of the items emitted by the source {@code SingleSource}s - * @throws NullPointerException if {@code source1} or {@code source2} is {@code null} - * @see ReactiveX operators documentation: Merge - * @see #mergeDelayError(SingleSource, SingleSource) - */ - @CheckReturnValue - @NonNull - @BackpressureSupport(BackpressureKind.FULL) - @SchedulerSupport(SchedulerSupport.NONE) - public static <@NonNull T> Flowable merge( - @NonNull SingleSource source1, @NonNull SingleSource source2 - ) { - Objects.requireNonNull(source1, "source1 is null"); - Objects.requireNonNull(source2, "source2 is null"); - return Flowable.fromArray(source1, source2).flatMapSingle(Functions.identity(), false, Integer.MAX_VALUE); - } - - /** - * Flattens three {@link SingleSource}s into one {@link Flowable} sequence, without any transformation. - *

- * - *

- * You can combine items emitted by multiple {@code SingleSource}s so that they appear as a single {@code Flowable}, by - * the {@code merge} method. - *

- *
Backpressure:
- *
The returned {@code Flowable} honors the backpressure of the downstream consumer.
- *
Scheduler:
- *
{@code merge} does not operate by default on a particular {@link Scheduler}.
- *
Error handling:
- *
If any of the source {@code SingleSource}s signal a {@link Throwable} via {@code onError}, the resulting - * {@code Flowable} terminates with that {@code Throwable} and all other source {@code SingleSource}s are disposed. - * If more than one {@code SingleSource} signals an error, the resulting {@code Flowable} may terminate with the - * first one's error or, depending on the concurrency of the sources, may terminate with a - * {@link CompositeException} containing two or more of the various error signals. - * {@code Throwable}s that didn't make into the composite will be sent (individually) to the global error handler via - * {@link RxJavaPlugins#onError(Throwable)} method as {@link UndeliverableException} errors. Similarly, {@code Throwable}s - * signaled by source(s) after the returned {@code Flowable} has been cancelled or terminated with a - * (composite) error will be sent to the same global error handler. - * Use {@link #mergeDelayError(SingleSource, SingleSource, SingleSource)} to merge sources and terminate only when all source {@code SingleSource}s - * have completed or failed with an error. - *
- *
- * - * @param the common value type - * @param source1 - * a {@code SingleSource} to be merged - * @param source2 - * a {@code SingleSource} to be merged - * @param source3 - * a {@code SingleSource} to be merged - * @return the new {@code Flowable} that emits all of the items emitted by the source {@code SingleSource}s - * @throws NullPointerException if {@code source1}, {@code source2} or {@code source3} is {@code null} - * @see ReactiveX operators documentation: Merge - * @see #mergeDelayError(SingleSource, SingleSource, SingleSource) - */ - @CheckReturnValue - @NonNull - @BackpressureSupport(BackpressureKind.FULL) - @SchedulerSupport(SchedulerSupport.NONE) - public static <@NonNull T> Flowable merge( - @NonNull SingleSource source1, @NonNull SingleSource source2, - @NonNull SingleSource source3 - ) { - Objects.requireNonNull(source1, "source1 is null"); - Objects.requireNonNull(source2, "source2 is null"); - Objects.requireNonNull(source3, "source3 is null"); - return Flowable.fromArray(source1, source2, source3).flatMapSingle(Functions.identity(), false, Integer.MAX_VALUE); - } - - /** - * Flattens four {@link SingleSource}s into one {@link Flowable} sequence, without any transformation. - *

- * - *

- * You can combine items emitted by multiple {@code SingleSource}s so that they appear as a single {@code Flowable}, by - * the {@code merge} method. - *

- *
Backpressure:
- *
The returned {@code Flowable} honors the backpressure of the downstream consumer.
- *
Scheduler:
- *
{@code merge} does not operate by default on a particular {@link Scheduler}.
- *
Error handling:
- *
If any of the source {@code SingleSource}s signal a {@link Throwable} via {@code onError}, the resulting - * {@code Flowable} terminates with that {@code Throwable} and all other source {@code SingleSource}s are disposed. - * If more than one {@code SingleSource} signals an error, the resulting {@code Flowable} may terminate with the - * first one's error or, depending on the concurrency of the sources, may terminate with a - * {@link CompositeException} containing two or more of the various error signals. - * {@code Throwable}s that didn't make into the composite will be sent (individually) to the global error handler via - * {@link RxJavaPlugins#onError(Throwable)} method as {@link UndeliverableException} errors. Similarly, {@code Throwable}s - * signaled by source(s) after the returned {@code Flowable} has been cancelled or terminated with a - * (composite) error will be sent to the same global error handler. - * Use {@link #mergeDelayError(SingleSource, SingleSource, SingleSource, SingleSource)} to merge sources and terminate only when all source {@code SingleSource}s - * have completed or failed with an error. - *
- *
- * - * @param the common value type - * @param source1 - * a {@code SingleSource} to be merged - * @param source2 - * a {@code SingleSource} to be merged - * @param source3 - * a {@code SingleSource} to be merged - * @param source4 - * a {@code SingleSource} to be merged - * @return the new {@code Flowable} that emits all of the items emitted by the source {@code SingleSource}s - * @throws NullPointerException if {@code source1}, {@code source2}, {@code source3} or {@code source4} is {@code null} - * @see ReactiveX operators documentation: Merge - * @see #mergeDelayError(SingleSource, SingleSource, SingleSource, SingleSource) - */ - @CheckReturnValue - @NonNull - @BackpressureSupport(BackpressureKind.FULL) - @SchedulerSupport(SchedulerSupport.NONE) - public static <@NonNull T> Flowable merge( - @NonNull SingleSource source1, @NonNull SingleSource source2, - @NonNull SingleSource source3, @NonNull SingleSource source4 - ) { - Objects.requireNonNull(source1, "source1 is null"); - Objects.requireNonNull(source2, "source2 is null"); - Objects.requireNonNull(source3, "source3 is null"); - Objects.requireNonNull(source4, "source4 is null"); - return Flowable.fromArray(source1, source2, source3, source4).flatMapSingle(Functions.identity(), false, Integer.MAX_VALUE); - } - /** * Merges an array of {@link SingleSource} instances into a single {@link Flowable} sequence, * running all {@code SingleSource}s at once. @@ -1478,15 +1200,12 @@ public abstract class Single<@NonNull T> implements SingleSource { * {@link RxJavaPlugins#onError(Throwable)} method as {@link UndeliverableException} errors. Similarly, {@code Throwable}s * signaled by source(s) after the returned {@code Flowable} has been cancelled or terminated with a * (composite) error will be sent to the same global error handler. - * Use {@link #mergeArrayDelayError(SingleSource...)} to merge sources and terminate only when all source {@code SingleSource}s - * have completed or failed with an error. * * * @param the common and resulting value type * @param sources the array sequence of {@code SingleSource} sources * @return the new {@code Flowable} instance * @throws NullPointerException if {@code sources} is {@code null} - * @see #mergeArrayDelayError(SingleSource...) */ @BackpressureSupport(BackpressureKind.FULL) @CheckReturnValue @@ -1518,19 +1237,21 @@ public abstract class Single<@NonNull T> implements SingleSource { * * * @param the common element base type - * @param sources - * the array of {@code SingleSource}s + * @param config the configuration record for this operator + * @param sources the array of {@code SingleSource}s * @return the new {@code Flowable} instance - * @throws NullPointerException if {@code sources} is {@code null} + * @throws NullPointerException if {@code sources} or {@code config} is {@code null} * @see ReactiveX operators documentation: Merge + * @since 4.0.0 */ @BackpressureSupport(BackpressureKind.FULL) @CheckReturnValue @SchedulerSupport(SchedulerSupport.NONE) @SafeVarargs @NonNull - public static <@NonNull T> Flowable mergeArrayDelayError(@NonNull SingleSource... sources) { - return Flowable.fromArray(sources).flatMapSingle(Functions.identity(), true, Math.max(1, sources.length)); + public static <@NonNull T> Flowable mergeArray(@NonNull SingleMergeConfig config, @NonNull SingleSource... sources) { + Objects.requireNonNull(config, "config is null"); + return Flowable.fromArray(sources).flatMapSingle(Functions.identity(), config.delayErrors(), config.maxConcurrency()); } /** @@ -1547,17 +1268,19 @@ public abstract class Single<@NonNull T> implements SingleSource { *

History: 2.1.9 - experimental * @param the common and resulting value type * @param sources the {@code Iterable} sequence of {@code SingleSource}s + * @param config the configuration record for this operator * @return the new {@code Flowable} instance - * @throws NullPointerException if {@code sources} is {@code null} + * @throws NullPointerException if {@code sources} or {@code config} is {@code null} * @see #merge(Iterable) - * @since 2.2 + * @since 4.0.0 */ @CheckReturnValue @NonNull @BackpressureSupport(BackpressureKind.FULL) @SchedulerSupport(SchedulerSupport.NONE) - public static <@NonNull T> Flowable mergeDelayError(@NonNull Iterable<@NonNull ? extends SingleSource> sources) { - return Flowable.fromIterable(sources).flatMapSingle(Functions.identity(), true, Integer.MAX_VALUE); + public static <@NonNull T> Flowable merge(@NonNull Iterable<@NonNull ? extends SingleSource> sources, @NonNull SingleMergeConfig config) { + Objects.requireNonNull(config, "config is null"); + return Flowable.fromIterable(sources).flatMapSingle(Functions.identity(), config.delayErrors(), config.maxConcurrency()); } /** @@ -1574,143 +1297,20 @@ public abstract class Single<@NonNull T> implements SingleSource { *

History: 2.1.9 - experimental * @param the common and resulting value type * @param sources the {@code Flowable} sequence of {@code SingleSource}s + * @param config the configuration record for this operator * @return the new {@code Flowable} instance - * @throws NullPointerException if {@code sources} is {@code null} - * @since 2.2 + * @throws NullPointerException if {@code sources} or {@code config} is {@code null} + * @since 4.0.0 * @see #merge(Publisher) */ @CheckReturnValue @NonNull @BackpressureSupport(BackpressureKind.FULL) @SchedulerSupport(SchedulerSupport.NONE) - public static <@NonNull T> Flowable mergeDelayError(@NonNull Publisher<@NonNull ? extends SingleSource> sources) { + public static <@NonNull T> Flowable merge(@NonNull Publisher<@NonNull ? extends SingleSource> sources, @NonNull SingleMergeConfig config) { Objects.requireNonNull(sources, "sources is null"); - return RxJavaPlugins.onAssembly(new FlowableFlatMapSinglePublisher<>(sources, Functions.identity(), true, Integer.MAX_VALUE)); - } - - /** - * Flattens two {@link SingleSource}s into one {@link Flowable}, without any transformation, delaying - * any error(s) until all sources succeed or fail. - *

- * - *

- * You can combine items emitted by multiple {@code SingleSource}s so that they appear as one {@code Flowable}, by - * using the {@code mergeDelayError} method. - *

- *
Backpressure:
- *
The returned {@code Flowable} honors the backpressure of the downstream consumer.
- *
Scheduler:
- *
{@code mergeDelayError} does not operate by default on a particular {@link Scheduler}.
- *
- *

History: 2.1.9 - experimental - * @param the common value type - * @param source1 - * a {@code SingleSource} to be merged - * @param source2 - * a {@code SingleSource} to be merged - * @return the new {@code Flowable} that emits all of the items emitted by the source {@code SingleSource}s - * @throws NullPointerException if {@code source1} or {@code source2} is {@code null} - * @see ReactiveX operators documentation: Merge - * @see #merge(SingleSource, SingleSource) - * @since 2.2 - */ - @CheckReturnValue - @NonNull - @BackpressureSupport(BackpressureKind.FULL) - @SchedulerSupport(SchedulerSupport.NONE) - public static <@NonNull T> Flowable mergeDelayError( - @NonNull SingleSource source1, @NonNull SingleSource source2 - ) { - Objects.requireNonNull(source1, "source1 is null"); - Objects.requireNonNull(source2, "source2 is null"); - return Flowable.fromArray(source1, source2).flatMapSingle(Functions.identity(), true, Integer.MAX_VALUE); - } - - /** - * Flattens two {@link SingleSource}s into one {@link Flowable}, without any transformation, delaying - * any error(s) until all sources succeed or fail. - *

- * - *

- * You can combine items emitted by multiple {@code SingleSource}s so that they appear as one {@code Flowable}, by - * the {@code mergeDelayError} method. - *

- *
Backpressure:
- *
The returned {@code Flowable} honors the backpressure of the downstream consumer.
- *
Scheduler:
- *
{@code mergeDelayError} does not operate by default on a particular {@link Scheduler}.
- *
- *

History: 2.1.9 - experimental - * @param the common value type - * @param source1 - * a {@code SingleSource} to be merged - * @param source2 - * a {@code SingleSource} to be merged - * @param source3 - * a {@code SingleSource} to be merged - * @return the new {@code Flowable} that emits all of the items emitted by the source {@code SingleSource}s - * @throws NullPointerException if {@code source1}, {@code source2} or {@code source3} is {@code null} - * @see ReactiveX operators documentation: Merge - * @see #merge(SingleSource, SingleSource, SingleSource) - * @since 2.2 - */ - @CheckReturnValue - @NonNull - @BackpressureSupport(BackpressureKind.FULL) - @SchedulerSupport(SchedulerSupport.NONE) - public static <@NonNull T> Flowable mergeDelayError( - @NonNull SingleSource source1, @NonNull SingleSource source2, - @NonNull SingleSource source3 - ) { - Objects.requireNonNull(source1, "source1 is null"); - Objects.requireNonNull(source2, "source2 is null"); - Objects.requireNonNull(source3, "source3 is null"); - return Flowable.fromArray(source1, source2, source3).flatMapSingle(Functions.identity(), true, Integer.MAX_VALUE); - } - - /** - * Flattens two {@link SingleSource}s into one {@link Flowable}, without any transformation, delaying - * any error(s) until all sources succeed or fail. - *

- * - *

- * You can combine items emitted by multiple {@code SingleSource}s so that they appear as one {@code Flowable}, by - * the {@code mergeDelayError} method. - *

- *
Backpressure:
- *
The returned {@code Flowable} honors the backpressure of the downstream consumer.
- *
Scheduler:
- *
{@code mergeDelayError} does not operate by default on a particular {@link Scheduler}.
- *
- *

History: 2.1.9 - experimental - * @param the common value type - * @param source1 - * a {@code SingleSource} to be merged - * @param source2 - * a {@code SingleSource} to be merged - * @param source3 - * a {@code SingleSource} to be merged - * @param source4 - * a {@code SingleSource} to be merged - * @return the new {@code Flowable} that emits all of the items emitted by the source {@code SingleSource}s - * @throws NullPointerException if {@code source1}, {@code source2}, {@code source3} or {@code source4} is {@code null} - * @see ReactiveX operators documentation: Merge - * @see #merge(SingleSource, SingleSource, SingleSource, SingleSource) - * @since 2.2 - */ - @CheckReturnValue - @NonNull - @BackpressureSupport(BackpressureKind.FULL) - @SchedulerSupport(SchedulerSupport.NONE) - public static <@NonNull T> Flowable mergeDelayError( - @NonNull SingleSource source1, @NonNull SingleSource source2, - @NonNull SingleSource source3, @NonNull SingleSource source4 - ) { - Objects.requireNonNull(source1, "source1 is null"); - Objects.requireNonNull(source2, "source2 is null"); - Objects.requireNonNull(source3, "source3 is null"); - Objects.requireNonNull(source4, "source4 is null"); - return Flowable.fromArray(source1, source2, source3, source4).flatMapSingle(Functions.identity(), true, Integer.MAX_VALUE); + Objects.requireNonNull(config, "config is null"); + return RxJavaPlugins.onAssembly(new FlowableFlatMapSinglePublisher<>(sources, Functions.identity(), config.delayErrors(), config.maxConcurrency())); } /** @@ -3962,7 +3562,8 @@ public final Single contains(@NonNull Object item, @NonNull BiPredicate @SchedulerSupport(SchedulerSupport.NONE) @NonNull public final Flowable mergeWith(@NonNull SingleSource other) { - return merge(this, other); + Objects.requireNonNull(other, "other is null"); + return mergeArray(this, other); } /** * Filters the items emitted by the current {@code Single}, only emitting its success value if that diff --git a/src/main/java/io/reactivex/rxjava4/core/config/SingleMergeConfig.java b/src/main/java/io/reactivex/rxjava4/core/config/SingleMergeConfig.java new file mode 100644 index 00000000000..5efeb5a8ee5 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/core/config/SingleMergeConfig.java @@ -0,0 +1,62 @@ +/* + * 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.core.config; + +import io.reactivex.rxjava4.internal.functions.ObjectHelper; + +/** + * Configuration record for Single.merge() operators. + * @param delayErrors should the error propagation be delayed? + * @param maxConcurrency the number of source sequences run concurrently + * @since 4.0.0 + */ +public record SingleMergeConfig(boolean delayErrors, int maxConcurrency) { + + /** + * The default config with no error delay and Integer#MAX_VALUE as the maximum concurrency setting. + */ + public static final SingleMergeConfig DEFAULT = new SingleMergeConfig(false, Integer.MAX_VALUE); + + /** + * The default config with error delay and Integer#MAX_VALUE as the maximum concurrency setting. + */ + public static final SingleMergeConfig DELAY_ERRORS = new SingleMergeConfig(true, Integer.MAX_VALUE); + + /** + * Constructs a configuration record. + * @param delayErrors should the error propagation be delayed? + */ + public SingleMergeConfig(boolean delayErrors) { + this(delayErrors, Integer.MAX_VALUE); + } + + /** + * Constructs a configuration record. + * @param maxConcurrency the number of source sequences run concurrently + */ + public SingleMergeConfig(int maxConcurrency) { + this(false, maxConcurrency); + } + + /** + * Constructs a configuration record. + * @param delayErrors should the error propagation be delayed? + * @param maxConcurrency the number of source sequences run concurrently + */ + public SingleMergeConfig(boolean delayErrors, int maxConcurrency) { + ObjectHelper.verifyPositive(maxConcurrency, "maxConcurrency"); + this.delayErrors = delayErrors; + this.maxConcurrency = maxConcurrency; + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableCombineLatestTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableCombineLatestTest.java index b3c61d1dd72..2ca1b6b8837 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableCombineLatestTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableCombineLatestTest.java @@ -995,16 +995,16 @@ public void combineLatestDelayErrorIterableOfSourcesWithError() { .assertFailure(TestException.class, "[1, 2]"); } - // @SuppressWarnings("unchecked") + @SuppressWarnings({ "unchecked", "rawtypes" }) @Test public void combineLatestArrayEmpty() { - assertSame(Flowable.empty(), Flowable.combineLatestArray(new Flowable[0], Functions.identity(), 16)); + assertSame(Flowable.empty(), Flowable.combineLatestArray(new Flowable[0], (Function)Functions.identity(), 16)); } - // @SuppressWarnings("unchecked") + @SuppressWarnings({ "unchecked", "rawtypes" }) @Test public void combineLatestDelayErrorEmpty() { - assertSame(Flowable.empty(), Flowable.combineLatestArrayDelayError(new Flowable[0], Functions.identity(), 16)); + assertSame(Flowable.empty(), Flowable.combineLatestArrayDelayError(new Flowable[0], (Function)Functions.identity(), 16)); } @Test diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/single/SingleMergeArrayTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/single/SingleMergeArrayTest.java index be18ca2b79a..bfbb35a28d0 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/single/SingleMergeArrayTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/single/SingleMergeArrayTest.java @@ -16,6 +16,7 @@ import org.junit.Test; import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.core.config.SingleMergeConfig; import io.reactivex.rxjava4.exceptions.TestException; public class SingleMergeArrayTest extends RxJavaTest { @@ -36,14 +37,14 @@ public void error() { @Test public void normalDelayError() { - Single.mergeArrayDelayError(Single.just(1), Single.just(2), Single.just(3)) + Single.mergeArray(SingleMergeConfig.DELAY_ERRORS, Single.just(1), Single.just(2), Single.just(3)) .test() .assertResult(1, 2, 3); } @Test public void errorDelayError() { - Single.mergeArrayDelayError(Single.just(1), Single.error(new TestException()), Single.just(3)) + Single.mergeArray(SingleMergeConfig.DELAY_ERRORS, Single.just(1), Single.error(new TestException()), Single.just(3)) .test() .assertFailure(TestException.class, 1, 3); } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/single/SingleMergeTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/single/SingleMergeTest.java index 6f3bac3cb50..6c8c7d7cb9a 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/single/SingleMergeTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/single/SingleMergeTest.java @@ -20,6 +20,7 @@ import org.junit.Test; import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.core.config.SingleMergeConfig; import io.reactivex.rxjava4.exceptions.TestException; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.testsupport.TestHelper; @@ -36,21 +37,21 @@ public void mergeSingleSingle() { @Test public void merge2() { - Single.merge(Single.just(1), Single.just(2)) + Single.mergeArray(Single.just(1), Single.just(2)) .test() .assertResult(1, 2); } @Test public void merge3() { - Single.merge(Single.just(1), Single.just(2), Single.just(3)) + Single.mergeArray(Single.just(1), Single.just(2), Single.just(3)) .test() .assertResult(1, 2, 3); } @Test public void merge4() { - Single.merge(Single.just(1), Single.just(2), Single.just(3), Single.just(4)) + Single.mergeArray(Single.just(1), Single.just(2), Single.just(3), Single.just(4)) .test() .assertResult(1, 2, 3, 4); } @@ -62,7 +63,7 @@ public void mergeErrors() { Single source1 = Single.error(new TestException("First")); Single source2 = Single.error(new TestException("Second")); - Single.merge(source1, source2) + Single.mergeArray(source1, source2) .to(TestHelper.testConsumer()) .assertFailureAndMessage(TestException.class, "First"); @@ -74,10 +75,10 @@ public void mergeErrors() { @Test public void mergeDelayErrorIterable() { - Single.mergeDelayError(Arrays.asList( + Single.merge(Arrays.asList( Single.just(1), Single.error(new TestException()), - Single.just(2)) + Single.just(2)), SingleMergeConfig.DELAY_ERRORS ) .test() .assertFailure(TestException.class, 1, 2); @@ -85,10 +86,10 @@ public void mergeDelayErrorIterable() { @Test public void mergeDelayErrorPublisher() { - Single.mergeDelayError(Flowable.just( + Single.merge(Flowable.just( Single.just(1), Single.error(new TestException()), - Single.just(2)) + Single.just(2)), SingleMergeConfig.DELAY_ERRORS ) .test() .assertFailure(TestException.class, 1, 2); @@ -96,7 +97,8 @@ public void mergeDelayErrorPublisher() { @Test public void mergeDelayError2() { - Single.mergeDelayError( + Single.mergeArray( + SingleMergeConfig.DELAY_ERRORS, Single.just(1), Single.error(new TestException()) ) @@ -106,7 +108,8 @@ public void mergeDelayError2() { @Test public void mergeDelayError2ErrorFirst() { - Single.mergeDelayError( + Single.mergeArray( + SingleMergeConfig.DELAY_ERRORS, Single.error(new TestException()), Single.just(1) ) @@ -116,7 +119,8 @@ public void mergeDelayError2ErrorFirst() { @Test public void mergeDelayError3() { - Single.mergeDelayError( + Single.mergeArray( + SingleMergeConfig.DELAY_ERRORS, Single.just(1), Single.error(new TestException()), Single.just(2) @@ -127,7 +131,8 @@ public void mergeDelayError3() { @Test public void mergeDelayError4() { - Single.mergeDelayError( + Single.mergeArray( + SingleMergeConfig.DELAY_ERRORS, Single.just(1), Single.error(new TestException()), Single.just(2), diff --git a/src/test/java/io/reactivex/rxjava4/single/SingleTest.java b/src/test/java/io/reactivex/rxjava4/single/SingleTest.java index 6c4ade0964d..9c704326385 100644 --- a/src/test/java/io/reactivex/rxjava4/single/SingleTest.java +++ b/src/test/java/io/reactivex/rxjava4/single/SingleTest.java @@ -99,7 +99,7 @@ public void merge() { Single a = Single.just("A"); Single b = Single.just("B"); - Single.merge(a, b).subscribe(ts); + Single.mergeArray(a, b).subscribe(ts); ts.assertValueSequence(Arrays.asList("A", "B")); } diff --git a/src/test/java/io/reactivex/rxjava4/validators/ParamValidationCheckerTest.java b/src/test/java/io/reactivex/rxjava4/validators/ParamValidationCheckerTest.java index 47f4a0c2b79..7d1f532ddb1 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/ParamValidationCheckerTest.java +++ b/src/test/java/io/reactivex/rxjava4/validators/ParamValidationCheckerTest.java @@ -622,6 +622,7 @@ public void checkParallelFlowable() { defaultValues.put(SingleConcatConfig.class, SingleConcatConfig.DEFAULT); defaultValues.put(SingleConcatEagerConfig.class, SingleConcatEagerConfig.DEFAULT); + defaultValues.put(SingleMergeConfig.class, SingleMergeConfig.DEFAULT); @SuppressWarnings("rawtypes") class MixedConverters implements FlowableConverter, ObservableConverter, SingleConverter, From 27daaf2e1a0f4eb08c074e8ccaa58dc1b65eea4d Mon Sep 17 00:00:00 2001 From: akarnokd Date: Tue, 23 Jun 2026 19:43:36 +0200 Subject: [PATCH 2/2] Test the SingleMergeConfig --- .../core/config/SingleMergeConfigTest.java | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) create mode 100644 src/test/java/io/reactivex/rxjava4/core/config/SingleMergeConfigTest.java diff --git a/src/test/java/io/reactivex/rxjava4/core/config/SingleMergeConfigTest.java b/src/test/java/io/reactivex/rxjava4/core/config/SingleMergeConfigTest.java new file mode 100644 index 00000000000..a18833f66cf --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/core/config/SingleMergeConfigTest.java @@ -0,0 +1,34 @@ +/* + * 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.core.config; + +import static org.junit.jupiter.api.Assertions.*; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.core.RxJavaTest; + +public class SingleMergeConfigTest extends RxJavaTest { + + @Test + public void validation() { + assertTrue(new SingleMergeConfig(true).delayErrors(), "delayErrors - true"); + assertFalse(new SingleMergeConfig(false).delayErrors(), "delayErrors - false"); + assertEquals(5, new SingleMergeConfig(5).maxConcurrency(), "maxConcurrency - 5"); + assertEquals(5, new SingleMergeConfig(true, 5).maxConcurrency(), "maxConcurrency both - true, 5"); + assertEquals(5, new SingleMergeConfig(false, 5).maxConcurrency(), "maxConcurrency both - false, 5"); + assertTrue(new SingleMergeConfig(true, 5).delayErrors(), "delayErrors both - true, 5"); + assertFalse(new SingleMergeConfig(false, 5).delayErrors(), "delayErrors both - false, 5"); + } +}