Add bufferOutput to HttpServer, HttpResponse and HttpClientRequest.

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;
 
 const int _OUTGOING_BUFFER_SIZE = 8 * 1024;
 
 class _HttpIncoming extends Stream<List<int>> {
   final int _transferLength;
@@ skipping 393 matching lines @@
     });
   }
 }
 
 
 abstract class _HttpOutboundMessage<T> extends _IOSinkImpl {
   // Used to mark when the body should be written. This is used for HEAD
   // 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;
 
   final _HttpHeaders headers;
 
   _HttpOutboundMessage(Uri uri,
                        String protocolVersion,
-                       _HttpOutgoing outgoing)
+                       _HttpOutgoing outgoing,
+                       this.bufferOutput)
       : super(outgoing, null),
         _uri = uri,
         headers = new _HttpHeaders(
             protocolVersion,
             defaultPortForScheme: uri.scheme == 'https' ?
                 HttpClient.DEFAULT_HTTPS_PORT :
                 HttpClient.DEFAULT_HTTP_PORT),
         _outgoing = outgoing {
     _outgoing.outbound = this;
     _encodingMutable = false;
@@ skipping 29 matching lines @@
 
   void write(Object obj) {
     if (!_encodingSet) {
       _encoding = encoding;
       _encodingSet = true;
     }
     super.write(obj);
   }
 
   void _writeHeader();
+
+  bool get _isConnectionClosed => false;
 }
 
 
 class _HttpResponse extends _HttpOutboundMessage<HttpResponse>
     implements HttpResponse {
   int _statusCode = 200;
   String _reasonPhrase;
   List<Cookie> _cookies;
   _HttpRequest _httpRequest;
   Duration _deadline;
   Timer _deadlineTimer;
 
   _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;
   }
 
   int get statusCode => _statusCode;
   void set statusCode(int statusCode) {
     if (_outgoing.headersWritten) throw new StateError("Header already sent");
     _statusCode = statusCode;
   }
@@ skipping 30 matching lines @@
   HttpConnectionInfo get connectionInfo => _httpRequest.connectionInfo;
 
   Duration get deadline => _deadline;
 
   void set deadline(Duration d) {
     if (_deadlineTimer != null) _deadlineTimer.cancel();
     _deadline = d;
 
     if (_deadline == null) return;
     _deadlineTimer = new Timer(_deadline, () {
-      _outgoing._socketError = true;
-      _outgoing.socket.destroy();
+      _httpRequest._httpConnection.destroy();
     });
   }
 
   void _writeHeader() {
     Uint8List buffer = new Uint8List(_OUTGOING_BUFFER_SIZE);
     int offset = 0;
 
     void write(List<int> bytes) {
       int len = bytes.length;
       for (int i = 0; i < len; i++) {
@@ skipping 129 matching lines @@
 
   // TODO(ajohnsen): Get default value from client?
   bool _followRedirects = true;
 
   int _maxRedirects = 5;
 
   List<RedirectInfo> _responseRedirects = [];
 
   _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") {
       contentLength = 0;
     } else {
       headers.chunkedTransferEncoding = true;
     }
   }
 
   Future<HttpClientResponse> get done {
@@ skipping 333 matching lines @@
           }
         });
   }
 
   Future close() {
     // If we are already closed, return that future.
     if (_closeFuture != null) return _closeFuture;
     // 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
         // request) and the content-length is unspecified, set contentLength to
         // 0.
         outbound.headers.chunkedTransferEncoding = false;
         outbound.headers.contentLength = 0;
       } else if (outbound.headers.contentLength > 0) {
         var error = new HttpException(
               "No content even though contentLength was specified to be "
@@ skipping 86 matching lines @@
                 _addChunk(data, _gzipAdd);
               }));
     }
   }
 
   bool _ignoreError(error)
     => (error is SocketException || error is TlsException) &&
        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));
       _gzipBuffer = new Uint8List(_OUTGOING_BUFFER_SIZE);
       _gzipBufferLength = 0;
     }
     if (chunk.length > _OUTGOING_BUFFER_SIZE) {
       add(chunk);
     } else {
       _gzipBuffer.setRange(_gzipBufferLength,
                            _gzipBufferLength + chunk.length,
                            chunk);
       _gzipBufferLength += chunk.length;
     }
   }
 
   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);
       _length = 0;
     }
     if (chunk.length > _OUTGOING_BUFFER_SIZE) {
       add(chunk);
     } else {
       _buffer.setRange(_length, _length + chunk.length, chunk);
       _length += chunk.length;
@@ skipping 717 matching lines @@
             if (closing) destroy();
           });
           // Only handle one incoming request at the time. Keep the
           // stream paused until the request has been send.
           _subscription.pause();
           _state = _ACTIVE;
           var outgoing = new _HttpOutgoing(_socket);
           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((_) {
                 response.deadline = null;
                 if (_state == _DETACHED) return;
                 if (response.persistentConnection &&
                     request.persistentConnection &&
                     incoming.fullBodyRead &&
                     !_httpParser.upgrade &&
                     !_httpServer.closed) {
@@ skipping 54 matching lines @@
   bool get _isActive => _state == _ACTIVE;
   bool get _isIdle => _state == _IDLE;
   bool get _isClosing => _state == _CLOSING;
   bool get _isDetached => _state == _DETACHED;
 }
 
 
 // HTTP server waiting for socket connections.
 class _HttpServer extends Stream<HttpRequest> implements HttpServer {
   String serverHeader;
+  bool bufferOutput = true;
 
   Duration _idleTimeout;
   Timer _idleTimer;
 
   static Future<HttpServer> bind(address, int port, int backlog) {
     return ServerSocket.bind(address, port, backlog: backlog).then((socket) {
       return new _HttpServer._(socket, true);
     });
   }
 
@@ skipping 586 matching lines @@
   const _RedirectInfo(this.statusCode, this.method, this.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)';
 }