Remove .writeStream, .consume and rewrite IOSink to correctly implement a (sane) well-defined behav…

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 308 matching lines @@
     return new Future.immediate(this);
   }
 }
 
 
 abstract class _HttpOutboundMessage<T> implements IOSink {
   // Used to mark when the body should be written. This is used for HEAD
   // requests and in error handling.
   bool _ignoreBody = false;
   bool _headersWritten = false;
+  bool _asGZip = false;
 
-  IOSink _ioSink;
+  IOSink _headersSink;
+  IOSink _dataSink;
+
   final _HttpOutgoing _outgoing;
 
   final _HttpHeaders headers;
 
   _HttpOutboundMessage(String protocolVersion, _HttpOutgoing outgoing)
       : _outgoing = outgoing,
-        _ioSink = new IOSink(outgoing, encoding: Encoding.ASCII),
-        headers = new _HttpHeaders(protocolVersion);
+        _headersSink = new IOSink(outgoing, encoding: Encoding.ASCII),
+        headers = new _HttpHeaders(protocolVersion) {
+    _dataSink = new IOSink(
+        new _HttpOutboundConsumer(_headersSink, _addStream, this));
+  }
 
   int get contentLength => headers.contentLength;
   void set contentLength(int contentLength) {
     headers.contentLength = contentLength;
   }
 
   bool get persistentConnection => headers.persistentConnection;
   void set persistentConnection(bool p) {
     headers.persistentConnection = p;
   }
@@ skipping 20 matching lines @@
     String string;
     if (obj is String) {
       string = obj;
     } else {
       string = obj.toString();
       if (string is! String) {
         throw new ArgumentError('toString() did not return a string');
       }
     }
     if (string.isEmpty) return;
-    _ioSink.write(string);
+    _dataSink.write(string);
   }
 
   void writeAll(Iterable objects, [String separator = ""]) {
     bool isFirst = true;
     for (Object obj in objects) {
       if (isFirst) {
         isFirst = false;
       } else {
         if (!separator.isEmpty) write(separator);
       }
       write(obj);
     }
   }
 
   void writeln([Object obj = ""]) {
     write(obj);
     write("\n");
   }
 
   void writeCharCode(int charCode) {
     write(new String.fromCharCode(charCode));
   }
 
   void add(List<int> data) {
     _writeHeaders();
     if (data.length == 0) return;
-    _ioSink.add(data);
+    _dataSink.add(data);
   }
 
   void addError(AsyncError error) {
     _writeHeaders();
-    _ioSink.addError(error);
-  }
-
-  Future<T> consume(Stream<List<int>> stream) {
-    _writeHeaders();
-    return _ioSink.consume(stream);
+    _dataSink.addError(error);
   }
 
   Future<T> addStream(Stream<List<int>> stream) {
     _writeHeaders();
-    return _ioSink.writeStream(stream).then((_) => this);
-  }
-
-  Future<T> writeStream(Stream<List<int>> stream) {
-    return addStream(stream);
+    return _dataSink.addStream(stream);
   }
 
   Future close() {
     // TODO(ajohnsen): Currently, contentLength, chunkedTransferEncoding and
     // persistentConnection is not guaranteed to be in sync.
     if (!_headersWritten && !_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;
     }
     _writeHeaders();
-    return _ioSink.close();
+    return _dataSink.close();
   }
 
-  Future<T> get done {
-    _writeHeaders();
-    return _ioSink.done;
-  }
+  Future<T> get done => _dataSink.done.then((_) => this);
 
   void _writeHeaders() {
     if (_headersWritten) return;
     _headersWritten = true;
-    _ioSink.encoding = Encoding.ASCII;
     headers._synchronize();  // Be sure the 'chunked' option is updated.
-    bool asGZip = false;
     bool isServerSide = this is _HttpResponse;
     if (isServerSide && headers.chunkedTransferEncoding) {
       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;
+        _asGZip = true;
       }
     }
     _writeHeader();
-    _ioSink = new IOSink(new _HttpOutboundConsumer(_ioSink, _consume, asGZip));
-    _ioSink.encoding = encoding;
   }
 
