Remove transformEvents and make StreamEventTransformer extend StreamTransformer.

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 191 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. 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)) {
     _FutureImpl result = new _FutureImpl();
     var value = initialValue;
     StreamSubscription subscription;
     subscription = this.listen(
       // TODO(ahe): Restore type when feature is implemented in dart2js
       // checked mode. http://dartbug.com/7733
       (/*T*/ element) {
         _runUserCode(
@@ skipping 683 matching lines @@
  * The target of a [Stream.pipe] call.
  *
  * The [Stream.pipe] call will pass itself to this object, and then return
  * the resulting [Future]. The pipe should complete the future when it's
  * done.
  */
 abstract class StreamConsumer<S, T> {
   Future<T> consume(Stream<S> stream);
 }
 
+
 /**
  * The target of a [Stream.transform] call.
  *
  * The [Stream.transform] call will pass itself to this object and then return
  * the resulting stream.
  */
 abstract class StreamTransformer<S, T> {
   /**
    * Create a [StreamTransformer] that delegates events to the given functions.
    *
-   * If a parameter is omitted, a default handler is used that forwards the
-   * event directly to the sink.
+   * This is actually a [StreamEventTransformer] where the event handling is
+   * performed by the function arguments.
+   * If an argument is omitted, it acts as the corresponding default method from
+   * [StreamEventTransformer].
    *
-   * Pauses on the returned stream are forwarded to the input stream as well.
+   * Example use:
+   *
+   *     stringStream.transform(new StreamTransformer<String, String>(
+   *         handleData: (Strung value, StreamSink<String> sink) {
+   *           sink.add(value);
+   *           sink.add(value);  // Duplicate the incoming events.
+   *         }));
+   *
    */
-  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);
+  factory StreamTransformer({
+      void handleData(S data, StreamSink<T> sink),
+      void handleError(AsyncError error, StreamSink<T> sink),
+      void handleDone(StreamSink<T> sink)}) {
+    return new _StreamTransformerImpl<S, T>(handleData,
+                                            handleError,
+                                            handleDone);
   }
 
   Stream<T> bind(Stream<S> stream);
 }
 
 
 /**
- * A transformer of stream events.
+ * Base class for transformers that modifies stream events.
  *
  * A [StreamEventTransformer] transforms incoming Stream
- * events of one kind into outgoing events of another kind.
+ * events of one kind into outgoing events of (possibly) another kind.
+ *
+ * Subscribing on the stream returned by [bind] is the same as subscribing on
+ * the source 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 forwarded to this stream.
+ *
+ * An example that duplicates all data events:
+ *
+ *     class DoubleTransformer<T> extends StreamEventTransformerBase<T, T> {
+ *       void handleData(T data, StreamSink<T> sink) {
+ *         sink.add(value);
+ *         sink.add(value);
+ *       }
+ *     }
+ *     someTypeStream.transform(new DoubleTransformer<Type>());
  *
  * 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.
+ * the events unmodified. If using the default [handleData] the generic type [T]
+ * needs to be assignable to [S].
  */
-abstract class StreamEventTransformer<S, T> {
+abstract class StreamEventTransformer<S, T> implements StreamTransformer<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);
+  Stream<T> bind(Stream<S> source) {
+    return new EventTransformStream<S, T>(source, this);
   }
 
-
   /**
    * 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);
   }
@@ skipping 12 matching lines @@
    * 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;
+  final Stream<S> _source;
+  final 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,
@@ skipping 35 matching lines @@
   }
 
   void cancel() {
     if (_subscription != null) {
       _subscription.cancel();
       _subscription = null;
     }
   }
 
   void _handleData(S data) {
-    _transformer.handleData(data, _sink);
+    try {
+      _transformer.handleData(data, _sink);
+    } catch (e, s) {
+      _sendError(_asyncError(e, s));
+    }
   }
 
   void _handleError(AsyncError error) {
-    _transformer.handleError(error, _sink);
+    try {
+      _transformer.handleError(error, _sink);
+    } catch (e, s) {
+      _sendError(_asyncError(e, s, error));
+    }
   }
 
   void _handleDone() {
-    _transformer.handleDone(_sink);
+    try {
+      _transformer.handleDone(_sink);
+    } catch (e, s) {
+      _sendError(_asyncError(e, s));
+    }
   }
 
   // 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();
 }