Make asBroadcastStream take two callbacks.

sdk/lib/async/stream_impl.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;
 
 /** Abstract and private interface for a place to put events. */
 abstract class _EventSink<T> {
   void _add(T data);
   void _addError(Object error);
@@ skipping 678 matching lines @@
 
   void _insertBefore(_BroadcastLinkedList newNext) {
     _BroadcastLinkedList newPrevious = newNext._previous;
     newPrevious._next = this;
     newNext._previous = _previous;
     _previous._next = newNext;
     _previous = newPrevious;
   }
 }
 
+typedef void _broadcastCallback(StreamSubscription subscription);
+
 class _AsBroadcastStream<T> extends Stream<T> {
   final Stream<T> _source;
+  final _broadcastCallback _onListenHandler;
+  final _broadcastCallback _onCancelHandler;
+  final _Zone _zone;
+
   _AsBroadcastStreamController<T> _controller;
   StreamSubscription<T> _subscription;
 
-  _AsBroadcastStream(this._source) {
-    _controller = new _AsBroadcastStreamController<T>(null, _onCancel);
+  _AsBroadcastStream(this._source,
+                     this._onListenHandler,
+                     this._onCancelHandler)
+    : _zone = _Zone.current {
+    _controller = new _AsBroadcastStreamController<T>(_onListen, _onCancel);
   }
 
   bool get isBroadcast => true;
 
   StreamSubscription<T> listen(void onData(T data),
                                { void onError(Object error),
                                  void onDone(),
                                  bool cancelOnError}) {
     if (_controller == null) {
       throw new StateError("Source stream has been closed.");
     }
     if (_subscription == null) {
       _subscription = _source.listen(_controller.add,
                                      onError: _controller.addError,
                                      onDone: _controller.close);
     }
     return _controller.stream.listen(onData, onError: onError, onDone: onDone,
                                      cancelOnError: cancelOnError);
   }
 
   void _onCancel() {
+    if (_onCancelHandler != null) {
+      _zone.executeCallback(() {

[floitsch, 2013/06/20 14:28:30]

I'm not sure if these should be run in the zone: they are synchronous calls
(like the cancel handler).
If we think they should be, then it would need to be:
executePeriodicCallback
(the zone doesn't expect a callback)

[Lasse Reichstein Nielsen, 2013/06/21 09:49:11]

Zone has been removed from the callback.
Uses current zone's handleUncaughtError for errors.

+        _onCancelHandler(new _BroadcastSubscriptionWrapper(this));
+      });
+    }
+  }
+
+  void _onListen() {
+    if (_onListenHandler != null) {
+      _zone.executeCallback(() {

[floitsch, 2013/06/20 14:28:30]

ditto.

[Lasse Reichstein Nielsen, 2013/06/21 09:49:11]

Done.

+        _onListenHandler(new _BroadcastSubscriptionWrapper(this));
+      });
+    }
+  }
+
+  // Methods called from _BroadcastSubscriptionWrapper.
+  void _cancelSubscription() {
+    if (_subscription == null) return;
     // Called by [_controller] when it has no subscribers left.
     StreamSubscription subscription = _subscription;
     _subscription = null;
     _controller = null;  // Marks the stream as no longer listenable.
     subscription.cancel();
   }
+
+  void _pauseSubscription(Future resumeSignal) {
+    if (_subscription == null) return;
+    _subscription.pause(resumeSignal);
+  }
+
+  void _resumeSubscription() {
+    if (_subscription == null) return;
+    _subscription.resume();
+  }
+
+  bool get _isSubscriptionPaused {
+    if (_subscription == null) return false;
+    return _subscription.isPaused;
+  }
 }
 
 /**
+ * Wrapper for subscription that disallows changing handlers.
+ *
+ * Then wrapper is only valid while a callback using it is running. After
+ * that it is invalidated, so a user can't hand on to it and change the

[floitsch, 2013/06/20 14:28:30]

Description seems to be wrong. Pause and resume keep _stream non-null.

[Lasse Reichstein Nielsen, 2013/06/21 09:49:11]

ACK, changed that but forgot the doc.
It stays valid. At some point the underlying subscription is canceled, and then
calling its pause/resume/cancel won't do anything, but that's all.

I've removed the "setting _stream to null" hackery since it won't do any good
anyway.

+ * underlying subscription at inopportune moments.
+ */
+class _BroadcastSubscriptionWrapper<T> implements StreamSubscription<T> {
+  _AsBroadcastStream _stream;
+
+  _BroadcastSubscriptionWrapper(this._stream);
+
+  void onData(void handleData(T data)) {
+    throw new UnsupportedError(
+        "Cannot change handlers of asBroadcastStream source subscription.");
+  }
+
+  void onError(void handleError(Object data)) {
+    throw new UnsupportedError(
+        "Cannot change handlers of asBroadcastStream source subscription.");
+  }
+
+  void onDone(void handleDone()) {
+    throw new UnsupportedError(
+        "Cannot change handlers of asBroadcastStream source subscription.");
+  }
+
+  void pause([Future resumeSignal]) {
+    if (_stream == null) return;

[floitsch, 2013/06/20 14:28:30]

throw.

[Lasse Reichstein Nielsen, 2013/06/21 09:49:11]

actually, I removed the line, and just call through in all cases.
The underlying subscription will ignore the calls when it is canceled anyway.

+    _stream._pauseSubscription(resumeSignal);

[floitsch, 2013/06/20 14:28:30]

This is a little bit strange: we allow a resume-signal, but we don't allow to
invoke 'resume' by ourselves?
For consistency I would say: no resumeSignal either.

[Lasse Reichstein Nielsen, 2013/06/21 09:49:11]

We do allow calling resume.

+  }
+
+  void resume() {
+    if (_stream == null) return;

[floitsch, 2013/06/20 14:28:30]

ditto (throw).

[Lasse Reichstein Nielsen, 2013/06/21 09:49:11]

Ditto, removed.

+    _stream._resumeSubscription();
+  }
+
+  void cancel() {
+    if (_stream == null) return;

[floitsch, 2013/06/20 14:28:30]

ditto.

[Lasse Reichstein Nielsen, 2013/06/21 09:49:11]

removed.

+    _AsBroadcastStream stream = _stream;
+    _stream = null;
+    stream._cancelSubscription();
+  }
+
+  bool get isPaused {
+    if (_stream == null) return false;

[floitsch, 2013/06/20 14:28:30]

ditto.

[Lasse Reichstein Nielsen, 2013/06/21 09:49:11]

Removed.

+    return _stream._isSubscriptionPaused;
+  }
+
+  Future asFuture([var futureValue]) {
+    throw new UnsupportedError(
+        "Cannot change handlers of asBroadcastStream source subscription.");
+  }
+}
+
+
+/**
  * Simple implementation of [StreamIterator].
  */
 class _StreamIteratorImpl<T> implements StreamIterator<T> {
   // Internal state of the stream iterator.
   // At any time, it is in one of these states.
   // The interpretation of the [_futureOrPrefecth] field depends on the state.
   // In _STATE_MOVING, the _data field holds the most recently returned
   // future.
   // When in one of the _STATE_EXTRA_* states, the it may hold the
   // next data/error object, and the subscription is paused.
@@ skipping 129 matching lines @@
       _FutureImpl<bool> hasNext = _futureOrPrefetch;
       _clear();
       hasNext._setValue(false);
       return;
     }
     _subscription.pause();
     _futureOrPrefetch = null;
     _state = _STATE_EXTRA_DONE;
   }
 }