Add transformer support to Converters.

sdk/lib/convert/chunked_conversion.dart

Index: sdk/lib/convert/chunked_conversion.dart
diff --git a/sdk/lib/convert/chunked_conversion.dart b/sdk/lib/convert/chunked_conversion.dart
index f8f049188acd62090aaced9796dd8af6f43e6b61..7674421fdf0a5bd75d14a622799795d9f372c03a 100644
--- a/sdk/lib/convert/chunked_conversion.dart
+++ b/sdk/lib/convert/chunked_conversion.dart
@@ -46,3 +46,103 @@ class _SimpleCallbackSink<T> extends ChunkedConversionSink<T> {
   void add(T chunk) { _accumulated.add(chunk); }
   void close() { _callback(_accumulated); }
 }
+
+/**
+ * This class wraps a [Converter] for use as a [StreamTransformer].
+ */
+class _ConverterTransformStream<S, T> extends EventTransformStream<S, T> {
+  final _ConverterStreamEventTransformer<S, T> _eventTransformer;
+
+  _ConverterTransformStream(Stream<S> source, Converter converter)
+      : this._withEventTransformer(

[Søren Gjesse, 2013/07/26 07:37:23]

Why do you need two constructors?

[floitsch, 2013/07/26 10:35:32]

Because I need to initialize a final field and super with the same variable
("new _ConverterStreamEventTransformer").
I don't know of a better way to do this.

+          source,
+          new _ConverterStreamEventTransformer<S, T>(converter));
+
+  _ConverterTransformStream._withEventTransformer(
+      Stream<S> source,
+      _ConverterStreamEventTransformer<S, T> eventTransformer)
+      : _eventTransformer = eventTransformer,
+        super(source, eventTransformer);
+
+  /**
+   * Starts listening to `this`.
+   *
+   * This starts the chunked conversion.
+   */
+  StreamSubscription<T> listen(void onData(T data),
+                               { void onError(error),
+                                 void onDone(),
+                                 bool cancelOnError }) {
+    _eventTransformer._startChunkedConversion();
+    return super.listen(onData, onError: onError, onDone: onDone,
+                        cancelOnError: cancelOnError);
+  }
+}
+
+/**
+ * This class converts implements the logic for a chunked conversion as a
+ * stream transformer.
+ *
+ * It is used as strategy in the [EventTransformStream].
+ *
+ * It also implements the [ChunkedConversionSink] interface so that it
+ * can be used as output sink in a chunked conversion.
+ */
+class _ConverterStreamEventTransformer<S, T>
+    implements ChunkedConversionSink<T>, StreamEventTransformer<S, T> {
+  final Converter _converter;
+
+  /** At every [handleData] this field is updated with the new event sink. */

[Søren Gjesse, 2013/07/26 07:37:23]

I don't think this should be a dartdoc comment.

[floitsch, 2013/07/26 10:35:32]

It's inside a private class, but done. (I initially hesitated too)

+  EventSink<T> _eventSink;
+
+  /**

[Søren Gjesse, 2013/07/26 07:37:23]

Ditto.

[floitsch, 2013/07/26 10:35:32]

Done.

+   * The input sink for new data. All data that is received with
+   * [handleData] is added into this sink.
+   */
+  ChunkedConversionSink _chunkedSink;
+
+  _ConverterStreamEventTransformer(this._converter);
+
+  /**
+   * Starts the chunked conversion.
+   */
+  void _startChunkedConversion() {
+    _chunkedSink = _converter.startChunkedConversion(this);
+  }
+
+  /**
+   * Not supported.
+   */
+  Stream bind(Stream otherStream) {
+    throw new UnsupportedError("Converter streams must not call bind");
+  }
+
+  void add(T o) => _eventSink.add(o);
+  void close() => _eventSink.close();
+
+  void handleData(S event, EventSink<T> eventSink) {
+    _eventSink = eventSink;
+    try {
+      _chunkedSink.add(event);
+    } catch(e) {
+      eventSink.addError(e);
+    } finally {
+      _eventSink = null;
+    }
+  }
+
+  void handleDone(EventSink<T> eventSink) {
+    _eventSink = eventSink;
+    try {
+      _chunkedSink.close();
+    } catch(e) {
+      eventSink.addError(e);
+    } finally {
+      _eventSink = null;
+    }
+  }
+
+  void handleError(var errorEvent, EventSink<T> eventSink) {
+    eventSink.addError(errorEvent);
+  }
+}