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

sdk/lib/async/stream.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;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 22 matching lines @@
  * A single-subscription stream allows only a single listener at a time.
  * It holds back events until it gets a listener, and it may exhaust
  * itself when the listener is unsubscribed, even if the stream wasn't done.
  *
  * Single-subscription streams are generally used for streaming parts of
  * contiguous data like file I/O.
  *
  * 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();
 
   /**
    * Creates a new single-subscription stream from the future.
    *
    * When the future completes, the stream will fire one event, either
    * data or error, and then close with a done-event.
    */
@@ skipping 27 matching lines @@
    * Returns a multi-subscription stream that produces the same events as this.
    *
    * If this stream is single-subscription, return a new stream that allows
    * multiple subscribers. It will subscribe to this stream when its first
    * subscriber is added, and unsubscribe again when the last subscription is
    * cancelled.
    *
    * If this stream is already a broadcast stream, it is returned unmodified.
    */
   Stream<T> asBroadcastStream() {
+    if (isBroadcast) return this;
     return new _SingleStreamMultiplexer<T>(this);
   }
 
   /**
    * Stream that outputs events from the [sources] in cyclic order.
    *
    * The merged streams are paused and resumed in order to ensure the proper
    * order of output events.
    */
   factory Stream.cyclic(Iterable<Stream> sources) {
@@ skipping 59 matching lines @@
    *
    * If this stream sends an error that matches [test], then it is intercepted
    * by the [handle] function.
    *
    * An [AsyncError] [:e:] is matched by a test function if [:test(e):] returns
    * true. If [test] is omitted, every error is considered matching.
    *
    * 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);
   }
 
   /**
    * Create a new stream from this stream that converts each element
    * into zero or more events.
    *
    * Each incoming event is converted to an [Iterable] of new events,
    * and each of these new events are then sent by the returned stream
@@ skipping 12 matching lines @@
 
   /**
    * Chain this stream as the input of the provided [StreamTransformer].
    *
    * Returns the result of [:streamTransformer.bind:] itself.
    */
   Stream transform(StreamTransformer<T, dynamic> streamTransformer) {
     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)) {
     _FutureImpl result = new _FutureImpl();
     var value = initialValue;
     StreamSubscription subscription;
     subscription = this.listen(
       (T element) {
         _runUserCode(
           () => combine(value, element),
@@ skipping 33 matching lines @@
    *
    * Completes the [Future] when the answer is known.
    * If this stream reports an error, the [Future] will report that error.
    */
   Future<bool> contains(T match) {
     _FutureImpl<bool> future = new _FutureImpl<bool>();
     StreamSubscription subscription;
     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)
           );
         },
         onError: future._setError,
         onDone: () {
           future._setValue(false);
         },
         unsubscribeOnError: true);
@@ skipping 620 matching lines @@
    */
   factory StreamTransformer.from({
       void onData(S data, StreamSink<T> sink),
       void onError(AsyncError error, StreamSink<T> sink),
       void onDone(StreamSink<T> sink)}) {
     return new _StreamTransformerImpl<S, T>(onData, onError, onDone);
   }
 
   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();
+}