-  Future _consume(IOSink ioSink, Stream<List<int>> stream, bool asGZip) {
+  Future _addStream(IOSink ioSink, Stream<List<int>> stream) {
     int contentLength = headers.contentLength;
     if (_ignoreBody) {
-      ioSink.close();
-      return stream.fold(null, (x, y) {}).then((_) => this);
+      stream.fold(null, (x, y) {}).catchError((_) {});
+      return ioSink.close().then((_) => this);
     }
     stream = stream.transform(new _BufferTransformer());
     if (headers.chunkedTransferEncoding) {
-      if (asGZip) {
+      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);
+    return ioSink.addStream(stream).then((_) => this);
   }
 
   void _writeHeader();  // TODO(ajohnsen): Better name.
 }
 
 
 class _HttpOutboundConsumer implements StreamConsumer {
-  Function _consume;
+  StreamController _controller;
+  StreamSubscription _subscription;
+  Function _addStream;
   IOSink _ioSink;
-  bool _asGZip;
+  Completer _closeCompleter = new Completer();
+  Completer _completer;
+
+  _HttpOutboundMessage _outbound;
   _HttpOutboundConsumer(IOSink this._ioSink,
-                        Function this._consume,
-                        bool this._asGZip);
+                        Function this._addStream,
+                        _HttpOutboundMessage this._outbound);
 
-  Future consume(var stream) => _consume(_ioSink, stream, _asGZip);
+  void _onPause() {
+    if (_controller.isPaused) {
+      _subscription.pause();
+    } else {
+      _subscription.resume();
+    }
+  }
+
+  void _onListen() {
+    if (!_controller.hasListener && _subscription != null) {
+      _subscription.cancel();
+    }
+  }
+
+  _ensureController() {
+    if (_controller != null) return;
+    _controller = new StreamController(onPauseStateChange: _onPause,
+                                       onSubscriptionStateChange: _onListen);
+    _addStream(_ioSink, _controller.stream)
+        .then((v) {

[Søren Gjesse, 2013/04/15 06:56:30]

v -> _

[Anders Johnsen, 2013/04/15 07:35:20]

Done.

+                _done();
+                _closeCompleter.complete(_outbound);
+              },
+              onError: (error) {
+                if (!_done(error)) {
+                  _closeCompleter.completeError(error);
+                }
+              });
+  }
+
+  bool _done([error]) {
+    if (_completer == null) return false;
+    var tmp = _completer;
+    _completer = null;
+    if (error != null) {
+      tmp.completeError(error);
+    } else {
+      tmp.complete(_outbound);
+    }
+    return true;
+  }
 
   Future addStream(var stream) {
-    throw new UnimplementedError("_HttpOutboundConsumer.addStream");
+    _ensureController();
+    _completer = new Completer();
+    _subscription = stream.listen(
+        (data) {
+          _controller.add(data);
+        },
+        onDone: () {
+          _done();
+        },
+        onError: (error) {
+          _done(error);
+        },
+        unsubscribeOnError: true);
+    return _completer.future;
   }
 
   Future close() {
-    throw new UnimplementedError("_HttpOutboundConsumer.close");
+    _ensureController();
+    _controller.close();
+    return _closeCompleter.future.then((_) {
+      return _ioSink.close().then((_) => _outbound);
+    });
   }
 }
 
 
 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();
 
@@ skipping 11 matching lines @@
     }
   }
 
   void handleDone(EventSink<List<int>> sink) {
     flush(sink);
     sink.close();
   }
 
   void flush(EventSink<List<int>> sink) {
     if (_buffer.length > 0) {
-      sink.add(_buffer.readBytes());
+      var data = _buffer.readBytes();

[Søren Gjesse, 2013/04/15 06:56:30]

Why this change?

[Anders Johnsen, 2013/04/15 07:35:20]

Done.

+      sink.add(data);
       _buffer.clear();
     }
   }
 }
 
 
 class _HttpResponse extends _HttpOutboundMessage<HttpResponse>
     implements HttpResponse {
   int statusCode = 200;
   String _reasonPhrase;
@@ skipping 15 matching lines @@
     _reasonPhrase = reasonPhrase;
   }
 
   Future<Socket> detachSocket() {
     if (_headersWritten) throw new StateError("Headers already sent");
     _writeHeaders();
     var future = _httpRequest._httpConnection.detachSocket();
     // Close connection so the socket is 'free'.
     close();
     done.catchError((_) {
-      // Catch any error on done, as they automatically will be propegated to
-      // the websocket.
+      // Catch any error on done, as they automatically will be
+      // propegated to the websocket.

[Søren Gjesse, 2013/04/15 06:56:30]

propagated

[Anders Johnsen, 2013/04/15 07:35:20]

Done.

     });
     return future;
   }
 
   HttpConnectionInfo get connectionInfo => _httpRequest.connectionInfo;
 
   void _writeHeader() {
     var buffer = new _BufferList();
     writeSP() => buffer.add(const [_CharCode.SP]);
     writeCRLF() => buffer.add(const [_CharCode.CR, _CharCode.LF]);
@@ skipping 36 matching lines @@
       _cookies.forEach((cookie) {
         headers.add(HttpHeaders.SET_COOKIE, cookie);
       });
     }
 
     headers._finalize();
 
     // Write headers.
     headers._write(buffer);
     writeCRLF();
