Fix bugs in StreamController.addStream.

sdk/lib/async/stream_controller.dart

 // Copyright (c) 2012, 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.
 
 part of dart.async;
 
 // -------------------------------------------------------------------
 // Controller for creating and adding events to a stream.
 // -------------------------------------------------------------------
 
@@ skipping 28 matching lines @@
  * Whether to invoke a callback depends only on the state before and after
  * a stream action, for example firing an event. If the state changes multiple
  * times during the action, and then ends up in the same state as before, no
  * callback is performed.
  *
  * If listeners are added after the stream has completed (sent a "done" event),
  * the listeners will be sent a "done" event eventually, but they won't affect
  * 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;
 
   /**
    * A controller with a [stream] that supports only one single subscriber.
    *
    * If [sync] is true, events may be passed directly to the stream's listener
    * during an [add], [addError] or [close] call. If [sync] is false, the event
    * will be passed to the listener at a later time, after the code creating
    * the event has returned.
@@ skipping 67 matching lines @@
                                       void onCancel(),
                                       bool sync: false}) {
     return sync
         ? new _SyncBroadcastStreamController<T>(onListen, onCancel)
         : new _AsyncBroadcastStreamController<T>(onListen, onCancel);
   }
 
   /**
    * 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.
    *
    * If true, the "done" event might not have fired yet, but it has been
    * scheduled, and it is too late to add more events.
    */
   bool get isClosed;
 
   /**
@@ skipping 14 matching lines @@
 
   /**
    * Send or enqueue an error event.
    *
    * Also allows an objection stack trace object, on top of what [EventSink]
    * allows.
    */
   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);
   void _recordPause(StreamSubscription<T> subscription) {}
   void _recordResume(StreamSubscription<T> subscription) {}
   Future _recordCancel(StreamSubscription<T> subscription) => null;
 }
 
 /**
  * Default implementation of [StreamController].
@@ skipping 75 matching lines @@
   _NotificationHandler get _onPause;
   _NotificationHandler get _onResume;
   _NotificationHandler get _onCancel;
 
   // Return a new stream every time. The streams are equal, but not identical.
   Stream<T> get stream => new _ControllerStream(this);
 
   /**
    * 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.
    *
    * After this, adding more events will be ignored.
    */
   bool get _isCanceled => (_state & _STATE_CANCELED) != 0;
 
   /** Whether there is an active listener. */
   bool get hasListener => (_state & _STATE_SUBSCRIBED) != 0;
@@ skipping 59 matching lines @@
    */
   Error _badEventState() {
     if (isClosed) {
       return new StateError("Cannot add event after closing");
     }
     assert(_isAddingStream);
     return new StateError("Cannot add event while adding a stream");
   }
 
   // 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;
   }
 
   Future get done => _ensureDoneFuture();
 
   Future _ensureDoneFuture() {
     if (_doneFuture == null) {
       _doneFuture = new _Future();
@@ skipping 85 matching lines @@
       throw new StateError("Stream has already been listened to.");
     }
     _ControllerSubscription subscription =
         new _ControllerSubscription(this, cancelOnError);
 
     _PendingEvents pendingEvents = _pendingEvents;
     _state |= _STATE_SUBSCRIBED;
     if (_isAddingStream) {
       _StreamControllerAddStreamState addState = _varData;
       addState.varData = subscription;
+      addState.resume();
     } else {
       _varData = subscription;
     }
     subscription._setPendingEvents(pendingEvents);
     subscription._guardCallback(() {
       _runGuarded(_onListen);
     });
 
     return subscription;
   }
@@ skipping 158 matching lines @@
     _controller._recordPause(this);
   }
 
   void _onResume() {
     _controller._recordResume(this);
   }
 }
 
 
 /** 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;
 }
 
 /**
  * Object containing the state used to handle [StreamController.addStream].
  */
 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();
   }
 
   void resume() {
     addSubscription.resume();
   }
 
   void cancel() {
     addSubscription.cancel();
     complete();
   }
 
   void complete() {
     addStreamFuture._asyncComplete(null);
   }
 }
 
 class _StreamControllerAddStreamState<T> extends _AddStreamState<T> {
   // The subscription or pending data of a _StreamController.
   // Stored here because we reuse the `_varData` field  in the _StreamController
   // to store this state object.
   var varData;
 
   _StreamControllerAddStreamState(_StreamController controller,
                                   this.varData,
-                                  Stream source) : super(controller, source) {
+                                  Stream source,
+                                  bool cancelOnError)
+      : super(controller, source, cancelOnError) {
     if (controller.isPaused) {
       addSubscription.pause();
     }
   }
 }