Redo StreamTransformers so they work with Stack traces.

sdk/lib/async/stream_transformers.dart

Index: sdk/lib/async/stream_transformers.dart
diff --git a/sdk/lib/async/stream_transformers.dart b/sdk/lib/async/stream_transformers.dart
new file mode 100644
index 0000000000000000000000000000000000000000..57627ba91b6f9cfbe7a5fe0cf324408f1aba9b97
--- /dev/null
+++ b/sdk/lib/async/stream_transformers.dart
@@ -0,0 +1,263 @@
+// Copyright (c) 2013, 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;
+
+/**
+ * Wraps an [_EventSink] so it exposes only the [EventSink] interface.
+ */
+class _EventSinkAdapter<T> implements EventSink<T> {
+  _EventSink _sink;
+  _EventSinkAdapter(this._sink);
+
+  void add(T data) { _sink._add(data); }
+  void addError(error, [StackTrace stackTrace]) {
+    _sink._addError(error, stackTrace);
+  }
+  void close() { _sink._close(); }
+}
+
+/**
+ * A StreamSubscription that pipes data through a sink.
+ *
+ * The constructor of this class takes a [_SinkMapper] which maps from
+ * [EventSink] to [EventSink]. The input to the mapper is the output of
+ * the transformation. The returned sink is the transformation's input.
+ */
+class _SinkTransformerStreamSubscription<S, T>
+    extends _BufferingStreamSubscription<T> {
+  /// EventSink wrapper for this class.
+  _EventSinkAdapter<T> _sink;
+
+  /// The transformer's input sink.
+  EventSink _transformerSink;
+
+  /// The subscription to the input stream.
+  StreamSubscription<S> _subscription;
+
+  _SinkTransformerStreamSubscription(Stream<S> source,
+                                     _SinkMapper mapper,
+                                     void onData(T data),
+                                     Function onError,
+                                     void onDone(),
+                                     bool cancelOnError)
+            // We set the adapter's target only when the user is allowed to
+            // send data.
+          : super(onData, onError, onDone, cancelOnError) {

[Lasse Reichstein Nielsen, 2013/10/04 14:35:53]

Indent by four only.

[floitsch, 2013/10/05 18:47:40]

Done.

+    _sink = new _EventSinkAdapter<T>(this);

[Lasse Reichstein Nielsen, 2013/10/04 14:35:53]

I don't think _sink is used anywhere else, so could it just be a local variable
here instead of a field.

[floitsch, 2013/10/05 18:47:40]

Done.

+    _transformerSink = mapper(_sink);
+    _subscription = source.listen(_handleData,
+                                  onError: _handleError,
+                                  onDone: _handleDone);
+  }
+
+  /** Whether this subscription is still subscribed to its source. */
+  bool get _isSubscribed => _subscription != null;
+
+  void _onPause() {
+    if (_isSubscribed) _subscription.pause();
+  }
+
+  void _onResume() {
+    if (_isSubscribed) _subscription.resume();
+  }
+
+  void _onCancel() {
+    if (_isSubscribed) {
+      StreamSubscription subscription = _subscription;
+      _subscription = null;
+      subscription.cancel();
+    }
+  }
+
+  void _handleData(S data) {
+    try {
+      _transformerSink.add(data);
+    } catch (e, s) {
+      _addError(_asyncError(e, s), s);
+    }
+  }
+
+  void _handleError(error, [stackTrace]) {
+    try {
+      _transformerSink.addError(error, stackTrace);
+    } catch (e, s) {
+      if (identical(e, error)) {
+        _addError(error, stackTrace);
+      } else {
+        _addError(_asyncError(e, s), s);
+      }
+    }
+  }
+
+  void _handleDone() {
+    try {
+      _subscription = null;
+      _transformerSink.close();
+    } catch (e, s) {
+      _addError(_asyncError(e, s), s);
+    }
+  }
+}
+
+
+typedef EventSink<S> _SinkMapper<S, T>(EventSink<T> output);
+
+/**
+ * A StreamTransformer for Sink-mappers.
+ *
+ * A Sink-mapper takes an [EventSink] (its output) and returns another
+ * EventSink (its input).
+ *
+ * Note that this class can be `const`.
+ */
+class _StreamSinkTransformer<S, T> implements StreamTransformer<S, T> {
+  final _SinkMapper<S, T> _sinkMapper;
+  const _StreamSinkTransformer(this._sinkMapper);
+
+  Stream<T> bind(Stream<S> stream)
+      => new _BoundSinkStream<S, T>(stream, _sinkMapper);
+}
+
+/**
+ * The result of binding a StreamTransformer for Sink-mappers.
+ *
+ * It contains the bound Stream and the sink-mapper. Only, when the user starts

[Lasse Reichstein Nielsen, 2013/10/04 14:35:53]

Only when the user starts listening to this stream is the sink-mapper invoked.

[floitsch, 2013/10/05 18:47:40]

Done.

+ * listening to this stream, the sink-mapper is invoked. The result is used
+ * to create a StreamSubscription that transforms events.
+ */
+class _BoundSinkStream<S, T> extends Stream<T> {
+  final _SinkMapper<S, T> _sinkMapper;
+  final Stream<S> _stream;
+
+  _BoundSinkStream(this._stream, this._sinkMapper);
+
+  StreamSubscription<T> listen(void onData(T event),
+                               { Function onError,
+                                 void onDone(),
+                                 bool cancelOnError }) {
+    cancelOnError = identical(true, cancelOnError);
+    return new _SinkTransformerStreamSubscription(
+        _stream, _sinkMapper, onData, onError, onDone, cancelOnError);
+  }
+}
+
+/// Data-handler coming from [StreamTransformer.fromHandlers].
+typedef void _TransformDataHandler<S, T>(S data, EventSink<T> sink);
+/// Error-handler coming from [StreamTransformer.fromHandlers].
+typedef void _TransformErrorHandler<T>(
+    Object error, StackTrace stackTrace, EventSink<T> sink);
+/// Done-handler coming from [StreamTransformer.fromHandlers].
+typedef void _TransformDoneHandler<T>(EventSink<T> sink);
+
+/**
+ * Wraps handlers (from [StreamTransformer.fromHandlers]) into an EventSink.

[Lasse Reichstein Nielsen, 2013/10/04 14:35:53]

`EventSink`

[floitsch, 2013/10/05 18:47:40]

Done.

+ *
+ * This way we can reuse the code from [_StreamSinkTransformer].
+ */
+class _HandlerEventSink<S, T> implements EventSink<T> {
+  final _TransformDataHandler<S, T> _handleData;
+  final _TransformErrorHandler<T> _handleError;
+  final _TransformDoneHandler<T> _handleDone;
+
+  /// The output sink where the handlers should send their data into.
+  final EventSink<T> _sink;
+
+  _HandlerEventSink(this._handleData, this._handleError, this._handleDone,
+                    this._sink);
+
+  void add(T data) => _handleData(data, _sink);
+  void addError(Object error, [StackTrace stackTrace])
+      => _handleError(error, stackTrace, _sink);
+  void close() => _handleDone(_sink);
+}
+
+/**
+ * A StreamTransformer that transformers events with the given handlers.
+ *
+ * Note that this transformer can only be used once.
+ */
+class _StreamHandlerTransformer<S, T> extends _StreamSinkTransformer<S, T> {
+  bool _hasBeenUsed = false;
+
+  _StreamHandlerTransformer({
+      void handleData(S data, EventSink<T> sink),
+      void handleError(Object error, StackTrace stackTrace, EventSink<T> sink),
+      void handleDone(EventSink<T> sink)})
+      : super((EventSink<S> outputSink) {
+        if (handleData == null) handleData = _defaultHandleData;
+        if (handleError == null) handleError = _defaultHandleError;
+        if (handleDone == null) handleDone = _defaultHandleDone;
+        return new _HandlerEventSink<S, T>(
+            handleData, handleError, handleDone, outputSink);
+      });
+
+  Stream<T> bind(Stream<S> stream) {
+    if (_hasBeenUsed) {
+      throw new StateError("Transformer has already been used.");
+    }
+    _hasBeenUsed = true;
+    return super.bind(stream);
+  }
+
+  /** Default data handler forwards all data. */
+  static void _defaultHandleData(var data, EventSink sink) {
+    sink.add(data);
+  }
+
+  /** Default error handler forwards all errors. */
+  static void _defaultHandleError(error, StackTrace stackTrace, EventSink sink) {

[Lasse Reichstein Nielsen, 2013/10/04 14:35:53]

long line.

[floitsch, 2013/10/05 18:47:40]

Done.

+    sink.addError(error);
+  }
+
+  /** Default done handler forwards done. */
+  static void _defaultHandleDone(EventSink sink) {
+    sink.close();
+  }
+}
+
+/// A closure mapping a stream and cancelOnError to a StreamSubscription.
+typedef StreamSubscription<T> _SubscriptionTransformer<S, T>(
+    Stream<S> stream, bool cancelOnError);
+
+/**
+ * This class is the most powerful [StreamTransformer].

[Lasse Reichstein Nielsen, 2013/10/04 14:35:53]

powerful -> generic?
"the most" sounds like it should be quantified. How about a fully generic stream
transformer, or an unrestricted one.

[floitsch, 2013/10/05 18:47:40]

"fully generic".
done.

+ *
+ * The given transformer closure maps from Stream, cancelOnError to a
+ * StreamSubscription. This allows it to do everything it needs. It can also

[Lasse Reichstein Nielsen, 2013/10/04 14:35:53]

do everything it needs -> interact with the entire stream life cycle.
Or something else, but the class doesn't have needs.

[floitsch, 2013/10/05 18:47:40]

Completely rewrote the comment.

+ * watch for cancels, which the other transformers above cannot do.
+ */
+class _StreamSubscriptionTransformer<S, T> implements StreamTransformer<S, T> {
+  final _SubscriptionTransformer<S, T> _transformer;
+
+  const _StreamSubscriptionTransformer(this._transformer);
+
+  Stream<T> bind(Stream<S> stream)
+      => new _BoundSubscriptionStream<S, T>(stream, _transformer);
+}
+
+/**
+ * The bound version of [_StreamSubscriptionTransformer].
+ *
+ * This class is a [Stream] that has already an input stream. The transformer
+ * closure is, however, only invoked when this stream is listened to.
+ */
+class _BoundSubscriptionStream<S, T> extends Stream<T> {
+  final _SubscriptionTransformer<S, T> _transformer;
+  final Stream<S> _stream;
+
+  _BoundSubscriptionStream(this._stream, this._transformer);
+
+  StreamSubscription<T> listen(void onData(T event),
+                               { Function onError,
+                                 void onDone(),
+                                 bool cancelOnError }) {
+    cancelOnError = identical(true, cancelOnError);
+    StreamSubscription<T> result = _transformer(_stream, cancelOnError);
+    result.onData(onData);
+    result.onError(onError);
+    result.onDone(onDone);
+    return result;
+  }
+}