Fix bad cancellation storm

Author: akarnokd

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

@@ -16,7 +16,6 @@
 import java.lang.reflect.InvocationTargetException;
 import java.util.*;
 import java.util.concurrent.*;
-import java.util.concurrent.atomic.AtomicInteger;
 
 import io.reactivex.rxjava4.annotations.*;
 import io.reactivex.rxjava4.disposables.*;
@@ -203,13 +202,10 @@ static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source, @NonNu
      */
     static <@NonNull T> Streamable<T> concat(Streamable<? extends Streamable<? extends T>> sources, ExecutorService exec) {
         return create(emitter -> {
-            var counter = new AtomicInteger();
             try (var mainSource = sources.forEach(item -> {
-                System.out.println(counter.incrementAndGet());
                 try (var innerSource = item.forEach(inner -> {
-                    System.out.println("> " + inner);
                     emitter.emit(inner);
-                }, emitter.canceller(), exec)) {
+                }, emitter.canceller().derive(), exec)) {
                     innerSource.await(emitter.canceller());
                 }
             }, emitter.canceller(), exec)) {

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

@@ -268,4 +268,14 @@ public void reset() {
             resources = null;
         }
     }
+
+    @Override
+    public DisposableContainer derive() {
+        var result = new CompositeDisposable();
+
+        add(result);
+        result.add(Disposable.fromRunnable(() -> delete(result)));
+
+        return result;
+    }
 }

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

@@ -56,6 +56,14 @@ 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.
@@ -133,5 +141,10 @@ public void reset() {
         public void clear() {
             // Who cares?
         }
+
+        @Override
+        public DisposableContainer derive() {
+            return NEVER;
+        }
     }
 }

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

@@ -198,4 +198,15 @@ public void reset() {
         }
     }
 
+
+    @Override
+    public DisposableContainer derive() {
+        var result = new ListCompositeDisposable();
+
+        add(result);
+        result.add(Disposable.fromRunnable(() -> delete(result)));
+
+        return result;
+    }
+
 }