Add public-facing method and class that allows intercepting stream events.

sdk/lib/async/stream.dart

Index: sdk/lib/async/stream.dart
diff --git a/sdk/lib/async/stream.dart b/sdk/lib/async/stream.dart
index ed4ab1a4198bc7314606924748a523c64f40ffc6..632bb6377ab4fa004044f257d1d9fe9aafc90f55 100644
--- a/sdk/lib/async/stream.dart
+++ b/sdk/lib/async/stream.dart
@@ -40,12 +40,12 @@ part of dart.async;
  * A broadcast stream allows any number of listeners, and it fires
  * its events when they are ready, whether there are listeners or not.
  *
- * Braodcast streams are used for independent events/observers.
+ * Broadcast streams are used for independent events/observers.
  *
- * The default implementation of [isBroadcast] and
- * [asBroadcastStream] are assuming this is a single-subscription stream
- * and a broadcast stream inheriting from [Stream] must override these
- * to return [:true:] and [:this:] respectively.
+ * The default implementation of [isBroadcast] is assuming this is a

[floitsch, 2013/01/28 14:45:09]

The default implementation of [isBroadcast] returns `false`.

[Lasse Reichstein Nielsen, 2013/01/29 08:42:54]

Done.

+ * single-subscription stream.
+ * A broadcast stream inheriting from [Stream] must override [isBroadcast]
+ * to return [:true:].
  */
 abstract class Stream<T> {
   Stream();
@@ -93,6 +93,7 @@ abstract class Stream<T> {
    * If this stream is already a broadcast stream, it is returned unmodified.
    */
   Stream<T> asBroadcastStream() {
+    if (isBroadcast) return this;
     return new _SingleStreamMultiplexer<T>(this);
   }
 
@@ -172,8 +173,11 @@ abstract class Stream<T> {
    * If the error is intercepted, the [handle] function can decide what to do
    * with it. It can throw if it wants to raise a new (or the same) error,
    * or simply return to make the stream forget the error.
+   *
+   * If you need to transform an error into a data event, use the more generic
+   * [Stream.transformEvent] to handle the event by writing a data event to
+   * the output sink
    */
-   // TODO(lrn): Say what to do if you want to convert the error to a value.
   Stream<T> handleError(void handle(AsyncError error), { bool test(error) }) {
     return new HandleErrorStream<T>(this, handle, test);
   }
@@ -206,6 +210,23 @@ abstract class Stream<T> {
     return streamTransformer.bind(this);
   }
 
+  /**
+   * Create a new stream from this by modifying events.
+   *
+   * Subscribing on the returned stream is the same as subscribing on
+   * this stream, except that events are passed through the [transformer]
+   * before being emitted. The transformer may generate any number and
+   * types of events for each incoming event.

[floitsch, 2013/01/28 14:45:09]

Mention how pauses are handled.

[Lasse Reichstein Nielsen, 2013/01/29 08:42:54]

Done.

+   *
+   * An example that adds one to each of a stream of integers:

[floitsch, 2013/01/28 14:45:09]

Example is bad, since it could be implemented with "mappedBy".

[Lasse Reichstein Nielsen, 2013/01/29 08:42:54]

Done.

+   *     intStream.transformEvents(new StreamEventTransformer.from(

[floitsch, 2013/01/28 14:45:09]

new line before the code.

[Lasse Reichstein Nielsen, 2013/01/29 08:42:54]

Done.

+   *       handleData: (int value, StreamSink sink) {
+   *         sink.add(value + 1);
+   *       }));
+   */
+  Stream transformEvents(StreamEventTransformer<T, dynamic> transformer) {

[floitsch, 2013/01/28 14:45:09]

As discussed. We should consider making this the default ".transform", and
rename the current transform to "chain".
But for backwards-compatibility we need to start this way anyways.

[Lasse Reichstein Nielsen, 2013/01/29 08:42:54]

Done.

+    return new EventTransformStream<T, dynamic>(this, transformer);
+  }
 
   /** Reduces a sequence of values by repeatedly applying [combine]. */
   Future reduce(var initialValue, combine(var previous, T element)) {
@@ -259,11 +280,11 @@ abstract class Stream<T> {
     subscription = this.listen(
         (T element) {
           _runUserCode(
-            () => match(element),
+            () => (element == match),
             (bool isMatch) {
               if (isMatch) {
                 subscription.cancel();
-                future._setValue(element);
+                future._setValue(true);

[floitsch, 2013/01/28 14:45:09]

please add tests.

               }
             },
             _cancelAndError(subscription, future)
@@ -904,3 +925,171 @@ abstract class StreamTransformer<S, T> {
 
   Stream<T> bind(Stream<S> stream);
 }
+
+
+/**
+ * A transformer of stream events.
+ *
+ * A [StreamEventTransformer] transforms incoming Stream
+ * events of one kind into outgoing events of another kind.
+ *
+ * The default implementations of the "handle" methods forward
+ * the events unmodified. This will not work for data events if the types

[floitsch, 2013/01/28 14:45:09]

... unmodified. In that case the generic type T needs to be assignable to S.

[Lasse Reichstein Nielsen, 2013/01/29 08:42:54]

Done.

+ * are different.
+ *
+ * You can use a [StreamEventTransformer] to modify a Stream's events using
+ * the [Stream.transformEvents] method.
+ */
+abstract class StreamEventTransformer<S, T> {
+  const StreamEventTransformer();
+
+  /**
+   * Create a [StreamEventTransformer] that delegates to the provided methods.
+   *
+   * The created transformer acts as if the provided functions were the
+   * methods of the same name.
+   */
+  factory StreamEventTransformer.from({

[floitsch, 2013/01/28 14:45:09]

remove the ".from".

[Lasse Reichstein Nielsen, 2013/01/29 08:42:54]

Not possible. That would conflict with the constructor above, which is needed
for classes extending StreamEventTransformer.

+    void handleData(S data, StreamSink<T> sink),
+    void handleError(AsyncError error, StreamSink<T> sink),
+    void handleDone(StreamSink<T> sink)
+  }) => new _StreamEventTransformerImpl<S, T>(handleData,

[floitsch, 2013/01/28 14:45:09]

too weird.
make it a "return".

[Lasse Reichstein Nielsen, 2013/01/29 08:42:54]

Done. It's still weird :)

+                                              handleError,
+                                              handleDone);
+
+  /**
+   * Act on incoming data event.
+   *
+   * The method may generate any number of events on the sink, but should
+   * not throw.

[floitsch, 2013/01/28 14:45:09]

what if it throws?
If we catch it, and propagate it as an error in the sink, then we might as well
"advertise" the behavior.

[Lasse Reichstein Nielsen, 2013/01/29 08:42:54]

We don't catch it. It'll be an uncaught error in the event firing of the
underlying stream, so it will be sent to the global error handler.
I've changed "should" to "must".

+   */
+  void handleData(S event, StreamSink<T> sink) {
+    var data = event;
+    sink.add(data);
+  }
+
+  /**
+   * Act on incoming error event.
+   *
+   * The method may generate any number of events on the sink, but should
+   * not throw.
+   */
+  void handleError(AsyncError error, StreamSink<T> sink) {
+    sink.signalError(error);
+  }
+
+  /**
+   * Act on incoming done event.
+   *
+   * The method may generate any number of events on the sink, but should
+   * not throw.
+   */
+  void handleDone(StreamSink<T> sink){
+    sink.close();
+  }
+}
+
+/**
+ * Stream that transforms another stream by intercepting and replacing events.
+ *
+ * This [Stream] is a transformation of a source stream. Listening on this
+ * stream is the same as listening on the source stream, except that events
+ * are intercepted and modified by a [StreamEventTransformer] before becoming
+ * events on this stream.
+ */
+class EventTransformStream<S, T> extends Stream<T> {
+  Stream<S> _source;
+  StreamEventTransformer _transformer;
+  EventTransformStream(Stream<S> source,
+                       StreamEventTransformer<S, T> transformer)
+      : _source = source, _transformer = transformer;
+
+  StreamSubscription<T> listen(void onData(T data),
+                               { void onError(AsyncError error),
+                                 void onDone(),
+                                 bool unsubscribeOnError }) {
+    return new _EventTransformStreamSubscription(_source, _transformer,
+                                                 onData, onError, onDone,
+                                                 unsubscribeOnError);
+  }
+}
+
+class _EventTransformStreamSubscription<S, T>
+    extends _BaseStreamSubscription<T>
+    implements _StreamOutputSink<T> {
+  /** The transformer used to transform events. */
+  final StreamEventTransformer<S, T> _transformer;
+  /** Whether to unsubscribe when emitting an error. */
+  final bool _unsubscribeOnError;
+  /** Source of incoming events. */
+  StreamSubscription<S> _subscription;
+  /** Cached StreamSink wrapper for this class. */
+  StreamSink<T> _sink;
+
+  _EventTransformStreamSubscription(Stream<S> source,
+                                    this._transformer,
+                                    void onData(T data),
+                                    void onError(AsyncError error),
+                                    void onDone(),
+                                    this._unsubscribeOnError)
+      : super(onData, onError, onDone) {
+    _sink = new _StreamOutputSinkWrapper<T>(this);
+    _subscription = source.listen(_handleData,
+                                  onError: _handleError,
+                                  onDone: _handleDone);
+  }
+
+  void pause([Future pauseSignal]) {
+    if (_subscription != null) _subscription.pause(pauseSignal);
+  }
+
+  void resume() {
+    if (_subscription != null) _subscription.resume();
+  }
+
+  void cancel() {
+    if (_subscription != null) {
+      _subscription.cancel();
+      _subscription = null;
+    }
+  }
+
+  void _handleData(S data) {
+    _transformer.handleData(data, _sink);
+  }
+
+  void _handleError(AsyncError error) {
+    _transformer.handleError(error, _sink);
+  }
+
+  void _handleDone() {
+    _transformer.handleDone(_sink);
+  }
+
+  // StreamOutputSink interface.
+  void _sendData(T data) {
+    _onData(data);
+  }
+
+  void _sendError(AsyncError error) {
+    _onError(error);
+    if (_unsubscribeOnError) {
+      cancel();
+    }
+  }
+
+  void _sendDone() {
+    // It's ok to cancel even if we have been unsubscribed already.
+    cancel();
+    _onDone();
+  }
+}
+
+class _StreamOutputSinkWrapper<T> implements StreamSink<T> {
+  _StreamOutputSink _sink;
+  _StreamOutputSinkWrapper(this._sink);
+
+  void add(T data) => _sink._sendData(data);
+  void signalError(AsyncError error) => _sink._sendError(error);
+  void close() => _sink._sendDone();
+}