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

sdk/lib/async/stream_pipe.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;
 
 /** Utility function to create an [AsyncError] if [error] isn't one already. */
 AsyncError _asyncError(Object error, Object stackTrace, [AsyncError cause]) {
   if (error is AsyncError) return error;
   if (cause == null) return new AsyncError(error, stackTrace);
@@ skipping 35 matching lines @@
  *
  * This class is intended for internal use only.
  */
 abstract class _ForwardingStream<S, T> extends Stream<T> {
   final Stream<S> _source;
 
   _ForwardingStream(this._source);
 
   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.
 
   void _handleData(S data, _StreamOutputSink<T> sink) {
     var outputData = data;
     sink._sendData(outputData);
   }
 
   void _handleError(AsyncError error, _StreamOutputSink<T> sink) {
     sink._sendError(error);
   }
 
   void _handleDone(_StreamOutputSink<T> sink) {
     sink._sendDone();
   }
 }
 
 /**
- * Abstract superclass for subscriptions that forward to other subscriptions.
+ * Basic implemnetation of the handler features of a [StreamSubscription].

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

implementation.
I don't understand the sentence.

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

Rewritten.

+ *
+ * Supports storing and updating of event handlers on a subscription.
  */
-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;
   }
 
   void onError(void handleError(AsyncError error)) {
     if (handleError == null) handleError = _nullErrorHandler;
     _onError = handleError;
   }
 
   void onDone(void handleDone()) {
     if (handleDone == null) handleDone = _nullDoneHandler;
     _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.

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

Took me way too long to understand the comment...
"The "unsubscribeOnError" flag is not forwarded to the source's subscription.
Only if an error is sent to the successors of this subscription is the stream
unsubscribed.
This allows the error-handler to intercept the error and correctly handle it
without being unsubscribed."

+    _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");
     }
     _subscription.pause(resumeSignal);
   }
 
   void resume() {
     if (_subscription == null) {
       throw new StateError("Subscription has been unsubscribed");
@@ skipping 10 matching lines @@
   }
 
   // _StreamOutputSink interface. Sends data to this subscription.
 
   void _sendData(T data) {
     _onData(data);
   }
 
   void _sendError(AsyncError error) {
     _onError(error);
+    if (_unsubscribeOnError) {
+      _subscription.cancel();
+      _subscription = null;
+    }
   }
 
   void _sendDone() {
     // If the transformation sends a done signal, we stop the subscription.
     if (_subscription != null) {
       _subscription.cancel();
       _subscription = null;
     }
     _onDone();
   }
@@ skipping 248 matching lines @@
         return null;
       }
       if (!isEqual) {
         sink._sendData(inputEvent);
         _previous = inputEvent;
       }
     }
   }
 }
 
+// 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.
  *
  * 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 _StreamTransformerImpl<S, T> implements StreamTransformer<S, T> {
@@ skipping 14 matching lines @@
     // Cache a Sink object to avoid creating a new one for each event.
     _sink = new _StreamImplSink(stream);
     source.listen(_handleData, onError: _handleError, onDone: _handleDone);
     return stream;
   }
 
   void _handleData(S data) {
     try {
       _onData(data, _sink);
     } catch (e, s) {
-      _stream._signalError(_asyncError(e, s));
+      _sink.signalError(_asyncError(e, s));
     }
   }
 
   void _handleError(AsyncError error) {
     try {
       _onError(error, _sink);
     } catch (e, s) {
-      _stream._signalError(_asyncError(e, s, error));
+      _sink.signalError(_asyncError(e, s, error));
     }
   }
 
   void _handleDone() {
     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. */
 class _StreamImplSink<T> implements StreamSink<T> {
   _StreamImpl<T> _target;
   _StreamImplSink(this._target);
   void add(T data) { _target._add(data); }
   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));
+    }
+  }
+}