Provide more error-handling customization. This makes it easier to wrap underlying channels that don't support errors at all. R=tjblasi@google.com Review URL: https://codereview.chromium.org//1669953002 .

diff --git a/pkgs/stream_channel/lib/src/guarantee_channel.dart b/pkgs/stream_channel/lib/src/guarantee_channel.dart
index 849cc2f..1047e14 100644
--- a/pkgs/stream_channel/lib/src/guarantee_channel.dart
+++ b/pkgs/stream_channel/lib/src/guarantee_channel.dart

@@ -30,8 +30,10 @@
   /// Whether the sink has closed, causing the underlying channel to disconnect.
   bool _disconnected = false;
 
-  GuaranteeChannel(Stream<T> innerStream, StreamSink<T> innerSink) {
-    _sink = new _GuaranteeSink<T>(innerSink, this);
+  GuaranteeChannel(Stream<T> innerStream, StreamSink<T> innerSink,
+      {bool allowSinkErrors: true}) {
+    _sink = new _GuaranteeSink<T>(innerSink, this,
+        allowErrors: allowSinkErrors);
 
     // Enforce the single-subscription guarantee by changing a broadcast stream
     // to single-subscription.
@@ -76,10 +78,13 @@
   /// The [GuaranteeChannel] this belongs to.
   final GuaranteeChannel<T> _channel;
 
-  Future get done => _inner.done;
+  Future get done => _doneCompleter.future;
+  final _doneCompleter = new Completer();
 
-  /// Whether the stream has emitted a done event, causing the underlying
-  /// channel to disconnect.
+  /// Whether connection is disconnected.
+  ///
+  /// This can happen because the stream has emitted a done event, or because
+  /// the user added an error when [_allowErrors] is `false`.
   bool _disconnected = false;
 
   /// Whether the user has called [close].
@@ -96,7 +101,14 @@
   /// Whether we're currently adding a stream with [addStream].
   bool get _inAddStream => _addStreamSubscription != null;
 
-  _GuaranteeSink(this._inner, this._channel);
+  /// Whether errors are passed on to the underlying sink.
+  ///
+  /// If this is `false`, any error passed to the sink is piped to [done] and
+  /// the underlying sink is closed.
+  final bool _allowErrors;
+
+  _GuaranteeSink(this._inner, this._channel, {bool allowErrors: true})
+      : _allowErrors = allowErrors;
 
   void add(T data) {
     if (_closed) throw new StateError("Cannot add event after closing.");
@@ -115,7 +127,20 @@
     }
     if (_disconnected) return;
 
-    _inner.addError(error, 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) {
@@ -141,11 +166,15 @@
       throw new StateError("Cannot close sink while adding stream.");
     }
 
+    if (_closed) return done;
     _closed = true;
-    if (_disconnected) return new Future.value();
 
-    _channel._onSinkDisconnected();
-    return _inner.close();
+    if (!_disconnected) {
+      _channel._onSinkDisconnected();
+      _doneCompleter.complete(_inner.close());
+    }
+
+    return done;
   }
 
   /// Called by [GuaranteeChannel] when the stream emits a done event.
@@ -154,8 +183,9 @@
   /// sink should stop forwarding events.
   void _onStreamDisconnected() {
     _disconnected = true;
-    if (!_inAddStream) return;
+    if (!_doneCompleter.isCompleted) _doneCompleter.complete();
 
+    if (!_inAddStream) return;
     _addStreamCompleter.complete(_addStreamSubscription.cancel());
     _addStreamCompleter = null;
     _addStreamSubscription = null;

diff --git a/pkgs/stream_channel/lib/src/stream_channel_controller.dart b/pkgs/stream_channel/lib/src/stream_channel_controller.dart
index 1812a0e..ad78323 100644
--- a/pkgs/stream_channel/lib/src/stream_channel_controller.dart
+++ b/pkgs/stream_channel/lib/src/stream_channel_controller.dart

@@ -47,12 +47,18 @@
   /// If [sync] is true, events added to either channel's sink are synchronously
   /// dispatched to the other channel's stream. This should only be done if the
   /// source of those events is already asynchronous.
-  StreamChannelController({bool sync: false}) {
+  ///
+  /// If [allowForeignErrors] is `false`, errors are not allowed to be passed to
+  /// the foreign channel's sink. If any are, the connection will close and the
+  /// error will be forwarded to the foreign channel's [Sink.done] future. This
+  /// guarantees that the local stream will never emit errors.
+  StreamChannelController({bool allowForeignErrors: true, bool sync: false}) {
     var localToForeignController = new StreamController<T>(sync: sync);
     var foreignToLocalController = new StreamController<T>(sync: sync);
     _local = new StreamChannel<T>.withGuarantees(
         foreignToLocalController.stream, localToForeignController.sink);
     _foreign = new StreamChannel<T>.withGuarantees(
-        localToForeignController.stream, foreignToLocalController.sink);
+        localToForeignController.stream, foreignToLocalController.sink,
+        allowSinkErrors: allowForeignErrors);
   }
 }

diff --git a/pkgs/stream_channel/lib/stream_channel.dart b/pkgs/stream_channel/lib/stream_channel.dart
index 8c4fad5..992f702 100644
--- a/pkgs/stream_channel/lib/stream_channel.dart
+++ b/pkgs/stream_channel/lib/stream_channel.dart

@@ -78,8 +78,13 @@
   /// [StreamChannel] documentation. This makes it somewhat less efficient than
   /// just wrapping a stream and a sink directly, so [new StreamChannel] should
   /// be used when the guarantees are provided natively.
-  factory StreamChannel.withGuarantees(Stream<T> stream, StreamSink<T> sink) =>
-      new GuaranteeChannel(stream, sink);
+  ///
+  /// If [allowSinkErrors] is `false`, errors are not allowed to be passed to
+  /// [sink]. If any are, the connection will close and the error will be
+  /// forwarded to [Sink.done].
+  factory StreamChannel.withGuarantees(Stream<T> stream, StreamSink<T> sink,
+          {bool allowSinkErrors: true}) =>
+      new GuaranteeChannel(stream, sink, allowSinkErrors: allowSinkErrors);
 
   /// Connects [this] to [other], so that any values emitted by either are sent
   /// directly to the other.

diff --git a/pkgs/stream_channel/pubspec.yaml b/pkgs/stream_channel/pubspec.yaml
index f0b830c..52ee521 100644
--- a/pkgs/stream_channel/pubspec.yaml
+++ b/pkgs/stream_channel/pubspec.yaml

@@ -1,5 +1,5 @@
 name: stream_channel
-version: 1.2.0-dev
+version: 1.2.0
 description: An abstraction for two-way communication channels.
 author: Dart Team <misc@dartlang.org>
 homepage: https://github.com/dart-lang/stream_channel

diff --git a/pkgs/stream_channel/test/stream_channel_controller_test.dart b/pkgs/stream_channel/test/stream_channel_controller_test.dart
index e45f090..cf8a31b 100644
--- a/pkgs/stream_channel/test/stream_channel_controller_test.dart
+++ b/pkgs/stream_channel/test/stream_channel_controller_test.dart

@@ -25,6 +25,17 @@
       controller.foreign.sink..add(1)..add(2)..add(3)..close();
       expect(controller.local.stream.toList(), completion(equals([1, 2, 3])));
     });
