Add bufferOutput to HttpServer, HttpResponse and HttpClientRequest.

sdk/lib/io/http_impl.dart

Index: sdk/lib/io/http_impl.dart
diff --git a/sdk/lib/io/http_impl.dart b/sdk/lib/io/http_impl.dart
index 207c08d104d4d423ffcc6ea8c197f211c02aa1cc..ba1fb08205aea7b68bded04e62b049a892e7fc1f 100644
--- a/sdk/lib/io/http_impl.dart
+++ b/sdk/lib/io/http_impl.dart
@@ -411,6 +411,8 @@ abstract class _HttpOutboundMessage<T> extends _IOSinkImpl {
   // requests and in error handling.
   bool _encodingSet = false;
 
+  bool bufferOutput;

[Søren Gjesse, 2014/04/08 11:54:25]

What happens if this is changed while the body is being written?

Should we only support it being changed before the first byte of the body is
written? If not it should flush the current buffer when turned off.

Should compression be turned off if this is set to false?

[Anders Johnsen, 2014/04/08 12:40:07]

Done. Sadly, this is needed, as zlib buffers as well.

+
   final Uri _uri;
   final _HttpOutgoing _outgoing;
 
@@ -418,7 +420,8 @@ abstract class _HttpOutboundMessage<T> extends _IOSinkImpl {
 
   _HttpOutboundMessage(Uri uri,
                        String protocolVersion,
-                       _HttpOutgoing outgoing)
+                       _HttpOutgoing outgoing,
+                       this.bufferOutput)
       : super(outgoing, null),
         _uri = uri,
         headers = new _HttpHeaders(
@@ -468,6 +471,8 @@ abstract class _HttpOutboundMessage<T> extends _IOSinkImpl {
   }
 
   void _writeHeader();
+
+  bool get _isConnectionClosed => false;
 }
 
 
@@ -483,11 +488,14 @@ class _HttpResponse extends _HttpOutboundMessage<HttpResponse>
   _HttpResponse(Uri uri,
                 String protocolVersion,
                 _HttpOutgoing outgoing,
-                String serverHeader)
-      : super(uri, protocolVersion, outgoing) {
+                String serverHeader,
+                bool bufferOutput)
+      : super(uri, protocolVersion, outgoing, bufferOutput) {
     if (serverHeader != null) headers._add('server', serverHeader);
   }
 
+  bool get _isConnectionClosed => _httpRequest._httpConnection._isClosing;
+
   List<Cookie> get cookies {
     if (_cookies == null) _cookies = new List<Cookie>();
     return _cookies;
@@ -538,8 +546,7 @@ class _HttpResponse extends _HttpOutboundMessage<HttpResponse>
 
     if (_deadline == null) return;
     _deadlineTimer = new Timer(_deadline, () {
-      _outgoing._socketError = true;
-      _outgoing.socket.destroy();
+      _httpRequest._httpConnection.destroy();
     });
   }
 
@@ -689,7 +696,7 @@ class _HttpClientRequest extends _HttpOutboundMessage<HttpClientResponse>
 
   _HttpClientRequest(_HttpOutgoing outgoing, Uri uri, this.method, this._proxy,
                      this._httpClient, this._httpClientConnection)
-      : super(uri, "1.1", outgoing),
+      : super(uri, "1.1", outgoing, true),
         uri = uri {
     // GET and HEAD have 'content-length: 0' by default.
     if (method == "GET" || method == "HEAD") {
@@ -1043,6 +1050,7 @@ class _HttpOutgoing implements StreamConsumer<List<int>> {
     // If we earlier saw an error, return immediate. The notification to
     // _Http*Connection is already done.
     if (_socketError) return new Future.value(outbound);
+    if (outbound._isConnectionClosed) return new Future.value(outbound);
     if (!headersWritten && !ignoreBody) {
       if (outbound.headers.contentLength == -1) {
         // If no body was written, ignoreBody is false (it's not a HEAD
@@ -1149,6 +1157,10 @@ class _HttpOutgoing implements StreamConsumer<List<int>> {
        outbound is HttpResponse;
 
   void _addGZipChunk(chunk, void add(List<int> data)) {
+    if (!outbound.bufferOutput) {
+      add(chunk);
+      return;
+    }
     if (chunk.length > _gzipBuffer.length - _gzipBufferLength) {
       add(new Uint8List.view(
           _gzipBuffer.buffer, 0, _gzipBufferLength));
@@ -1166,6 +1178,15 @@ class _HttpOutgoing implements StreamConsumer<List<int>> {
   }
 
   void _addChunk(chunk, void add(List<int> data)) {
+    if (!outbound.bufferOutput) {
+      if (_buffer != null) {
+        add(new Uint8List.view(_buffer.buffer, 0, _length));

[Søren Gjesse, 2014/04/08 11:54:25]

Please add a comment on why you are doing this.

[Anders Johnsen, 2014/04/08 12:40:07]

Done.

+        _buffer = null;
+        _length = 0;
+      }
+      add(chunk);
+      return;
+    }
     if (chunk.length > _buffer.length - _length) {
       add(new Uint8List.view(_buffer.buffer, 0, _length));
       _buffer = new Uint8List(_OUTGOING_BUFFER_SIZE);
@@ -1903,7 +1924,8 @@ class _HttpConnection extends LinkedListEntry<_HttpConnection> {
           var response = new _HttpResponse(incoming.uri,
                                            incoming.headers.protocolVersion,
                                            outgoing,
-                                           _httpServer.serverHeader);
+                                           _httpServer.serverHeader,
+                                           _httpServer.bufferOutput);
           var request = new _HttpRequest(response, incoming, _httpServer, this);
           _streamFuture = outgoing.done
               .then((_) {
@@ -1978,6 +2000,7 @@ class _HttpConnection extends LinkedListEntry<_HttpConnection> {
 // HTTP server waiting for socket connections.
 class _HttpServer extends Stream<HttpRequest> implements HttpServer {
   String serverHeader;
+  bool bufferOutput = true;
 
   Duration _idleTimeout;
   Timer _idleTimer;