-    _ioSink.add(buffer.readBytes());
+    _headersSink.add(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 78 matching lines @@
                      _HttpClientConnection this._httpClientConnection)
       : super("1.1", outgoing) {
     // GET and HEAD have 'content-length: 0' by default.
     if (method == "GET" || method == "HEAD") {
       contentLength = 0;
     }
   }
 
   Future<HttpClientResponse> get done {
     if (_response == null) {
-      _response = Future.wait([_responseCompleter.future, super.done])
+      _response = Future.wait([_responseCompleter.future,
+                               super.done])
         .then((list) => list[0]);
     }
     return _response;
   }
 
-  Future<HttpClientResponse> consume(Stream<List<int>> stream) {
-    super.consume(stream);
-    return done;
-  }
-
   Future<HttpClientResponse> close() {
     super.close();
     return done;
   }
 
   int get maxRedirects => _maxRedirects;
   void set maxRedirects(int maxRedirects) {
     if (_headersWritten) throw new StateError("Request already sent");
     _maxRedirects = maxRedirects;
   }
@@ skipping 75 matching lines @@
         sb.write(cookies[i].value);
       }
       headers.add(HttpHeaders.COOKIE, sb.toString());
     }
 
     headers._finalize();
 
     // Write headers.
     headers._write(buffer);
     writeCRLF();