+
+    test("with allowForeignErrors: false, shuts down the connection if an "
+        "error is added to the foreign channel", () {
+      controller = new StreamChannelController(allowForeignErrors: false);
+
+      controller.foreign.sink.addError("oh no");
+      expect(controller.foreign.sink.done, throwsA("oh no"));
+      expect(controller.foreign.stream.toList(), completion(isEmpty));
+      expect(controller.local.sink.done, completes);
+      expect(controller.local.stream.toList(), completion(isEmpty));
+    });
   });
 
   group("synchronously", () {

diff --git a/pkgs/stream_channel/test/with_guarantees_test.dart b/pkgs/stream_channel/test/with_guarantees_test.dart
index 0e95f9d..38afefa 100644
--- a/pkgs/stream_channel/test/with_guarantees_test.dart
+++ b/pkgs/stream_channel/test/with_guarantees_test.dart

@@ -107,4 +107,44 @@
     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();
+  });
+
+  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");
+      expect(channel.sink.done, throwsA("oh no"));
+      sinkController.stream.listen(null,
+          onError: expectAsync((_) {}, count: 0));
+    });
+
+    test("adding an error causes the stream to emit a done event", () {
+      expect(channel.sink.done, throwsA("oh no"));
+
+      streamController.add(1);
+      streamController.add(2);
+      streamController.add(3);
+
+      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));
+    });
+  });
 }