Merge pull request #28 from Workiva/rap-1780-single-subscription-streams

RAP-1780 Handle single-subscription stream controllers better

lib/src/disposable/disposable.dart

@@ -30,10 +30,10 @@ class _InternalDisposable implements _Disposable {
 
   @override
   Future<Null> dispose() {
-    var disposeFuture = _disposer();
+    var disposeFuture = _disposer != null ? _disposer() : null;
     _disposer = null;
     if (disposeFuture == null) {
-      return new Future(() => null);
+      return new Future.value();
     }
     return disposeFuture.then((_) => null);
   }
@@ -300,12 +300,27 @@ class Disposable implements _Disposable, DisposableManagerV3 {
   @override
   void manageStreamController(StreamController controller) {
     _throwOnInvalidCall('manageStreamController', 'controller', controller);
-    _internalDisposables.add(new _InternalDisposable(() {
-      if (!controller.hasListener) {
+    // If a single-subscription stream has a subscription and that
+    // subscription is subsequently canceled, the `done` future will
+    // complete, but there is no other way for us to tell that this
+    // is what has happened. If we then listen to the stream (since
+    // closing a stream that was never listened to never completes) we'll
+    // get an exception. This workaround allows us to "know" when a
+    // subscription has been canceled so we don't bother trying to
+    // listen to the stream before closing it.
+    bool isDone = false;
+    var disposable = new _InternalDisposable(() {
+      if (!controller.hasListener && !controller.isClosed && !isDone) {
         controller.stream.listen((_) {});
       }
       return controller.close();
-    }));
+    });
+    controller.done.then((_) {
+      isDone = true;
+      _internalDisposables.remove(disposable);
+      disposable.dispose();
+    });
+    _internalDisposables.add(disposable);
   }
 
   @mustCallSuper

test/unit/vm/disposable_test.dart

@@ -262,8 +262,29 @@ void main() {
       });
 
       test(
-          'should close a single-subscription stream with no listener'
-          'when parent is disposed', () async {
+          'should complete normally for a single-subscription stream, with '
+          'a listener, that has been closed when parent is disposed', () async {
+        var controller = new StreamController();
+        var sub = controller.stream.listen(expectAsync1((_) {}, count: 0));
+        thing.manageStreamController(controller);
+        await controller.close();
+        await thing.dispose();
+        await sub.cancel();
+      });
+
+      test(
+          'should complete normally for a single-subscription stream with a '
+          'canceled listener when parent is disposed', () async {
+        var controller = new StreamController();
+        var sub = controller.stream.listen(expectAsync1((_) {}, count: 0));
+        thing.manageStreamController(controller);
+        await sub.cancel();
+        await thing.dispose();
+      });
+
+      test(
+          'should close a single-subscription stream that never had a '
+          'listener when parent is disposed', () async {
         var controller = new StreamController();
         thing.manageStreamController(controller);
         expect(controller.isClosed, isFalse);