4.x: Streamable.stream() DisposableContainer -> StreamerCancellation (#8221)

* 4.x: Streamable.stream() DisposableContainer -> StreamerCancellation

* remove unused never disposable container, coverage

Author: akarnokd

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

@@ -33,7 +33,7 @@
 ///
 /// The lifecycle of the sequence is as follows:
 ///
-/// - consumer calls {@link #stream(DisposableContainer)}
+/// - consumer calls {@link #stream(StreamerCancellation)}
 /// - consumer calls {@link Streamer#next()} in a loop and consumer checks if what
 ///   the {@link CompletionStage} outcome is:
 ///   - if it succeeded with a boolean {@code true}, it is safe to call {@link Streamer#current()}.
@@ -44,7 +44,7 @@
 /// It is always necessary to have the consumer call {@code finish} because that is responsible for cleaning up
 /// resources of the upstream.
 ///
-/// Downstream cancellations are signaled via the [DisposableContainer], where operators can register their own
+/// Downstream cancellations are signaled via the [StreamerCancellation], where operators can register their own
 /// [Disposable]s that get disposed. Because dispose can happen at any time and asynchronously to the consumption loop,
 /// the sensitive sources must complete their waiting `CompletionStage` returned by `next` exceptionally via a
 /// [CancellationException]. This will unblock the loops and invoke the `finish` method of the lifecycle at
@@ -72,12 +72,24 @@ public interface Streamable<@NonNull T> {
 
     /**
      * Realizes the stream and returns an interface that lets one consume it.
-     * @param cancellation where to register and listen for cancellation calls.
+     * <p>
+     * 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
+     * the parent {@code cancellation} instance by accident.
+     * <p>
+     * It is recommened you {@link StreamerCancellation#remove(Disposable)} or
+     * {@link StreamerCancellation#delete(Disposable)} any extra resources you added
+     * via {@link StreamerCancellation#add(Disposable)} when your operator
+     * reaches the {@link Streamer#finish()} call.
+     * @param cancellation where to register, check and listen for cancellation calls via
+     *                     its {@link StreamerCancellation#isDisposed()}
      * @return the Streamer instance to consume.
      */
     @CheckReturnValue
     @NonNull
-    Streamer<T> stream(@NonNull DisposableContainer cancellation);
+    Streamer<T> stream(@NonNull StreamerCancellation cancellation);
 
     // oooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooo
     // Data sources and wrappers
@@ -561,7 +573,7 @@ static Streamable<Long> timer(long delay, TimeUnit unit, ExecutorService executo
      * once the {@code Streamable} terminated.
      * @param <T> the element type of the sequence
      * @param <R> the resource type
-     * @param resourceSupplier supplies a resource object per {@link #stream(DisposableContainer)} call
+     * @param resourceSupplier supplies a resource object per {@link #stream(StreamerCancellation)} call
      * @param resourceMapper maps the supplied resource into a {@code Streamable} source
      * @param resourceCleaner cleans up the supplied resource
      * @return the new {@code Streamable} instance
@@ -725,7 +737,7 @@ default Streamable<T> intercept(StreamableInterceptConfig<T> config) {
     /**
      * <strong>This method requires advanced knowledge about building operators, please consider
      * other standard composition methods first;</strong>
-     * Returns a {@code Streamable} instance which when its {@link #stream(DisposableContainer)} is invoked,
+     * Returns a {@code Streamable} instance which when its {@link #stream(StreamerCancellation)} is invoked,
      * applies the specified operator callback to the upstream {@link Streamer} to produce
      * an actual {@code Streamer} instance to be handed downstream.
      * <p>

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

@@ -14,7 +14,7 @@
 package io.reactivex.rxjava4.core;
 
 import io.reactivex.rxjava4.annotations.NonNull;
-import io.reactivex.rxjava4.disposables.DisposableContainer;
+import io.reactivex.rxjava4.disposables.StreamerCancellation;
 
 /**
  * Interface to map/wrap an upstream {@link Streamer} to an downstream {@code Streamer}.
@@ -27,12 +27,12 @@
 public interface StreamableOperator<@NonNull T, @NonNull R> {
     /**
      * Applies a function to the upstream {@link Streamer} and returns a new downstream {@code Streamer}.
-     * @param container the {@link DisposableContainer} handling the cancellation propagation for the downstream
+     * @param container the {@link StreamerCancellation} handling the cancellation propagation for the downstream
      * @param streamer the upstream {@code Streamer} instance
      * @return the downstream {@code Streamer} instance
      * @throws Throwable on failure
      */
     @NonNull
-    Streamer<? extends R> apply(@NonNull DisposableContainer container,
+    Streamer<? extends R> apply(@NonNull StreamerCancellation container,
             @NonNull Streamer<? extends T> streamer) throws Throwable;
 }

src/main/java/io/reactivex/rxjava4/core/config/StreamableInterceptConfig.java

@@ -18,30 +18,31 @@
 
 import io.reactivex.rxjava4.annotations.NonNull;
 import io.reactivex.rxjava4.core.*;
-import io.reactivex.rxjava4.disposables.DisposableContainer;
+import io.reactivex.rxjava4.disposables.*;
 import io.reactivex.rxjava4.functions.*;
 
 /**
  * Configuration record the intercept() operator with various lifecylce-stage transforming callbacks
  * @param <T> the element type of the sequence
- * @param onStream called when the {@link Streamable#stream(io.reactivex.rxjava4.disposables.DisposableContainer)} is invoked
+ * @param onStream called when the {@link Streamable#stream(StreamerCancellation)} is invoked
  * @param onNext called when the {@link Streamer#next()} is invoked
  * @param onCurrent called when the {@link Streamer#current()} is invoked
  * @param onFinish called when the {@link Streamer#finish()} is invoked
  * @since 4.0.0
  */
 public record StreamableInterceptConfig<T>(
-        @NonNull BiFunction<? super DisposableContainer, ? super Streamer<? extends T>, ? extends Streamer<? extends T>> onStream,
-        @NonNull BiFunction<? super DisposableContainer, ? super CompletionStage<Boolean>, ? extends CompletionStage<Boolean>> onNext,
+        @NonNull BiFunction<? super StreamerCancellation, ? super Streamer<? extends T>, ? extends Streamer<? extends T>> onStream,
+        @NonNull BiFunction<? super StreamerCancellation, ? super CompletionStage<Boolean>, ? extends CompletionStage<Boolean>> onNext,
         @NonNull Function<? super T, ? extends T> onCurrent,
-        @NonNull BiFunction<? super DisposableContainer, ? super CompletionStage<Void>, ? extends CompletionStage<Void>> onFinish
+        @NonNull BiFunction<? super StreamerCancellation, ? super CompletionStage<Void>, ? extends CompletionStage<Void>> onFinish
 ) {
 
     /**
      * Constructs a configuration with a custom {@link #onNext()} intercept and everything else is pass-through.
      * @param onNext the callback for intercepting the {@code next()} calls
      */
-    public StreamableInterceptConfig(@NonNull BiFunction<? super DisposableContainer, ? super CompletionStage<Boolean>, ? extends CompletionStage<Boolean>> onNext) {
+    public StreamableInterceptConfig(
+            @NonNull BiFunction<? super StreamerCancellation, ? super CompletionStage<Boolean>, ? extends CompletionStage<Boolean>> onNext) {
         this((_, v) -> v, onNext, v -> v, (_, v) -> v);
     }
 
@@ -55,7 +56,7 @@ public StreamableInterceptConfig(@NonNull Function<? super T, ? extends T> onCur
 
     /**
      * Constructs a fully configured record.
-     * @param onStream called when the {@link Streamable#stream(io.reactivex.rxjava4.disposables.DisposableContainer)} is invoked
+     * @param onStream called when the {@link Streamable#stream(StreamerCancellation)} is invoked
      * @param onNext called when the {@link Streamer#next()} is invoked
      * @param onCurrent called when the {@link Streamer#current()} is invoked
      * @param onFinish called when the {@link Streamer#finish()} is invoked

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

@@ -274,8 +274,21 @@ public DisposableContainer derive() {
         var result = new CompositeDisposable();
 
         add(result);
-        result.add(Disposable.fromRunnable(() -> delete(result)));
+        result.add(new DerivedCleaner(this, result));
 
         return result;
     }
+
+    record DerivedCleaner(DisposableContainer parent, Disposable child) implements Disposable {
+
+        @Override
+        public void dispose() {
+            parent.delete(child);
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return child.isDisposed();
+        }
+    }
 }

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

@@ -14,34 +14,10 @@
 package io.reactivex.rxjava4.disposables;
 
 /**
- * Common interface to add and remove disposables from a container.
+ * Common interface to add and remove {@link Disposable}s from a container.
  * @since 2.0
  */
-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(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(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(Disposable d);
+public interface DisposableContainer extends Disposable, StreamerCancellation {
 
     /**
      * Removes all contained {@link Disposable}s without disposing them, making this
@@ -56,95 +32,4 @@ public interface DisposableContainer extends Disposable {
      */
     void clear();
 
-    /**
-     * Create a derived sub container that can get cancelled by this container,
-     * but cancelling the subcontainer does not cancel this container.
-     * @return the derived subcontainer
-     * @since 4.0
-     */
-    DisposableContainer derive();
-
-    /**
-     * Registers a {@link Disposable} with this container so that it can be removed and disposed
-     * via a simple {@link #dispose()} call to the returned Disposable.
-     * @param d the disposable to register
-     * @return the Disposable to trigger a {@link #remove(Disposable)}
-     * @see #subscribe(Disposable) for non-disposing removal.
-     * @since 4.0.0
-     */
-    default Disposable register(Disposable d) {
-        add(d);
-        return Disposable.fromRunnable(() -> remove(d));
-    }
-
-    /**
-     * Registers a {@link Disposable} with this container so that it can be deleted, not disposed
-     * via a simple {@link #dispose()} call to the returned Disposable.
-     * @param d the disposable to register
-     * @return the Disposable to trigger a {@link #remove(Disposable)}
-     * @see #subscribe(Disposable) for non-disposing removal.
-     * @since 4.0.0
-     */
-    default Disposable subscribe(Disposable d) {
-        add(d);
-        return Disposable.fromRunnable(() -> delete(d));
-    }
-
-    /**
-     * The container implementation that just ignores everything, for
-     * cases where the dispose signal has no side effects to work with.
-     * @since 4.0.0
-     */
-    DisposableContainer NEVER = new NeverDisposableContainer();
-
-    /**
-     * Implementation of a never disposable container.
-     * @since 4.0.0
-     */
-    record NeverDisposableContainer() implements DisposableContainer {
-
-        @Override
-        public void dispose() {
-            // Deliberately empty
-        }
-
-        @Override
-        public boolean isDisposed() {
-            // Who cares?
-            return false;
-        }
-
-        @Override
-        public boolean add(Disposable d) {
-            // Who cares?
-            return false;
-        }
-
-        @Override
-        public boolean remove(Disposable d) {
-            // Who cares?
-            return false;
-        }
-
-        @Override
-        public boolean delete(Disposable d) {
-            // Who cares?
-            return false;
-        }
-
-        @Override
-        public void reset() {
-            // Who cares?
-        }
-
-        @Override
-        public void clear() {
-            // Who cares?
-        }
-
-        @Override
-        public DisposableContainer derive() {
-            return NEVER;
-        }
-    }
 }

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

@@ -0,0 +1,75 @@
+/*
+ * 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;
+
+import io.reactivex.rxjava4.annotations.NonNull;
+import io.reactivex.rxjava4.core.Streamable;
+
+/**
+ * Represents non-disposable view of a {@link DisposableContainer}
+ * 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()}
+ * in {@link Streamable#stream(StreamerCancellation)} implementations because
+ * disposing a stream is the privilege of the caller/downstream.
+ * <p>
+ * Use the {@link #derive()} to create a sub-container with full disposability access.
+ * <p>
+ * This interface doesn't support {@link DisposableContainer#reset()} nor
+ * {@link DisposableContainer#clear()} because it would allow accidentally removing another
+ * operator's added/registered {@code Disposable}s.
+ * @since 4.0.0
+ */
+public interface StreamerCancellation {
+    /**
+     * Returns true if this resource has been disposed.
+     * @return true if this resource has been disposed
+     */
+    boolean isDisposed();
+
+    /**
+     * 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);
+
+    /**
+     * 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
+     */
+    @NonNull
+    DisposableContainer derive();
+
+}

src/main/java/io/reactivex/rxjava4/internal/disposables/ListCompositeDisposable.java

@@ -22,7 +22,7 @@
 /**
  * A disposable container that can hold onto multiple other disposables.
  */
-public final class ListCompositeDisposable implements Disposable, DisposableContainer {
+public final class ListCompositeDisposable implements Disposable, DisposableContainer, StreamerCancellation {
 
     List<Disposable> resources;
 

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

@@ -20,7 +20,7 @@
 
 import io.reactivex.rxjava4.annotations.NonNull;
 import io.reactivex.rxjava4.core.*;
-import io.reactivex.rxjava4.disposables.DisposableContainer;
+import io.reactivex.rxjava4.disposables.StreamerCancellation;
 import io.reactivex.rxjava4.internal.fuseable.HasUpstreamStreamableSource;
 
 public record StreamableCollector<T, A, R>(
@@ -29,7 +29,7 @@ public record StreamableCollector<T, A, R>(
 ) implements Streamable<R>, HasUpstreamStreamableSource<T> {
 
     @Override
-    public @NonNull Streamer<@NonNull R> stream(@NonNull DisposableContainer cancellation) {
+    public @NonNull Streamer<@NonNull R> stream(@NonNull StreamerCancellation cancellation) {
         return new CollectorStreamable<>(
                 source.stream(cancellation),
                 collector.supplier().get(),