diff --git a/src/main/java/io/reactivex/rxjava4/core/StreamSink.java b/src/main/java/io/reactivex/rxjava4/core/StreamSink.java index ffcb40f5b5..535243b04f 100644 --- a/src/main/java/io/reactivex/rxjava4/core/StreamSink.java +++ b/src/main/java/io/reactivex/rxjava4/core/StreamSink.java @@ -60,20 +60,20 @@ public interface StreamSink<@NonNull T> { * @return the {@code DisposableContainer} */ @NonNull - default DisposableContainer cancellation() { + default DisposableStreamerCancellation cancellation() { return new CompositeDisposable(); } /** - * Returns a new {@link StreamSink} that returns the given {@link DisposableContainer} + * Returns a new {@link StreamSink} that returns the given {@link DisposableStreamerCancellation} * in its {@link #cancellation()}, allowing overriding the cancellation management * of this {@code StreamSink} - * @param cancellation the {@link DisposableContainer} to use as cancellation management + * @param cancellation the {@link DisposableStreamerCancellation} to use as cancellation management * @return the new {@code StreamSink} instance * @throws NullPointerException if {@code cancellation} is {@code null} */ @NonNull - default StreamSink withCancellation(DisposableContainer cancellation) { + default StreamSink withCancellation(DisposableStreamerCancellation cancellation) { Objects.requireNonNull(cancellation, "cancellation is null"); return new StreamSinkWithCancellation<>(this, cancellation); } diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamable.java b/src/main/java/io/reactivex/rxjava4/core/Streamable.java index ef64a500d9..924342b0fc 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamable.java @@ -76,7 +76,7 @@ public interface Streamable<@NonNull T> { * Implementations are not meant to dispose the {@code cancellation} as it is the * privilege of whatever calls the {@code stream} method. If you need to talk to other * {@code Streamable}s, use the {@link StreamerCancellation#derive()} to gain access to - * a fully fledged {@link DisposableContainer} where you can dispose it without disposing + * a fully fledged {@link DisposableStreamerCancellation} where you can dispose it without disposing * the parent {@code cancellation} instance by accident. *

* It is recommened you {@link StreamerCancellation#remove(Disposable)} or @@ -987,7 +987,8 @@ default CompletionStageDisposable forEach(@NonNull Consumer con */ @CheckReturnValue @NonNull - default CompletionStageDisposable forEach(@NonNull Consumer consumer, @NonNull DisposableContainer canceller) { + default CompletionStageDisposable forEach(@NonNull Consumer consumer, + @NonNull DisposableStreamerCancellation canceller) { return forEach(consumer, canceller, Executors.newVirtualThreadPerTaskExecutor()); } @@ -1014,7 +1015,8 @@ default CompletionStageDisposable forEach(@NonNull Consumer con */ @CheckReturnValue @NonNull - default CompletionStageDisposable forEach(@NonNull Consumer consumer, @NonNull DisposableContainer canceller, @NonNull ExecutorService executor) { + default CompletionStageDisposable forEach(@NonNull Consumer consumer, + @NonNull DisposableStreamerCancellation canceller, @NonNull ExecutorService executor) { Objects.requireNonNull(consumer, "consumer is null"); Objects.requireNonNull(canceller, "canceller is null"); Objects.requireNonNull(executor, "executor is null"); @@ -1033,7 +1035,7 @@ default CompletionStageDisposable forEach(@NonNull Consumer con @NonNull default CompletionStageDisposable forEach( @NonNull BiConsumer consumer, - @NonNull DisposableContainer canceller, + @NonNull DisposableStreamerCancellation canceller, @NonNull ExecutorService executor) { Objects.requireNonNull(consumer, "consumer is null"); Objects.requireNonNull(canceller, "canceller is null"); diff --git a/src/main/java/io/reactivex/rxjava4/core/VirtualEmitter.java b/src/main/java/io/reactivex/rxjava4/core/VirtualEmitter.java index 97a3e6bccf..542aea5fe8 100644 --- a/src/main/java/io/reactivex/rxjava4/core/VirtualEmitter.java +++ b/src/main/java/io/reactivex/rxjava4/core/VirtualEmitter.java @@ -33,5 +33,5 @@ public interface VirtualEmitter { * Returns a disposable container to relay cancellation notifications while awaiting the run. * @return a new Disposable Container instance */ - DisposableContainer canceller(); + DisposableStreamerCancellation canceller(); } \ No newline at end of file diff --git a/src/main/java/io/reactivex/rxjava4/disposables/CompositeDisposable.java b/src/main/java/io/reactivex/rxjava4/disposables/CompositeDisposable.java index ea6996a837..0539ce82a6 100644 --- a/src/main/java/io/reactivex/rxjava4/disposables/CompositeDisposable.java +++ b/src/main/java/io/reactivex/rxjava4/disposables/CompositeDisposable.java @@ -24,7 +24,8 @@ * offers O(1) time complexity for {@link #add(Disposable)}, {@link #remove(Disposable)} and {@link #delete(Disposable)} * operations. */ -public final class CompositeDisposable implements Disposable, DisposableContainer { +public final class CompositeDisposable +implements Disposable, DisposableContainer, DisposableStreamerCancellation { OpenHashSet resources; @@ -270,7 +271,7 @@ public void reset() { } @Override - public DisposableContainer derive() { + public DisposableStreamerCancellation derive() { var result = new CompositeDisposable(); add(result); diff --git a/src/main/java/io/reactivex/rxjava4/disposables/DisposableContainer.java b/src/main/java/io/reactivex/rxjava4/disposables/DisposableContainer.java index 380fd3d5ad..a843db20ce 100644 --- a/src/main/java/io/reactivex/rxjava4/disposables/DisposableContainer.java +++ b/src/main/java/io/reactivex/rxjava4/disposables/DisposableContainer.java @@ -13,11 +13,37 @@ package io.reactivex.rxjava4.disposables; +import io.reactivex.rxjava4.annotations.NonNull; + /** * Common interface to add and remove {@link Disposable}s from a container. * @since 2.0 */ -public interface DisposableContainer extends Disposable, StreamerCancellation { +public interface DisposableContainer extends Disposable { + + /** + * Adds a disposable to this container or disposes it if the + * container has been disposed. + * @param d the disposable to add, not null + * @return true if successful, false if this container has been disposed + */ + boolean add(@NonNull Disposable d); + + /** + * Removes and disposes the given disposable if it is part of this + * container. + * @param d the disposable to remove and dispose, not null + * @return true if the operation was successful + */ + boolean remove(@NonNull Disposable d); + + /** + * Removes but does not dispose the given disposable if it is part of this + * container. + * @param d the disposable to remove, not null + * @return true if the operation was successful + */ + boolean delete(@NonNull Disposable d); /** * Removes all contained {@link Disposable}s without disposing them, making this diff --git a/src/main/java/io/reactivex/rxjava4/disposables/DisposableStreamerCancellation.java b/src/main/java/io/reactivex/rxjava4/disposables/DisposableStreamerCancellation.java new file mode 100644 index 0000000000..65910a0726 --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/disposables/DisposableStreamerCancellation.java @@ -0,0 +1,26 @@ +/* + * 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.disposables; + +/** + * Represents the full, disposable cancellation interface for {@code Streamer} + * operations. + *

+ * It was decided to have a separate set of interfaces instead of trying to + * retrofit {@link DisposableContainer}. + * @since 4.0.0 + */ +public interface DisposableStreamerCancellation extends StreamerCancellation, Disposable { + +} diff --git a/src/main/java/io/reactivex/rxjava4/disposables/StreamerCancellation.java b/src/main/java/io/reactivex/rxjava4/disposables/StreamerCancellation.java index 18caaa0647..7ce0bcfeef 100644 --- a/src/main/java/io/reactivex/rxjava4/disposables/StreamerCancellation.java +++ b/src/main/java/io/reactivex/rxjava4/disposables/StreamerCancellation.java @@ -17,12 +17,12 @@ import io.reactivex.rxjava4.core.Streamable; /** - * Represents non-disposable view of a {@link DisposableContainer} + * Represents non-disposable view of a {@link DisposableStreamerCancellation} * that allows synchronous testing for disposed state as well as allow * adding and removing {@link Disposable} resources to be * cleaned up when the full container is disposed. *

- * This view is provided to prevent calling {@link DisposableContainer#dispose()} + * This view is provided to prevent calling {@link DisposableStreamerCancellation#dispose()} * in {@link Streamable#stream(StreamerCancellation)} implementations because * disposing a stream is the privilege of the caller/downstream. *

@@ -65,11 +65,12 @@ public interface StreamerCancellation { boolean delete(@NonNull Disposable d); /** - * Create a derived sub-container that can get cancelled by this container, - * but disposing the sub-container does not dispose this container. - * @return the derived sub-container + * Create a derived sub-cancellation management interface that can get disposed + * by this cancellation management object but disposing the sub-cancellation + * manager does not dispose this current {@code StreamerCancellation} instance.. + * @return the derived sub-cancellation management object */ @NonNull - DisposableContainer derive(); + DisposableStreamerCancellation derive(); } diff --git a/src/main/java/io/reactivex/rxjava4/internal/disposables/ListCompositeDisposable.java b/src/main/java/io/reactivex/rxjava4/internal/disposables/ListCompositeDisposable.java index 9d586e8585..422f7c3fac 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/disposables/ListCompositeDisposable.java +++ b/src/main/java/io/reactivex/rxjava4/internal/disposables/ListCompositeDisposable.java @@ -22,7 +22,8 @@ /** * A disposable container that can hold onto multiple other disposables. */ -public final class ListCompositeDisposable implements Disposable, DisposableContainer, StreamerCancellation { +public final class ListCompositeDisposable +implements Disposable, DisposableContainer, DisposableStreamerCancellation { List resources; @@ -199,7 +200,7 @@ public void reset() { } @Override - public DisposableContainer derive() { + public DisposableStreamerCancellation derive() { var result = new ListCompositeDisposable(); add(result); diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamSinkWithCancellation.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamSinkWithCancellation.java index 4185786525..1887db315b 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamSinkWithCancellation.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamSinkWithCancellation.java @@ -16,12 +16,13 @@ import java.util.Objects; import java.util.concurrent.CompletionStage; -import io.reactivex.rxjava4.annotations.*; +import io.reactivex.rxjava4.annotations.NonNull; +import io.reactivex.rxjava4.annotations.Nullable; import io.reactivex.rxjava4.core.StreamSink; -import io.reactivex.rxjava4.disposables.DisposableContainer; +import io.reactivex.rxjava4.disposables.DisposableStreamerCancellation; /** - * Wraps a {@link StreamSink} and uses the given {@link DisposableContainer} to be + * Wraps a {@link StreamSink} and uses the given {@link DisposableStreamerCancellation} to be * returned via {@link #cancellation()}. * @param the element type of the stream * @param downstream the {@code StreamSink} to relay events to @@ -29,7 +30,7 @@ * @since 4.0.0 */ public record StreamSinkWithCancellation<@NonNull T>( - @NonNull StreamSink downstream, @NonNull DisposableContainer cancellation) + @NonNull StreamSink downstream, @NonNull DisposableStreamerCancellation cancellation) implements StreamSink { public StreamSinkWithCancellation { diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterable.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterable.java index 61ec76f048..0ac897fde2 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterable.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterable.java @@ -41,7 +41,7 @@ static final class ConcatIteratorStreamer extends AtomicInteger implements St final StreamerCancellation cancellation; - DisposableContainer currentCancellation; + DisposableStreamerCancellation currentCancellation; Streamer upstream; diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFlatMap.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFlatMap.java index 31add965bf..2b01168ccc 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFlatMap.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFlatMap.java @@ -189,13 +189,14 @@ static final class InnerStreamer extends AtomicInteger implements BiConsum final FlatMapMainStreamer parent; - final DisposableContainer canceller; + final DisposableStreamerCancellation canceller; volatile boolean done; boolean finishing; - InnerStreamer(Streamer inner, FlatMapMainStreamer parent, DisposableContainer canceller) { + InnerStreamer(Streamer inner, FlatMapMainStreamer parent, + DisposableStreamerCancellation canceller) { this.inner = inner; this.parent = parent; this.canceller = canceller; diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java index 6a1d437f86..e9d80f663a 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java @@ -31,7 +31,7 @@ public record StreamableForEach() { public static CompletionStageDisposable forEach( @NonNull Streamable me, @NonNull Consumer consumer, - @NonNull DisposableContainer canceller, + @NonNull DisposableStreamerCancellation canceller, @NonNull ExecutorService executor) { var future = CompletableFuture.supplyAsync(() -> { Throwable finallyCrash = null; @@ -71,7 +71,7 @@ public static CompletionStageDisposable forEach( public static CompletionStageDisposable forEach( @NonNull Streamable me, @NonNull BiConsumer consumer, - @NonNull DisposableContainer canceller, + @NonNull DisposableStreamerCancellation canceller, @NonNull ExecutorService executor) { var future = CompletableFuture.supplyAsync(() -> { var str = me.stream(canceller); @@ -110,7 +110,8 @@ public static CompletionStageDisposable forEach( return new CompletionStageDisposable<>(future, canceller); } - public static CompletionStage forEach(Streamable me, StreamSink consumer, ExecutorService executor) { + public static CompletionStage forEach( + Streamable me, StreamSink consumer, ExecutorService executor) { var cf = new CompletableFuture(); CompletableFuture.runAsync(() -> { Throwable error = null; 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 index 8ed7b8cb38..f61ee9bf4f 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeUntil.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeUntil.java @@ -13,13 +13,18 @@ package io.reactivex.rxjava4.internal.operators.streamable; -import static io.reactivex.rxjava4.internal.operators.streamable.StreamableHelper.*; +import static io.reactivex.rxjava4.internal.operators.streamable.StreamableHelper.andThenSupply; +import static io.reactivex.rxjava4.internal.operators.streamable.StreamableHelper.suppressCancel; +import static io.reactivex.rxjava4.internal.operators.streamable.StreamableHelper.suppressValueAndCancel; +import static io.reactivex.rxjava4.internal.operators.streamable.StreamableHelper.whenBoth; import java.util.concurrent.CompletionStage; import io.reactivex.rxjava4.annotations.NonNull; -import io.reactivex.rxjava4.core.*; -import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.core.Streamer; +import io.reactivex.rxjava4.disposables.DisposableStreamerCancellation; +import io.reactivex.rxjava4.disposables.StreamerCancellation; import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; public record StreamableTakeUntil( @@ -48,9 +53,9 @@ static final class TakeUntilMainStreamer implements Streamer { final Streamer otherStreamer; - final DisposableContainer mainCancellation; + final DisposableStreamerCancellation mainCancellation; - final DisposableContainer otherCancellation; + final DisposableStreamerCancellation otherCancellation; final CompletionStage otherNext; @@ -59,8 +64,8 @@ static final class TakeUntilMainStreamer implements Streamer { public TakeUntilMainStreamer( Streamer upstream, Streamer otherStreamer, - DisposableContainer mainCancellation, - DisposableContainer otherCancellation, + DisposableStreamerCancellation mainCancellation, + DisposableStreamerCancellation otherCancellation, CompletionStage otherNext ) { this.upstream = upstream; diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableToObservable.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableToObservable.java index af0d2c9b4b..84bf5d60f5 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableToObservable.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableToObservable.java @@ -13,12 +13,19 @@ package io.reactivex.rxjava4.internal.operators.streamable; -import java.util.concurrent.atomic.*; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.BiConsumer; import io.reactivex.rxjava4.annotations.NonNull; -import io.reactivex.rxjava4.core.*; -import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.core.Observable; +import io.reactivex.rxjava4.core.Observer; +import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.core.Streamer; +import io.reactivex.rxjava4.disposables.CompositeDisposable; +import io.reactivex.rxjava4.disposables.Disposable; +import io.reactivex.rxjava4.disposables.DisposableStreamerCancellation; import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource; import io.reactivex.rxjava4.internal.util.ExceptionHelper; @@ -46,8 +53,13 @@ protected void subscribeActual(@NonNull Observer observer) { sto.drain(); } - record StreamToObserver(Streamer streamer, Observer observer, DisposableContainer cancellation, - AtomicInteger wip, AtomicBoolean done, AtomicReference mainError, AtomicBoolean disposed) + record StreamToObserver(Streamer streamer, + Observer observer, + DisposableStreamerCancellation cancellation, + AtomicInteger wip, + AtomicBoolean done, + AtomicReference mainError, + AtomicBoolean disposed) implements BiConsumer, Disposable { void drain() { diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableZip.java b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableZip.java index 7be0e4f134..d5216e8e3a 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableZip.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableZip.java @@ -13,14 +13,19 @@ package io.reactivex.rxjava4.internal.operators.streamable; -import java.util.*; -import java.util.concurrent.*; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiConsumer; import io.reactivex.rxjava4.annotations.NonNull; -import io.reactivex.rxjava4.core.*; -import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.core.Streamer; +import io.reactivex.rxjava4.disposables.DisposableStreamerCancellation; +import io.reactivex.rxjava4.disposables.StreamerCancellation; import io.reactivex.rxjava4.internal.util.AtomicThrowable; public record StreamableZip( @@ -50,7 +55,7 @@ public record StreamableZip( static final class ZipStreamer implements Streamer>, BiConsumer { - final DisposableContainer innerCancellation; + final DisposableStreamerCancellation innerCancellation; final List> streamers; @@ -72,7 +77,7 @@ static final class ZipStreamer volatile boolean done; - ZipStreamer(List> streamers, DisposableContainer innerCancellation) { + ZipStreamer(List> streamers, DisposableStreamerCancellation innerCancellation) { this.streamers = streamers; this.innerCancellation = innerCancellation; this.wip = new AtomicInteger(); diff --git a/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualCreateExecutor.java b/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualCreateExecutor.java index 2fe02df5c0..16d21e4d14 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualCreateExecutor.java +++ b/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualCreateExecutor.java @@ -15,14 +15,22 @@ import java.io.Serial; import java.util.Objects; -import java.util.concurrent.*; -import java.util.concurrent.Flow.*; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Flow.Subscriber; +import java.util.concurrent.Flow.Subscription; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.atomic.AtomicLong; -import io.reactivex.rxjava4.core.*; -import io.reactivex.rxjava4.disposables.*; -import io.reactivex.rxjava4.exceptions.*; -import io.reactivex.rxjava4.internal.util.*; +import io.reactivex.rxjava4.core.Flowable; +import io.reactivex.rxjava4.core.Scheduler; +import io.reactivex.rxjava4.core.VirtualEmitter; +import io.reactivex.rxjava4.core.VirtualGenerator; +import io.reactivex.rxjava4.disposables.CompositeDisposable; +import io.reactivex.rxjava4.disposables.DisposableStreamerCancellation; +import io.reactivex.rxjava4.exceptions.Exceptions; +import io.reactivex.rxjava4.internal.util.BackpressureHelper; +import io.reactivex.rxjava4.internal.util.ExceptionHelper; /** * Runs a generator callback on a virtual thread backed by a Worker of the given scheduler @@ -83,7 +91,7 @@ static final class ExecutorVirtualCreateSubscription extends AtomicLong Scheduler.Worker worker; - final DisposableContainer canceller; + final DisposableStreamerCancellation canceller; ExecutorVirtualCreateSubscription(Subscriber downstream, VirtualGenerator generator) { this.downstream = downstream; @@ -153,7 +161,7 @@ public void emit(T item) throws Throwable { } @Override - public DisposableContainer canceller() { + public DisposableStreamerCancellation canceller() { return canceller; } } diff --git a/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualTransformExecutor.java b/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualTransformExecutor.java index 262399f1ea..5acb6f3ba3 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualTransformExecutor.java +++ b/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualTransformExecutor.java @@ -96,7 +96,7 @@ static final class ExecutorVirtualTransformSubscriber extends AtomicLong Worker worker; - final DisposableContainer canceller; + final DisposableStreamerCancellation canceller; volatile boolean stopped; @@ -246,7 +246,7 @@ public Void call() { } @Override - public DisposableContainer canceller() { + public DisposableStreamerCancellation canceller() { return canceller; } diff --git a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java index f2b63c8826..89541434e0 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java +++ b/src/test/java/io/reactivex/rxjava4/validators/CheckParamValidationTest.java @@ -13,26 +13,123 @@ package io.reactivex.rxjava4.validators; -import java.lang.reflect.*; +import java.lang.reflect.Method; +import java.lang.reflect.Modifier; import java.time.Duration; -import java.util.*; -import java.util.concurrent.*; -import java.util.concurrent.Flow.*; -import java.util.stream.*; - -import org.junit.jupiter.api.*; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.Callable; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Flow.Publisher; +import java.util.concurrent.Flow.Subscriber; +import java.util.concurrent.Flow.Subscription; +import java.util.concurrent.Future; +import java.util.concurrent.FutureTask; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collector; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; import io.reactivex.rxjava4.annotations.NonNull; -import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.core.BackpressureOverflowStrategy; +import io.reactivex.rxjava4.core.BackpressureStrategy; +import io.reactivex.rxjava4.core.Completable; +import io.reactivex.rxjava4.core.CompletableConverter; +import io.reactivex.rxjava4.core.CompletableEmitter; +import io.reactivex.rxjava4.core.CompletableObserver; +import io.reactivex.rxjava4.core.CompletableOnSubscribe; +import io.reactivex.rxjava4.core.CompletableOperator; +import io.reactivex.rxjava4.core.CompletableSource; +import io.reactivex.rxjava4.core.CompletableTransformer; +import io.reactivex.rxjava4.core.CompletionStageDisposable; +import io.reactivex.rxjava4.core.Flowable; +import io.reactivex.rxjava4.core.FlowableConverter; +import io.reactivex.rxjava4.core.FlowableEmitter; +import io.reactivex.rxjava4.core.FlowableOnSubscribe; +import io.reactivex.rxjava4.core.FlowableOperator; +import io.reactivex.rxjava4.core.FlowableSubscriber; +import io.reactivex.rxjava4.core.FlowableTransformer; +import io.reactivex.rxjava4.core.Maybe; +import io.reactivex.rxjava4.core.MaybeConverter; +import io.reactivex.rxjava4.core.MaybeEmitter; +import io.reactivex.rxjava4.core.MaybeObserver; +import io.reactivex.rxjava4.core.MaybeOnSubscribe; +import io.reactivex.rxjava4.core.MaybeOperator; +import io.reactivex.rxjava4.core.MaybeSource; +import io.reactivex.rxjava4.core.MaybeTransformer; import io.reactivex.rxjava4.core.Observable; +import io.reactivex.rxjava4.core.ObservableConverter; +import io.reactivex.rxjava4.core.ObservableEmitter; +import io.reactivex.rxjava4.core.ObservableOnSubscribe; +import io.reactivex.rxjava4.core.ObservableOperator; +import io.reactivex.rxjava4.core.ObservableSource; +import io.reactivex.rxjava4.core.ObservableTransformer; import io.reactivex.rxjava4.core.Observer; -import io.reactivex.rxjava4.core.config.*; -import io.reactivex.rxjava4.disposables.*; +import io.reactivex.rxjava4.core.RxJavaTest; +import io.reactivex.rxjava4.core.Scheduler; +import io.reactivex.rxjava4.core.Single; +import io.reactivex.rxjava4.core.SingleConverter; +import io.reactivex.rxjava4.core.SingleEmitter; +import io.reactivex.rxjava4.core.SingleObserver; +import io.reactivex.rxjava4.core.SingleOnSubscribe; +import io.reactivex.rxjava4.core.SingleOperator; +import io.reactivex.rxjava4.core.SingleSource; +import io.reactivex.rxjava4.core.SingleTransformer; +import io.reactivex.rxjava4.core.StreamSink; +import io.reactivex.rxjava4.core.Streamable; +import io.reactivex.rxjava4.core.StreamableConverter; +import io.reactivex.rxjava4.core.StreamableOperator; +import io.reactivex.rxjava4.core.Streamer; +import io.reactivex.rxjava4.core.VirtualGenerator; +import io.reactivex.rxjava4.core.VirtualTransformer; +import io.reactivex.rxjava4.core.config.OnBackpressureBufferConfig; +import io.reactivex.rxjava4.core.config.SampleConfig; +import io.reactivex.rxjava4.core.config.SequenceEqualConfig; +import io.reactivex.rxjava4.core.config.StandardBufferedConfig; +import io.reactivex.rxjava4.core.config.StandardConcurrentBufferedConfig; +import io.reactivex.rxjava4.core.config.StandardConcurrentConfig; +import io.reactivex.rxjava4.core.config.StreamableInterceptConfig; +import io.reactivex.rxjava4.disposables.CompositeDisposable; +import io.reactivex.rxjava4.disposables.Disposable; +import io.reactivex.rxjava4.disposables.DisposableContainer; +import io.reactivex.rxjava4.disposables.DisposableStreamerCancellation; +import io.reactivex.rxjava4.disposables.StreamerCancellation; import io.reactivex.rxjava4.exceptions.TestException; -import io.reactivex.rxjava4.functions.*; +import io.reactivex.rxjava4.functions.Action; +import io.reactivex.rxjava4.functions.BiConsumer; +import io.reactivex.rxjava4.functions.BiFunction; +import io.reactivex.rxjava4.functions.BiPredicate; +import io.reactivex.rxjava4.functions.BooleanSupplier; +import io.reactivex.rxjava4.functions.Consumer; +import io.reactivex.rxjava4.functions.Function; +import io.reactivex.rxjava4.functions.Function3; +import io.reactivex.rxjava4.functions.Function4; +import io.reactivex.rxjava4.functions.Function5; +import io.reactivex.rxjava4.functions.Function6; +import io.reactivex.rxjava4.functions.Function7; +import io.reactivex.rxjava4.functions.Function8; +import io.reactivex.rxjava4.functions.Function9; +import io.reactivex.rxjava4.functions.LongConsumer; +import io.reactivex.rxjava4.functions.Predicate; +import io.reactivex.rxjava4.functions.Supplier; import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.internal.operators.streamable.StreamableNever; -import io.reactivex.rxjava4.parallel.*; +import io.reactivex.rxjava4.parallel.ParallelFailureHandling; +import io.reactivex.rxjava4.parallel.ParallelFlowable; +import io.reactivex.rxjava4.parallel.ParallelFlowableConverter; +import io.reactivex.rxjava4.parallel.ParallelTransformer; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.processors.DispatchStreamProcessor; import io.reactivex.rxjava4.schedulers.Schedulers; @@ -595,6 +692,8 @@ public void checkStreamable() { defaultValues.put(ParallelFailureHandling.class, ParallelFailureHandling.ERROR); defaultValues.put(DisposableContainer.class, new CompositeDisposable()); + defaultValues.put(DisposableStreamerCancellation.class, new CompositeDisposable()); + defaultValues.put(StreamerCancellation.class, new CompositeDisposable()); // JDK 8 types