Adapt streams for additional stackTrace argument.

sdk/lib/io/http_impl.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.io;
 
 class _HttpIncoming extends Stream<List<int>> {
   final int _transferLength;
   final Completer _dataCompleter = new Completer();
   Stream<List<int>> _stream;
@@ skipping 19 matching lines @@
   // the length of the massage body is not known due to transfer
   // codings.
   int get transferLength => _transferLength;
 
   _HttpIncoming(_HttpHeaders this.headers,
                 int this._transferLength,
                 Stream<List<int>> this._stream) {
   }
 
   StreamSubscription<List<int>> listen(void onData(List<int> event),
-                                       {void onError(error),
+                                       {Function onError,
                                         void onDone(),
                                         bool cancelOnError}) {
     hasSubscriber = true;
     return _stream
         .handleError((error) {
           throw new HttpException(error.message, uri: uri);
         })
         .listen(onData,
                 onError: onError,
                 onDone: onDone,
@@ skipping 53 matching lines @@
         _session = _httpServer._sessionManager.getSession(sessionId);
         if (_session != null) {
           _session._markSeen();
           break;
         }
       }
     }
   }
 
   StreamSubscription<List<int>> listen(void onData(List<int> event),
-                                       {void onError(error),
+                                       {Function onError,
                                         void onDone(),
                                         bool cancelOnError}) {
     return _incoming.listen(onData,
                             onError: onError,
                             onDone: onDone,
                             cancelOnError: cancelOnError);
   }
 
   Uri get uri => _incoming.uri;
 
@@ skipping 106 matching lines @@
         .then((request) {
           request._responseRedirects.addAll(this.redirects);
           request._responseRedirects.add(new _RedirectInfo(statusCode,
                                                            method,
                                                            url));
           return request.close();
         });
   }
 
   StreamSubscription<List<int>> listen(void onData(List<int> event),
-                                       {void onError(error),
+                                       {Function onError,
                                         void onDone(),
                                         bool cancelOnError}) {
     var stream = _incoming;
     if (headers.value(HttpHeaders.CONTENT_ENCODING) == "gzip") {
       stream = stream.transform(GZIP.decoder);
     }
     return stream.listen(onData,
                          onError: onError,
                          onDone: onDone,
                          cancelOnError: cancelOnError);
@@ skipping 327 matching lines @@
                                        onPause: () => _subscription.pause(),
                                        onResume: () => _subscription.resume(),
                                        onListen: () => _subscription.resume(),
                                        onCancel: _cancel);
     _outbound._addStream(_controller.stream)
         .then((_) {
                 _cancel();
                 _done();
                 _closeCompleter.complete(_outbound);
               },
-              onError: (error) {
+              onError: (error, [StackTrace stackTrace]) {
                 _socketError = true;
                 if (_ignoreError(error)) {
                   _cancel();
                   _done();
                   _closeCompleter.complete(_outbound);
                 } else {
                   if (!_done(error)) {
-                    _closeCompleter.completeError(error);
+                    _closeCompleter.completeError(error, stackTrace);
                   }
                 }
               });
   }
 
-  bool _done([error]) {
+  bool _done([error, StackTrace stackTrace]) {
     if (_completer == null) return false;
     if (error != null) {
-      _completer.completeError(error);
+      _completer.completeError(error, stackTrace);
     } else {
       _completer.complete(_outbound);
     }
     _completer = null;
     return true;
   }
 
   Future addStream(var stream) {
     // If we saw a socket error subscribe and then cancel, to ignore any data
     // on the stream.
     if (_socketError) {
       stream.listen(null).cancel();
       return new Future.value(_outbound);
     }
     _completer = new Completer();
     _subscription = stream.listen(
         (data) {
           _controller.add(data);
         },
-        onDone: () {
-          _done();
-        },
-        onError: (error) {
-          _done(error);
-        },
+        onDone: _done,
+        onError: _done,
         cancelOnError: true);
     // Pause the first request.
     if (_controller == null) _subscription.pause();
     _ensureController();
     return _completer.future;
   }
 
   Future close() {
     Future closeOutbound() {
       if (_socketError) return new Future.value(_outbound);
@@ skipping 300 matching lines @@
       }
     } else if (response._shouldAuthenticateProxy) {
       future = response._authenticate(true);
     } else if (response._shouldAuthenticate) {
       future = response._authenticate(false);
     } else {
       future = new Future<HttpClientResponse>.value(response);
     }
     future.then(
         (v) => _responseCompleter.complete(v),
-        onError: (e) {
-          _responseCompleter.completeError(e);
-        });
+        onError: _responseCompleter.completeError);
   }
 
-  void _onError(error) {
-    _responseCompleter.completeError(error);
+  void _onError(error, StackTrace stackTrace) {
+    _responseCompleter.completeError(error, stackTrace);
   }
 
   // Generate the request URI based on the method and proxy.
   String _requestUri() {
     // Generate the request URI starting from the path component.
     String uriStartingFromPath() {
       String result = uri.path;
       if (result.length == 0) result = "/";
       if (uri.query != "") {
         if (uri.fragment != "") {
@@ skipping 204 matching lines @@
           // Only handle one incoming response at the time. Keep the
           // stream paused until the response have been processed.
           _subscription.pause();
           // We assume the response is not here, until we have send the request.
           if (_nextResponseCompleter == null) {
             throw new HttpException("Unexpected response.", uri: _currentUri);
           }
           _nextResponseCompleter.complete(incoming);
           _nextResponseCompleter = null;
         },
-        onError: (error) {
+        onError: (error, [StackTrace stackTrace]) {
           if (_nextResponseCompleter != null) {
             _nextResponseCompleter.completeError(
-                new HttpException(error.message, uri: _currentUri));
+                new HttpException(error.message, uri: _currentUri),
+                stackTrace);
             _nextResponseCompleter = null;
           }
         },
         onDone: () {
           if (_nextResponseCompleter != null) {
             _nextResponseCompleter.completeError(new HttpException(
                 "Connection closed before response was received",
                 uri: _currentUri));
             _nextResponseCompleter = null;
           }
@@ skipping 101 matching lines @@
                   }
                 }
                 request._onIncoming(incoming);
               })
               // If we see a state error, we failed to get the 'first'
               // element.
               .catchError((error) {
                 throw new HttpException(
                     "Connection closed before data was received", uri: uri);
               }, test: (error) => error is StateError)
-              .catchError((error) {
+              .catchError((error, stackTrace) {
                 // We are done with the socket.
                 destroy();
-                request._onError(error);
+                request._onError(error, stackTrace);
               });
 
           // Resume the parser now we have a handler.
           _subscription.resume();
           return s;
         }, onError: (e) {
           destroy();
         });
     return request;
   }
@@ skipping 641 matching lines @@
                                                     onCancel: close);
   }
 
   _HttpServer.listenOn(ServerSocket this._serverSocket)
       : _closeServer = false {
     _controller = new StreamController<HttpRequest>(sync: true,
                                                     onCancel: close);
   }
 
   StreamSubscription<HttpRequest> listen(void onData(HttpRequest event),
-                                         {void onError(error),
+                                         {Function onError,
                                          void onDone(),
                                          bool cancelOnError}) {
     _serverSocket.listen(
         (Socket socket) {
           socket.setOption(SocketOption.TCP_NODELAY, true);
           // Accept the client connection.
           _HttpConnection connection = new _HttpConnection(socket, this);
           _connections.add(connection);
         },
         onError: _controller.addError,
@@ skipping 190 matching lines @@
 }
 
 
 class _DetachedSocket extends Stream<List<int>> implements Socket {
   final Stream<List<int>> _incoming;
   final Socket _socket;
 
   _DetachedSocket(this._socket, this._incoming);
 
   StreamSubscription<List<int>> listen(void onData(List<int> event),
-                                       {void onError(error),
+                                       {Function onError,
                                         void onDone(),
                                         bool cancelOnError}) {
     return _incoming.listen(onData,
                             onError: onError,
                             onDone: onDone,
                             cancelOnError: cancelOnError);
   }
 
   Encoding get encoding => _socket.encoding;
 
@@ skipping 280 matching lines @@
   final Uri location;
 }
 
 String _getHttpVersion() {
   var version = Platform.version;
   // Only include major and minor version numbers.
   int index = version.indexOf('.', version.indexOf('.') + 1);
   version = version.substring(0, index);
   return 'Dart/$version (dart:io)';
 }