| 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(
|
| + 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. */
|
| + EventSink<T> _eventSink;
|
| +
|
| + /**
|
| + * 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);
|
| + }
|
| +}
|
|
|