Adapt streams for additional stackTrace argument.

sdk/lib/async/stream.dart

 // 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;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 173 matching lines @@
 
   /**
    * Adds a subscription to this stream.
    *
    * On each data event from this stream, the subscriber's [onData] handler
    * is called. If [onData] is null, nothing happens.
    *
    * On errors from this stream, the [onError] handler is given a
    * object describing the error.
    *
+   * The [onError] callback must be of type `void onError(error)` or
+   * `void onError(error, StackTrace stackTrace)`. Depending on the function
+   * type the the stream either invokes [onError] with or without a stack
+   * trace. The stack trace argument might be `null` if the stream itself
+   * received an error without stack trace.
+   *
    * If this stream closes, the [onDone] handler is called.
    *
    * If [cancelOnError] is true, the subscription is ended when
    * the first error is reported. The default is false.
    */
   StreamSubscription<T> listen(void onData(T event),
-                               { void onError(error),
+                               { Function onError,
                                  void onDone(),
                                  bool cancelOnError});
 
   /**
    * Creates a new stream from this stream that discards some data events.
    *
    * The new stream sends the same error and done events as this stream,
    * but it only sends the data events that satisfy the [test].
    */
   Stream<T> where(bool test(T event)) {
     return new _WhereStream<T>(this, test);
   }
 
   /**
    * Creates a new stream that converts each element of this stream
    * to a new value using the [convert] function.
    */
   Stream map(convert(T event)) {
     return new _MapStream<T, dynamic>(this, convert);
   }
 
   /**
    * Creates a wrapper Stream that intercepts some errors from this stream.
    *
    * If this stream sends an error that matches [test], then it is intercepted
    * by the [handle] function.
    *
+   * The [onError] callback must be of type `void onError(error)` or
+   * `void onError(error, StackTrace stackTrace)`. Depending on the function
+   * type the the stream either invokes [onError] with or without a stack
+   * trace. The stack trace argument might be `null` if the stream itself
+   * received an error without stack trace.

[Lasse Reichstein Nielsen, 2013/10/04 08:45:17]

"without a stack trace".

What happens if onError is of none of the two types? E.g., void onError(int x,
int y) ?

How about:
"If [onError] has the latter type, accepting a second `stackTrace` argument, it
is called with two arguments, otherwise it is called with just the error
object."

[floitsch, 2013/10/05 18:11:48]

That's an error ;)
   * The [onError] callback must be of type `void onError(error)` or
   * `void onError(error, StackTrace stackTrace)`. If [onError] accepts
   * two arguments it is called with the stack trace (which could be `null` if
   * the stream itself received an error without stack trace).
   * Otherwise it is called with just the error object.
done.

[Lasse Reichstein Nielsen, 2013/10/07 11:55:48]

Ah, it's an error, but we don't say which error it is.
Perfectly fine.

From "if .. accepts two arguments", I'd assume that it gets called with two
arguments, even if the types don't match, and in unchecked mode, it'll work.
Which is better than what I asked for - no need to make it complicated for
ourselves.

If onError requires three arguments, does it then "accept two arguments"? (I'd
argue "no", and again it's not something we have to write out, it's just an
error).

+   *
    * An [AsyncError] [:e:] is matched by a test function if [:test(e):] returns
    * true. If [test] is omitted, every error is considered matching.
    *
    * If the error is intercepted, the [handle] function can decide what to do
    * with it. It can throw if it wants to raise a new (or the same) error,
    * or simply return to make the stream forget the error.
    *
    * If you need to transform an error into a data event, use the more generic
    * [Stream.transform] to handle the event by writing a data event to
    * the output sink
    */
-  Stream<T> handleError(void handle( error), { bool test(error) }) {
-    return new _HandleErrorStream<T>(this, handle, test);
+  Stream<T> handleError(Function onError, { bool test(error) }) {
+    return new _HandleErrorStream<T>(this, onError, test);
   }
 
   /**
    * Creates a new stream from this stream that converts each element
    * into zero or more events.
    *
    * Each incoming event is converted to an [Iterable] of new events,
    * and each of these new events are then sent by the returned stream
    * in order.
    */
@@ skipping 55 matching lines @@
     var value = initialValue;
     StreamSubscription subscription;
     subscription = this.listen(
       (T element) {
         _runUserCode(
           () => combine(value, element),
           (newValue) { value = newValue; },
           _cancelAndError(subscription, result)
         );
       },
-      onError: (e) {
-        result._completeError(e);
+      onError: (e, st) {
+        result._completeError(e, st);
       },
       onDone: () {
         result._complete(value);
       },
       cancelOnError: true);
     return result;
   }
 
   /**
    * Collects string of data events' string representations.
@@ skipping 552 matching lines @@
    *
    * If an event is currently firing, this unsubscription will only
    * take effect after all subscribers have received the current event.
    */
   void cancel();
 
   /** Set or override the data event handler of this subscription. */
   void onData(void handleData(T data));
 
   /** Set or override the error event handler of this subscription. */
-  void onError(void handleError(error));
+  void onError(Function handleError);
 
   /** Set or override the done event handler of this subscription. */
   void onDone(void handleDone());
 
   /**
    * Request that the stream pauses events until further notice.
    *
    * If [resumeSignal] is provided, the stream will undo the pause
    * when the future completes. If the future completes with an error,
    * it will not be handled!
@@ skipping 50 matching lines @@
 
   StreamView(this._stream);
 
   bool get isBroadcast => _stream.isBroadcast;
 
   Stream<T> asBroadcastStream({void onListen(StreamSubscription subscription),
                                void onCancel(StreamSubscription subscription)})
       => _stream.asBroadcastStream(onListen: onListen, onCancel: onCancel);
 
   StreamSubscription<T> listen(void onData(T value),
-                               { void onError(error),
+                               { Function onError,
                                  void onDone(),
                                  bool cancelOnError }) {
     return _stream.listen(onData, onError: onError, onDone: onDone,
                           cancelOnError: cancelOnError);
   }
 }
 
 
 /**
  * The target of a [Stream.pipe] call.
@@ skipping 62 matching lines @@
    *
    *     stringStream.transform(new StreamTransformer<String, String>(
    *         handleData: (String value, EventSink<String> sink) {
    *           sink.add(value);
    *           sink.add(value);  // Duplicate the incoming events.
    *         }));
    *
    */
   factory StreamTransformer({
       void handleData(S data, EventSink<T> sink),
-      void handleError(error, EventSink<T> sink),
+      Function handleError,
       void handleDone(EventSink<T> sink)}) {
     return new _StreamTransformerImpl<S, T>(handleData,
                                             handleError,
                                             handleDone);
   }
 
   Stream<T> bind(Stream<S> stream);
 }
 
 
@@ skipping 72 matching lines @@
  * events on this stream.
  */
 class EventTransformStream<S, T> extends Stream<T> {
   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(error),
+                               { Function onError,
                                  void onDone(),
                                  bool cancelOnError }) {
     if (onData == null) onData = _nullDataHandler;
     if (onError == null) onError = _nullErrorHandler;
     if (onDone == null) onDone = _nullDoneHandler;
     cancelOnError = identical(true, cancelOnError);
     return new _EventTransformStreamSubscription(_source, _transformer,
                                                  onData, onError, onDone,
                                                  cancelOnError);
   }
 }
 
 class _EventTransformStreamSubscription<S, T>
     extends _BufferingStreamSubscription<T> {
   /** The transformer used to transform events. */
   final StreamEventTransformer<S, T> _transformer;
 
   /** Whether this stream has sent a done event. */
   bool _isClosed = false;
 
   /** Source of incoming events. */
   StreamSubscription<S> _subscription;
 
   /** Cached EventSink wrapper for this class. */
   EventSink<T> _sink;
 
   _EventTransformStreamSubscription(Stream<S> source,
                                     this._transformer,
                                     void onData(T data),
-                                    void onError(error),
+                                    Function onError,
                                     void onDone(),
                                     bool cancelOnError)
       : super(onData, onError, onDone, cancelOnError) {
     _sink = new _EventSinkAdapter<T>(this);
     _subscription = source.listen(_handleData,
                                   onError: _handleError,
                                   onDone: _handleDone);
   }
 
   /** Whether this subscription is still subscribed to its source. */
@@ skipping 13 matching lines @@
       _subscription = null;
       subscription.cancel();
     }
     _isClosed = true;
   }
 
   void _handleData(S data) {
     try {
       _transformer.handleData(data, _sink);
     } catch (e, s) {
-      _addError(_asyncError(e, s));
+      _addError(_asyncError(e, s), s);
     }
   }
 
