4.x: Enhance Streamable/Streamer and DisposableContainer APIs

Author: akarnokd

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

@@ -17,6 +17,7 @@
 import java.util.concurrent.CompletionStage;
 
 import io.reactivex.rxjava4.annotations.NonNull;
+import io.reactivex.rxjava4.disposables.*;
 
 /**
  * A realized stream which can then be consumed asynchronously in steps.
@@ -29,10 +30,21 @@ public interface Streamer<@NonNull T> extends AutoCloseable {
 
     /**
      * Determine if there are more elements available from the source.
+     * @param cancellation ability to perform cancellation on a per-virtual-pull request.
      * @return eventually true or false, indicating availability or termination
      */
     @NonNull
-    CompletionStage<Boolean> next();
+    CompletionStage<Boolean> next(@NonNull DisposableContainer cancellation);
+
+    /**
+     * Determine if there are more elements available from the source.
+     * Uses a default, individual {@link CompositeDisposable} to manage cancellation.
+     * @return eventually true or false, indicating availability or termination
+     */
+    @NonNull
+    default CompletionStage<Boolean> next() {
+        return next(new CompositeDisposable());
+    }
 
     /**
      * Returns the current element if {@link #next()} yielded {@code true}.
@@ -52,7 +64,7 @@ public interface Streamer<@NonNull T> extends AutoCloseable {
     CompletionStage<Void> cancel();
 
     /**
-     * Make this Streamer a resource and a Closeable.
+     * Make this Streamer a resource and a Closeable, allowing virtually blocking closing.
      */
     default void close() {
         cancel().toCompletableFuture().join();

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

@@ -255,4 +255,17 @@ void dispose(@Nullable OpenHashSet<Disposable> set) {
             throw new CompositeException(errors);
         }
     }
+
+    @Override
+    public void reset() {
+        if (disposed) {
+            return;
+        }
+        synchronized (this) {
+            if (disposed) {
+                return;
+            }
+            resources = null;
+        }
+    }
 }

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

@@ -42,4 +42,40 @@ public interface DisposableContainer extends Disposable {
      * @return true if the operation was successful
      */
     boolean delete(Disposable d);
+
+    /**
+     * Removes all contained {@link Disposable}s without disposing them, making this
+     * container fresh.
+     */
+    void reset();
+
+    /**
+     * Removes and disposes all contained {@link Disposable}s, making this container fresh
+     * without disposing the entire container.
+     */
+    void clear();
+
+    /**
+     * 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.
+     */
+    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.
+     */
+    default Disposable subscribe(Disposable d) {
+        add(d);
+        return Disposable.fromRunnable(() -> delete(d));
+    }
 }

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

@@ -184,4 +184,18 @@ void dispose(List<Disposable> set) {
             throw new CompositeException(errors);
         }
     }
+
+    @Override
+    public void reset() {
+        if (disposed) {
+            return;
+        }
+        synchronized (this) {
+            if (disposed) {
+                return;
+            }
+            resources = null;
+        }
+    }
+
 }

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

@@ -18,7 +18,7 @@
 
 import io.reactivex.rxjava4.annotations.NonNull;
 import io.reactivex.rxjava4.core.*;
-import io.reactivex.rxjava4.disposables.DisposableContainer;
+import io.reactivex.rxjava4.disposables.*;
 
 public final class StreamableEmpty<T> extends Streamable<T> {
 
@@ -30,7 +30,7 @@ public final class StreamableEmpty<T> extends Streamable<T> {
     static final class EmptyStreamer<T> implements Streamer<T> {
 
         @Override
-        public @NonNull CompletionStage<Boolean> next() {
+        public @NonNull CompletionStage<Boolean> next(DisposableContainer cancellation) {
             return CompletableFuture.completedStage(false); // TODO would constant stages work here or is that contention?
         }
 

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

@@ -48,7 +48,7 @@ static final class JustStreamer<T> implements Streamer<T>, Disposable {
         }
 
         @Override
-        public @NonNull CompletionStage<Boolean> next() {
+        public @NonNull CompletionStage<Boolean> next(DisposableContainer cancellation) {
             if (stage == 0) {
                 stage = 1;
                 return CompletableFuture.completedStage(true);