Make IsolateChannel use StreamChannelCompleter.

test/with_guarantees_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';
@@ skipping 96 matching lines @@
     channel.sink.addError("error");
     expect(sinkController.stream.first, throwsA("error"));
   });
 
   test("Sink.done completes once the stream is done", () {
     channel.stream.listen(null);
     expect(channel.sink.done, completes);
     streamController.close();
   });
 
+  test("events can't be added to an explicitly-closed sink", () {
+    sinkController.stream.listen(null); // Work around sdk#19095.
+
+    expect(channel.sink.close(), completes);
+    expect(() => channel.sink.add(1), throwsStateError);
+    expect(() => channel.sink.addError("oh no"), throwsStateError);
+    expect(() => channel.sink.addStream(new Stream.fromIterable([])),
+        throwsStateError);
+  });
+
+  test("events can't be added while a stream is being added", () {
+    var controller = new StreamController();
+    channel.sink.addStream(controller.stream);
+
+    expect(() => channel.sink.add(1), throwsStateError);
+    expect(() => channel.sink.addError("oh no"), throwsStateError);
+    expect(() => channel.sink.addStream(new Stream.fromIterable([])),
+        throwsStateError);
+    expect(() => channel.sink.close(), throwsStateError);
+
+    controller.close();
+  });
+
   group("with allowSinkErrors: false", () {
     setUp(() {
       streamController = new StreamController();
       sinkController = new StreamController();
       channel = new StreamChannel.withGuarantees(
           streamController.stream, sinkController.sink, allowSinkErrors: false);
     });
 
     test("forwards errors to Sink.done but not the stream", () {
       channel.sink.addError("oh no");
@@ skipping 12 matching lines @@
       expect(channel.stream.listen(expectAsync((event) {
         if (event == 2) channel.sink.addError("oh no");
       }, count: 2)).asFuture(), completes);
     });
 
     test("adding an error closes the inner sink", () {
       channel.sink.addError("oh no");
       expect(channel.sink.done, throwsA("oh no"));
       expect(sinkController.stream.toList(), completion(isEmpty));
     });
+
+    test("adding an error via via addStream causes the stream to emit a done "
+        "event", () async {
+      var canceled = false;
+      var controller = new StreamController(onCancel: () {
+        canceled = true;
+      });
+
+      // This future shouldn't get the error, because it's sent to [Sink.done].
+      expect(channel.sink.addStream(controller.stream), completes);
+
+      controller.addError("oh no");
+      expect(channel.sink.done, throwsA("oh no"));
+      await pumpEventQueue();
+      expect(canceled, isTrue);
+
+      // Even though the sink is closed, this shouldn't throw an error because
+      // the user didn't explicitly close it.
+      channel.sink.add(1);
+    });
   });
 }