Redo StreamTransformers so they work with Stack traces.

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 495 matching lines @@
   }
 
   Future _addStream(Stream<List<int>> stream) {
     return _writeHeaders()
         .then((_) {
           int contentLength = headers.contentLength;
           if (_ignoreBody) {
             stream.drain().catchError((_) {});
             return _headersSink.close();
           }
-          stream = stream.transform(new _BufferTransformer());
+          stream = stream.transform(const _BufferTransformer());
           if (headers.chunkedTransferEncoding) {
             if (_asGZip) {
               stream = stream.transform(GZIP.encoder);
             }
-            stream = stream.transform(new _ChunkedTransformer());
+            stream = stream.transform(const _ChunkedTransformer());
           } else if (contentLength >= 0) {
             stream = stream.transform(
                 new _ContentLengthValidator(contentLength, _uri));
           }
           return _headersSink.addStream(stream);
         });
   }
 
   Future _close() {
     // TODO(ajohnsen): Currently, contentLength, chunkedTransferEncoding and
@@ skipping 107 matching lines @@
           .catchError((_) {}, test: _ignoreError)
           .then((_) => _outbound);
     }
     if (_controller == null) return closeOutbound();
     _controller.close();
     return _closeCompleter.future.then((_) => closeOutbound());
   }
 }
 
 
