Make IsolateChannel use StreamChannelCompleter.

lib/src/guarantee_channel.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:async/async.dart';
 
 import '../stream_channel.dart';
 
@@ skipping 109 matching lines @@
     _inner.add(data);
   }
 
   void addError(error, [StackTrace stackTrace]) {
     if (_closed) throw new StateError("Cannot add event after closing.");
     if (_inAddStream) {
       throw new StateError("Cannot add event while adding stream.");
     }
     if (_disconnected) return;
 
+    _addError(error, stackTrace);
+  }
+
+  /// Like [addError], but doesn't check to ensure that an error can be added.
+  ///
+  /// This is called from [addStream], so it shouldn't fail if a stream is being
+  /// added.
+  void _addError(error, [StackTrace stackTrace]) {
     if (_allowErrors) {
       _inner.addError(error, stackTrace);
       return;
     }
 
     _doneCompleter.completeError(error, stackTrace);
 
     // Treat an error like both the stream and sink disconnecting.
     _onStreamDisconnected();
     _channel._onSinkDisconnected();
 
     // Ignore errors from the inner sink. We're already surfacing one error, and
     // if the user handles it we don't want them to have another top-level.
     _inner.close().catchError((_) {});
   }
 
   Future addStream(Stream<T> stream) {
     if (_closed) throw new StateError("Cannot add stream after closing.");
     if (_inAddStream) {
       throw new StateError("Cannot add stream while adding stream.");
     }
     if (_disconnected) return new Future.value();
 
     _addStreamCompleter = new Completer.sync();
     _addStreamSubscription = stream.listen(
         _inner.add,
-        onError: _inner.addError,
+        onError: _addError,
         onDone: _addStreamCompleter.complete);
     return _addStreamCompleter.future.then((_) {
       _addStreamCompleter = null;
       _addStreamSubscription = null;
     });
   }
 
   Future close() {
     if (_inAddStream) {
       throw new StateError("Cannot close sink while adding stream.");
@@ skipping 17 matching lines @@
   void _onStreamDisconnected() {
     _disconnected = true;
     if (!_doneCompleter.isCompleted) _doneCompleter.complete();
 
     if (!_inAddStream) return;
     _addStreamCompleter.complete(_addStreamSubscription.cancel());
     _addStreamCompleter = null;
     _addStreamSubscription = null;
   }
 }