-  void _handleError(error) {
+  void _handleError(error, [stackTrace]) {
     try {
       _transformer.handleError(error, _sink);
     } catch (e, s) {
-      _addError(_asyncError(e, s));
+      if (identical(e, error)) {
+        _addError(error, stackTrace);
+      } else {
+        _addError(_asyncError(e, s), s);
+      }
     }
   }
 
   void _handleDone() {
     try {
       _subscription = null;
       _transformer.handleDone(_sink);
     } catch (e, s) {
-      _addError(_asyncError(e, s));
+      _addError(_asyncError(e, s), s);
     }
   }
 }
 
 class _EventSinkAdapter<T> implements EventSink<T> {
   _EventSink _sink;
   _EventSinkAdapter(this._sink);
 
   void add(T data) { _sink._add(data); }
-  void addError(error) { _sink._addError(error); }
+  void addError(error, [StackTrace stackTrace]) {
+    _sink._addError(error, stackTrace);
+  }
   void close() { _sink._close(); }
 }
 
 
 /**
  * An [Iterable] like interface for the values of a [Stream].
  *
  * This wraps a [Stream] and a subscription on the stream. It listens
  * on the stream, and completes the future returned by [moveNext] when the
  * next value becomes available.
@@ skipping 35 matching lines @@
    *
    * If a [moveNext] call has been made, it will complete with `false` as value,
    * as will all further calls to [moveNext].
    *
    * If you need to stop listening for values before the stream iterator is
    * automatically closed, you must call [cancel] to ensure that the stream
    * is properly closed.
    */
   void cancel();
 }