Discard HTTP data written after a socket error

runtime/bin/http_impl.dart

 // Copyright (c) 2012, 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.
 
 class _HttpRequestResponseBase {
   _HttpRequestResponseBase(_HttpConnectionBase this._httpConnection)
       : _contentLength = -1,
         _keepAlive = false,
         _headers = new Map();
 
   int get contentLength() => _contentLength;
   bool get keepAlive() => _keepAlive;
   Map get headers() => _headers;
 
   void _setHeader(String name, String value) {
     _headers[name.toLowerCase()] = value;
   }
 
   bool _write(List<int> data, bool copyBuffer) {
     bool allWritten = true;
     if (data.length > 0) {
       if (_contentLength < 0) {
         // Write chunk size if transfer encoding is chunked.
         _writeHexString(data.length);
         _writeCRLF();
-        _httpConnection.outputStream.write(data, copyBuffer);
+        _httpConnection._write(data, copyBuffer);
         allWritten = _writeCRLF();
       } else {
-        allWritten = _httpConnection.outputStream.write(data, copyBuffer);
+        allWritten = _httpConnection._write(data, copyBuffer);
       }
     }
     return allWritten;
   }
 
   bool _writeList(List<int> data, int offset, int count) {
     bool allWritten = true;
     if (count > 0) {
       if (_contentLength < 0) {
         // Write chunk size if transfer encoding is chunked.
         _writeHexString(count);
         _writeCRLF();
-        _httpConnection.outputStream.writeFrom(data, offset, count);
+        _httpConnection._writeFrom(data, offset, count);
         allWritten = _writeCRLF();
       } else {
-        allWritten = _httpConnection.outputStream.writeFrom(data, offset, count);
+        allWritten = _httpConnection._writeFrom(data, offset, count);
       }
     }
     return allWritten;
   }
 
   bool _writeDone() {
     bool allWritten = true;
     if (_contentLength < 0) {
       // Terminate the content if transfer encoding is chunked.
-      allWritten = _httpConnection.outputStream.write(_Const.END_CHUNKED);
+      allWritten = _httpConnection._write(_Const.END_CHUNKED);
     }
     return allWritten;
   }
 
   bool _writeHeaders() {
     List<int> data;
 
     // Format headers.
     _headers.forEach((String name, String value) {
       data = name.charCodes();
-      _httpConnection.outputStream.write(data);
+      _httpConnection._write(data);
       data = ": ".charCodes();
-      _httpConnection.outputStream.write(data);
+      _httpConnection._write(data);
       data = value.charCodes();
-      _httpConnection.outputStream.write(data);
+      _httpConnection._write(data);
       _writeCRLF();
     });
     // Terminate header.
     return _writeCRLF();
   }
 
   bool _writeHexString(int x) {
     final List<int> hexDigits = [0x30, 0x31, 0x32, 0x33, 0x34,
                                  0x35, 0x36, 0x37, 0x38, 0x39,
                                  0x41, 0x42, 0x43, 0x44, 0x45, 0x46];
     ByteArray hex = new ByteArray(10);
     int index = hex.length;
     while (x > 0) {
       index--;
       hex[index] = hexDigits[x % 16];
       x = x >> 4;
     }
-    return _httpConnection.outputStream.writeFrom(hex, index, hex.length - index);
+    return _httpConnection._writeFrom(hex, index, hex.length - index);
   }
 
   bool _writeCRLF() {
     final CRLF = const [_CharCode.CR, _CharCode.LF];
-    return _httpConnection.outputStream.write(CRLF);
+    return _httpConnection._write(CRLF);
   }
 
   bool _writeSP() {
     final SP = const [_CharCode.SP];
-    return _httpConnection.outputStream.write(SP);
+    return _httpConnection._write(SP);
   }
 
   _HttpConnectionBase _httpConnection;
   Map<String, String> _headers;
 
   // Length of the content body. If this is set to -1 (default value)
   // when starting to send data chunked transfer encoding will be
   // used.
   int _contentLength;
   bool _keepAlive;
@@ skipping 157 matching lines @@
   }
 
   bool _streamWriteFrom(List<int> buffer, int offset, int len) {
     return _writeList(buffer, offset, len);
   }
 
   void _streamClose() {
     _httpConnection._phase = _HttpConnectionBase.PHASE_IDLE;
     _state = DONE;
     // Stop tracking no pending write events.
-    _httpConnection.outputStream.onNoPendingWrites = null;
+    _httpConnection._onNoPendingWrites = null;
     // Ensure that any trailing data is written.
     _writeDone();
     // If the connection is closing then close the output stream to
     // fully close the socket.
     if (_httpConnection._closing) {
-      _httpConnection.outputStream.close();
+      _httpConnection._close();
     }
   }
 
   void _streamSetNoPendingWriteHandler(callback()) {
     if (_state != DONE) {
-      _httpConnection.outputStream.onNoPendingWrites = callback;
+      _httpConnection._onNoPendingWrites = callback;
     }
   }
 
   void _streamSetCloseHandler(callback()) {
     // TODO(sgjesse): Handle this.
   }
 
   void _streamSetErrorHandler(callback(Exception e)) {
     _streamErrorHandler = callback;
   }
