Enable throttleing of HttpClient per host, with default being 6.

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;
 
 const int _OUTGOING_BUFFER_SIZE = 8 * 1024;
 
 class _HttpIncoming extends Stream<List<int>> {
   final int _transferLength;
@@ skipping 1504 matching lines @@
 
 
 class _ConnectionTarget {
   // Unique key for this connection target.
   final String key;
   final String host;
   final int port;
   final bool isSecure;
   final Set<_HttpClientConnection> _idle = new HashSet();
   final Set<_HttpClientConnection> _active = new HashSet();
+  final Queue _pending = new ListQueue();
   int _connecting = 0;
 
   _ConnectionTarget(this.key, this.host, this.port, this.isSecure);
 
   bool get isEmpty => _idle.isEmpty && _active.isEmpty && _connecting == 0;
 
   bool get hasIdle => _idle.isNotEmpty;
 
-  bool get hasActive => _active.isNotEmpty && _connecting > 0;
+  bool get hasActive => _active.isNotEmpty || _connecting > 0;
 
   _HttpClientConnection takeIdle() {
     assert(hasIdle);
     _HttpClientConnection connection = _idle.first;
     _idle.remove(connection);
     connection.stopTimer();
     _active.add(connection);
     return connection;
   }
 
+  _checkPending() {
+    if (_pending.isNotEmpty) {
+      _pending.removeFirst()();
+    }
+  }
+
   void addNewActive(_HttpClientConnection connection) {
     _active.add(connection);
   }
 
   void returnConnection(_HttpClientConnection connection) {
     assert(_active.contains(connection));
     _active.remove(connection);
     _idle.add(connection);
+    connection.startTimer();
+    _checkPending();
   }
 
   void connectionClosed(_HttpClientConnection connection) {
     assert(!_active.contains(connection) || !_idle.contains(connection));
     _active.remove(connection);
     _idle.remove(connection);
+    _checkPending();
   }
 
   void close(bool force) {
     for (var c in _idle.toList()) {
       c.close();
     }
     if (force) {
       for (var c in _active.toList()) {
         c.destroy();
       }
     }
   }
 
   Future<_ConnectionInfo> connect(String uriHost,
                                    int uriPort,
                                    _Proxy proxy,
                                    _HttpClient client) {
     if (hasIdle) {
       var connection = takeIdle();
       client._updateTimers();
       return new Future.value(new _ConnectionInfo(connection, proxy));
     }
+    if (client.maxConnectionsPerHost != null &&
+        _active.length + _connecting >= client.maxConnectionsPerHost) {
+      var completer = new Completer();
+      _pending.add(() {
+        connect(uriHost, uriPort, proxy, client)
+            .then(completer.complete, onError: completer.completeError);
+      });
+      return completer.future;
+    }
     var currentBadCertificateCallback = client._badCertificateCallback;
     bool callback(X509Certificate certificate) =>
         currentBadCertificateCallback == null ? false :
         currentBadCertificateCallback(certificate, uriHost, uriPort);
     Future socketFuture = (isSecure && proxy.isDirect
         ? SecureSocket.connect(host,
                                port,
                                sendClientCertificate: true,
                                onBadCertificate: callback)
         : Socket.connect(host, port));
     _connecting++;
     return socketFuture.then((socket) {
         _connecting--;
         socket.setOption(SocketOption.TCP_NODELAY, true);
         var connection = new _HttpClientConnection(key, socket, client);
         if (isSecure && !proxy.isDirect) {
           connection._dispose = true;
           return connection.createProxyTunnel(uriHost, uriPort, proxy, callback)
               .then((tunnel) {
                 client._getConnectionTarget(uriHost, uriPort, true)
                     .addNewActive(tunnel);
                 return new _ConnectionInfo(tunnel, proxy);
               });
         } else {
           addNewActive(connection);
           return new _ConnectionInfo(connection, proxy);
         }
+      }, onError: (error) {
+        _connecting--;
+        _checkPending();
+        throw error;
       });
   }
 }
 
 
 class _HttpClient implements HttpClient {
   bool _closing = false;
   final Map<String, _ConnectionTarget> _connectionTargets
       = new HashMap<String, _ConnectionTarget>();
   final List<_Credentials> _credentials = [];
   final List<_ProxyCredentials> _proxyCredentials = [];
   Function _authenticate;
   Function _authenticateProxy;
   Function _findProxy = HttpClient.findProxyFromEnvironment;
   Duration _idleTimeout = const Duration(seconds: 15);
   Function _badCertificateCallback;
 
   Timer _noActiveTimer;
 
   Duration get idleTimeout => _idleTimeout;
 
+  int maxConnectionsPerHost;
+
   String userAgent = _getHttpVersion();
 
   void set idleTimeout(Duration timeout) {
     _idleTimeout = timeout;
     for (var c in _connectionTargets.values) {
       for (var idle in c.idle) {
         // Reset timer. This is fine, as it's not happening often.
         idle.stopTimer();
         idle.startTimer();
       }
@@ skipping 149 matching lines @@
           }
           return request
               ..headers.chunkedTransferEncoding = false
               ..contentLength = 0;
         });
   }
 
   // Return a live connection to the idle pool.
   void _returnConnection(_HttpClientConnection connection) {
     _connectionTargets[connection.key].returnConnection(connection);
-    connection.startTimer();
     _updateTimers();
   }
 
   // Remove a closed connnection from the active set.
   void _connectionClosed(_HttpClientConnection connection) {
     connection.stopTimer();
     var connectionTarget = _connectionTargets[connection.key];
     if (connectionTarget != null) {
       connectionTarget.connectionClosed(connection);
       if (connectionTarget.isEmpty) {
@@ skipping 921 matching lines @@
   const _RedirectInfo(this.statusCode, this.method, this.location);
 }
 
 String _getHttpVersion() {
   var version = Platform.version;
   // Only include major and minor version numbers.
   int index = version.indexOf('.', version.indexOf('.') + 1);
   version = version.substring(0, index);
   return 'Dart/$version (dart:io)';
 }