-    _ioSink.add(buffer.readBytes());
+    _headersSink.add(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 55 matching lines @@
           " $_bytesWritten bytes written while expected "
           "$expectedContentLength.")));
     }
     sink.close();
   }
 }
 
 
 // Extends StreamConsumer as this is an internal type, only used to pipe to.
 class _HttpOutgoing implements StreamConsumer<List<int>> {
-  Function _onStream;
-  final Completer _consumeCompleter = new Completer();
+  final Completer _doneCompleter = new Completer();
+  final StreamConsumer _consumer;
 
-  Future onStream(Future callback(Stream<List<int>> stream)) {
-    _onStream = callback;
-    return _consumeCompleter.future;
-  }
-
-  Future consume(Stream<List<int>> stream) {
-    _onStream(stream)
-        .then((_) => _consumeCompleter.complete(),
-              onError: _consumeCompleter.completeError);
-    // Use .then to ensure a Future branch.
-    return _consumeCompleter.future.then((_) => this);
-  }
+  _HttpOutgoing(StreamConsumer this._consumer);
 
   Future addStream(Stream<List<int>> stream) {
-    throw new UnimplementedError("_HttpOutgoing.addStream");
+    return _consumer.addStream(stream)
+        .catchError((error) {
+          _doneCompleter.completeError(error);
+          throw error;
+        });
   }
 
   Future close() {
-    throw new UnimplementedError("_HttpOutgoing.close");
+    _doneCompleter.complete(_consumer);
+    return new Future.immediate(null);
   }
+
+  Future get done => _doneCompleter.future;
 }
 
 
 class _HttpClientConnection {
   final String key;
   final Socket _socket;
   final _HttpParser _httpParser;
   StreamSubscription _subscription;
   final _HttpClient _httpClient;
 
   Completer<_HttpIncoming> _nextResponseCompleter;
   Future _streamFuture;
 
   _HttpClientConnection(String this.key,
                         Socket this._socket,
                         _HttpClient this._httpClient)
       : _httpParser = new _HttpParser.responseParser() {
     _socket.pipe(_httpParser);
-    _socket.done.catchError((e) { destroy(); });
 
     // Set up handlers on the parser here, so we are sure to get 'onDone' from
     // the parser.
     _subscription = _httpParser.listen(
         (incoming) {
           // Only handle one incoming response at the time. Keep the
           // stream paused until the response have been processed.
           _subscription.pause();
           // We assume the response is not here, until we have send the request.
           assert(_nextResponseCompleter != null);
           var completer = _nextResponseCompleter;
           _nextResponseCompleter = null;
           completer.complete(incoming);
         },
         onError: (error) {
           if (_nextResponseCompleter != null) {
             _nextResponseCompleter.completeError(error);
             _nextResponseCompleter = null;
           }
         },
         onDone: () {
           close();
         });
   }
 
   _HttpClientRequest send(Uri uri, int port, String method, bool isDirect) {
     // Start with pausing the parser.
     _subscription.pause();
-    var outgoing = new _HttpOutgoing();
+    var outgoing = new _HttpOutgoing(_socket);
     // Create new request object, wrapping the outgoing connection.
     var request = new _HttpClientRequest(outgoing,
                                          uri,
                                          method,
                                          !isDirect,
                                          _httpClient,
                                          this);
     request.headers.host = uri.domain;
     request.headers.port = port;
     request.headers.set(HttpHeaders.ACCEPT_ENCODING, "gzip");
     if (uri.userInfo != null && !uri.userInfo.isEmpty) {
       // If the URL contains user information use that for basic
       // authorization
       String auth =
           CryptoUtils.bytesToBase64(_encodeString(uri.userInfo));
       request.headers.set(HttpHeaders.AUTHORIZATION, "Basic $auth");
     } else {
       // Look for credentials.
       _Credentials cr = _httpClient._findCredentials(uri);
       if (cr != null) {
         cr.authorize(request);
       }
     }
     // Start sending the request (lazy, delayed until the user provides
     // data).
     _httpParser.responseToMethod = method;
-    _streamFuture = outgoing.onStream((stream) {
-        return _socket.writeStream(stream)
-            .then((s) {
-              // Request sent, set up response completer.
-              _nextResponseCompleter = new Completer();
+    _streamFuture = outgoing.done
+        .then((s) {
+          // Request sent, set up response completer.
+          _nextResponseCompleter = new Completer();
 
-              // Listen for response.
-              _nextResponseCompleter.future
-                  .then((incoming) {
-                    incoming.dataDone.then((_) {
-                      if (incoming.headers.persistentConnection &&
-                          request.persistentConnection) {
-                        // Return connection, now we are done.
-                        _httpClient._returnConnection(this);
-                        _subscription.resume();
-                      } else {
-                        destroy();
-                      }
-                    });
-                    request._onIncoming(incoming);
-                  })
-                  // If we see a state error, we failed to get the 'first'
-                  // element.
-                  // Transform the error to a HttpParserException, for
-                  // consistency.
-                  .catchError((error) {
-                    throw new HttpParserException(
-                        "Connection closed before data was received");
-                  }, test: (error) => error is StateError)
-                  .catchError((error) {
-                    // We are done with the socket.
+          // Listen for response.
+          _nextResponseCompleter.future
+              .then((incoming) {
+                incoming.dataDone.then((_) {
+                  if (incoming.headers.persistentConnection &&
+                      request.persistentConnection) {
+                    // Return connection, now we are done.
+                    _httpClient._returnConnection(this);
+                    _subscription.resume();
+                  } else {
                     destroy();
-                    request._onError(error);
-                  });
+                  }
+                });
+                request._onIncoming(incoming);
+              })
+              // If we see a state error, we failed to get the 'first'
+              // element.
+              // Transform the error to a HttpParserException, for
+              // consistency.
+              .catchError((error) {
+                throw new HttpParserException(
+                    "Connection closed before data was received");
+              }, test: (error) => error is StateError)
+              .catchError((error) {
+                // We are done with the socket.
+                destroy();
+                request._onError(error);
+              });
 
-              // Resume the parser now we have a handler.
-              _subscription.resume();
-              return s;
-            }, onError: (e) {
-              destroy();
-              throw e;
-            });
-    });
+          // Resume the parser now we have a handler.
+          _subscription.resume();
+          return s;
+        }, onError: (e) {
+          destroy();
+        });
     return request;
   }
 
   Future<Socket> detachSocket() {
-    return _streamFuture
-        .then((_) => new _DetachedSocket(_socket, _httpParser.detachIncoming()),
-              onError: (_) {});
+    return _streamFuture.then(
+        (_) => new _DetachedSocket(_socket, _httpParser.detachIncoming()));
   }
 
   void destroy() {
     _httpClient._connectionClosed(this);
     _socket.destroy();
   }
 
   void close() {
     _httpClient._connectionClosed(this);
     _streamFuture
           // TODO(ajohnsen): Add timeout.
-        .then((_) => _socket.destroy(),
-              onError: (_) {});
+        .then((_) => _socket.destroy());
   }
 
   HttpConnectionInfo get connectionInfo => _HttpConnectionInfo.create(_socket);
 }
 
 class _ConnnectionInfo {
   _ConnnectionInfo(_HttpClientConnection this.connection, _Proxy this.proxy);
   final _HttpClientConnection connection;
   final _Proxy proxy;
 }
