Support auto-drain of HttpRequest data.

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;
 
   bool fullBodyRead = false;
 
   // Common properties.
   final _HttpHeaders headers;
   bool upgraded = false;
 
   // ClientResponse properties.
   int statusCode;
   String reasonPhrase;
 
   // Request properties.
   String method;
   Uri uri;
 
+  bool hasSubscriber = false;
+
   // The transfer length if the length of the message body as it
   // appears in the message (RFC 2616 section 4.4). This can be -1 if
   // 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),
                                         void onDone(),
                                         bool cancelOnError}) {
+    hasSubscriber = true;
     return _stream.listen(onData,
                           onError: onError,
                           onDone: onDone,
                           cancelOnError: cancelOnError);
   }
 
   // Is completed once all data have been received.
   Future get dataDone => _dataCompleter.future;
 
   void close(bool closing) {
     fullBodyRead = true;
+    hasSubscriber = true;
     _dataCompleter.complete(closing);
   }
 }
 
 abstract class _HttpInboundMessage extends Stream<List<int>> {
   final _HttpIncoming _incoming;
   List<Cookie> _cookies;
 
   _HttpInboundMessage(_HttpIncoming this._incoming);
 
@@ skipping 205 matching lines @@
   bool get _shouldAuthenticate {
     // Only try to authenticate if there is a challenge in the response.
     List<String> challenge = headers[HttpHeaders.WWW_AUTHENTICATE];
     return statusCode == HttpStatus.UNAUTHORIZED &&
         challenge != null && challenge.length == 1;
   }
 
   Future<HttpClientResponse> _authenticate(bool proxyAuth) {
     Future<HttpClientResponse> retry() {
       // Drain body and retry.
-      return fold(null, (x, y) {}).then((_) {
+      return drain().then((_) {
           return _httpClient._openUrlFromRequest(_httpRequest.method,
                                                  _httpRequest.uri,
                                                  _httpRequest)
               .then((request) => request.close());
           });
     }
 
     List<String> authChallenge() {
       if (proxyAuth) {
         return headers[HttpHeaders.PROXY_AUTHENTICATE];
@@ skipping 175 matching lines @@
   Future<T> addStream(Stream<List<int>> stream) {
     return _dataSink.addStream(stream);
   }
 
   Future close() {
     return _dataSink.close();
   }
 
   Future<T> get done => _dataSink.done;
 
-  void _writeHeaders() {
-    if (_headersWritten) return;
+  Future _writeHeaders({drainRequest: true}) {
+    if (_headersWritten) return new Future.value();
     _headersWritten = true;
     headers._synchronize();  // Be sure the 'chunked' option is updated.
     bool isServerSide = this is _HttpResponse;
-    if (isServerSide && headers.chunkedTransferEncoding) {
+    if (isServerSide) {
       var response = this;
-      List acceptEncodings =
-          response._httpRequest.headers[HttpHeaders.ACCEPT_ENCODING];
-      List contentEncoding = headers[HttpHeaders.CONTENT_ENCODING];
-      if (acceptEncodings != null &&
-          acceptEncodings
-              .expand((list) => list.split(","))
-              .any((encoding) => encoding.trim().toLowerCase() == "gzip") &&
-          contentEncoding == null) {
-        headers.set(HttpHeaders.CONTENT_ENCODING, "gzip");
-        _asGZip = true;
+      if (headers.chunkedTransferEncoding) {
+        List acceptEncodings =
+            response._httpRequest.headers[HttpHeaders.ACCEPT_ENCODING];
+        List contentEncoding = headers[HttpHeaders.CONTENT_ENCODING];
+        if (acceptEncodings != null &&
+            acceptEncodings
+                .expand((list) => list.split(","))
+                .any((encoding) => encoding.trim().toLowerCase() == "gzip") &&
+            contentEncoding == null) {
+          headers.set(HttpHeaders.CONTENT_ENCODING, "gzip");
+          _asGZip = true;
+        }
+      }
+      if (drainRequest && !response._httpRequest._incoming.hasSubscriber) {
+        return response._httpRequest.drain()
+            // TODO(ajohnsen): Timeout on drain?
+            .catchError((_) {})  // Ignore errors.
+            .then((_) => _writeHeader());
       }
     }
-    _writeHeader();
+    return new Future.sync(_writeHeader);
   }
 
   Future _addStream(Stream<List<int>> stream) {
-    _writeHeaders();
-    int contentLength = headers.contentLength;
-    if (_ignoreBody) {
-      stream.fold(null, (x, y) {}).catchError((_) {});
-      return _headersSink.close();
-    }
-    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 _headersSink.addStream(stream);
+    return _writeHeaders()
+        .then((_) {
+          int contentLength = headers.contentLength;
+          if (_ignoreBody) {
+            stream.drain().catchError((_) {});
+            return _headersSink.close();
+          }
+          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 _headersSink.addStream(stream);
+        });
   }
 
   Future _close() {
     // TODO(ajohnsen): Currently, contentLength, chunkedTransferEncoding and
     // persistentConnection is not guaranteed to be in sync.
     if (!_headersWritten) {
       if (!_ignoreBody && 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.
         headers.chunkedTransferEncoding = false;
         headers.contentLength = 0;
       } else if (!_ignoreBody && headers.contentLength > 0) {
         _headersSink.close().catchError((_) {});
         return new Future.error(new HttpException(
             "No content while contentLength was specified to be greater "
             " than 0: ${headers.contentLength}."));
       }
     }
-    _writeHeaders();
-    return _headersSink.close();
+    return _writeHeaders().then((_) => _headersSink.close());
   }
 
   void _writeHeader();  // TODO(ajohnsen): Better name.
 }
 
 
 class _HttpOutboundConsumer implements StreamConsumer {
   final _HttpOutboundMessage _outbound;
   StreamController _controller;
   StreamSubscription _subscription;
   Completer _closeCompleter = new Completer();
   Completer _completer;
   bool _socketError = false;
 
   _HttpOutboundConsumer(_HttpOutboundMessage this._outbound);
 
   void _cancel() {
     if (_subscription != null) {
       _subscription.cancel();
     }
   }
 
   _ensureController() {
     if (_controller != null) return;
     _controller = new StreamController(onPause: () => _subscription.pause(),
                                        onResume: () => _subscription.resume(),
+                                       onListen: () => _subscription.resume(),
                                        onCancel: _cancel);
     _outbound._addStream(_controller.stream)
         .then((_) {
                 _cancel();
                 _done();
                 _closeCompleter.complete(_outbound);
               },
               onError: (error) {
                 _socketError = true;
                 if (error is SocketIOException &&
@@ skipping 32 matching lines @@
         (data) {
           _controller.add(data);
         },
         onDone: () {
           _done();
         },
         onError: (error) {
           _done(error);
         },
         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);
       return _outbound._close().then((_) => _outbound);
     }
     if (_controller == null) return closeOutbound();
@@ skipping 54 matching lines @@
   }
 
   String get reasonPhrase => _findReasonPhrase(statusCode);
   void set reasonPhrase(String reasonPhrase) {
     if (_headersWritten) throw new StateError("Header already sent");
     _reasonPhrase = reasonPhrase;
   }
 
   Future<Socket> detachSocket() {
     if (_headersWritten) throw new StateError("Headers already sent");
-    _writeHeaders();
     var future = _httpRequest._httpConnection.detachSocket();
+    _writeHeaders(drainRequest: false).then((_) => close());
     // Close connection so the socket is 'free'.
     close();
     done.catchError((_) {
       // Catch any error on done, as they automatically will be
       // propagated to the websocket.
     });
     return future;
   }
 
   HttpConnectionInfo get connectionInfo => _httpRequest.connectionInfo;
@@ skipping 174 matching lines @@
   HttpConnectionInfo get connectionInfo => _httpClientConnection.connectionInfo;
 
   void _onIncoming(_HttpIncoming incoming) {
     var response = new _HttpClientResponse(incoming,
                                            this,
                                           _httpClient);
     Future<HttpClientResponse> future;
     if (followRedirects && response.isRedirect) {
       if (response.redirects.length < maxRedirects) {
         // Redirect and drain response.
-        future = response.fold(null, (x, y) {})
+        future = response.drain()
           .then((_) => response.redirect());
       } else {
         // End with exception, too many redirects.
-        future = response.fold(null, (x, y) {})
+        future = response.drain()
             .then((_) => new Future.error(
                 new RedirectLimitExceededException(response.redirects)));
       }
     } else if (response._shouldAuthenticateProxy) {
       future = response._authenticate(true);
     } else if (response._shouldAuthenticate) {
       future = response._authenticate(false);
     } else {
       future = new Future<HttpClientResponse>.value(response);
     }
@@ skipping 1423 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;
 }