Buffer the entire http header to one packet, and buffer other data in chunks of 4-16 kb.

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 431 matching lines @@
     _ioSink = new IOSink(new _HttpOutboundConsumer(_ioSink, _consume, asGZip));
     _ioSink.encoding = encoding;
   }
 
   Future _consume(IOSink ioSink, Stream<List<int>> stream, bool asGZip) {
     int contentLength = headers.contentLength;
     if (_ignoreBody) {
       ioSink.close();
       return stream.reduce(null, (x, y) {}).then((_) => this);
     }
+    stream = stream.transform(new _BufferTransformer());
     if (headers.chunkedTransferEncoding) {
       if (asGZip) {
         stream = stream.transform(new ZLibDeflater(gzip: true, level: 6));
       }
       stream = stream.transform(new _ChunkedTransformer());
     } else if (contentLength >= 0) {
       stream = stream.transform(new _ContentLengthValidator(contentLength));
     }
     return stream.pipe(ioSink).then((_) => this);
   }
 
   void _writeHeader();  // TODO(ajohnsen): Better name.
 }
 
 
 class _HttpOutboundConsumer implements StreamConsumer {
   Function _consume;
   IOSink _ioSink;
   bool _asGZip;
   _HttpOutboundConsumer(IOSink this._ioSink,
                         Function this._consume,
                         bool this._asGZip);
 
   Future consume(var stream) => _consume(_ioSink, stream, _asGZip);
 }
 
