- CompositeSubscription's dispose, clear, and remove methods no…

…w return a completion future

- Fixed an issue where a stream not present in CompositeSubscription was canceled
- Added the ability not to cancel the stream when it is removed from CompositeSubscription

lib/src/utils/composite_subscription.dart

@@ -48,26 +48,32 @@ class CompositeSubscription {
     return subscription;
   }
 
-  /// Cancels subscription and removes it from this composite.
-  void remove(StreamSubscription<dynamic> subscription) {
-    subscription.cancel();
-    _subscriptionsList.remove(subscription);
+  /// Remove the subscription from this composite and cancel it if it has been removed.
+  Future<void>? remove(
+    StreamSubscription<dynamic> subscription, {
+    bool shouldCancel = true,
+  }) {
+    if (_subscriptionsList.remove(subscription) && shouldCancel) {
+      return subscription.cancel();
+    }
   }
 
   /// Cancels all subscriptions added to this composite. Clears subscriptions collection.
   ///
   /// This composite can be reused after calling this method.
-  void clear() {
-    _subscriptionsList.cancelAll();
+  Future<void>? clear() {
+    final cancelAllDone = _subscriptionsList.cancelAll();
     _subscriptionsList.clear();
+    return cancelAllDone;
   }
 
   /// Cancels all subscriptions added to this composite. Disposes this.
   ///
   /// This composite can't be reused after calling this method.
-  void dispose() {
-    clear();
+  Future<void>? dispose() {
+    final clearDone = clear();
     _isDisposed = true;
+    return clearDone;
   }
 
   /// Pauses all subscriptions added to this composite.

test/streams/composite_subscription_test.dart

@@ -5,54 +5,96 @@ import 'package:test/test.dart';
 
 void main() {
   group('CompositeSubscription', () {
-    test('should cancel all subscriptions on clear()', () {
-      final stream = Stream.fromIterable(const [1, 2, 3]).shareValue();
-      final composite = CompositeSubscription();
+    group('Rx.compositeSubscription.clear', () {
+      test('should cancel all subscriptions', () {
+        final stream = Stream.fromIterable(const [1, 2, 3]).shareValue();
+        final composite = CompositeSubscription();
+
+        composite
+          ..add(stream.listen(null))
+          ..add(stream.listen(null))
+          ..add(stream.listen(null));
+
+        final done = composite.clear();
 
-      composite
-        ..add(stream.listen(null))
-        ..add(stream.listen(null))
-        ..add(stream.listen(null));
+        expect(stream, neverEmits(anything));
+        expect(done, isA<Future>());
+      });
+      test('should return null since no subscription has been canceled clear()',
+          () {
+        final composite = CompositeSubscription();
 
-      composite.clear();
+        final done = composite.clear();
 
-      expect(stream, neverEmits(anything));
+        expect(done, null);
+      });
     });
-    test('should cancel all subscriptions on dispose()', () {
-      final stream = Stream.fromIterable(const [1, 2, 3]).shareValue();
-      final composite = CompositeSubscription();
 
-      composite
-        ..add(stream.listen(null))
-        ..add(stream.listen(null))
-        ..add(stream.listen(null));
+    group('Rx.compositeSubscription.onDispose', () {
+      test('should cancel all subscriptions', () {
+        final stream = Stream.fromIterable(const [1, 2, 3]).shareValue();
+        final composite = CompositeSubscription();
 
-      composite.dispose();
+        composite
+          ..add(stream.listen(null))
+          ..add(stream.listen(null))
+          ..add(stream.listen(null));
 
-      expect(stream, neverEmits(anything));
-    });
-    test(
-        'should throw exception if trying to add subscription to disposed composite',
-        () {
-      final stream = Stream.fromIterable(const [1, 2, 3]).shareValue();
-      final composite = CompositeSubscription();
+        final done = composite.dispose();
 
-      composite.dispose();
+        expect(stream, neverEmits(anything));
+        expect(done, isA<Future>());
+      });
+      test(
+          'should return null since no subscription has been canceled on dispose()',
+          () {
+        final composite = CompositeSubscription();
 
-      expect(() => composite.add(stream.listen(null)), throwsA(anything));
-    });
-    test('should cancel subscription on if it is removed from composite', () {
-      const value = 1;
-      final stream = Stream.fromIterable([value]).shareValue();
-      final composite = CompositeSubscription();
-      final subscription = stream.listen(null);
+        final done = composite.dispose();
+
+        expect(done, null);
+      });
+      test(
+          'should throw exception if trying to add subscription to disposed composite',
+          () {
+        final stream = Stream.fromIterable(const [1, 2, 3]).shareValue();
+        final composite = CompositeSubscription();
 
-      composite.add(subscription);
-      composite.remove(subscription);
+        composite.dispose();
 
-      expect(stream, neverEmits(anything));
+        expect(() => composite.add(stream.listen(null)), throwsA(anything));
+      });
     });
-    test('Rx.compositeSubscription.pauseAndResume', () {
+
+    group('Rx.compositeSubscription.remove', () {
+      test('should cancel subscription on if it is removed from composite', () {
+        const value = 1;
+        final stream = Stream.fromIterable([value]).shareValue();
+        final composite = CompositeSubscription();
+        final subscription = stream.listen(null);
+
+        composite.add(subscription);
+        final done = composite.remove(subscription);
+
+        expect(stream, neverEmits(anything));
+        expect(done, isA<Future>());
+      });
+      test(
+          'should not cancel the subscription since it is not present in the composite',
+          () {
+        const value = 1;
+        final stream = Stream.fromIterable([value]).shareValue();
+        final composite = CompositeSubscription();
+        final subscription = stream.listen(null);
+
+        final done = composite.remove(subscription);
+
+        expect(stream, emits(anything));
+        expect(done, null);
+      });
+    });
+
+    test('Rx.compositeSubscription.pauseAndResume()', () {
       final composite = CompositeSubscription();
       final s1 = Stream.fromIterable(const [1, 2, 3]).listen(null),
           s2 = Stream.fromIterable(const [4, 5, 6]).listen(null);