diff --git a/gradle/javadoc_cleanup.gradle b/gradle/javadoc_cleanup.gradle index 9ee8d2ecb9..e1fd47d827 100644 --- a/gradle/javadoc_cleanup.gradle +++ b/gradle/javadoc_cleanup.gradle @@ -6,8 +6,8 @@ task javadocCleanup(dependsOn: "javadoc") doLast { fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/core/Maybe.html')) fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/core/Completable.html')) - fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/flowables/ConnectableFlowable.html')) - fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/observables/ConnectableObservable.html')) + fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/core/ConnectableFlowable.html')) + fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/core/ConnectableObservable.html')) fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/subjects/ReplaySubject.html')) fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/processors/ReplayProcessor.html')) diff --git a/src/jmh/java/io/reactivex/rxjava4/parallel/ParallelPerf.java b/src/jmh/java/io/reactivex/rxjava4/parallel/ParallelPerf.java index e4f3621602..d6d7ecf678 100644 --- a/src/jmh/java/io/reactivex/rxjava4/parallel/ParallelPerf.java +++ b/src/jmh/java/io/reactivex/rxjava4/parallel/ParallelPerf.java @@ -22,7 +22,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.core.config.StandardConcurrentBufferedConfig; -import io.reactivex.rxjava4.flowables.GroupedFlowable; import io.reactivex.rxjava4.functions.Function; import io.reactivex.rxjava4.schedulers.Schedulers; diff --git a/src/main/java/io/reactivex/rxjava4/flowables/ConnectableFlowable.java b/src/main/java/io/reactivex/rxjava4/core/ConnectableFlowable.java similarity index 99% rename from src/main/java/io/reactivex/rxjava4/flowables/ConnectableFlowable.java rename to src/main/java/io/reactivex/rxjava4/core/ConnectableFlowable.java index f64042f20a..eb0f23048e 100644 --- a/src/main/java/io/reactivex/rxjava4/flowables/ConnectableFlowable.java +++ b/src/main/java/io/reactivex/rxjava4/core/ConnectableFlowable.java @@ -11,15 +11,13 @@ * the License for the specific language governing permissions and limitations under the License. */ -package io.reactivex.rxjava4.flowables; +package io.reactivex.rxjava4.core; import java.util.Objects; +import java.util.concurrent.Flow.*; import java.util.concurrent.TimeUnit; -import static java.util.concurrent.Flow.*; - import io.reactivex.rxjava4.annotations.*; -import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.functions.Consumer; import io.reactivex.rxjava4.internal.functions.*; diff --git a/src/main/java/io/reactivex/rxjava4/observables/ConnectableObservable.java b/src/main/java/io/reactivex/rxjava4/core/ConnectableObservable.java similarity index 99% rename from src/main/java/io/reactivex/rxjava4/observables/ConnectableObservable.java rename to src/main/java/io/reactivex/rxjava4/core/ConnectableObservable.java index 8c4948f2e9..9fc1d50810 100644 --- a/src/main/java/io/reactivex/rxjava4/observables/ConnectableObservable.java +++ b/src/main/java/io/reactivex/rxjava4/core/ConnectableObservable.java @@ -11,13 +11,12 @@ * the License for the specific language governing permissions and limitations under the License. */ -package io.reactivex.rxjava4.observables; +package io.reactivex.rxjava4.core; import java.util.Objects; import java.util.concurrent.TimeUnit; import io.reactivex.rxjava4.annotations.*; -import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.functions.Consumer; import io.reactivex.rxjava4.internal.functions.*; diff --git a/src/main/java/io/reactivex/rxjava4/core/Flowable.java b/src/main/java/io/reactivex/rxjava4/core/Flowable.java index 083b562caa..7b65f90cdf 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Flowable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Flowable.java @@ -22,7 +22,6 @@ import io.reactivex.rxjava4.core.config.*; import io.reactivex.rxjava4.disposables.*; import io.reactivex.rxjava4.exceptions.*; -import io.reactivex.rxjava4.flowables.*; import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.functions.*; import io.reactivex.rxjava4.internal.jdk8.*; diff --git a/src/main/java/io/reactivex/rxjava4/flowables/GroupedFlowable.java b/src/main/java/io/reactivex/rxjava4/core/GroupedFlowable.java similarity index 96% rename from src/main/java/io/reactivex/rxjava4/flowables/GroupedFlowable.java rename to src/main/java/io/reactivex/rxjava4/core/GroupedFlowable.java index 0f8c70dd0b..adb42a6e2a 100644 --- a/src/main/java/io/reactivex/rxjava4/flowables/GroupedFlowable.java +++ b/src/main/java/io/reactivex/rxjava4/core/GroupedFlowable.java @@ -11,10 +11,9 @@ * the License for the specific language governing permissions and limitations under the License. */ -package io.reactivex.rxjava4.flowables; +package io.reactivex.rxjava4.core; import io.reactivex.rxjava4.annotations.Nullable; -import io.reactivex.rxjava4.core.Flowable; /** * A {@link Flowable} that has been grouped by key, the value of which can be obtained with {@link #getKey()}. diff --git a/src/main/java/io/reactivex/rxjava4/observables/GroupedObservable.java b/src/main/java/io/reactivex/rxjava4/core/GroupedObservable.java similarity index 95% rename from src/main/java/io/reactivex/rxjava4/observables/GroupedObservable.java rename to src/main/java/io/reactivex/rxjava4/core/GroupedObservable.java index be0833b4e0..17e7249310 100644 --- a/src/main/java/io/reactivex/rxjava4/observables/GroupedObservable.java +++ b/src/main/java/io/reactivex/rxjava4/core/GroupedObservable.java @@ -11,10 +11,9 @@ * the License for the specific language governing permissions and limitations under the License. */ -package io.reactivex.rxjava4.observables; +package io.reactivex.rxjava4.core; import io.reactivex.rxjava4.annotations.Nullable; -import io.reactivex.rxjava4.core.Observable; /** * An {@link Observable} that has been grouped by key, the value of which can be obtained with {@link #getKey()}. diff --git a/src/main/java/io/reactivex/rxjava4/core/GroupedStreamable.java b/src/main/java/io/reactivex/rxjava4/core/GroupedStreamable.java new file mode 100644 index 0000000000..464aa81ec7 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/core/GroupedStreamable.java @@ -0,0 +1,50 @@ +/* + * 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; + +import io.reactivex.rxjava4.annotations.*; + +/** + * An {@link Streamable} that has been grouped by key, the value of which can be obtained with {@link #getKey()}. + * @param + * the type of the key, can be null + * @param + * the type of the items emitted by the {@code GroupedStreamable} + * @see Streamable#groupBy(io.reactivex.rxjava4.functions.Function) + * @see ReactiveX documentation: GroupBy + * @since 4.0.0 + */ +public abstract class GroupedStreamable<@Nullable K, @NonNull T> implements Streamable { + + final K key; + + /** + * Constructs a GroupedStreamable with the given key. + * @param key the key + */ + protected GroupedStreamable(@Nullable K key) { + this.key = key; + } + + /** + * Returns the key that identifies the group of items emitted by this {@code GroupedStreamable}. + * + * @return the key that the items emitted by this {@code GroupedStreamable} were grouped by + */ + @Nullable + public K getKey() { + return key; + } + +} diff --git a/src/main/java/io/reactivex/rxjava4/core/Observable.java b/src/main/java/io/reactivex/rxjava4/core/Observable.java index fe6234bedd..fb68d40c57 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Observable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Observable.java @@ -32,7 +32,6 @@ import io.reactivex.rxjava4.internal.operators.observable.*; import io.reactivex.rxjava4.internal.operators.single.SingleToObservable; import io.reactivex.rxjava4.internal.util.*; -import io.reactivex.rxjava4.observables.*; import io.reactivex.rxjava4.observers.*; import io.reactivex.rxjava4.operators.ScalarSupplier; import io.reactivex.rxjava4.plugins.RxJavaPlugins; diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamable.java b/src/main/java/io/reactivex/rxjava4/core/Streamable.java index 2480799b92..13c23591f1 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamable.java @@ -572,6 +572,20 @@ default Streamable flatMap(@NonNull Function(this, mapper, config.maxConcurrency())); } + /** + * Maps each upstream item into a {@code GroupedStreamable} group, emits those groups and keeps + * relaying the upstream items into those groups. + * @param the key type, {@code null}s allowed + * @param keySelector the function that receives the upstream item and returns a key that determines + * which group the item will go into + * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code keySelector} is {@code null} + */ + default <@Nullable K> Streamable> groupBy(Function keySelector) { + Objects.requireNonNull(keySelector, "keySelector is null"); + return RxJavaPlugins.onAssembly(new StreamableGroupBy<>(this, keySelector)); + } + /** * Hides the identity of this {@code Streamable} and its {@link Streamer}. *

@@ -688,6 +702,21 @@ default Streamable takeWhile(@NonNull Predicate predicate) { return RxJavaPlugins.onAssembly(new StreamableTakeWhile<>(this, predicate)); } + /** + * Relays items from this {@code Streamable} until the other {@code Streamable} signals + * an item or completes. + * @param the element type of the other {@code Streamable} + * @param other the {@code Streamable} expected to signal when to stop taking items from this {@code Streamable} + * @return the new {@code Streamable} instance + * @throws NullPointerException if {@code other} is {@code null} + */ + @CheckReturnValue + @NonNull + default Streamable takeUntil(@NonNull Streamable other) { + Objects.requireNonNull(other, "other is null"); + return RxJavaPlugins.onAssembly(new StreamableTakeUntil<>(this, other)); + } + /** * Calls the specified converter function during assembly time and returns its resulting value. *

diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamer.java b/src/main/java/io/reactivex/rxjava4/core/Streamer.java index c5dcc5c087..8aef3ec26b 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamer.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamer.java @@ -39,7 +39,7 @@ public interface Streamer<@NonNull T> { *

    *
  • `true` indicates there is an item available for consumption via {@link #current()} *
  • `false` indicates there are no more items available - *< li>`Throwable` indicates there was an upstream error + *
  • `Throwable` indicates there was an upstream error *
*/ @NonNull @@ -101,17 +101,17 @@ default void awaitFinish() { * Use this constant in {@link #next()} to indicate * the next value is available, synchronously. */ - CompletionStage NEXT_TRUE = CompletableFuture.completedStage(true); + CompletableFuture NEXT_TRUE = CompletableFuture.completedFuture(true); /** * Use this constant in {@link #next()} to indicate * no more values will be available, synchronously. */ - CompletionStage NEXT_FALSE = CompletableFuture.completedStage(false); + CompletableFuture NEXT_FALSE = CompletableFuture.completedFuture(false); /** * Use this constant in {@link #finish()} to indicate * the cleanup was done synchronously. */ - CompletionStage FINISHED = CompletableFuture.completedStage(null); + CompletableFuture FINISHED = CompletableFuture.completedFuture(null); } diff --git a/src/main/java/io/reactivex/rxjava4/core/package-info.java b/src/main/java/io/reactivex/rxjava4/core/package-info.java index 8431d69d7f..cbe0b0c099 100644 --- a/src/main/java/io/reactivex/rxjava4/core/package-info.java +++ b/src/main/java/io/reactivex/rxjava4/core/package-info.java @@ -48,6 +48,14 @@ * consumers to subscribe to them and receive events.

*

Usage examples can be found on the {@link io.reactivex.rxjava4.core.Flowable}/{@link io.reactivex.rxjava4.core.Observable} * and {@link java.util.concurrent.Flow.Subscriber} classes.

+ *

+ * Classes supporting the Flowable base reactive class: + * {@link io.reactivex.rxjava4.core.ConnectableFlowable} and + * {@link io.reactivex.rxjava4.core.GroupedFlowable}. + *

+ * Classes supporting the Observable base reactive class: + * {@link io.reactivex.rxjava4.core.ConnectableObservable} and + * {@link io.reactivex.rxjava4.core.GroupedObservable}. */ package io.reactivex.rxjava4.core; diff --git a/src/main/java/io/reactivex/rxjava4/exceptions/ThrowableWrapper.java b/src/main/java/io/reactivex/rxjava4/exceptions/ThrowableWrapper.java index f23c10844e..ad0ed95953 100644 --- a/src/main/java/io/reactivex/rxjava4/exceptions/ThrowableWrapper.java +++ b/src/main/java/io/reactivex/rxjava4/exceptions/ThrowableWrapper.java @@ -14,6 +14,7 @@ package io.reactivex.rxjava4.exceptions; import java.io.Serial; +import java.util.concurrent.CompletionException; /** * A runtime exception to sneak around checked exceptions. @@ -22,7 +23,7 @@ * at the right place. * @since 4.0.0 */ -public final class ThrowableWrapper extends RuntimeException { +public final class ThrowableWrapper extends CompletionException { @Serial private static final long serialVersionUID = -5280780582536857320L; diff --git a/src/main/java/io/reactivex/rxjava4/flowables/package-info.java b/src/main/java/io/reactivex/rxjava4/flowables/package-info.java deleted file mode 100644 index dd2c8dc86c..0000000000 --- a/src/main/java/io/reactivex/rxjava4/flowables/package-info.java +++ /dev/null @@ -1,19 +0,0 @@ -/* - * 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. - */ - -/** - * Classes supporting the Flowable base reactive class: - * {@link io.reactivex.rxjava4.flowables.ConnectableFlowable} and - * {@link io.reactivex.rxjava4.flowables.GroupedFlowable}. - */ -package io.reactivex.rxjava4.flowables; diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableAutoConnect.java b/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableAutoConnect.java index 5342cf93d4..f22a50a434 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableAutoConnect.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableAutoConnect.java @@ -13,13 +13,11 @@ package io.reactivex.rxjava4.internal.operators.flowable; +import java.util.concurrent.Flow.Subscriber; import java.util.concurrent.atomic.AtomicInteger; -import static java.util.concurrent.Flow.*; - -import io.reactivex.rxjava4.core.Flowable; +import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.Disposable; -import io.reactivex.rxjava4.flowables.ConnectableFlowable; import io.reactivex.rxjava4.functions.Consumer; /** diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableGroupBy.java b/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableGroupBy.java index db58e5aae2..c3c2304c7b 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableGroupBy.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableGroupBy.java @@ -16,14 +16,12 @@ import java.io.Serial; import java.util.*; import java.util.concurrent.*; +import java.util.concurrent.Flow.*; import java.util.concurrent.atomic.*; -import static java.util.concurrent.Flow.*; - import io.reactivex.rxjava4.annotations.Nullable; import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.exceptions.*; -import io.reactivex.rxjava4.flowables.GroupedFlowable; import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.subscriptions.*; import io.reactivex.rxjava4.internal.util.*; diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableInternalHelper.java b/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableInternalHelper.java index 9b99d402ca..4eca40257e 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableInternalHelper.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableInternalHelper.java @@ -14,14 +14,12 @@ package io.reactivex.rxjava4.internal.operators.flowable; import java.util.Objects; +import java.util.concurrent.Flow.*; import java.util.concurrent.TimeUnit; -import static java.util.concurrent.Flow.*; - import io.reactivex.rxjava4.core.*; -import io.reactivex.rxjava4.flowables.ConnectableFlowable; import io.reactivex.rxjava4.functions.*; -import io.reactivex.rxjava4.internal.functions.*; +import io.reactivex.rxjava4.internal.functions.Functions; /** * Helper utility class to support Flowable with inner classes. diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowablePublish.java b/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowablePublish.java index 72becc1281..46c4ffe962 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowablePublish.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowablePublish.java @@ -14,21 +14,17 @@ package io.reactivex.rxjava4.internal.operators.flowable; import java.io.Serial; +import java.util.concurrent.Flow.*; import java.util.concurrent.atomic.*; -import static java.util.concurrent.Flow.*; - -import io.reactivex.rxjava4.core.FlowableSubscriber; +import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.exceptions.*; -import io.reactivex.rxjava4.flowables.ConnectableFlowable; import io.reactivex.rxjava4.functions.Consumer; -import io.reactivex.rxjava4.internal.fuseable.*; +import io.reactivex.rxjava4.internal.fuseable.HasUpstreamPublisher; import io.reactivex.rxjava4.internal.subscriptions.SubscriptionHelper; import io.reactivex.rxjava4.internal.util.*; -import io.reactivex.rxjava4.operators.QueueSubscription; -import io.reactivex.rxjava4.operators.SimpleQueue; -import io.reactivex.rxjava4.operators.SpscArrayQueue; +import io.reactivex.rxjava4.operators.*; import io.reactivex.rxjava4.plugins.RxJavaPlugins; /** diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableRefCount.java b/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableRefCount.java index f47c6970eb..8470c57458 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableRefCount.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableRefCount.java @@ -14,14 +14,12 @@ package io.reactivex.rxjava4.internal.operators.flowable; import java.io.Serial; +import java.util.concurrent.Flow.*; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.*; -import static java.util.concurrent.Flow.*; - import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.Disposable; -import io.reactivex.rxjava4.flowables.ConnectableFlowable; import io.reactivex.rxjava4.functions.Consumer; import io.reactivex.rxjava4.internal.disposables.*; import io.reactivex.rxjava4.internal.subscriptions.SubscriptionHelper; diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableReplay.java b/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableReplay.java index 5e8c01eebf..74baae5b3a 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableReplay.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableReplay.java @@ -15,15 +15,13 @@ import java.io.Serial; import java.util.*; +import java.util.concurrent.Flow.*; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.*; -import static java.util.concurrent.Flow.*; - import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.exceptions.Exceptions; -import io.reactivex.rxjava4.flowables.ConnectableFlowable; import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.fuseable.HasUpstreamPublisher; import io.reactivex.rxjava4.internal.subscribers.SubscriberResourceWrapper; diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableAutoConnect.java b/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableAutoConnect.java index 7dbd8ae2d2..b6aa66b6a1 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableAutoConnect.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableAutoConnect.java @@ -18,7 +18,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.functions.Consumer; -import io.reactivex.rxjava4.observables.ConnectableObservable; /** * Wraps a ConnectableObservable and calls its connect() method once diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableGroupBy.java b/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableGroupBy.java index 1d780cf720..a181b2fe44 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableGroupBy.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableGroupBy.java @@ -18,13 +18,12 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.*; -import io.reactivex.rxjava4.core.ObservableSource; +import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.core.Observer; import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.exceptions.Exceptions; import io.reactivex.rxjava4.functions.Function; import io.reactivex.rxjava4.internal.disposables.*; -import io.reactivex.rxjava4.observables.GroupedObservable; import io.reactivex.rxjava4.operators.SpscLinkedArrayQueue; public final class ObservableGroupBy extends AbstractObservableWithUpstream> { diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableInternalHelper.java b/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableInternalHelper.java index b110fcb470..28695a4893 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableInternalHelper.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableInternalHelper.java @@ -18,8 +18,7 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.functions.*; -import io.reactivex.rxjava4.internal.functions.*; -import io.reactivex.rxjava4.observables.ConnectableObservable; +import io.reactivex.rxjava4.internal.functions.Functions; /** * Helper utility class to support Observable with inner classes. diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservablePublish.java b/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservablePublish.java index 4c7e7b6d61..723c36d830 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservablePublish.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservablePublish.java @@ -23,7 +23,6 @@ import io.reactivex.rxjava4.internal.disposables.DisposableHelper; import io.reactivex.rxjava4.internal.fuseable.HasUpstreamObservableSource; import io.reactivex.rxjava4.internal.util.ExceptionHelper; -import io.reactivex.rxjava4.observables.ConnectableObservable; import io.reactivex.rxjava4.plugins.RxJavaPlugins; /** diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableRefCount.java b/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableRefCount.java index 9aad2986e2..05c9c6e8ab 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableRefCount.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableRefCount.java @@ -21,7 +21,6 @@ import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.functions.Consumer; import io.reactivex.rxjava4.internal.disposables.*; -import io.reactivex.rxjava4.observables.ConnectableObservable; import io.reactivex.rxjava4.plugins.RxJavaPlugins; /** diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableReplay.java b/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableReplay.java index a565bd8e29..381cfeccaf 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableReplay.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableReplay.java @@ -27,7 +27,6 @@ import io.reactivex.rxjava4.internal.disposables.*; import io.reactivex.rxjava4.internal.fuseable.HasUpstreamObservableSource; import io.reactivex.rxjava4.internal.util.*; -import io.reactivex.rxjava4.observables.ConnectableObservable; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.schedulers.Timed; diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StageResumable.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StageResumable.java new file mode 100644 index 0000000000..6a5a43dadc --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StageResumable.java @@ -0,0 +1,67 @@ +/* + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.rxjava4.internal.operators.streamable; + +import java.io.Serial; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BiConsumer; + +import io.reactivex.rxjava4.annotations.*; + +public final class StageResumable extends AtomicReference> +implements BiConsumer { + + @Serial + private static final long serialVersionUID = -7518852864146380895L; + + @CheckReturnValue + @NonNull + public CompletableFuture ready() { + CompletableFuture cf; + for (;;) { + cf = get(); + if (cf != null) { + break; + } + cf = new CompletableFuture<>(); + if (compareAndSet(null, cf)) { + break; + } + } + return cf; + } + + @CheckReturnValue + @NonNull + public CompletableFuture await() { + CompletableFuture cf; + for (;;) { + cf = get(); + if (cf != null) { + break; + } + cf = new CompletableFuture<>(); + if (compareAndSet(null, cf)) { + break; + } + } + return cf.whenComplete(this); + } + + @Override + public void accept(T t, Throwable u) { + getAndSet(null); + } +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDefer.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDefer.java index d1833cb9fb..93730b9fbf 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDefer.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDefer.java @@ -13,6 +13,8 @@ package io.reactivex.rxjava4.internal.operators.streamable; +import java.util.Objects; + import io.reactivex.rxjava4.annotations.NonNull; import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.DisposableContainer; @@ -25,7 +27,8 @@ public record StreamableDefer(Supplier> sup @Override public @NonNull Streamer<@NonNull T> stream(@NonNull DisposableContainer cancellation) { try { - return (Streamer)(supplier.get().stream(cancellation)); + return (Streamer)(Objects.requireNonNull(supplier.get(), "The supplier returned a null Streamable") + .stream(cancellation)); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); return StreamableError.createFailed(ex); diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableGroupBy.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableGroupBy.java new file mode 100644 index 0000000000..2af74e9d9b --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableGroupBy.java @@ -0,0 +1,303 @@ +/* + * Copyright (c) 2016-present, RxJava Contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in + * compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is + * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See + * the License for the specific language governing permissions and limitations under the License. + */ + +package io.reactivex.rxjava4.internal.operators.streamable; + +import java.util.Map; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; +import java.util.function.BiConsumer; + +import io.reactivex.rxjava4.annotations.*; +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.exceptions.Exceptions; +import io.reactivex.rxjava4.functions.Function; +import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; + +public record StreamableGroupBy(Streamable source, Function keySelector) +implements Streamable>, HasUpstreamStreamableSource { + + @Override + public @NonNull Streamer> stream(@NonNull DisposableContainer cancellation) { + var streamer = new GroupByStreamer(source.stream(cancellation), keySelector); + streamer.drain(); + return streamer; + } + + static final class GroupByStreamer + implements Streamer>, + BiConsumer, java.util.function.Function { + + final Streamer upstream; + + final Function keySelector; + + final AtomicInteger wip; + + final Map> groups; + + final CompletableFuture onFinish; + + volatile boolean done; + + volatile GroupedStreamable currentGroup; + + volatile GroupedStreamable currentNext; + + final StageResumable mainNext; + + GroupByStreamer(Streamer upstream, Function keySelector) { + this.upstream = upstream; + this.keySelector = keySelector; + this.wip = new AtomicInteger(); + this.groups = new ConcurrentHashMap<>(); + this.mainNext = new StageResumable<>(); + this.onFinish = new CompletableFuture<>(); + } + + @Override + public @NonNull CompletionStage next() { + var cs = mainNext.await().thenApply(this); + return cs; + } + + @Override + public Boolean apply(Boolean v) { + if (v) { + currentNext = currentGroup; + } else { + currentNext = null; + } + currentGroup = null; + return v; + } + + @Override + public @NonNull GroupedStreamable current() { + return currentNext; + } + + @Override + public @NonNull CompletionStage finish() { + done = true; + drain(); + return onFinish; + } + + void drain() { + if (wip.getAndIncrement() != 0) { + return; + } + + do { + if (done) { + upstream.finish().whenComplete(this); + break; + } else { + upstream.next().whenComplete(this); + } + } while (wip.decrementAndGet() != 0); + } + + @Override + public void accept(Object t, Throwable u) { + if (done) { + upstream.finish().whenComplete((v, e) -> { + currentNext = null; + currentGroup = null; + if (e != null) { + onFinish.completeExceptionally(e); + } else { + onFinish.complete(v); + } + }); + } else { + if (u != null) { + done = true; + for (var g : groups.values()) { + g.terminate(u); // TODO whenComplete + } + groups.clear(); + mainNext.ready().completeExceptionally(u); + } else + if ((Boolean)t) { + try { + var c = upstream.current(); + var key = keySelector.apply(c); + + var g = groups.get(key); + if (g == null) { + g = new AsyncGroup<>(key, this); + groups.put(key, g); + currentGroup = g; + mainNext.ready().complete(true); + } + g.send(c).whenComplete((_, _) -> drain()); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + done = true; + for (var g : groups.values()) { + g.terminate(ex); + } + groups.clear(); + mainNext.ready().completeExceptionally(ex); + } + } else { + done = true; + for (var g : groups.values()) { + g.terminate(null); + } + groups.clear(); + mainNext.ready().complete(false); + } + } + } + + @SuppressWarnings("unchecked") + void delete(K key) { + groups.replace(key, (BasicGroupedStreamable)TombstoneGroup.INSTANCE); + } + + boolean isDeleted(K key) { + var e = groups.get(key); + return e == null || e == TombstoneGroup.INSTANCE; + } + } + + static abstract class BasicGroupedStreamable extends GroupedStreamable { + BasicGroupedStreamable(K key) { + super(key); + } + + abstract CompletionStage send(@NonNull T value); + + abstract CompletionStage terminate(@Nullable Throwable throwable); + } + + static final class AsyncGroup extends BasicGroupedStreamable + implements Streamer, Disposable, java.util.function.Function { + + final AtomicBoolean once; + + final GroupByStreamer parent; + + final StageResumable sendCanProgress; + + final StageResumable nextCanProgress; + + volatile T item; + + volatile T current; + + DisposableContainer cancellation; + + AsyncGroup(K key, GroupByStreamer parent) { + super(key); + this.once = new AtomicBoolean(); + this.parent = parent; + this.sendCanProgress = new StageResumable<>(); + this.nextCanProgress = new StageResumable<>(); + } + + @Override + public @NonNull Streamer<@NonNull T> stream(@NonNull DisposableContainer cancellation) { + if (once.compareAndSet(false, true)) { + this.cancellation = cancellation; + cancellation.add(this); + return this; + } + return StreamableError.createFailed(new IllegalStateException("Only one streamer is allowed!")); + } + + @Override + CompletionStage send(T value) { + return sendCanProgress.await().thenAccept(_ -> { + item = value; + nextCanProgress.ready().complete(true); + }); + } + + @Override + CompletionStage terminate(@Nullable Throwable throwable) { + return sendCanProgress.await().thenAccept(_ -> { + if (throwable == null) { + nextCanProgress.ready().complete(false); + } else { + nextCanProgress.ready().completeExceptionally(throwable); + } + }); + } + + @Override + public @NonNull CompletionStage next() { + sendCanProgress.ready().complete(true); + return nextCanProgress.await().thenApply(this); + } + + @Override + public @NonNull Boolean apply(@NonNull Boolean b) { + current = item; + item = null; + return b; + } + + @Override + public @NonNull T current() { + return current; + } + + @Override + public @NonNull CompletionStage finish() { + this.cancellation.remove(this); + return FINISHED; + } + + @Override + public void dispose() { + parent.delete(getKey()); + } + + @Override + public boolean isDisposed() { + // TODO Auto-generated method stub + return parent.isDeleted(getKey()); + } + } + + static final class TombstoneGroup extends BasicGroupedStreamable { + + static final Object TOMBSTONE_KEY = new Object(); + + static final TombstoneGroup INSTANCE = new TombstoneGroup(); + + TombstoneGroup() { + super(TOMBSTONE_KEY); + } + + @Override + public @NonNull Streamer<@NonNull Object> stream(@NonNull DisposableContainer cancellation) { + return StreamableError.createFailed(new CancellationException("TOMBSTONE")); + } + + @Override + CompletionStage send(Object value) { + return Streamer.FINISHED; + } + + @Override + CompletionStage terminate(@Nullable Throwable throwable) { + return Streamer.FINISHED; + } + } +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelper.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelper.java new file mode 100644 index 0000000000..30ce177cef --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelper.java @@ -0,0 +1,211 @@ +/* + * 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 java.util.concurrent.atomic.AtomicInteger; +import java.util.function.*; + +import io.reactivex.rxjava4.annotations.*; +import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.disposables.Disposable; +import io.reactivex.rxjava4.exceptions.CompositeException; +import io.reactivex.rxjava4.internal.util.AtomicThrowable; + +/** + * Helper static methods for {@link Streamable}s, {@link CompletableFuture}s and {@link CompletionStage}s. + * @since 4.0.0 + */ +public enum StreamableHelper { + ; + + /** + * Checks which source completes first, calls the given acceptor with 1 or 2 indicating the winner, + * then terminates the resulting {@link CompletableFuture} with said terminal event. + *

+ * Use the {@code thenAcceptor} to do something about the winner and/or loser before the terminal + * signal is propagated further + * @param the common element type of the stages + * @param first the first stage to wait for + * @param second the second stage to wait for + * @param thenAcceptor the callback that receives who won + * @return the new {@code CompletableFuture} that gets completed with the winner's event. + */ + @CheckReturnValue + @NonNull + public static CompletableFuture race(CompletionStage first, CompletionStage second, IntConsumer thenAcceptor) { + var result = new CompletableFuture(); + var winner = new AtomicInteger(); + first.whenComplete((v, e) -> { + if (winner.compareAndSet(0, 1)) { + thenAcceptor.accept(1); + if (e != null) { + result.completeExceptionally(e); + } else { + result.complete(v); + } + } + }); + second.whenComplete((v, e) -> { + if (winner.compareAndSet(0, 2)) { + thenAcceptor.accept(2); + if (e != null) { + result.completeExceptionally(e); + } else { + result.complete(v); + } + } + }); + + return result; + } + + /** + * Creates a {@link CompletableFuture} that completes when both sources complete in a way, + * relaying one or the other's exception, or both exceptions via {@link CompositeException}. + * @param one a stage to wait for completion + * @param two a stage to wait for completion + * @return then new {@code CompletableFuture} instance + */ + @CheckReturnValue + @NonNull + public static CompletableFuture whenBoth(CompletionStage one, CompletionStage two) { + var result = new CompletableFuture(); + var errors = new AtomicThrowable(); + var wip = new AtomicInteger(2); + BiConsumer handler = (_, e) -> { + if (e != null) { + errors.tryAddThrowableOrReport(e); + } + if (wip.decrementAndGet() == 0) { + var err = errors.get(); + if (err != null) { + result.completeExceptionally(err); + } else { + result.complete(null); + } + } + }; + one.whenComplete(handler); + two.whenComplete(handler); + return result; + } + + /** + * Creates a {@link CompletableFuture} that completes when the second completes after the first completes, + * chaining their completion into a sequence and relays either or both's exception as is or via a + * {@link CompositeException}. + * @param first the first stage to wait for completion + * @param supplier produces the next stage when the first stage completes in any way + * @return the new {@code CompletableFuture} instance + */ + @CheckReturnValue + @NonNull + public static CompletableFuture andThenSupply(@NonNull CompletionStage first, + @NonNull Supplier<@NonNull ? extends CompletionStage> supplier) { + var result = new CompletableFuture(); + var errors = new AtomicThrowable(); + first.whenComplete((_, e) -> { + if (e != null) { + errors.tryAddThrowableOrReport(e); + } + supplier.get().whenComplete((_, e1) -> { + if (e1 != null) { + errors.tryAddThrowableOrReport(e1); + } + var err = errors.get(); + if (err != null) { + result.completeExceptionally(err); + } else { + result.complete(null); + } + }); + }); + return result; + } + + /** + * Check if the throwable is itself a {@link CancellationException} or + * it is a {@link CompletionException} caused by a {@code CancellationException}. + *

+ * Why is that {@code completableFuture.completeExceptionally(new CancellationException())} + * results in the wrapping? no idea. + * @param e the throwable to check + * @return true if the throwable holds a cancellation exception in some fashion + */ + @CheckReturnValue + static boolean isCancellation(@Nullable Throwable e) { + return e instanceof CancellationException + || (e instanceof CompletionException && e.getCause() instanceof CancellationException); + } + + /** + * Creates a {@link CompletableFuture} that relays the terminal events of the source {@code stage} + * except any {@link CancellationException} if the given {@link Disposable#isDisposed()} returns {@code true}. + *

+ * Use it to suppress expected cancellation errors. + * @param the element type of the stage + * @param stage the original stage to gate the cancellation exception of + * @param disposable the {@code Disposable} to check + * @param defaultValue to signal if the cancellation exception was gated + * @return the new {@code CompletableFuture} instance + */ + @CheckReturnValue + @NonNull + public static CompletableFuture suppressCancel( + @NonNull CompletionStage stage, @NonNull Disposable disposable, T defaultValue) { + var result = new CompletableFuture(); + stage.whenComplete((v, e) -> { + if (disposable.isDisposed() && isCancellation(e)) { + result.complete(defaultValue); + } else + if (e != null) { + result.completeExceptionally(e); + } else { + result.complete(v); + } + }); + return result; + } + + /** + * Creates a {@link CompletableFuture} that suppresses the source value or cancellation exception and + * replaces them with a normal completion signaling the {@code defaultValue}. + *

+ * Use it to suppress expected cancellation errors and any source value. + * @param the element type of the stage + * @param stage the original stage to gate the cancellation exception of + * @param disposable the {@code Disposable} to check + * @param defaultValue to signal if the cancellation exception was gated + * @return the new {@code CompletableFuture} instance + */ + @CheckReturnValue + @NonNull + public static CompletableFuture suppressValueAndCancel( + @NonNull CompletionStage stage, + @NonNull Disposable disposable, T defaultValue) { + var result = new CompletableFuture(); + stage.whenComplete((_, e) -> { + if (disposable.isDisposed() && isCancellation(e)) { // FIXME coverage possible even? + result.complete(defaultValue); + } else + if (e != null) { + result.completeExceptionally(e); + } else { + result.complete(defaultValue); + } + }); + return result; + } +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeUntil.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeUntil.java new file mode 100644 index 0000000000..9eb1c7dd1e --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeUntil.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 static io.reactivex.rxjava4.internal.operators.streamable.StreamableHelper.*; + +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.internal.fuseable.HasUpstreamStreamableSource; + +public record StreamableTakeUntil( + Streamable source, + Streamable other +) implements Streamable, HasUpstreamStreamableSource { + + @Override + public @NonNull Streamer<@NonNull T> stream(@NonNull DisposableContainer cancellation) { + var otherCancellation = cancellation.derive(); + var otherStreamer = other.stream(otherCancellation); + var mainCancellation = cancellation.derive(); + var mainStreamer = source.stream(mainCancellation); + + var streamer = new TakeUntilMainStreamer<>( + mainStreamer, otherStreamer, mainCancellation, otherCancellation, + suppressValueAndCancel(otherStreamer.next(), otherCancellation, false) + ); + + return streamer; + } + + static final class TakeUntilMainStreamer implements Streamer { + + final Streamer upstream; + + final Streamer otherStreamer; + + final DisposableContainer mainCancellation; + + final DisposableContainer otherCancellation; + + final CompletionStage otherNext; + + CompletionStage mainNext; + + public TakeUntilMainStreamer( + Streamer upstream, + Streamer otherStreamer, + DisposableContainer mainCancellation, + DisposableContainer otherCancellation, + CompletionStage otherNext + ) { + this.upstream = upstream; + this.otherStreamer = otherStreamer; + this.mainCancellation = mainCancellation; + this.otherCancellation = otherCancellation; + this.otherNext = otherNext; + } + + @Override + public @NonNull CompletionStage next() { + mainNext = suppressCancel(upstream.next(), mainCancellation, false); + return StreamableHelper.race(mainNext, otherNext, w -> { + if (w == 2) { + mainCancellation.dispose(); + } + }); + } + + @Override + public @NonNull T current() { + return upstream.current(); + } + + @Override + public @NonNull CompletionStage finish() { + otherCancellation.dispose(); + return andThenSupply( + whenBoth(mainNext, otherNext), + () -> whenBoth(upstream.finish(), otherStreamer.finish()) + ); + } + } +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/util/AtomicThrowable.java b/src/main/java/io/reactivex/rxjava4/internal/util/AtomicThrowable.java index 8b18324913..33d3e1e763 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/util/AtomicThrowable.java +++ b/src/main/java/io/reactivex/rxjava4/internal/util/AtomicThrowable.java @@ -131,7 +131,7 @@ public void tryTerminateConsumer(MaybeObserver consumer) { /** * Tries to terminate this atomic throwable (by swapping in the TERMINATED indicator) - * and notifies the consumer if there was no error (onComplete) or there was a + * and notifies the consumer if there was a * non-null, non-indicator exception contained before (onError). * If there was a terminated indicator, the consumer is not signaled. * @param consumer the consumer to notify diff --git a/src/main/java/io/reactivex/rxjava4/observables/package-info.java b/src/main/java/io/reactivex/rxjava4/observables/package-info.java deleted file mode 100644 index 6ffba99b70..0000000000 --- a/src/main/java/io/reactivex/rxjava4/observables/package-info.java +++ /dev/null @@ -1,19 +0,0 @@ -/* - * 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. - */ - -/** - * Classes supporting the Observable base reactive class: - * {@link io.reactivex.rxjava4.observables.ConnectableObservable} and - * {@link io.reactivex.rxjava4.observables.GroupedObservable}. - */ -package io.reactivex.rxjava4.observables; diff --git a/src/main/java/io/reactivex/rxjava4/observers/BaseTestConsumer.java b/src/main/java/io/reactivex/rxjava4/observers/BaseTestConsumer.java index f410eb7d66..abf78fdd44 100644 --- a/src/main/java/io/reactivex/rxjava4/observers/BaseTestConsumer.java +++ b/src/main/java/io/reactivex/rxjava4/observers/BaseTestConsumer.java @@ -86,6 +86,30 @@ public final List values() { return values; } + /** + * Returns a shared list of received {@code onError} errors or the single {@code onError} value. + *

+ * Note that accessing the items via certain methods of the {@link List} + * interface while the upstream is still actively emitting + * more items may result in a {@code ConcurrentModificationException}. + *

+ * The {@link List#size()} method will return the number of items + * already received by this {@code TestObserver}/{@code TestSubscriber} in a thread-safe + * manner that can be read via {@link List#get(int)}) method + * (index range of 0 to {@code List.size() - 1}). + *

+ * A view of the returned List can be created via {@link List#subList(int, int)} + * by using the bounds 0 (inclusive) to {@link List#size()} (exclusive) which, + * when accessed in a read-only fashion, should be also thread-safe and not throw any + * {@code ConcurrentModificationException}. + * @return a list of received onError values + * @since 4.0.0 + */ + @NonNull + public final List errors() { + return errors; + } + /** * Fail with the given message and add the sequence of errors as suppressed ones. *

Note this is deliberately the only fail method. Usually an assertion @@ -97,6 +121,20 @@ public final List values() { */ @NonNull protected final AssertionError fail(@NonNull String message) { + + AssertionError ae = new AssertionError(createStateText(message).toString()); + if (!errors.isEmpty()) { + if (errors.size() == 1) { + ae.initCause(errors.getFirst()); + } else { + CompositeException ce = new CompositeException(errors); + ae.initCause(ce); + } + } + return ae; + } + + private StringBuilder createStateText(String message) { StringBuilder b = new StringBuilder(64 + message.length()); b.append(message); @@ -124,17 +162,12 @@ protected final AssertionError fail(@NonNull String message) { b .append(')') ; + return b; + } - AssertionError ae = new AssertionError(b.toString()); - if (!errors.isEmpty()) { - if (errors.size() == 1) { - ae.initCause(errors.getFirst()); - } else { - CompositeException ce = new CompositeException(errors); - ae.initCause(ce); - } - } - return ae; + @Override + public String toString() { + return createStateText(getClass().getSimpleName()).toString(); } /** diff --git a/src/main/java/io/reactivex/rxjava4/observers/TestObserver.java b/src/main/java/io/reactivex/rxjava4/observers/TestObserver.java index 96689bb5a9..968b6ee53a 100644 --- a/src/main/java/io/reactivex/rxjava4/observers/TestObserver.java +++ b/src/main/java/io/reactivex/rxjava4/observers/TestObserver.java @@ -32,7 +32,8 @@ * *

Since 4.0.0, the {@code TestObserver} does implements {@link Disposable} * anymore. Use {@link #asDisposable()} to create a wrapper that calls the {@link #dispose()}. - * @implNote Avoids all the resource warnings because {@code Disposable} implements {@link AutoCloseable} now. + * Implementation note
+ * Avoids all the resource warnings because {@code Disposable} implements {@link AutoCloseable} now. * * @param the value type * @see io.reactivex.rxjava4.subscribers.TestSubscriber diff --git a/src/main/java/io/reactivex/rxjava4/plugins/RxJavaPlugins.java b/src/main/java/io/reactivex/rxjava4/plugins/RxJavaPlugins.java index 6a67462488..e770f34876 100644 --- a/src/main/java/io/reactivex/rxjava4/plugins/RxJavaPlugins.java +++ b/src/main/java/io/reactivex/rxjava4/plugins/RxJavaPlugins.java @@ -21,11 +21,9 @@ import io.reactivex.rxjava4.annotations.*; import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.exceptions.*; -import io.reactivex.rxjava4.flowables.ConnectableFlowable; import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.schedulers.*; import io.reactivex.rxjava4.internal.util.ExceptionHelper; -import io.reactivex.rxjava4.observables.ConnectableObservable; import io.reactivex.rxjava4.parallel.ParallelFlowable; import io.reactivex.rxjava4.schedulers.Schedulers; /** diff --git a/src/main/java/io/reactivex/rxjava4/schedulers/BlockingScheduler.java b/src/main/java/io/reactivex/rxjava4/schedulers/BlockingScheduler.java index 5c49a54e02..bb7aec03bb 100644 --- a/src/main/java/io/reactivex/rxjava4/schedulers/BlockingScheduler.java +++ b/src/main/java/io/reactivex/rxjava4/schedulers/BlockingScheduler.java @@ -21,7 +21,9 @@ /** * Holds onto a blocking scheduler instance and provides access to its {@link #execute()} * method and a way to obtain a pure {@link Scheduler} instance to be used as parameter. - * @implNote No need to instantiate this record by client applications, it serves as a way to + *

+ * Implementation note
+ * No need to instantiate this record by client applications, it serves as a way to * give access to the {@code Scheduler} interface as well as the blocking-specific * {@link #execute()} methods. * @param backingScheduler the scheduler instance diff --git a/src/main/module/module-info.java b/src/main/module/module-info.java index 3c702749f5..9035221c1d 100644 --- a/src/main/module/module-info.java +++ b/src/main/module/module-info.java @@ -17,9 +17,7 @@ exports io.reactivex.rxjava4.core.config; exports io.reactivex.rxjava4.disposables; exports io.reactivex.rxjava4.exceptions; - exports io.reactivex.rxjava4.flowables; exports io.reactivex.rxjava4.functions; - exports io.reactivex.rxjava4.observables; exports io.reactivex.rxjava4.observers; exports io.reactivex.rxjava4.operators; exports io.reactivex.rxjava4.parallel; diff --git a/src/test/java/io/reactivex/rxjava4/flowable/FlowableCovarianceTest.java b/src/test/java/io/reactivex/rxjava4/flowable/FlowableCovarianceTest.java index 1c232612c7..ea2f067148 100644 --- a/src/test/java/io/reactivex/rxjava4/flowable/FlowableCovarianceTest.java +++ b/src/test/java/io/reactivex/rxjava4/flowable/FlowableCovarianceTest.java @@ -21,7 +21,6 @@ import org.junit.jupiter.api.Test; import io.reactivex.rxjava4.core.*; -import io.reactivex.rxjava4.flowables.GroupedFlowable; import io.reactivex.rxjava4.functions.Function; import io.reactivex.rxjava4.testsupport.TestSubscriberEx; diff --git a/src/test/java/io/reactivex/rxjava4/flowable/FlowableGroupByTests.java b/src/test/java/io/reactivex/rxjava4/flowable/FlowableGroupByTests.java index 4b6301446d..f87afe3ad7 100644 --- a/src/test/java/io/reactivex/rxjava4/flowable/FlowableGroupByTests.java +++ b/src/test/java/io/reactivex/rxjava4/flowable/FlowableGroupByTests.java @@ -20,7 +20,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.core.config.StandardBufferedConfig; import io.reactivex.rxjava4.flowable.FlowableEventStream.Event; -import io.reactivex.rxjava4.flowables.GroupedFlowable; import io.reactivex.rxjava4.functions.Function; import io.reactivex.rxjava4.subscribers.TestSubscriber; diff --git a/src/test/java/io/reactivex/rxjava4/flowable/FlowableTests.java b/src/test/java/io/reactivex/rxjava4/flowable/FlowableTests.java index ee5dbbb1de..cce33ba12d 100644 --- a/src/test/java/io/reactivex/rxjava4/flowable/FlowableTests.java +++ b/src/test/java/io/reactivex/rxjava4/flowable/FlowableTests.java @@ -28,7 +28,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.exceptions.*; -import io.reactivex.rxjava4.flowables.ConnectableFlowable; import io.reactivex.rxjava4.functions.Predicate; import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.internal.subscriptions.BooleanSubscription; diff --git a/src/test/java/io/reactivex/rxjava4/flowable/FlowableZipTests.java b/src/test/java/io/reactivex/rxjava4/flowable/FlowableZipTests.java index d06f6b3a6f..6c58881f1a 100644 --- a/src/test/java/io/reactivex/rxjava4/flowable/FlowableZipTests.java +++ b/src/test/java/io/reactivex/rxjava4/flowable/FlowableZipTests.java @@ -24,7 +24,6 @@ import io.reactivex.rxjava4.core.config.StandardBufferedConfig; import io.reactivex.rxjava4.flowable.FlowableCovarianceTest.*; import io.reactivex.rxjava4.flowable.FlowableEventStream.Event; -import io.reactivex.rxjava4.flowables.GroupedFlowable; import io.reactivex.rxjava4.functions.*; public class FlowableZipTests extends RxJavaTest { diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableDoOnEachTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableDoOnEachTest.java index e037ef75d5..271a050a37 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableDoOnEachTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableDoOnEachTest.java @@ -26,7 +26,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.exceptions.*; -import io.reactivex.rxjava4.flowables.ConnectableFlowable; import io.reactivex.rxjava4.functions.Function; import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.internal.subscriptions.BooleanSubscription; diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableGroupByTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableGroupByTest.java index 79b21b08d2..93dc71eb2f 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableGroupByTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableGroupByTest.java @@ -33,7 +33,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.core.config.*; import io.reactivex.rxjava4.exceptions.*; -import io.reactivex.rxjava4.flowables.GroupedFlowable; import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.internal.schedulers.ImmediateThinScheduler; diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowablePublishTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowablePublishTest.java index 7a6ff4cfa2..e1fa92db5b 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowablePublishTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowablePublishTest.java @@ -25,7 +25,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.exceptions.*; -import io.reactivex.rxjava4.flowables.ConnectableFlowable; import io.reactivex.rxjava4.functions.Function; import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.internal.fuseable.HasUpstreamPublisher; diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableRefCountTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableRefCountTest.java index e907500da3..f903d926bf 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableRefCountTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableRefCountTest.java @@ -30,7 +30,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.exceptions.*; -import io.reactivex.rxjava4.flowables.ConnectableFlowable; import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.internal.operators.flowable.FlowableRefCount.RefConnection; diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableReplayEagerTruncateTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableReplayEagerTruncateTest.java index 8a19a42fb9..ee477fc6ec 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableReplayEagerTruncateTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableReplayEagerTruncateTest.java @@ -33,7 +33,6 @@ import io.reactivex.rxjava4.core.config.StandardBufferedConfig; import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.exceptions.TestException; -import io.reactivex.rxjava4.flowables.ConnectableFlowable; import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.internal.fuseable.HasUpstreamPublisher; diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableReplayTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableReplayTest.java index c5fab5bfec..3169e9e1de 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableReplayTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableReplayTest.java @@ -33,7 +33,6 @@ import io.reactivex.rxjava4.core.config.StandardBufferedConfig; import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.exceptions.TestException; -import io.reactivex.rxjava4.flowables.ConnectableFlowable; import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.internal.fuseable.HasUpstreamPublisher; diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableRetryTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableRetryTest.java index 3d9c5f488c..c6ad02a66f 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableRetryTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableRetryTest.java @@ -29,7 +29,6 @@ import io.reactivex.rxjava4.core.config.StandardConcurrentBufferedConfig; import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.exceptions.TestException; -import io.reactivex.rxjava4.flowables.GroupedFlowable; import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.internal.subscriptions.BooleanSubscription; diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableTimerTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableTimerTest.java index 3755e78e18..9b918b0b7f 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableTimerTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableTimerTest.java @@ -27,7 +27,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.exceptions.*; -import io.reactivex.rxjava4.flowables.ConnectableFlowable; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.schedulers.*; import io.reactivex.rxjava4.subscribers.*; diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableCacheTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableCacheTest.java index bacf81d55d..f65c467ccf 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableCacheTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableCacheTest.java @@ -28,7 +28,6 @@ import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.exceptions.TestException; import io.reactivex.rxjava4.functions.Action; -import io.reactivex.rxjava4.observables.ConnectableObservable; import io.reactivex.rxjava4.observers.TestObserver; import io.reactivex.rxjava4.schedulers.Schedulers; import io.reactivex.rxjava4.subjects.PublishSubject; diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableGroupByTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableGroupByTest.java index 0471e51931..84ca971a5f 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableGroupByTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableGroupByTest.java @@ -32,7 +32,6 @@ import io.reactivex.rxjava4.exceptions.TestException; import io.reactivex.rxjava4.functions.Function; import io.reactivex.rxjava4.internal.functions.Functions; -import io.reactivex.rxjava4.observables.GroupedObservable; import io.reactivex.rxjava4.observers.*; import io.reactivex.rxjava4.schedulers.Schedulers; import io.reactivex.rxjava4.subjects.*; diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservablePublishTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservablePublishTest.java index 1902946576..6aeb89bb57 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservablePublishTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservablePublishTest.java @@ -29,7 +29,6 @@ import io.reactivex.rxjava4.functions.Function; import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.internal.fuseable.HasUpstreamObservableSource; -import io.reactivex.rxjava4.observables.ConnectableObservable; import io.reactivex.rxjava4.observers.TestObserver; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.schedulers.*; diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableRefCountTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableRefCountTest.java index 1078db4878..8c65527e46 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableRefCountTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableRefCountTest.java @@ -26,16 +26,15 @@ import org.junit.jupiter.api.Test; import org.mockito.InOrder; +import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.core.Observable; import io.reactivex.rxjava4.core.Observer; -import io.reactivex.rxjava4.core.RxJavaTest; import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.exceptions.*; import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.internal.operators.observable.ObservableRefCount.RefConnection; import io.reactivex.rxjava4.internal.util.ExceptionHelper; -import io.reactivex.rxjava4.observables.ConnectableObservable; import io.reactivex.rxjava4.observers.TestObserver; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.schedulers.*; diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableReplayEagerTruncateTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableReplayEagerTruncateTest.java index 2c96807d27..65b186e76e 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableReplayEagerTruncateTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableReplayEagerTruncateTest.java @@ -37,7 +37,6 @@ import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.internal.fuseable.HasUpstreamObservableSource; import io.reactivex.rxjava4.internal.operators.observable.ObservableReplay.*; -import io.reactivex.rxjava4.observables.ConnectableObservable; import io.reactivex.rxjava4.observers.TestObserver; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.schedulers.*; diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableReplayTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableReplayTest.java index 011d55a2b0..828de3a25f 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableReplayTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableReplayTest.java @@ -37,7 +37,6 @@ import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.internal.fuseable.HasUpstreamObservableSource; import io.reactivex.rxjava4.internal.operators.observable.ObservableReplay.*; -import io.reactivex.rxjava4.observables.ConnectableObservable; import io.reactivex.rxjava4.observers.TestObserver; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.schedulers.*; diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableRetryTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableRetryTest.java index 27061d5e3c..0387a82ad7 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableRetryTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableRetryTest.java @@ -32,7 +32,6 @@ import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.internal.subscriptions.BooleanSubscription; -import io.reactivex.rxjava4.observables.GroupedObservable; import io.reactivex.rxjava4.observers.*; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.schedulers.Schedulers; diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableTimerTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableTimerTest.java index 5b19add94f..209a9c9403 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableTimerTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableTimerTest.java @@ -28,7 +28,6 @@ import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.exceptions.TestException; import io.reactivex.rxjava4.internal.operators.observable.ObservableTimer.TimerObserver; -import io.reactivex.rxjava4.observables.ConnectableObservable; import io.reactivex.rxjava4.observers.*; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.schedulers.*; diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableGroupByTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableGroupByTest.java new file mode 100644 index 0000000000..d75fa25c2e --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableGroupByTest.java @@ -0,0 +1,283 @@ +/* + * 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.List; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.jupiter.api.*; + +import io.reactivex.rxjava4.annotations.NonNull; +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.core.config.StandardConcurrentConfig; +import io.reactivex.rxjava4.disposables.Disposable; +import io.reactivex.rxjava4.exceptions.*; +import io.reactivex.rxjava4.processors.PublishProcessor; + +public class StreamableGroupByTest extends StreamableBaseTest { + + @Test + public void normal() throws Throwable { + Streamable.range(1, 5) + .groupBy(v -> v) + .flatMap(v -> v, StandardConcurrentConfig.DEFAULT) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void empty() throws Throwable { + Streamable.empty() + .groupBy(v -> v) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void oneGroup() throws Throwable { + Streamable.range(1, 5) + .groupBy(_ -> 1) + .flatMap(v -> v, StandardConcurrentConfig.DEFAULT) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void crash() throws Throwable { + Streamable.error(new TestException()) + .groupBy(v -> v) + .flatMap(v -> v, StandardConcurrentConfig.DEFAULT) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void oneGroupDebug() throws Throwable { + withCachedExecutor(exec -> { + Streamable.range(1, 5) + .groupBy(_ -> 1) + .flatMap(v -> v, StandardConcurrentConfig.DEFAULT) + .test(exec) + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + }); + } + + @Test + public void crashDebug() throws Throwable { + withCachedExecutor(exec -> { + Streamable.error(new TestException()) + .groupBy(v -> v) + .flatMap(v -> v, StandardConcurrentConfig.DEFAULT) + .test(exec) + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + }); + } + + @Test + public void mod2() throws Throwable { + Streamable.range(1, 5) + .groupBy(v -> v % 2) + .flatMap(v -> v, StandardConcurrentConfig.DEFAULT) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void mapperCrash() throws Throwable { + Streamable.error(new TestException()) + .groupBy(_ -> { throw new TestException(); }) + .flatMap(v -> v, StandardConcurrentConfig.DEFAULT) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void mapperCrash2() throws Throwable { + Streamable.range(1, 5) + .groupBy(v -> { + if (v == 1) { + return 1; + } + throw new TestException(); + }) + .flatMap(v -> v, StandardConcurrentConfig.DEFAULT) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(CompositeException.class, 1); + } + + @Test + public void tombstone() { + StreamableGroupBy.TombstoneGroup.INSTANCE + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(CancellationException.class); + } + + @Test + public void tombstone2() { + assertSame(Streamer.FINISHED, StreamableGroupBy.TombstoneGroup.INSTANCE.send(true), "send"); + assertSame(Streamer.FINISHED, StreamableGroupBy.TombstoneGroup.INSTANCE.terminate(null), "terminate"); + } + + @Test + public void innerMultiTest() { + var ts = Streamable.just(1) + .groupBy(v -> v) + .map(g -> List.of(g.test(), g.test())) + .test() + .awaitDone(5, TimeUnit.SECONDS); + + var ts1 = ts.values().get(0).get(0) + .withTag("1") + .awaitDone(5, TimeUnit.SECONDS); + + var ts2 = ts.values().get(0).get(1) + .withTag("2") + .awaitDone(5, TimeUnit.SECONDS); + + if (ts1.errors().isEmpty() && !ts2.errors().isEmpty()) { + ts1.assertResult(1); + ts2.assertFailure(IllegalStateException.class); + } else + if (!ts1.errors().isEmpty() && ts2.errors().isEmpty()) { + ts2.assertResult(1); + ts1.assertFailure(IllegalStateException.class); + } else { + fail("Wrong failure state. " + ts1.toString() + " | " + ts2.toString()); + } + } + + @Test + public void innerMultiTestDebug() throws Throwable { + withCachedExecutor(exec -> { + var ts = Streamable.just(1) + .groupBy(v -> v) + .map(g -> List.of(g.test(exec), g.test(exec))) + .test(exec) + .awaitDone(5, TimeUnit.SECONDS); + + var ts1 = ts.values().get(0).get(0) + .withTag("1") + .awaitDone(5, TimeUnit.SECONDS); + + var ts2 = ts.values().get(0).get(1) + .withTag("2") + .awaitDone(5, TimeUnit.SECONDS); + + if (ts1.errors().isEmpty() && !ts2.errors().isEmpty()) { + ts1.assertResult(1); + ts2.assertFailure(IllegalStateException.class); + } else + if (!ts1.errors().isEmpty() && ts2.errors().isEmpty()) { + ts2.assertResult(1); + ts1.assertFailure(IllegalStateException.class); + } else { + fail("Wrong failure state. " + ts1.toString() + " | " + ts2.toString()); + } + }); + } + + @Test + public void upstreamErrorWithOneGroup() { + Streamable.range(1, 2) + .map(v -> { + if (v == 2) { + throw new TestException(); + } + return v; + }) + .groupBy(v -> v) + .flatMap(v -> v, StandardConcurrentConfig.DEFAULT) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(CompositeException.class, 1) + ; + } + + @Test + public void groupDisposeTest() throws Throwable { + var pp = PublishProcessor.create(); + var gr = new AtomicReference>(); + + var ts = pp.toStreamable() + .groupBy(v -> v) + .map(g -> { + gr.lazySet(g); + return g.test(); + }) + .test(); + + while (!ts.hasSubscription()) { + Thread.sleep(1); + } + + while (!pp.hasSubscribers()) { + Thread.sleep(1); + } + + pp.onNext(1); + + while (gr.get() == null) { + Thread.sleep(1); + } + + assertFalse(((Disposable)gr.get()).isDisposed(), "Group already disposed?"); + + pp.onComplete(); + + ts.awaitDone(5, TimeUnit.SECONDS) + .assertValueCount(1); + + assertTrue(((Disposable)gr.get()).isDisposed(), "Group not disposed"); + } + + @Test + public void finishFails() { + Streamable str = _ -> new Streamer<>() /* NFI */ { + @Override + public @NonNull CompletionStage next() { + return NEXT_FALSE; + } + + @Override + public @NonNull Integer current() { + return null; + } + + @Override + public @NonNull CompletionStage finish() { + return CompletableFuture.failedFuture(new TestException()); + } + }; + + str + .groupBy(v -> v) + .flatMap(v -> v, StandardConcurrentConfig.DEFAULT) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelperTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelperTest.java new file mode 100644 index 0000000000..48d718ec3e --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelperTest.java @@ -0,0 +1,49 @@ +/* + * 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.*; + +import org.junit.jupiter.api.Test; + +import io.reactivex.rxjava4.exceptions.TestException; +import io.reactivex.rxjava4.testsupport.TestHelper; + +public class StreamableHelperTest extends StreamableBaseTest { + + @Test + public void utilityClass() throws Throwable { + TestHelper.checkEnum(StreamableHelper.class); + } + + @Test + public void andThenSupplyOtherFails() { + var e = assertThrows(CompletionException.class, () -> { + StreamableHelper.andThenSupply(CompletableFuture.completedStage(1), + () -> CompletableFuture.failedFuture(new TestException())) + .join(); + }); + + assertTrue(e.getCause() instanceof TestException, e.getCause().toString()); + } + + @Test + public void isCancellationUnwraps() { + assertTrue(StreamableHelper.isCancellation(new CompletionException(new CancellationException())), "complete-cancel"); + assertTrue(StreamableHelper.isCancellation(new CancellationException()), "cancel"); + assertFalse(StreamableHelper.isCancellation(new CompletionException(new TestException())), "complete-test"); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeUntilTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeUntilTest.java new file mode 100644 index 0000000000..bf97c8aa80 --- /dev/null +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeUntilTest.java @@ -0,0 +1,113 @@ +/* + * 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; +import io.reactivex.rxjava4.schedulers.Schedulers; + +public class StreamableTakeUntilTest extends StreamableBaseTest { + + @Test + public void passthrough() throws Throwable { + Streamable.range(1, 5) + .takeUntil(Streamable.never()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } + + @Test + public void passthroughError() throws Throwable { + Streamable.error(new TestException()) + .takeUntil(Streamable.never()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void timer() throws Throwable { + Streamable.never() + .takeUntil(Streamable.timer(10, TimeUnit.MILLISECONDS, Schedulers.single())) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void empty() throws Throwable { + Streamable.never() + .takeUntil(Streamable.empty()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void otherError() throws Throwable { + Streamable.never() + .takeUntil(Streamable.error(new TestException())) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertFailure(TestException.class); + } + + @Test + public void realtime() { + withRetry(3, () -> { + Streamable.intervalRange(1, 5, 0, 100, TimeUnit.MILLISECONDS, Schedulers.single()) + .takeUntil(Streamable.timer(250, TimeUnit.MILLISECONDS, Schedulers.virtual())) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1L, 2L, 3L); + }); + } + + @Test + public void realtimeDebug() throws Throwable { + withCachedExecutor(exec -> { + withRetry(3, () -> { + Streamable.intervalRange(1, 5, 0, 100, TimeUnit.MILLISECONDS, Schedulers.single()) + .takeUntil(Streamable.timer(250, TimeUnit.MILLISECONDS, Schedulers.virtual())) + .test(exec) + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1L, 2L, 3L); + }); + }); + } + + @Test + public void neverNeverEMpty() throws Throwable { + Streamable.never() + .takeUntil(Streamable.never()) + .takeUntil(Streamable.empty()) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(); + } + + @Test + public void longTimer() throws Throwable { + Streamable.range(1, 5) + .takeUntil(Streamable.timer(1, TimeUnit.MINUTES, Schedulers.single())) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertResult(1, 2, 3, 4, 5); + } +} diff --git a/src/test/java/io/reactivex/rxjava4/observable/ObservableCovarianceTest.java b/src/test/java/io/reactivex/rxjava4/observable/ObservableCovarianceTest.java index f6913a711d..519f1a7e22 100644 --- a/src/test/java/io/reactivex/rxjava4/observable/ObservableCovarianceTest.java +++ b/src/test/java/io/reactivex/rxjava4/observable/ObservableCovarianceTest.java @@ -22,7 +22,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.core.Observable; import io.reactivex.rxjava4.functions.Function; -import io.reactivex.rxjava4.observables.GroupedObservable; import io.reactivex.rxjava4.testsupport.TestObserverEx; /** diff --git a/src/test/java/io/reactivex/rxjava4/observable/ObservableGroupByTests.java b/src/test/java/io/reactivex/rxjava4/observable/ObservableGroupByTests.java index 9d37e5082d..8798d3d6c2 100644 --- a/src/test/java/io/reactivex/rxjava4/observable/ObservableGroupByTests.java +++ b/src/test/java/io/reactivex/rxjava4/observable/ObservableGroupByTests.java @@ -18,7 +18,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.functions.Function; import io.reactivex.rxjava4.observable.ObservableEventStream.Event; -import io.reactivex.rxjava4.observables.GroupedObservable; public class ObservableGroupByTests extends RxJavaTest { diff --git a/src/test/java/io/reactivex/rxjava4/observable/ObservableTest.java b/src/test/java/io/reactivex/rxjava4/observable/ObservableTest.java index 46ea171d98..2832a6dfe8 100644 --- a/src/test/java/io/reactivex/rxjava4/observable/ObservableTest.java +++ b/src/test/java/io/reactivex/rxjava4/observable/ObservableTest.java @@ -30,7 +30,6 @@ import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.functions.Functions; -import io.reactivex.rxjava4.observables.ConnectableObservable; import io.reactivex.rxjava4.observers.*; import io.reactivex.rxjava4.schedulers.*; import io.reactivex.rxjava4.subjects.*; diff --git a/src/test/java/io/reactivex/rxjava4/observable/ObservableZipTests.java b/src/test/java/io/reactivex/rxjava4/observable/ObservableZipTests.java index ab77e627ec..1090dbdd8f 100644 --- a/src/test/java/io/reactivex/rxjava4/observable/ObservableZipTests.java +++ b/src/test/java/io/reactivex/rxjava4/observable/ObservableZipTests.java @@ -19,13 +19,12 @@ import org.junit.jupiter.api.Test; +import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.core.Observable; -import io.reactivex.rxjava4.core.RxJavaTest; import io.reactivex.rxjava4.core.config.StandardBufferedConfig; import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.observable.ObservableCovarianceTest.*; import io.reactivex.rxjava4.observable.ObservableEventStream.Event; -import io.reactivex.rxjava4.observables.GroupedObservable; public class ObservableZipTests extends RxJavaTest { diff --git a/src/test/java/io/reactivex/rxjava4/plugins/RxJavaPluginsTest.java b/src/test/java/io/reactivex/rxjava4/plugins/RxJavaPluginsTest.java index 6eae8c8c5b..2f5f984bb4 100644 --- a/src/test/java/io/reactivex/rxjava4/plugins/RxJavaPluginsTest.java +++ b/src/test/java/io/reactivex/rxjava4/plugins/RxJavaPluginsTest.java @@ -30,7 +30,6 @@ import io.reactivex.rxjava4.core.Scheduler.Worker; import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.exceptions.*; -import io.reactivex.rxjava4.flowables.ConnectableFlowable; import io.reactivex.rxjava4.functions.*; import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.internal.operators.completable.CompletableError; @@ -43,7 +42,6 @@ import io.reactivex.rxjava4.internal.schedulers.ImmediateThinScheduler; import io.reactivex.rxjava4.internal.subscriptions.ScalarSubscription; import io.reactivex.rxjava4.internal.util.ExceptionHelper; -import io.reactivex.rxjava4.observables.ConnectableObservable; import io.reactivex.rxjava4.parallel.ParallelFlowable; import io.reactivex.rxjava4.schedulers.Schedulers; import io.reactivex.rxjava4.testsupport.TestHelper; diff --git a/src/test/java/io/reactivex/rxjava4/subscribers/TestSubscriberTest.java b/src/test/java/io/reactivex/rxjava4/subscribers/TestSubscriberTest.java index ebf0c568bb..96f06b40fa 100644 --- a/src/test/java/io/reactivex/rxjava4/subscribers/TestSubscriberTest.java +++ b/src/test/java/io/reactivex/rxjava4/subscribers/TestSubscriberTest.java @@ -1707,4 +1707,16 @@ public void assertValueSetFail() { ts.assertValueSet(1, 2); }); } + + @Test + public void toStringValue() { + var ts = new TestSubscriber<>(); + ts.onSubscribe(new BooleanSubscription()); + ts.onNext(2); + ts.onNext(1); + ts.onNext(3); + ts.onComplete(); + + assertEquals("TestSubscriber (latch = 0, values = 3, errors = 0, completions = 1)", ts.toString()); + } } diff --git a/src/test/java/io/reactivex/rxjava4/testsupport/BaseTestConsumerEx.java b/src/test/java/io/reactivex/rxjava4/testsupport/BaseTestConsumerEx.java index 1f0e681835..12c3403baf 100644 --- a/src/test/java/io/reactivex/rxjava4/testsupport/BaseTestConsumerEx.java +++ b/src/test/java/io/reactivex/rxjava4/testsupport/BaseTestConsumerEx.java @@ -13,15 +13,13 @@ package io.reactivex.rxjava4.testsupport; -import java.util.List; +import java.util.Objects; import io.reactivex.rxjava4.functions.Predicate; import io.reactivex.rxjava4.internal.util.ExceptionHelper; import io.reactivex.rxjava4.observers.BaseTestConsumer; import io.reactivex.rxjava4.operators.QueueFuseable; -import java.util.Objects; - /** * Base class with shared infrastructure to support TestSubscriber and TestObserver. * @param the value type consumed @@ -205,14 +203,6 @@ public final U assertFailureAndMessage(Class error, .assertNotComplete(); } - /** - * Returns the internal shared list of errors. - * @return Returns the internal shared list of errors. - */ - public final List errors() { - return errors; - } - /** * Returns true if this test consumer has terminated in any fashion. * @return true if this test consumer has terminated in any fashion diff --git a/src/test/java/io/reactivex/rxjava4/validators/CheckBaseTypeAnnotationsTest.java b/src/test/java/io/reactivex/rxjava4/validators/CheckBaseTypeAnnotationsTest.java index 5d3af49077..5449d83520 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/CheckBaseTypeAnnotationsTest.java +++ b/src/test/java/io/reactivex/rxjava4/validators/CheckBaseTypeAnnotationsTest.java @@ -23,8 +23,6 @@ import io.reactivex.rxjava4.annotations.*; import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.disposables.DisposableContainer; -import io.reactivex.rxjava4.flowables.ConnectableFlowable; -import io.reactivex.rxjava4.observables.ConnectableObservable; import io.reactivex.rxjava4.parallel.ParallelFlowable; import io.reactivex.rxjava4.processors.*; import io.reactivex.rxjava4.subjects.*; diff --git a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationNamingTest.java b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationNamingTest.java index 3c9e7ca741..4bcdabed08 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationNamingTest.java +++ b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationNamingTest.java @@ -22,8 +22,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.core.Observable; import io.reactivex.rxjava4.disposables.*; -import io.reactivex.rxjava4.flowables.ConnectableFlowable; -import io.reactivex.rxjava4.observables.ConnectableObservable; import io.reactivex.rxjava4.parallel.ParallelFlowable; import io.reactivex.rxjava4.processors.*; import io.reactivex.rxjava4.schedulers.Schedulers; diff --git a/src/test/java/io/reactivex/rxjava4/validators/CheckSourceAnnotationTest.java b/src/test/java/io/reactivex/rxjava4/validators/CheckSourceAnnotationTest.java index 81c53053df..b649446722 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/CheckSourceAnnotationTest.java +++ b/src/test/java/io/reactivex/rxjava4/validators/CheckSourceAnnotationTest.java @@ -22,8 +22,6 @@ import io.reactivex.rxjava4.core.*; import io.reactivex.rxjava4.core.Observable; import io.reactivex.rxjava4.disposables.Disposable; -import io.reactivex.rxjava4.flowables.ConnectableFlowable; -import io.reactivex.rxjava4.observables.ConnectableObservable; import io.reactivex.rxjava4.parallel.ParallelFlowable; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.processors.*;