Support canceling an incoming message, while the other end will be notified (by write-error from th…

sdk/lib/io/http_parser.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;
 
 // Global constants.
 class _Const {
   // Bytes for "HTTP".
   static const HTTP = const [72, 84, 84, 80];
@@ skipping 856 matching lines @@
       return byte - 0x61 + 10;  // a - f
     } else {
       throw new HttpParserException("Failed to parse HTTP");
     }
   }
 
   void _createIncoming(int transferLength) {
     assert(_incoming == null);
     assert(_bodyController == null);
     _bodyController = new StreamController<List<int>>(
-        onSubscriptionStateChange: _updateParsePauseState,
+        onSubscriptionStateChange: _bodySubscriptionStateChange,
         onPauseStateChange: _updateParsePauseState);
     _incoming = new _HttpIncoming(
         _headers, transferLength, _bodyController.stream);
     _pauseParsing();  // Needed to handle detaching - don't start on the body!
   }
 
   void _closeIncoming() {
     assert(_incoming != null);
     var tmp = _incoming;
     _incoming = null;
     tmp.close();
     if (_bodyController != null) {
       _bodyController.close();
       _bodyController = null;
     }
     _updateParsePauseState();
   }
 
   void _continueParsing() {
     _paused = false;
     if (!_parserCalled && _buffer != null) _parse();
   }
 
   void _pauseParsing() {
     _paused = true;
   }
 
+  void _bodySubscriptionStateChange() {
+    if (_incoming != null && !_bodyController.hasSubscribers) {
+      _closeIncoming();
+    } else {
+      _updateParsePauseState();
+    }
+  }
+
   void _updateParsePauseState() {
     if (_bodyController != null) {
       if (_bodyController.hasSubscribers && !_bodyController.isPaused) {
         _continueParsing();
       } else {
         _pauseParsing();
       }
     } else {
       if (_controller.hasSubscribers && !_controller.isPaused) {
         _continueParsing();
@@ skipping 47 matching lines @@
   StreamController<_HttpIncoming> _controller;
   StreamController<List<int>> _bodyController;
 }
 
 
 class HttpParserException implements Exception {
   const HttpParserException([String this.message = ""]);
   String toString() => "HttpParserException: $message";
   final String message;
 }