Fix bugs in StreamController.addStream.

sdk/lib/async/stream_controller.dart

Index: sdk/lib/async/stream_controller.dart
diff --git a/sdk/lib/async/stream_controller.dart b/sdk/lib/async/stream_controller.dart
index ba3fd617bf5f0642102dcb44984d9dba2de147cb..2405a1227c8319dbe20824887df45aa77f0771db 100644
--- a/sdk/lib/async/stream_controller.dart
+++ b/sdk/lib/async/stream_controller.dart
@@ -46,7 +46,7 @@ part of dart.async;
  * the stream at all, and won't trigger callbacks. From the controller's point
  * of view, the stream is completely inert when has completed.
  */
-abstract class StreamController<T> implements StreamSink<T> {
+abstract class StreamController<T> implements StreamControllerSink<T> {
   /** The stream that this controller is controlling. */
   Stream<T> get stream;
 
@@ -134,7 +134,7 @@ abstract class StreamController<T> implements StreamSink<T> {
   /**
    * Returns a view of this object that only exposes the [StreamSink] interface.
    */
-  StreamSink<T> get sink;
+  StreamControllerSink<T> get sink;
 
   /**
    * Whether the stream is closed for adding more events.
@@ -169,6 +169,17 @@ abstract class StreamController<T> implements StreamSink<T> {
   void addError(Object error, [StackTrace stackTrace]);
 }
 
+/**
+ * Extension of [StreamSink] with optional `cancelOnError` parameter
+ * on `addStream`.
+ *
+ * This interface is implemented by [StreamController] and by the
+ * `StreamSink` returned by [StreamController.sink].
+ */
+abstract class StreamControllerSink<T> extends StreamSink<T> {

[Anders Johnsen, 2013/10/28 13:44:26]

Is this worth a hole new type? dart:async is already quite bloated. Maybe we
should consider if this should be the default interface?

[Anders Johnsen, 2013/10/28 13:44:26]

Thinking some more about it, I think it's a property of the StreamController,
not of the sink/addStream.

[Lasse Reichstein Nielsen, 2013/10/29 08:49:45]

You are right, it's too much overhead just so that controller.sink could accept
an extra argument on addStream. Which is silly, because you can only use it if
you know you have a controller, and then you can use the controller directly.
The extra type has been removed.

+  Future addStream(Stream<T> source, { bool cancelOnError: true });
+}
+
 
 abstract class _StreamControllerLifecycle<T> {
   StreamSubscription<T> _subscribe(bool cancelOnError);
@@ -264,7 +275,7 @@ abstract class _StreamController<T> implements StreamController<T>,
   /**
    * Returns a view of this object that only exposes the [StreamSink] interface.
    */
-  StreamSink<T> get sink => new _StreamSinkWrapper<T>(this);
+  StreamControllerSink<T> get sink => new _StreamSinkWrapper<T>(this);
 
   /**
    * Whether a listener has existed and been canceled.
@@ -344,11 +355,30 @@ abstract class _StreamController<T> implements StreamController<T>,
   }
 
   // StreamSink interface.
-  Future addStream(Stream<T> source) {
+  /**
+   * Receives events from [source] and puts them into this controller's stream.
+   *
+   * Returns a future which completes when the stream adding is done.

[Anders Johnsen, 2013/10/28 13:44:26]

... when the source stream is done.

[Lasse Reichstein Nielsen, 2013/10/29 08:49:45]

Done.

+   *
+   * Events must not be added directly to this controller using [add],
+   * [addError], [close] or [addStream], until the returned future
+   * is complete.
+   *
+   * Data and error events are forwarded to this controller's stream. A done
+   * event on the source will end the `addStream` operation and complete the
+   * returned future.
+   *
+   * If [cancelOnError] is true, only the first error on [source] is
+   * forwarded to the controller's stream, and the `addStream` ends
+   * after theis. If [cancelOnError] is false, all errors are forwarded
+   * and only a done event from
+   */
+  Future addStream(Stream<T> source, { bool cancelOnError: true }) {
     if (!_mayAddEvent) throw _badEventState();
     if (_isCanceled) return new _Future.immediate(null);
     _StreamControllerAddStreamState addState =
-        new _StreamControllerAddStreamState(this, _varData, source);
+        new _StreamControllerAddStreamState(this, _varData, source,
+                                            cancelOnError);

[Anders Johnsen, 2013/10/28 13:44:26]

Nit: argument should be on one line or each on a line.

[Lasse Reichstein Nielsen, 2013/10/29 08:49:45]

That's silly.
But done.

     _varData = addState;
     _state |= _STATE_ADDSTREAM;
     return addState.addStreamFuture;
@@ -454,6 +484,7 @@ abstract class _StreamController<T> implements StreamController<T>,
     if (_isAddingStream) {
       _StreamControllerAddStreamState addState = _varData;
       addState.varData = subscription;
+      addState.resume();
     } else {
       _varData = subscription;
     }
@@ -632,15 +663,16 @@ class _ControllerSubscription<T> extends _BufferingStreamSubscription<T> {
 
 
 /** A class that exposes only the [StreamSink] interface of an object. */
-class _StreamSinkWrapper<T> implements StreamSink<T> {
-  final StreamSink _target;
+class _StreamSinkWrapper<T> implements StreamControllerSink<T> {
+  final StreamController _target;
   _StreamSinkWrapper(this._target);
   void add(T data) { _target.add(data); }
   void addError(Object error, [StackTrace stackTrace]) {
     _target.addError(error);
   }
   Future close() => _target.close();
-  Future addStream(Stream<T> source) => _target.addStream(source);
+  Future addStream(Stream<T> source, { bool cancelOnError: true})
+      => _target.addStream(source, cancelOnError: cancelOnError);
   Future get done => _target.done;
 }
 
@@ -649,17 +681,25 @@ class _StreamSinkWrapper<T> implements StreamSink<T> {
  */
 class _AddStreamState<T> {
   // [_Future] returned by call to addStream.
-  _Future addStreamFuture;
+  final _Future addStreamFuture;
 
   // Subscription on stream argument to addStream.
-  StreamSubscription addSubscription;
+  final StreamSubscription addSubscription;
 
-  _AddStreamState(_EventSink<T> controller, Stream source)
+  _AddStreamState(_EventSink<T> controller, Stream source, bool cancelOnError)
       : addStreamFuture = new _Future(),
         addSubscription = source.listen(controller._add,
-                                        onError: controller._addError,
+                                        onError: cancelOnError
+                                             ? makeErrorHandler(controller)
+                                             : controller._addError,
                                         onDone: controller._close,
-                                        cancelOnError: true);
+                                        cancelOnError: cancelOnError);
+
+  static makeErrorHandler(_EventSink<T> controller) =>
+      (e, StackTrace s) {
+        controller._addError(e, s);
+        controller._close();
+      };
 
   void pause() {
     addSubscription.pause();
@@ -687,7 +727,9 @@ class _StreamControllerAddStreamState<T> extends _AddStreamState<T> {
 
   _StreamControllerAddStreamState(_StreamController controller,
                                   this.varData,
-                                  Stream source) : super(controller, source) {
+                                  Stream source,
+                                  bool cancelOnError)
+      : super(controller, source, cancelOnError) {
     if (controller.isPaused) {
       addSubscription.pause();
     }