4.x: Streamable ~ separate streamer cancellation & DisposableContainer

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<T> withCancellation(DisposableContainer cancellation) {
+    default StreamSink<T> withCancellation(DisposableStreamerCancellation cancellation) {
         Objects.requireNonNull(cancellation, "cancellation is null");
         return new StreamSinkWithCancellation<>(this, cancellation);
     }

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.
      * <p>
      * It is recommened you {@link StreamerCancellation#remove(Disposable)} or
@@ -987,7 +987,8 @@ default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> con
      */
     @CheckReturnValue
     @NonNull
-    default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> consumer, @NonNull DisposableContainer canceller) {
+    default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> consumer,
+            @NonNull DisposableStreamerCancellation canceller) {
         return forEach(consumer, canceller, Executors.newVirtualThreadPerTaskExecutor());
     }
 
@@ -1014,7 +1015,8 @@ default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> con
      */
     @CheckReturnValue
     @NonNull
-    default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> consumer, @NonNull DisposableContainer canceller, @NonNull ExecutorService executor) {
+    default CompletionStageDisposable<Void> forEach(@NonNull Consumer<? super T> 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<Void> forEach(@NonNull Consumer<? super T> con
     @NonNull
     default CompletionStageDisposable<Void> forEach(
             @NonNull BiConsumer<? super T, ? super Disposable> consumer,
-            @NonNull DisposableContainer canceller,
+            @NonNull DisposableStreamerCancellation canceller,
             @NonNull ExecutorService executor) {
         Objects.requireNonNull(consumer, "consumer is null");
         Objects.requireNonNull(canceller, "canceller is null");

src/main/java/io/reactivex/rxjava4/core/VirtualEmitter.java

@@ -33,5 +33,5 @@ public interface VirtualEmitter<T> {
      * Returns a disposable container to relay cancellation notifications while awaiting the run.
      * @return a new Disposable Container instance
      */
-    DisposableContainer canceller();
+    DisposableStreamerCancellation canceller();
 }

src/main/java/io/reactivex/rxjava4/disposables/CompositeDisposable.java

@@ -24,7 +24,8 @@
  * offers <em>O(1)</em> 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<Disposable> resources;
 
@@ -270,7 +271,7 @@ public void reset() {
     }
 
     @Override
-    public DisposableContainer derive() {
+    public DisposableStreamerCancellation derive() {
         var result = new CompositeDisposable();
 
         add(result);

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

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.
+ * <p>
+ * 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 {
+
+}

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.
  * <p>
- * 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.
  * <p>
@@ -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();
 
 }

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<Disposable> resources;
 
@@ -199,7 +200,7 @@ public void reset() {
     }
 
     @Override
-    public DisposableContainer derive() {
+    public DisposableStreamerCancellation derive() {
         var result = new ListCompositeDisposable();
 
         add(result);

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamSinkWithCancellation.java

@@ -16,20 +16,21 @@
 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 <T> the element type of the stream
  * @param downstream the {@code StreamSink} to relay events to
  * @param cancellation the {@code DisposableContainer} to be used for indicating cancellation
  * @since 4.0.0
  */
 public record StreamSinkWithCancellation<@NonNull T>(
-        @NonNull StreamSink<T> downstream, @NonNull DisposableContainer cancellation)
+        @NonNull StreamSink<T> downstream, @NonNull DisposableStreamerCancellation cancellation)
 implements StreamSink<T> {
 
     public StreamSinkWithCancellation {

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableConcatIterable.java

@@ -41,7 +41,7 @@ static final class ConcatIteratorStreamer<T> extends AtomicInteger implements St
 
         final StreamerCancellation cancellation;
 
-        DisposableContainer currentCancellation;
+        DisposableStreamerCancellation currentCancellation;
 
         Streamer<? extends T> upstream;
 

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFlatMap.java

@@ -189,13 +189,14 @@ static final class InnerStreamer<T, R> extends AtomicInteger implements BiConsum
 
         final FlatMapMainStreamer<T, R> parent;
 
-        final DisposableContainer canceller;
+        final DisposableStreamerCancellation canceller;
 
         volatile boolean done;
 
         boolean finishing;
 
-        InnerStreamer(Streamer<? extends R> inner, FlatMapMainStreamer<T, R> parent, DisposableContainer canceller) {
+        InnerStreamer(Streamer<? extends R> inner, FlatMapMainStreamer<T, R> parent,
+                DisposableStreamerCancellation canceller) {
             this.inner = inner;
             this.parent = parent;
             this.canceller = canceller;

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java

@@ -31,7 +31,7 @@ public record StreamableForEach() {
     public static <T> CompletionStageDisposable<Void> forEach(
             @NonNull Streamable<T> me,
             @NonNull Consumer<? super T> consumer,
-            @NonNull DisposableContainer canceller,
+            @NonNull DisposableStreamerCancellation canceller,
             @NonNull ExecutorService executor) {
         var future = CompletableFuture.<Void>supplyAsync(() -> {
             Throwable finallyCrash = null;
@@ -71,7 +71,7 @@ public static <T> CompletionStageDisposable<Void> forEach(
     public static <T> CompletionStageDisposable<Void> forEach(
             @NonNull Streamable<T> me,
             @NonNull BiConsumer<? super T, ? super Disposable> consumer,
-            @NonNull DisposableContainer canceller,
+            @NonNull DisposableStreamerCancellation canceller,
             @NonNull ExecutorService executor) {
         var future = CompletableFuture.<Void>supplyAsync(() -> {
             var str = me.stream(canceller);
@@ -110,7 +110,8 @@ public static <T> CompletionStageDisposable<Void> forEach(
         return new CompletionStageDisposable<>(future, canceller);
     }
 
-    public static <T> CompletionStage<Void> forEach(Streamable<T> me, StreamSink<? super T> consumer, ExecutorService executor) {
+    public static <T> CompletionStage<Void> forEach(
+            Streamable<T> me, StreamSink<? super T> consumer, ExecutorService executor) {
         var cf = new CompletableFuture<Void>();
         CompletableFuture.runAsync(() -> {
             Throwable error = null;

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<T, U>(
@@ -48,9 +53,9 @@ static final class TakeUntilMainStreamer<T, U> implements Streamer<T> {
 
         final Streamer<? extends U> otherStreamer;
 
-        final DisposableContainer mainCancellation;
+        final DisposableStreamerCancellation mainCancellation;
 
-        final DisposableContainer otherCancellation;
+        final DisposableStreamerCancellation otherCancellation;
 
         final CompletionStage<Boolean> otherNext;
 
@@ -59,8 +64,8 @@ static final class TakeUntilMainStreamer<T, U> implements Streamer<T> {
         public TakeUntilMainStreamer(
                 Streamer<? extends T> upstream,
                 Streamer<? extends U> otherStreamer,
-                DisposableContainer mainCancellation,
-                DisposableContainer otherCancellation,
+                DisposableStreamerCancellation mainCancellation,
+                DisposableStreamerCancellation otherCancellation,
                 CompletionStage<Boolean> otherNext
         ) {
             this.upstream = upstream;

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<? super @NonNull T> observer) {
         sto.drain();
     }
 
-    record StreamToObserver<T>(Streamer<T> streamer, Observer<? super T> observer, DisposableContainer cancellation,
-            AtomicInteger wip, AtomicBoolean done, AtomicReference<Throwable> mainError, AtomicBoolean disposed)
+    record StreamToObserver<T>(Streamer<T> streamer,
+            Observer<? super T> observer,
+            DisposableStreamerCancellation cancellation,
+            AtomicInteger wip,
+            AtomicBoolean done,
+            AtomicReference<Throwable> mainError,
+            AtomicBoolean disposed)
     implements BiConsumer<Object, Throwable>, Disposable {
 
         void drain() {