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

This allows intercepting events at the subscription level instead of creating
a new full stream and adding the events to that.

Also fix a number of typos and bugs detected by analyzer.

Review URL: https://codereview.chromium.org//11953103

git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart@17746 260f80e4-7a28-3924-810f-c04153c831b5

sdk/lib/async/future.dart

@@ -113,7 +113,7 @@ abstract class Future<T> {
    * See [Completer]s, for futures with values that are computed asynchronously.
    */
   factory Future.delayed(int milliseconds, T value()) {
-    _FutureImpl<T> future = new _ThenFuture<dynamic, T>((_) => value());
+    _ThenFuture<dynamic, T> future = new _ThenFuture<dynamic, T>((_) => value());
     new Timer(milliseconds, (_) => future._sendValue(null));
     return future;
   }

sdk/lib/async/merge_stream.dart

@@ -159,29 +159,34 @@ class _CycleEntry<T> {
   Stream source;
   /** The active subscription, if any. */
   StreamSubscription subscription = null;
+  /** Whether the subscription is currently paused. */
+  bool isPaused = false;
   /** Next entry in a linked list of entries. */
   _CycleEntry next;
 
   _CycleEntry(this.stream, this.source);
 
   void cancel() {
-    // This method may be called event if this entry has never been activated.
+    // This method may be called even if this entry has never been activated.
     if (subscription != null) {
       subscription.cancel();
       subscription = null;
+      isPaused = false;
     }
   }
 
   void pause() {
     ensureSubscribed();
-    if (!subscription.isPaused) {
+    if (!isPaused) {
       subscription.pause();
+      isPaused = true;
     }
   }
 
   void activate() {
     ensureSubscribed();
-    if (subscription.isPaused) {
+    if (isPaused) {
+      isPaused = false;
       subscription.resume();
     }
   }

sdk/lib/async/stream.dart

@@ -40,12 +40,11 @@ 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] returns false.
+ * A broadcast stream inheriting from [Stream] must override [isBroadcast]
+ * to return [:true:].
  */
 abstract class Stream<T> {
   Stream();
@@ -93,6 +92,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 +172,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 +209,26 @@ 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. Pauses on the returned
+   * subscription are pauses on this stream.
+   *
+   * An example that duplicates all data events:
+   *
+   *     someStream.transformEvents(new StreamEventTransformer.from(
+   *       handleData: (var value, StreamSink sink) {
+   *         sink.add(value);
+   *         sink.add(value);
+   *       }));
+   */
+  Stream transformEvents(StreamEventTransformer<T, dynamic> transformer) {
+    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)) {
@@ -263,11 +286,11 @@ abstract class Stream<T> {
         // checked mode. http://dartbug.com/7733
         (/*T*/ element) {
           _runUserCode(
-            () => match(element),
+            () => (element == match),
             (bool isMatch) {
               if (isMatch) {
                 subscription.cancel();
-                future._setValue(element);
+                future._setValue(true);
               }
             },
             _cancelAndError(subscription, future)
@@ -934,3 +957,174 @@ 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. In that case the generic type [T] needs to be
+ * assignable to [S].
+ *
+ * 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({
+    void handleData(S data, StreamSink<T> sink),
+    void handleError(AsyncError error, StreamSink<T> sink),
+    void handleDone(StreamSink<T> sink)
+  }) {
+    return new _StreamEventTransformerImpl<S, T>(handleData,
+                                                 handleError,
+                                                 handleDone);
+  }
+
+
+  /**
+   * Act on incoming data event.
+   *
+   * The method may generate any number of events on the sink, but should
+   * not throw.
+   */
+  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();
+}

sdk/lib/async/stream_impl.dart

@@ -62,10 +62,10 @@ abstract class _StreamImpl<T> extends Stream<T> {
   // ------------------------------------------------------------------
   // Stream interface.
 
-  StreamSubscription listen(void onData(T data),
-                            { void onError(AsyncError error),
-                              void onDone(),
-                              bool unsubscribeOnError }) {
+  StreamSubscription<T> listen(void onData(T data),
+                               { void onError(AsyncError error),
+                                 void onDone(),
+                                 bool unsubscribeOnError }) {
     if (_isComplete) {
       return new _DoneSubscription(onDone);
     }
@@ -73,7 +73,7 @@ abstract class _StreamImpl<T> extends Stream<T> {
     if (onError == null) onError = _nullErrorHandler;
     if (onDone == null) onDone = _nullDoneHandler;
     unsubscribeOnError = identical(true, unsubscribeOnError);
-    _StreamListener subscription =
+    _StreamSubscriptionImpl subscription =
         _createSubscription(onData, onError, onDone, unsubscribeOnError);
     _addListener(subscription);
     return subscription;
@@ -1092,8 +1092,10 @@ class _DoneSubscription<T> implements StreamSubscription<T> {
   bool get _isComplete => _timer == null && _pauseCount == 0;
 
   void onData(void handleAction(T value)) {}
-  void onError(void handleError(StateError error)) {}
-  void onDone(void handleDone(T value)) {
+
+  void onError(void handleError(AsyncError error)) {}
+
+  void onDone(void handleDone()) {
     _handler = handleDone;
   }
 

sdk/lib/async/stream_pipe.dart

@@ -53,20 +53,23 @@ abstract class _ForwardingStream<S, T> extends Stream<T> {
 
   bool get isBroadcast => _source.isBroadcast;
 
-  bool asBroadcastStream() => _source.asBroadcastStream;
-
-  StreamSubscription listen(void onData(T value),
-                            { void onError(AsyncError error),
-                              void onDone(),
-                              bool unsubscribeOnError }) {
+  StreamSubscription<T> listen(void onData(T value),
+                              { void onError(AsyncError error),
+                                void onDone(),
+                                bool unsubscribeOnError }) {
     if (onData == null) onData = _nullDataHandler;
     if (onError == null) onError = _nullErrorHandler;
     if (onDone == null) onDone = _nullDoneHandler;
     unsubscribeOnError = identical(true, unsubscribeOnError);
-    StreamSubscription subscription =
-        new _ForwardingStreamSubscription<S, T>(
-            this, onData, onError, onDone, unsubscribeOnError);
-    return subscription;
+    return _createSubscription(onData, onError, onDone, unsubscribeOnError);
+  }
+
+  StreamSubscription<T> _createSubscription(void onData(T value),
+                                            void onError(AsyncError error),
+                                            void onDone(),
+                                            bool unsubscribeOnError) {
+    return new _ForwardingStreamSubscription<S, T>(
+        this, onData, onError, onDone, unsubscribeOnError);
   }
 
   // Override the following methods in subclasses to change the behavior.
@@ -86,33 +89,26 @@ abstract class _ForwardingStream<S, T> extends Stream<T> {
 }
 
 /**
- * Abstract superclass for subscriptions that forward to other subscriptions.
+ * Common behavior of [StreamSubscription] classes.
+ *
+ * Stores and allows updating of the event handlers of a [StreamSubscription].
  */
-class _ForwardingStreamSubscription<S, T>
-    implements StreamSubscription<T>, _StreamOutputSink<T> {
-  final _ForwardingStream<S, T> _stream;
+abstract class _BaseStreamSubscription<T> implements StreamSubscription<T> {
   // TODO(ahe): Restore type when feature is implemented in dart2js
   // checked mode. http://dartbug.com/7733
   var /* _DataHandler<T> */ _onData;
   _ErrorHandler _onError;
   _DoneHandler _onDone;
 
-  StreamSubscription<S> _subscription;
-
-  _ForwardingStreamSubscription(this._stream,
-                                this._onData,
-                                this._onError,
-                                this._onDone,
-                                bool unsubscribeOnError) {
-    _subscription =
-        _stream._source.listen(_handleData,
-                               onError: _handleError,
-                               onDone: _handleDone,
-                               unsubscribeOnError: unsubscribeOnError);
+  _BaseStreamSubscription(this._onData,
+                          this._onError,
+                          this._onDone) {
+    if (_onData == null) _onData = _nullDataHandler;
+    if (_onError == null) _onError = _nullErrorHandler;
+    if (_onDone == null) _onDone = _nullDoneHandler;
   }
 
   // StreamSubscription interface.
-
   void onData(void handleData(T event)) {
     if (handleData == null) handleData = _nullDataHandler;
     _onData = handleData;
@@ -128,6 +124,39 @@ class _ForwardingStreamSubscription<S, T>
     _onDone = handleDone;
   }
 
+  void pause([Future resumeSignal]);
+
+  void resume();
+
+  void cancel();
+}
+
+
+/**
+ * Abstract superclass for subscriptions that forward to other subscriptions.
+ */
+class _ForwardingStreamSubscription<S, T>
+    extends _BaseStreamSubscription<T> implements _StreamOutputSink<T> {
+  final _ForwardingStream<S, T> _stream;
+  final bool _unsubscribeOnError;
+
+  StreamSubscription<S> _subscription;
+
+  _ForwardingStreamSubscription(this._stream,
+                                void onData(T data),
+                                void onError(AsyncError error),
+                                void onDone(),
+                                this._unsubscribeOnError)
+      : super(onData, onError, onDone) {
+    // Don't unsubscribe on incoming error, only if we send an error forwards.
+    _subscription =
+        _stream._source.listen(_handleData,
+                               onError: _handleError,
+                               onDone: _handleDone);
+  }
+
+  // StreamSubscription interface.
+
   void pause([Future resumeSignal]) {
     if (_subscription == null) {
       throw new StateError("Subscription has been unsubscribed");
@@ -158,6 +187,10 @@ class _ForwardingStreamSubscription<S, T>
 
   void _sendError(AsyncError error) {
     _onError(error);
+    if (_unsubscribeOnError) {
+      _subscription.cancel();
+      _subscription = null;
+    }
   }
 
   void _sendDone() {
@@ -428,11 +461,28 @@ class _DistinctStream<T> extends _ForwardingStream<T, T> {
   }
 }
 
+// Stream transformations and event transformations.
 
 typedef void _TransformDataHandler<S, T>(S data, StreamSink<T> sink);
 typedef void _TransformErrorHandler<T>(AsyncError data, StreamSink<T> sink);
 typedef void _TransformDoneHandler<T>(StreamSink<T> sink);
 
+/** Default data handler forwards all data. */
+void _defaultHandleData(var data, StreamSink sink) {
+  sink.add(data);
+}
+
+/** Default error handler forwards all errors. */
+void _defaultHandleError(AsyncError error, StreamSink sink) {
+  sink.signalError(error);
+}
+
+/** Default done handler forwards done. */
+void _defaultHandleDone(StreamSink sink) {
+  sink.close();
+}
+
+
 /**
  * A stream transformer that intercepts all events and can generate any event as
  * output.
@@ -467,7 +517,7 @@ class _StreamTransformerImpl<S, T> implements StreamTransformer<S, T> {
     try {
       _onData(data, _sink);
     } catch (e, s) {
-      _stream._signalError(_asyncError(e, s));
+      _sink.signalError(_asyncError(e, s));
     }
   }
 
@@ -475,7 +525,7 @@ class _StreamTransformerImpl<S, T> implements StreamTransformer<S, T> {
     try {
       _onError(error, _sink);
     } catch (e, s) {
-      _stream._signalError(_asyncError(e, s, error));
+      _sink.signalError(_asyncError(e, s, error));
     }
   }
 
@@ -483,22 +533,9 @@ class _StreamTransformerImpl<S, T> implements StreamTransformer<S, T> {
     try {
       _onDone(_sink);
     } catch (e, s) {
-      _stream._signalError(_asyncError(e, s));
+      _sink.signalError(_asyncError(e, s));
     }
   }
-
-  /** Default data handler forwards all data. */
-  static void _defaultHandleData(var data, StreamSink sink) {
-    sink.add(data);
-  }
-  /** Default error handler forwards all errors. */
-  static void _defaultHandleError(AsyncError error, StreamSink sink) {
-    sink.signalError(error);
-  }
-  /** Default done handler forwards done. */
-  static void _defaultHandleDone(StreamSink sink) {
-    sink.close();
-  }
 }
 
 /** Creates a [StreamSink] from a [_StreamImpl]'s input methods. */
@@ -509,3 +546,51 @@ class _StreamImplSink<T> implements StreamSink<T> {
   void signalError(AsyncError error) { _target._signalError(error); }
   void close() { _target._close(); }
 }
+
+
+/**
+ * A stream transformer that intercepts all events and can generate any event as
+ * output.
+ *
+ * Each incoming event on the source stream is passed to the corresponding
+ * provided event handler, along with a [StreamSink] linked to the output
+ * Stream.
+ * The handler can then decide exactly which events to send to the output.
+ */
+class _StreamEventTransformerImpl<S, T>
+    implements StreamEventTransformer<S, T> {
+  final _TransformDataHandler<S, T> _handleData;
+  final _TransformErrorHandler<T> _handleError;
+  final _TransformDoneHandler<T> _handleDone;
+
+  _StreamEventTransformerImpl(void onData(S data, StreamSink<T> sink),
+                              void onError(AsyncError data, StreamSink<T> sink),
+                              void onDone(StreamSink<T> sink))
+      : this._handleData = (onData == null ? _defaultHandleData : onData),
+        this._handleError = (onError == null ? _defaultHandleError : onError),
+        this._handleDone = (onDone == null ? _defaultHandleDone : onDone);
+
+  void handleData(S data, StreamSink<T> sink) {
+    try {
+      _handleData(data, sink);
+    } catch (e, s) {
+      sink.signalError(_asyncError(e, s));
+    }
+  }
+
+  void handleError(AsyncError error, StreamSink<T> sink) {
+    try {
+      _handleError(error, sink);
+    } catch (e, s) {
+      sink.signalError(_asyncError(e, s, error));
+    }
+  }
+
+  void handleDone(StreamSink<T> sink) {
+    try {
+      _handleDone(sink);
+    } catch (e, s) {
+      sink.signalError(_asyncError(e, s));
+    }
+  }
+}

tests/lib/async/stream_controller_test.dart

@@ -240,6 +240,42 @@ testSingleController() {
   c.add(9);
   c.close();
 
+  // test contains.
+  {
+    c = new StreamController();
+    // Error after match is not important.
+    sentEvents = new Events()..add("a")..add("x")..error("FAIL")..close();
+    Future<bool> contains = c.stream.contains("x");
+    contains.then((var c) {
+      Expect.isTrue(c);
+    });
+    sentEvents.replay(c);
+  }
+
+  {
+    c = new StreamController();
+    // Not matching is ok.
+    sentEvents = new Events()..add("a")..add("x")..add("b")..close();
+    Future<bool> contains = c.stream.contains("y");
+    contains.then((var c) {
+      Expect.isFalse(c);
+    });
+    sentEvents.replay(c);
+  }
+
+  {
+    c = new StreamController();
+    // Error before match makes future err.
+    sentEvents = new Events()..add("a")..error("FAIL")..add("b")..close();
+    Future<bool> contains = c.stream.contains("b");
+    contains.then((var c) {
+      Expect.fail("no value expected");
+    }).catchError((AsyncError e) {
+      Expect.equals("FAIL", e.error);
+    });
+    sentEvents.replay(c);
+  }
+
   // Test transform.
   c = new StreamController();
   sentEvents = new Events()..add("a")..error(42)..add("b")..close();

tests/lib/async/stream_event_transform_test.dart

@@ -0,0 +1,77 @@
+// Copyright (c) 2011, 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.
+
+import 'dart:async';
+import '../../../pkg/unittest/lib/unittest.dart';
+import 'event_helper.dart';
+
+void handleData(int data, StreamSink<int> sink) {
+  sink.signalError(new AsyncError("$data"));
+  sink.add(data + 1);
+}
+void handleError(AsyncError e, StreamSink<int> sink) {
+  String value = e.error;
+  int data = int.parse(value);
+  sink.add(data);
+  sink.signalError(new AsyncError("${data + 1}"));
+}
+
+void handleDone(StreamSink<int> sink) {
+  sink.add(99);
+  sink.close();
+}
+
+class EventTransformer extends StreamEventTransformer<int,int> {
+  void handleData(int data, StreamSink<int> sink) {
+    sink.signalError(new AsyncError("$data"));
+    sink.add(data + 1);
+  }
+  void handleError(AsyncError e, StreamSink<int> sink) {
+    String value = e.error;
+    int data = int.parse(value);
+    sink.add(data);
+    sink.signalError(new AsyncError("${data + 1}"));
+  }
+
+  void handleDone(StreamSink<int> sink) {
+    sink.add(99);
+    sink.close();
+  }
+}
+
+main() {
+  {
+    StreamController c = new StreamController();
+    Events expected = new Events()..error("0")..add(1)
+                                  ..error("1")..add(2)
+                                  ..add(3)..error("4")
+                                  ..add(99)..close();
+    Events input = new Events()..add(0)..add(1)..error("3")..close();
+    Events actual = new Events.capture(
+        c.stream.transformEvents(new EventTransformer()));
+    actual.onDone(() {
+      Expect.listEquals(expected.events, actual.events);
+    });
+    input.replay(c);
+  }
+
+  {
+    StreamController c = new StreamController();
+    Events expected = new Events()..error("0")..add(1)
+                                  ..error("1")..add(2)
+                                  ..add(3)..error("4")
+                                  ..add(99)..close();
+    Events input = new Events()..add(0)..add(1)..error("3")..close();
+    Events actual = new Events.capture(
+        c.stream.transformEvents(new StreamEventTransformer.from(
+            handleData: handleData,
+            handleError: handleError,
+            handleDone: handleDone
+        )));
+    actual.onDone(() {
+      Expect.listEquals(expected.events, actual.events);
+    });
+    input.replay(c);
+  }
+}