+class _BufferTransformer extends StreamEventTransformer<List<int>, List<int>> {
+  const int MIN_CHUNK_SIZE = 4 * 1024;
+  const int MAX_BUFFER_SIZE = 16 * 1024;
+
+  final _BufferList _buffer = new _BufferList();
+
+  void handleData(List<int> data, EventSink<List<int>> sink) {
+    // TODO(ajohnsen): Use timeout?
+    if (data.length == 0) return;
+    if (data.length >= MIN_CHUNK_SIZE) {
+      flush(sink);
+      sink.add(data);
+    } else {
+      _buffer.add(data);
+      if (_buffer.length >= MAX_BUFFER_SIZE) {
+        flush(sink);
+      }
+    }
+  }
+
+  void handleDone(EventSink<List<int>> sink) {
+    flush(sink);
+    sink.close();
+  }
+
+  void flush(EventSink<List<int>> sink) {
+    if (_buffer.length > 0) {
+      sink.add(_buffer.readBytes());
+      _buffer.clear();
+    }
+  }
+}
+
 
 class _HttpResponse extends _HttpOutboundMessage<HttpResponse>
     implements HttpResponse {
   int statusCode = 200;
   String _reasonPhrase;
   List<Cookie> _cookies;
   _HttpRequest _httpRequest;
 
   _HttpResponse(String protocolVersion,
                 _HttpOutgoing _outgoing)
@@ skipping 19 matching lines @@
     done.catchError((_) {
       // Catch any error on done, as they automatically will be propegated to
       // the websocket.
     });
     return future;
   }
 
   HttpConnectionInfo get connectionInfo => _httpRequest.connectionInfo;
 
   void _writeHeader() {
-    writeSP() => _ioSink.writeBytes([_CharCode.SP]);
-    writeCRLF() => _ioSink.writeBytes([_CharCode.CR, _CharCode.LF]);
+    var buffer = new _BufferList();
+    writeSP() => buffer.add(const [_CharCode.SP]);
+    writeCRLF() => buffer.add(const [_CharCode.CR, _CharCode.LF]);
 
     // Write status line.
     if (headers.protocolVersion == "1.1") {
-      _ioSink.writeBytes(_Const.HTTP11);
+      buffer.add(_Const.HTTP11);
     } else {
-      _ioSink.writeBytes(_Const.HTTP10);
+      buffer.add(_Const.HTTP10);
     }
     writeSP();
-    _ioSink.write(statusCode.toString());
+    buffer.add(statusCode.toString().codeUnits);
     writeSP();
-    _ioSink.write(reasonPhrase);
+    buffer.add(reasonPhrase.codeUnits);
     writeCRLF();
 
     var session = _httpRequest._session;
     if (session != null && !session._destroyed) {
       // Mark as not new.
       session._isNew = false;
       // Make sure we only send the current session id.
       bool found = false;
       for (int i = 0; i < cookies.length; i++) {
         if (cookies[i].name.toUpperCase() == _DART_SESSION_ID) {
@@ skipping 13 matching lines @@
     // Add all the cookies set to the headers.
     if (_cookies != null) {
       _cookies.forEach((cookie) {
         headers.add(HttpHeaders.SET_COOKIE, cookie);
       });
     }
 
     headers._finalize();
 
     // Write headers.
-    headers._write(_ioSink);
+    headers._write(buffer);
     writeCRLF();
+    _ioSink.writeBytes(buffer.readBytes());
   }
 
   String _findReasonPhrase(int statusCode) {
     if (_reasonPhrase != null) {
       return _reasonPhrase;
     }
 
     switch (statusCode) {
       case HttpStatus.CONTINUE: return "Continue";
       case HttpStatus.SWITCHING_PROTOCOLS: return "Switching Protocols";
@@ skipping 128 matching lines @@
         onError: (e) {
           _responseCompleter.completeError(e);
         });
   }
 
   void _onError(AsyncError error) {
     _responseCompleter.completeError(error);
   }
 
   void _writeHeader() {
-    writeSP() => _ioSink.writeBytes([_CharCode.SP]);
-    writeCRLF() => _ioSink.writeBytes([_CharCode.CR, _CharCode.LF]);
+    var buffer = new _BufferList();

[Søren Gjesse, 2013/03/12 11:30:55]

I am not sure that using a _BufferList for the header this way is a good idea,
but it is easy to change later.

E.g. adding CR LF is now adding a full Queue entry in the _BufferList for two
bytes of output. One option for the header is to calculate its size and then
create and fill a UInt8List.

[Anders Johnsen, 2013/03/12 12:11:30]

Yes, we have to look into this once we really dive into the performance story.

+    writeSP() => buffer.add(const [_CharCode.SP]);
+    writeCRLF() => buffer.add(const [_CharCode.CR, _CharCode.LF]);
 
-    _ioSink.write(method);
+    buffer.add(method.codeUnits);
     writeSP();
     // Send the path for direct connections and the whole URL for
     // proxy connections.
     if (!_usingProxy) {
       String path = uri.path;
       if (path.length == 0) path = "/";
       if (uri.query != "") {
         if (uri.fragment != "") {
           path = "${path}?${uri.query}#${uri.fragment}";
         } else {
           path = "${path}?${uri.query}";
         }
       }
-      _ioSink.write(path);
+      buffer.add(path.codeUnits);
     } else {
-      _ioSink.write(uri.toString());
+      buffer.add(uri.toString().codeUnits);
     }
     writeSP();
-    _ioSink.writeBytes(_Const.HTTP11);
+    buffer.add(_Const.HTTP11);
     writeCRLF();
 
     // Add the cookies to the headers.
     if (!cookies.isEmpty) {
       StringBuffer sb = new StringBuffer();
       for (int i = 0; i < cookies.length; i++) {
         if (i > 0) sb.write("; ");
         sb.write(cookies[i].name);
         sb.write("=");
         sb.write(cookies[i].value);
       }
       headers.add(HttpHeaders.COOKIE, sb.toString());
     }
 
     headers._finalize();
 
     // Write headers.
-    headers._write(_ioSink);
+    headers._write(buffer);
     writeCRLF();
+    _ioSink.writeBytes(buffer.readBytes());
   }
 }
 
 
 // Transformer that transforms data to HTTP Chunked Encoding.
 class _ChunkedTransformer extends StreamEventTransformer<List<int>, List<int>> {
   void handleData(List<int> data, EventSink<List<int>> sink) {
     sink.add(_chunkHeader(data.length));
     if (data.length > 0) sink.add(data);
     sink.add(_chunkFooter);
@@ skipping 905 matching lines @@
 
 
 class _RedirectInfo implements RedirectInfo {
   const _RedirectInfo(int this.statusCode,
                       String this.method,
                       Uri this.location);
   final int statusCode;
   final String method;
   final Uri location;
 }