Make StreamController be a StreamSink, not just an EventSink.

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 107 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 [EventSink] interface.
    */
-  EventSink<T> get sink;
+  StreamSink<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 39 matching lines @@
                                                _EventDispatch<T> {
   static const int _STATE_OPEN = 0;
   static const int _STATE_CANCELLED = 1;
   static const int _STATE_CLOSED = 2;
 
   final _NotificationHandler _onListen;
   final _NotificationHandler _onPause;
   final _NotificationHandler _onResume;
   final _NotificationHandler _onCancel;
   _StreamImpl<T> _stream;
+  /**
+   * Cached value returned by [sink].
+   *
+   * Used to pause the stream if necessary.
+   */
+  _ControllerStreamSink _sink;
 
   // An active subscription on the stream, or null if no subscripton is active.
   _ControllerSubscription<T> _subscription;
 
   // Whether we have sent a "done" event.
   int _state = _STATE_OPEN;
 
   // Events added to the stream before it has an active subscription.
   _PendingEvents _pendingEvents = null;
 
   _StreamController(this._onListen,
                     this._onPause,
                     this._onResume,
                     this._onCancel) {
     _stream = new _ControllerStream<T>(this);
   }
 
   Stream<T> get stream => _stream;
 
   /**
    * Returns a view of this object that only exposes the [EventSink] interface.
    */
-  EventSink<T> get sink => new _EventSinkView<T>(this);
+  StreamSink<T> get sink =>
+      (_sink != null) ? _sink
+                      : _sink = new _ControllerStreamSink<T>(this);
 
   /**
    * Whether a listener has existed and been cancelled.
    *
    * After this, adding more events will be ignored.
    */
   bool get _isCancelled => (_state & _STATE_CANCELLED) != 0;
 
   bool get isClosed => (_state & _STATE_CLOSED) != 0;
 
@@ skipping 70 matching lines @@
     _pendingEvents = null;
     subscription._guardCallback(() {
       _runGuarded(_onListen);
     });
   }
 
   void _recordCancel(StreamSubscription<T> subscription) {
     assert(identical(_subscription, subscription));
     _subscription = null;
     _state |= _STATE_CANCELLED;
+    if (_sink != null) _sink._cancel();
     _runGuarded(_onCancel);
   }
 
   void _recordPause(StreamSubscription<T> subscription) {
+    if (_sink != null) _sink._pause();
     _runGuarded(_onPause);

[floitsch, 2013/06/06 15:08:29]

Do we want to call onPause when there is an addStream active?

   }
 
   void _recordResume(StreamSubscription<T> subscription) {
+    if (_sink != null) _sink._resume();
     _runGuarded(_onResume);

[floitsch, 2013/06/06 15:08:29]

ditto.

   }
 }
 
 class _SyncStreamController<T> extends _StreamController<T> {
   _SyncStreamController(void onListen(),
                         void onPause(),
                         void onResume(),
                         void onCancel())
       : super(onListen, onPause, onResume, onCancel);
 
@@ skipping 168 matching lines @@
   final _NotificationHandler _onListen;
   final _NotificationHandler _onCancel;
 
   // State of the controller.
   int _state;
 
   // Double-linked list of active listeners.
   _BroadcastSubscriptionLink _next;
   _BroadcastSubscriptionLink _previous;
 
+  // Cached return value of [sink]. Used to cancel a `sink.addStream`

[floitsch, 2013/06/06 15:08:29]

cancel, pause and resume a `sink.addStream`.

+  // when the stream ends.
+  _ControllerStreamSink _sink;
+
   _BroadcastStreamController(this._onListen, this._onCancel)
       : _state = _STATE_INITIAL {
     _next = _previous = this;
   }
 
   // StreamController interface.
 
   Stream<T> get stream => new _BroadcastStream<T>(this);
 
-  EventSink<T> get sink => new _EventSinkView<T>(this);
+  StreamSink<T> get sink =>
+      (_sink != null) ? _sink : _sink = new _ControllerStreamSink<T>(this);
 
   bool get isClosed => (_state & _STATE_CLOSED) != 0;
 
   /**
    * A broadcast controller is never paused.
    *
    * Each receiving stream may be paused individually, and they handle their
    * own buffering.
    */
   bool get isPaused => false;
@@ skipping 11 matching lines @@
   /** Adds subscription to linked list of active listeners. */
   void _addListener(_BroadcastSubscription<T> subscription) {
     _BroadcastSubscriptionLink previous = _previous;
     previous._next = subscription;
     _previous = subscription._previous;
     subscription._previous._next = this;
     subscription._previous = previous;
     subscription._eventState = (_state & _STATE_EVENT_ID);
   }
 
-  void _removeListener(_BroadcastSubscription<T> subscription) {
+  bool _removeListener(_BroadcastSubscription<T> subscription) {
     assert(identical(subscription._controller, this));
     assert(!identical(subscription._next, subscription));
     subscription._previous._next = subscription._next;
     subscription._next._previous = subscription._previous;
     subscription._next = subscription._previous = subscription;
+    return true;
   }
 
   // _StreamControllerLifecycle interface.
 
   void _recordListen(_BroadcastSubscription<T> subscription) {
     _addListener(subscription);
     if (identical(_next, _previous)) {
       // Only one listener, so it must be the first listener.
       _runGuarded(_onListen);
     }
   }
 
   void _recordCancel(_BroadcastSubscription<T> subscription) {
+    // If already removed by the stream, don't remove it again.
+    if (identical(subscription._next, subscription)) return;
     if (subscription._isFiring) {
       subscription._setRemoveAfterFiring();
     } else {
       _removeListener(subscription);
       // If we are currently firing an event, the empty-check is performed at
       // the end of the listener loop instead of here.
       if ((_state & _STATE_FIRING) == 0 && _isEmpty) {
         _callOnCancel();
       }
     }
@@ skipping 61 matching lines @@
       }
     }
     _state &= ~_STATE_FIRING;
 
     if (_isEmpty) {
       _callOnCancel();
     }
   }
 
   void _callOnCancel() {
+    if (_sink != null && isClosed) {

[floitsch, 2013/06/06 15:08:29]

add comment explaining why you look at `isClosed`.

+      _sink._cancel();
+    }
     _runGuarded(_onCancel);
   }
 }
 
 class _SyncBroadcastStreamController<T> extends _BroadcastStreamController<T> {
   _SyncBroadcastStreamController(void onListen(), void onCancel())
       : super(onListen, onCancel);
 
   // EventDispatch interface.
 
@@ skipping 116 matching lines @@
   }
 
   void _callOnCancel() {
     if (_hasPending) {
       _pending.clear();
       _pending = null;
     }
     super._callOnCancel();
   }
 }
