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.
+ * 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;
   }
 
   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.
+    _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 250 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));
+    }
+  }
+}