Make MultiChannel follow the stream channel rules.

test/multi_channel_test.dart

 // Copyright (c) 2016, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 import 'dart:async';
 
 import 'package:stream_channel/stream_channel.dart';
 import 'package:test/test.dart';
 
 import 'utils.dart';
 
 void main() {
-  var oneToTwo;
-  var twoToOne;
+  var controller;
   var channel1;
   var channel2;
   setUp(() {
-    oneToTwo = new StreamController();
-    twoToOne = new StreamController();
-    channel1 = new MultiChannel(
-        new StreamChannel(twoToOne.stream, oneToTwo.sink));
-    channel2 = new MultiChannel(
-        new StreamChannel(oneToTwo.stream, twoToOne.sink));
+    controller = new StreamChannelController();
+    channel1 = new MultiChannel(controller.local);
+    channel2 = new MultiChannel(controller.foreign);
   });
 
   group("the default virtual channel", () {
     test("begins connected", () {
       var first = true;
       channel2.stream.listen(expectAsync((message) {
         if (first) {
           expect(message, equals("hello"));
           first = false;
         } else {
@@ skipping 25 matching lines @@
 
       channel1.stream.listen((_) {}).cancel();
 
       // Ensure that there's enough time for the channel to close if it's going
       // to.
       return pumpEventQueue();
     });
 
     test("closes the underlying channel when it closes without any other "
         "virtual channels", () {
-      expect(oneToTwo.done, completes);
-      expect(twoToOne.done, completes);
+      expect(controller.local.sink.done, completes);
+      expect(controller.foreign.sink.done, completes);
 
       channel1.sink.close();
     });
 
     test("doesn't close the underlying channel when it closes with other "
         "virtual channels", () {
-      oneToTwo.done.then(expectAsync((_) {}, count: 0));
-      twoToOne.done.then(expectAsync((_) {}, count: 0));
+      controller.local.sink.done.then(expectAsync((_) {}, count: 0));
+      controller.foreign.sink.done.then(expectAsync((_) {}, count: 0));
 
       // Establish another virtual connection which should keep the underlying
       // connection open.
       channel2.virtualChannel(channel1.virtualChannel().id);
       channel1.sink.close();
 
       // Ensure that there's enough time for the underlying channel to complete
       // if it's going to.
       return pumpEventQueue();
     });
@@ skipping 53 matching lines @@
       return pumpEventQueue();
     });
 
     test("closes the underlying channel when it closes without any other "
         "virtual channels", () async {
       // First close the default channel so we can test the new channel as the
       // last living virtual channel.
       channel1.sink.close();
 
       await channel2.stream.toList();
-      expect(oneToTwo.done, completes);
-      expect(twoToOne.done, completes);
+      expect(controller.local.sink.done, completes);
+      expect(controller.foreign.sink.done, completes);
 
       virtual1.sink.close();
     });
 
     test("doesn't close the underlying channel when it closes with other "
         "virtual channels", () {
-      oneToTwo.done.then(expectAsync((_) {}, count: 0));
-      twoToOne.done.then(expectAsync((_) {}, count: 0));
+      controller.local.sink.done.then(expectAsync((_) {}, count: 0));
+      controller.foreign.sink.done.then(expectAsync((_) {}, count: 0));
 
       virtual1.sink.close();
 
       // Ensure that there's enough time for the underlying channel to complete
       // if it's going to.
       return pumpEventQueue();
     });
 
     test("doesn't conflict with a remote virtual channel", () {
       var virtual3 = channel2.virtualChannel();
@@ skipping 67 matching lines @@
       return pumpEventQueue();
     });
 
     test("closes the underlying channel when it closes without any other "
         "virtual channels", () async {
       // First close the default channel so we can test the new channel as the
       // last living virtual channel.
       channel2.sink.close();
 
       await channel1.stream.toList();
-      expect(oneToTwo.done, completes);
-      expect(twoToOne.done, completes);
+      expect(controller.local.sink.done, completes);
+      expect(controller.foreign.sink.done, completes);
 
       virtual2.sink.close();
     });
 
     test("doesn't close the underlying channel when it closes with other "
         "virtual channels", () {
-      oneToTwo.done.then(expectAsync((_) {}, count: 0));
-      twoToOne.done.then(expectAsync((_) {}, count: 0));
+      controller.local.sink.done.then(expectAsync((_) {}, count: 0));
+      controller.foreign.sink.done.then(expectAsync((_) {}, count: 0));
 
       virtual2.sink.close();
 
       // Ensure that there's enough time for the underlying channel to complete
       // if it's going to.
       return pumpEventQueue();
     });
 
     test("doesn't allow another virtual channel with the same id", () {
       expect(() => channel2.virtualChannel(virtual1.id),
@@ skipping 12 matching lines @@
     test("closes, all virtual channels close", () {
       expect(channel1.stream.toList(), completion(isEmpty));
       expect(channel1.sink.done, completes);
       expect(channel2.stream.toList(), completion(isEmpty));
       expect(channel2.sink.done, completes);
       expect(virtual1.stream.toList(), completion(isEmpty));
       expect(virtual1.sink.done, completes);
       expect(virtual2.stream.toList(), completion(isEmpty));
       expect(virtual2.sink.done, completes);
 
-      oneToTwo.close();
+      controller.local.sink.close();
     });
 
-    test("closes, no more virtual channels may be created", () {
-      expect(channel1.sink.done.then((_) => channel1.virtualChannel()),
-          throwsStateError);
-      expect(channel2.sink.done.then((_) => channel2.virtualChannel()),
-          throwsStateError);
+    test("closes, more virtual channels are created closed", () async {
+      channel2.sink.close();
+      virtual2.sink.close();
 
-      oneToTwo.close();
+      // Wait for the existing channels to emit done events.
+      await channel1.stream.toList();
+      await virtual1.stream.toList();
+
+      var virtual = channel1.virtualChannel();
+      expect(virtual.stream.toList(), completion(isEmpty));
+      expect(virtual.sink.done, completes);
+
+      virtual = channel1.virtualChannel();
+      expect(virtual.stream.toList(), completion(isEmpty));
+      expect(virtual.sink.done, completes);
     });
 
     test("emits an error, the error is sent only to the default channel", () {
       channel1.stream.listen(expectAsync((_) {}, count: 0),
           onError: expectAsync((error) => expect(error, equals("oh no"))));
       virtual1.stream.listen(expectAsync((_) {}, count: 0),
           onError: expectAsync((_) {}, count: 0));
 
-      twoToOne.addError("oh no");
+      controller.foreign.sink.addError("oh no");
+    });
+  });
+
+  group("stream channel rules", () {
+    group("for the main stream:", () {
+      test("closing the sink causes the stream to close before it emits any more "
+          "events", () {
+        channel1.sink.add(1);
+        channel1.sink.add(2);
+        channel1.sink.add(3);
+
+        channel2.stream.listen(expectAsync((message) {
+          expect(message, equals(1));
+          channel2.sink.close();
+        }, count: 1));
+      });
+
+      test("after the stream closes, the sink ignores events", () async {
+        channel1.sink.close();
+
+        // Wait for the done event to be delivered.
+        await channel2.stream.toList();
+        channel2.sink.add(1);
+        channel2.sink.add(2);
+        channel2.sink.add(3);
+        channel2.sink.close();
+
+        // None of our channel.sink additions should make it to the other endpoint.
+        channel1.stream.listen(expectAsync((_) {}, count: 0));
+        await pumpEventQueue();
+      });
+
+      test("canceling the stream's subscription has no effect on the sink",
+          () async {
+        channel1.stream.listen(null).cancel();
+        await pumpEventQueue();
+
+        channel1.sink.add(1);
+        channel1.sink.add(2);
+        channel1.sink.add(3);
+        channel1.sink.close();
+        expect(channel2.stream.toList(), completion(equals([1, 2, 3])));
+      });
+
+      test("canceling the stream's subscription doesn't stop a done event",
+          () async {
+        channel1.stream.listen(null).cancel();
+        await pumpEventQueue();
+
+        channel2.sink.close();
+        await pumpEventQueue();
+
+        channel1.sink.add(1);
+        channel1.sink.add(2);
+        channel1.sink.add(3);
+        channel1.sink.close();
+
+        // The sink should be ignoring events because the channel closed.
+        channel2.stream.listen(expectAsync((_) {}, count: 0));
+        await pumpEventQueue();
+      });
+    });
+
+    group("for a virtual channel:", () {
+      var virtual1;
+      var virtual2;
+      setUp(() {
+        virtual1 = channel1.virtualChannel();
+        virtual2 = channel2.virtualChannel(virtual1.id);
+      });
+
+      test("closing the sink causes the stream to close before it emits any more "
+          "events", () {
+        virtual1.sink.add(1);
+        virtual1.sink.add(2);
+        virtual1.sink.add(3);
+
+        virtual2.stream.listen(expectAsync((message) {
+          expect(message, equals(1));
+          virtual2.sink.close();
+        }, count: 1));
+      });
+
+      test("after the stream closes, the sink ignores events", () async {
+        virtual1.sink.close();
+
+        // Wait for the done event to be delivered.
+        await virtual2.stream.toList();
+        virtual2.sink.add(1);
+        virtual2.sink.add(2);
+        virtual2.sink.add(3);
+        virtual2.sink.close();
+
+        // None of our virtual.sink additions should make it to the other endpoint.
+        virtual1.stream.listen(expectAsync((_) {}, count: 0));
+        await pumpEventQueue();
+      });
+
+      test("canceling the stream's subscription has no effect on the sink",
+          () async {
+        virtual1.stream.listen(null).cancel();
+        await pumpEventQueue();
+
+        virtual1.sink.add(1);
+        virtual1.sink.add(2);
+        virtual1.sink.add(3);
+        virtual1.sink.close();
+        expect(virtual2.stream.toList(), completion(equals([1, 2, 3])));
+      });
+
+      test("canceling the stream's subscription doesn't stop a done event",
+          () async {
+        virtual1.stream.listen(null).cancel();
+        await pumpEventQueue();
+
+        virtual2.sink.close();
+        await pumpEventQueue();
+
+        virtual1.sink.add(1);
+        virtual1.sink.add(2);
+        virtual1.sink.add(3);
+        virtual1.sink.close();
+
+        // The sink should be ignoring events because the stream closed.
+        virtual2.stream.listen(expectAsync((_) {}, count: 0));
+        await pumpEventQueue();
+      });
     });
   });
 }