+
+
+/**
+ * [EventSink] wrapper that only exposes a [StreamSink] interface.
+ */
+class _ControllerStreamSink<T> implements StreamSink<T> {
+  final EventSink<T> _sink;
+  // Future completed when then controller stream is closed.
+  _FutureImpl _doneFuture;
+  // [_FutureImpl] returned by latest call to addStream.
+  // Set to null while not processing an [addStream] stream.
+  _FutureImpl _addStreamFuture;
+  // Subscription of latest call to addStream.
+  // Set to null while not processing an [addStream] stream.
+  StreamSubscription _subscription;
+
+  _ControllerStreamSink(this._sink);
+
+  bool get _isAddStreamActive => _addStreamFuture != null;
+
+  void _pause() {
+    if (_subscription != null) _subscription.pause();
+  }
+
+  void _resume() {
+    if (_subscription != null) _subscription.resume();
+  }
+
+  void _cancel() {
+    if (_isAddStreamActive) {
+      StreamSubscription subscription = _subscription;
+      _FutureImpl future = _addStreamFuture;
+      _subscription = null;
+      _addStreamFuture = null;
+      subscription.cancel();
+      future._setValue(null);
+    }
+    if (_doneFuture != null) {
+      _doneFuture._setValue(null);
+    }
+  }
+
+  void add(T value) {
+    if (_isAddStreamActive) {
+      throw new StateError("Cannot add events while addStream is running.");
+    }
+    _sink.add(value);
+  }
+
+  void addError(error) {
+    if (_isAddStreamActive) {
+      throw new StateError("Cannot add events while addStream is running.");
+    }
+    _sink.addError(error);
+  }
+
+  Future close() {
+    if (_isAddStreamActive) {
+      throw new StateError("Cannot add events while addStream is running.");
+    }
+    if (_doneFuture == null) _doneFuture = new _FutureImpl();
+    _sink.close();
+    return _doneFuture;
+  }
+
+  Future addStream(Stream<T> stream) {
+    if (_isAddStreamActive) {
+      throw new StateError("Cannot add a new stream while "
+                          "addStream is running.");
+    }
+    _addStreamFuture = new _FutureImpl();
+    _subscription = stream.listen(
+      _sink.add,
+      onError: (error) {
+        _FutureImpl future = _addStreamFuture;
+        _addStreamFuture = null;
+        _subscription = null;
+        future._setError(error);
+      },
+      onDone: () {
+        _FutureImpl future = _addStreamFuture;
+        _addStreamFuture = null;
+        _subscription = null;
+        future._setValue(null);
+      },
+      cancelOnError: true
+    );
+    return _addStreamFuture;
+  }
+
+  Future get done =>
+      (_addStreamFuture != null) ? _addStreamFuture
+                                 : new _FutureImpl.immediate(null);
+}