Make new StreamController be async by default.

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 89 matching lines @@
   final StreamSubscription subscription;
 
   List<int> carryOverData;
   bool paused;
 
   Completer resumeCompleter;
 
   _HttpDetachedIncoming(StreamSubscription this.subscription,
                         List<int> this.carryOverData) {
     controller = new StreamController<List<int>>(
+        sync: true,
         onListen: resume,
         onPause: pause,
         onResume: resume,
         onCancel: () => subscription.cancel());
     if (subscription == null) {
       // Socket was already closed.
       if (carryOverData != null) controller.add(carryOverData);
       controller.close();
     } else {
       pause();
@@ skipping 68 matching lines @@
   factory _HttpParser.requestParser() {
     return new _HttpParser._(true);
   }
 
   factory _HttpParser.responseParser() {
     return new _HttpParser._(false);
   }
 
   _HttpParser._(this._requestParser) {
     _controller = new StreamController<_HttpIncoming>(
-          onListen: () {
-            _socketSubscription.resume();
-            _paused = false;
-          },
-          onPause: () {
-            _paused = true;
-            _pauseStateChanged();
-          },
-          onResume: () {
-            _paused = false;
-            _pauseStateChanged();
-          },
-          onCancel: () {
-            try {
-              _socketSubscription.cancel();
-            } catch (e) {
-            }
-          });
+        sync: true,
+        onListen: () {
+          _socketSubscription.resume();
+          _paused = false;
+        },
+        onPause: () {
+          _paused = true;
+          _pauseStateChanged();
+        },
+        onResume: () {
+          _paused = false;
+          _pauseStateChanged();
+        },
+        onCancel: () {
+          try {
+            _socketSubscription.cancel();
+          } catch (e) {
+          }
+        });
     _reset();
   }
 
 
   StreamSubscription<_HttpIncoming> listen(void onData(_HttpIncoming event),
                                            {void onError(error),
                                             void onDone(),
                                             bool cancelOnError}) {
     return _controller.stream.listen(onData,
                                      onError: onError,
@@ skipping 650 matching lines @@
       throw new HttpParserException("Failed to parse HTTP");
     }
   }
 
   void _createIncoming(int transferLength) {
     assert(_incoming == null);
     assert(_bodyController == null);
     assert(!_bodyPaused);
     var incoming;
     _bodyController = new StreamController<List<int>>(
+        sync: true,
         onListen: () {
           if (incoming != _incoming) return;
           assert(_bodyPaused);
           _bodyPaused = false;
           _pauseStateChanged();
         },
         onPause: () {
           if (incoming != _incoming) return;
           assert(!_bodyPaused);
           _bodyPaused = true;
@@ skipping 89 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;
 }