@@ skipping 48 matching lines @@
       case HttpStatus.SERVICE_UNAVAILABLE: return "Service Unavailable";
       case HttpStatus.GATEWAY_TIMEOUT: return "Gateway Time-out";
       case HttpStatus.HTTP_VERSION_NOT_SUPPORTED:
         return "Http Version not supported";
       default: return "Status $statusCode";
     }
   }
 
   bool _writeHeader() {
     List<int> data;
-    OutputStream stream = _httpConnection.outputStream;
 
     // Write status line.
-    stream.write(_Const.HTTP11);
+    _httpConnection._write(_Const.HTTP11);
     _writeSP();
     data = _statusCode.toString().charCodes();
-    stream.write(data);
+    _httpConnection._write(data);
     _writeSP();
     data = reasonPhrase.charCodes();
-    stream.write(data);
+    _httpConnection._write(data);
     _writeCRLF();
 
     // Determine the value of the "Connection" header
     // based on the keep alive state.
     setHeader("Connection", keepAlive ? "keep-alive" : "close");
     // Determine the value of the "Transfer-Encoding" header based on
     // whether the content length is known.
     if (_contentLength >= 0) {
       setHeader("Content-Length", _contentLength.toString());
     } else {
@@ skipping 102 matching lines @@
                           _httpParser = new _HttpParser();
 
   void _connectionEstablished(Socket socket) {
     _socket = socket;
     // Register handler for socket events.
     _socket.onData = _onData;
     _socket.onClosed = _onClosed;
     _socket.onError = _onError;
   }
 
-  OutputStream get outputStream() {
-    return _socket.outputStream;
+  bool _write(List<int> data, [bool copyBuffer = false]) {
+    if (!_error) {
+      return _socket.outputStream.write(data, copyBuffer);
+    }
+  }
+
+  bool _writeFrom(List<int> buffer, [int offset, int len]) {
+    if (!_error) {
+      return _socket.outputStream.writeFrom(buffer, offset, len);
+    }
+  }
+
+  bool _close() {
+    _socket.close();
   }
 
   void _onData() {
     int available = _socket.available();
     if (available == 0) {
       return;
     }
 
     ByteArray buffer = new ByteArray(available);
     int bytesRead = _socket.readList(buffer, 0, available);
     if (bytesRead > 0) {
       int parsed = _httpParser.writeList(buffer, 0, bytesRead);
       if (parsed != bytesRead) {
         // TODO(sgjesse): Error handling.
         _socket.close();
       }
     }
   }
 
   void _onClosed() {

[Anders Johnsen, 2012/04/02 08:42:32]

Could the socket be closed normally from the client, when not all data have been
written? (E.g. the state where all current data is flushed, and the socket is
waiting for more data).

[Søren Gjesse, 2012/04/02 09:14:31]

The client can half-close the socket when finished writing the request. That
will result in an onClosed event but the server can still write the full
response before closing the socket for writing.

However you have a point that the client can do a full close, and then writing
will fail. It should still go through onError though.

I will try to add a test.

     if (_phase != PHASE_IDLE) {
       // Client closed socket for writing. Socket should still be open
       // for writing the response.
       _closing = true;
     } else {
       // The connection is currently not used by any request just close it.
       _socket.close();
     }
     if (_onDisconnectCallback != null) _onDisconnectCallback();
   }
 
   void _onError(Exception e) {
     // If an error occurs, make sure to close the socket if one is associated.
+    _error = true;
     if (_socket != null) {
       _socket.close();
     }
     if (_onErrorCallback != null) {
       _onErrorCallback(e);
     }
     _propagateError(e);
   }
 
   abstract void _propagateError(Exception e);
 
   void set onDisconnect(void callback()) {
     _onDisconnectCallback = callback;
   }
 
   void set onError(void callback(Exception e)) {
     _onErrorCallback = callback;
   }
 
+  void set _onNoPendingWrites(void callback()) {
+    if (!_error) {
+      _socket.outputStream.onNoPendingWrites = callback;
+    }
+  }
+
   int hashCode() => _socket.hashCode();
 
   int _phase;
   Socket _socket;
   bool _closing = false;  // Is the socket closed by the client?
+  bool _error = false;  // Is the socket closed due to an error?
   _HttpParser _httpParser;
 
   Queue _sendBuffers;
 
   Function _onDisconnectCallback;
   Function _onErrorCallback;
 }
 
 
 // HTTP server connection over a socket.
@@ skipping 190 matching lines @@
     return _write(buffer, copyBuffer);
   }
 
   bool _streamWriteFrom(List<int> buffer, int offset, int len) {
     return _writeList(buffer, offset, len);
   }
 
   void _streamClose() {
     _state = DONE;
     // Stop tracking no pending write events.
-    _httpConnection.outputStream.onNoPendingWrites = null;
+    _httpConnection._onNoPendingWrites = null;
     // Ensure that any trailing data is written.
     _writeDone();
     // If the connection is closing then close the output stream to
     // fully close the socket.
     if (_httpConnection._closing) {
-      _httpConnection.outputStream.close();
+      _httpConnection._close();
     }
   }
 
   void _streamSetNoPendingWriteHandler(callback()) {
     if (_state != DONE) {
-      _httpConnection.outputStream.onNoPendingWrites = callback;
+      _httpConnection._onNoPendingWrites = callback;
     }
   }
 
   void _streamSetCloseHandler(callback()) {
     // TODO(sgjesse): Handle this.
   }
 
   void _streamSetErrorHandler(callback(Exception e)) {
     _streamErrorHandler = callback;
   }
 
   void _writeHeader() {
     List<int> data;
-    OutputStream stream = _httpConnection.outputStream;
 
     // Write request line.
     data = _method.toString().charCodes();
-    stream.write(data);
+    _httpConnection._write(data);
     _writeSP();
     data = _uri.toString().charCodes();
-    stream.write(data);
+    _httpConnection._write(data);
     _writeSP();
-    stream.write(_Const.HTTP11);
+    _httpConnection._write(_Const.HTTP11);
     _writeCRLF();
 
     // Determine the value of the "Connection" header
     // based on the keep alive state.
     setHeader("Connection", keepAlive ? "keep-alive" : "close");
     // Determine the value of the "Transfer-Encoding" header based on
     // whether the content length is known.
     if (_contentLength >= 0) {
       setHeader("Content-Length", _contentLength.toString());
     } else {
@@ skipping 390 matching lines @@
     sockets.addFirst(socketConn);
     socketConn._markReturned();
   }
 
   Function _onOpen;
   Map<String, Queue<_SocketConnection>> _openSockets;
   Set<_SocketConnection> _activeSockets;
   Timer _evictionTimer;
   bool _shutdown;  // Has this HTTP client been shutdown?
 }