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.
    *
+   * `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.
+   *
    * 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.
+   *
    * 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();
 }