-class _BufferTransformer extends StreamEventTransformer<List<int>, List<int>> {
+class _BufferTransformerSink implements EventSink<List<int>> {
   static const int MIN_CHUNK_SIZE = 4 * 1024;
   static const int MAX_BUFFER_SIZE = 16 * 1024;
 
   final BytesBuilder _builder = new BytesBuilder();
+  final EventSink<List<int>> _outSink;
 
-  void handleData(List<int> data, EventSink<List<int>> sink) {
+  _BufferTransformerSink(this._outSink);
+
+  void add(List<int> data) {
     // TODO(ajohnsen): Use timeout?
     if (data.length == 0) return;
     if (data.length >= MIN_CHUNK_SIZE) {
-      flush(sink);
-      sink.add(data);
+      flush();
+      _outSink.add(data);
     } else {
       _builder.add(data);
       if (_builder.length >= MAX_BUFFER_SIZE) {
-        flush(sink);
+        flush();
       }
     }
   }
 
-  void handleDone(EventSink<List<int>> sink) {
-    flush(sink);
-    sink.close();
+  void addError(Object error, [StackTrace stackTrace]) {
+    _outSink.addError(error, stackTrace);
   }
 
-  void flush(EventSink<List<int>> sink) {
+  void close() {
+    flush();
+    _outSink.close();
+  }
+
+  void flush() {
     if (_builder.length > 0) {
       // takeBytes will clear the BytesBuilder.
-      sink.add(_builder.takeBytes());
+      _outSink.add(_builder.takeBytes());
     }
   }
 }
 
+class _BufferTransformer implements StreamTransformer<List<int>, List<int>> {
+  const _BufferTransformer();
+
+  Stream<List<int>> bind(Stream<List<int>> stream) {
+    return new Stream<List<int>>.eventTransformed(
+        stream,
+        (EventSink outSink) => new _BufferTransformerSink(outSink));
+  }
+}
+
 
 class _HttpResponse extends _HttpOutboundMessage<HttpResponse>
     implements HttpResponse {
   int statusCode = 200;
   String _reasonPhrase;
   List<Cookie> _cookies;
   _HttpRequest _httpRequest;
   Duration _deadline;
   Timer _deadlineTimer;
 
@@ skipping 326 matching lines @@
     headers._finalize();
 
     // Write headers.
     headers._write(builder);
     writeCRLF();
     _headersSink.add(builder.takeBytes());
   }
 }
 
 
-// Transformer that transforms data to HTTP Chunked Encoding.
-class _ChunkedTransformer extends StreamEventTransformer<List<int>, List<int>> {
+class _ChunkedTransformerSink implements EventSink<List<int>> {
+
   int _pendingFooter = 0;
+  final EventSink<List<int>> _outSink;
 
-  void handleData(List<int> data, EventSink<List<int>> sink) {
-    sink.add(_chunkHeader(data.length));
-    if (data.length > 0) sink.add(data);
+  _ChunkedTransformerSink(this._outSink);
+
+  void add(List<int> data) {
+    _outSink.add(_chunkHeader(data.length));
+    if (data.length > 0) _outSink.add(data);
     _pendingFooter = 2;
   }
 
-  void handleDone(EventSink<List<int>> sink) {
-    handleData(const [], sink);
-    sink.close();
+  void addError(Object error, [StackTrace stackTrace]) {
+    _outSink.addError(error, stackTrace);
+  }
+
+  void close() {
+    add(const []);
+    _outSink.close();
   }
 
   List<int> _chunkHeader(int length) {
     const hexDigits = const [0x30, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37,
                              0x38, 0x39, 0x41, 0x42, 0x43, 0x44, 0x45, 0x46];
     if (length == 0) {
       if (_pendingFooter == 2) return _footerAndChunk0Length;
       return _chunk0Length;
     }
     int size = _pendingFooter;
@@ skipping 19 matching lines @@
   }
 
   static List<int> get _footerAndChunk0Length => new Uint8List.fromList(
       const [_CharCode.CR, _CharCode.LF, 0x30, _CharCode.CR, _CharCode.LF,
              _CharCode.CR, _CharCode.LF]);
 
   static List<int> get _chunk0Length => new Uint8List.fromList(
       const [0x30, _CharCode.CR, _CharCode.LF, _CharCode.CR, _CharCode.LF]);
 }
 
+// Transformer that transforms data to HTTP Chunked Encoding.
+class _ChunkedTransformer implements StreamTransformer<List<int>, List<int>> {
+  const _ChunkedTransformer();
+
+  Stream<List<int>> bind(Stream<List<int>> stream) {
+    return new Stream<List<int>>.eventTransformed(
+        stream,
+        (EventSink<List<int>> sink) => new _ChunkedTransformerSink(sink));
+  }
+}
 
 // Transformer that validates the content length.
 class _ContentLengthValidator
-    extends StreamEventTransformer<List<int>, List<int>> {
+    implements StreamTransformer<List<int>, List<int>>, EventSink<List<int>> {
   final int expectedContentLength;
   final Uri uri;
   int _bytesWritten = 0;
 
+  EventSink<List<int>> _outSink;
+
   _ContentLengthValidator(int this.expectedContentLength, Uri this.uri);
 
-  void handleData(List<int> data, EventSink<List<int>> sink) {
+  Stream<List<int>> bind(Stream<List<int>> stream) {
+    return new Stream.eventTransformed(
+        stream,
+        (EventSink sink) {
+          if (_outSink != null) {
+            throw new StateError("Validator transformer already used");
+          }
+          _outSink = sink;
+          return this;
+        });
+  }
+
+  void add(List<int> data) {
     _bytesWritten += data.length;
     if (_bytesWritten > expectedContentLength) {
-      sink.addError(new HttpException(
+      _outSink.addError(new HttpException(
           "Content size exceeds specified contentLength. "
           "$_bytesWritten bytes written while expected "
           "$expectedContentLength. "
           "[${new String.fromCharCodes(data)}]",
           uri: uri));
-      sink.close();
+      _outSink.close();
     } else {
-      sink.add(data);
+      _outSink.add(data);
     }
   }
 
-  void handleDone(EventSink<List<int>> sink) {
+  void addError(Object error, [StackTrace stackTrace]) {
+    _outSink.addError(error, stackTrace);
+  }
+
+  void close() {
     if (_bytesWritten < expectedContentLength) {
-      sink.addError(new HttpException(
+      _outSink.addError(new HttpException(
           "Content size below specified contentLength. "
           " $_bytesWritten bytes written while expected "
           "$expectedContentLength.",
           uri: uri));
     }
-    sink.close();
+    _outSink.close();
   }
 }
 
 
 // Extends StreamConsumer as this is an internal type, only used to pipe to.
 class _HttpOutgoing implements StreamConsumer<List<int>> {
   final Completer _doneCompleter = new Completer();
   final Socket socket;
 
   _HttpOutgoing(Socket this.socket);
@@ skipping 1318 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)';
 }