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:
+   *
+   *     someTypeStream.transform(new StreamTransformer<Type, Type>(

[floitsch, 2013/01/29 13:37:04]

Make it a real example.
stringStream.transform(new StreamTransformer<String, String>(
    handleData: (String value, StreamSink<String> sink) {
      sink.add(value);
      sink.add(value);  // Double the incoming events.
    }));

[Lasse Reichstein Nielsen, 2013/01/29 14:04:14]

Done.

+   *         handleData: (Type value, StreamSink<Type> sink) {
+   *           sink.add(value);
+   *           sink.add(value);
+   *         }));
+   *
    */
-  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

[floitsch, 2013/01/29 13:37:04]

The pause needs more description:
- does the StreamEventTransformer buffer if necessary?
- I guess that your sentence means that pauses are forwarded to the stream's
input-stream.

[Lasse Reichstein Nielsen, 2013/01/29 14:04:14]

The latter. The returned stream can't buffer at all, it can just wrap a
subscription in another and transform events on the way. All pauses and
unsubscribes are forwarded.
Reworded slightly.

+ * subscription are pauses on 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 73 matching lines @@
 }
 
 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();
 }