@@ skipping 284 matching lines @@
   final Socket _socket;
   final _HttpServer _httpServer;
   final _HttpParser _httpParser;
   StreamSubscription _subscription;
 
   Future _streamFuture;
 
   _HttpConnection(Socket this._socket, _HttpServer this._httpServer)
       : _httpParser = new _HttpParser.requestParser() {
     _socket.pipe(_httpParser);
-    _socket.done.catchError((e) => destroy());
     _subscription = _httpParser.listen(
         (incoming) {
           // 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();
+          var outgoing = new _HttpOutgoing(_socket);
           var response = new _HttpResponse(incoming.headers.protocolVersion,
                                            outgoing);
           var request = new _HttpRequest(response, incoming, _httpServer, this);
-          outgoing.onStream((stream) {
-            return _streamFuture = _socket.writeStream(stream)
-                .then((_) {
-                  if (_state == _DETACHED) return;
-                  if (response.persistentConnection &&
-                      request.persistentConnection &&
-                      incoming.fullBodyRead) {
-                    _state = _IDLE;
-                    // Resume the subscription for incoming requests as the
-                    // request is now processed.
-                    _subscription.resume();
-                  } else {
-                    // Close socket, keep-alive not used or body sent before
-                    // received data was handled.
-                    destroy();
-                  }
-                })
-                .catchError((e) {
+          _streamFuture = outgoing.done
+              .then((_) {
+                if (_state == _DETACHED) return;
+                if (response.persistentConnection &&
+                    request.persistentConnection &&
+                    incoming.fullBodyRead) {
+                  _state = _IDLE;
+                  // Resume the subscription for incoming requests as the
+                  // request is now processed.
+                  _subscription.resume();
+                } else {
+                  // Close socket, keep-alive not used or body sent before
+                  // received data was handled.
                   destroy();
-                  throw e;
-                });
-          });
+                }
+              })
+              .catchError((e) {
+                destroy();
+              });
           response._ignoreBody = request.method == "HEAD";
           response._httpRequest = request;
           _httpServer._handleRequest(request);
         },
         onDone: () {
           destroy();
         },
         onError: (error) {
           _httpServer._handleError(error);
           destroy();
@@ skipping 261 matching lines @@
   void writeCharCode(int charCode) => _socket.writeCharCode(charCode);
 
   void writeAll(Iterable objects, [String separator = ""]) {
     _socket.writeAll(objects, separator);
   }
 
   void add(List<int> bytes) => _socket.add(bytes);
 
   void addError(AsyncError error) => _socket.addError(error);
 
-  Future<Socket> consume(Stream<List<int>> stream) {
-    return _socket.consume(stream);
-  }
-
   Future<Socket> addStream(Stream<List<int>> stream) {
     return _socket.addStream(stream);
   }
 
-  Future<Socket> writeStream(Stream<List<int>> stream) {
-    return _socket.writeStream(stream);
-  }
-
   void destroy() => _socket.destroy();
 
   Future close() => _socket.close();
 
   Future<Socket> get done => _socket.done;
 
   int get port => _socket.port;
 
   String get remoteHost => _socket.remoteHost;
 
@@ skipping 110 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;
 }