feat(http): support coalescing http requests

To enable this, configure the HttpConfig module with a coalescing
duration.  Doing so will enable all HTTP responses that arrive within
that interval to all be processed in a single digest.  Typically, this
is most helpful if you have many XHRs that are all going to be satisfied
by cached responses causing them to happen in quick succession.

e.g. Use something like this to configure a 50ms interval.

    bind(HttpConfig, toValue: new HttpConfig(
        coalesceDuration: new Duration(milliseconds: 50)));

lib/core/module.dart

@@ -42,6 +42,7 @@ export "package:angular/core_dom/module_internal.dart" show
     EventHandler,
     Http,
     HttpBackend,
+    HttpConfig,
     HttpDefaultHeaders,
     HttpDefaults,
     HttpInterceptor,

lib/core_dom/http.dart

@@ -38,6 +38,11 @@ typedef RequestInterceptor(HttpResponseConfig);
 typedef RequestErrorInterceptor(dynamic);
 typedef Response(HttpResponse);
 typedef ResponseError(dynamic);
+typedef _CompleteResponse(HttpResponse);
+typedef _RunCoaleced(fn());
+
+_runNow(fn()) => fn();
+_identity(x) => x;
 
 /**
 * HttpInterceptors are used to modify the Http request. They can be added to
@@ -369,23 +374,29 @@ class HttpDefaults {
  */
 @Injectable()
 class Http {
-  var _pendingRequests = new HashMap<String, async.Future<HttpResponse>>();
-  BrowserCookies _cookies;
-  LocationWrapper _location;
-  UrlRewriter _rewriter;
-  HttpBackend _backend;
-  HttpInterceptors _interceptors;
+  final _pendingRequests = new HashMap<String, async.Future<HttpResponse>>();
+  final BrowserCookies _cookies;
+  final LocationWrapper _location;
+  final UrlRewriter _rewriter;
+  final HttpBackend _backend;
+  final HttpInterceptors _interceptors;
+  final RootScope _rootScope;
+  final HttpConfig _httpConfig;
+  final VmTurnZone _zone;
+
+  final _responseQueue = <Function>[];
+  async.Timer _responseQueueTimer;
 
   /**
    * The defaults for [Http]
    */
-  HttpDefaults defaults;
+  final HttpDefaults defaults;
 
   /**
    * Constructor, useful for DI.
    */
-  Http(this._cookies, this._location, this._rewriter, this._backend,
-       this.defaults, this._interceptors);
+  Http(this._cookies, this._location, this._rewriter, this._backend, this.defaults,
+       this._interceptors, this._rootScope, this._httpConfig, this._zone);
 
   /**
    * Parse a [requestUrl] and determine whether this is a same-origin request as
@@ -482,29 +493,25 @@ class Http {
         return new async.Future.value(new HttpResponse.copy(cachedResponse));
       }
 
-      var result = _backend.request(url,
-                                    method: method,
-                                    requestHeaders: config.headers,
-                                    sendData: config.data,
-                                    withCredentials: withCredentials).then((dom.HttpRequest value) {
-        // TODO: Uncomment after apps migrate off of this class.
-        // assert(value.status >= 200 && value.status < 300);
-
-        var response = new HttpResponse(value.status, value.responseText,
-            parseHeaders(value), config);
-
-        if (cache != null) cache.put(url, response);
-        _pendingRequests.remove(url);
-        return response;
-      }, onError: (error) {
-        if (error is! dom.ProgressEvent) throw error;
-        dom.ProgressEvent event = error;
-        _pendingRequests.remove(url);
-        dom.HttpRequest request = event.currentTarget;
-        return new async.Future.error(
-            new HttpResponse(request.status, request.response, parseHeaders(request), config));
-      });
-      return _pendingRequests[url] = result;
+      requestFromBackend(runCoalesced, onComplete, onError) => _backend.request(
+          url,
+          method: method,
+          requestHeaders: config.headers,
+          sendData: config.data,
+          withCredentials: withCredentials
+      ).then((dom.HttpRequest req) => _onResponse(req, runCoalesced, onComplete, config, cache, url),
+             onError: (e) => _onError(e, runCoalesced, onError, config, url));
+
+      async.Future responseFuture;
+      if (_httpConfig.coalesceDuration != null) {
+        async.Completer completer = new async.Completer();
+        responseFuture = completer.future;
+        _zone.runOutsideAngular(() => requestFromBackend(
+            _coalesce, completer.complete, completer.completeError));
+      } else {
+        responseFuture = requestFromBackend(_runNow, _identity, _identity);
+      }
+      return _pendingRequests[url] = responseFuture;
     };
 
     var chain = [[serverRequest, null]];
@@ -650,11 +657,50 @@ class Http {
              xsrfCookieName: xsrfCookieName, interceptors: interceptors, cache: cache,
              timeout: timeout);
 
+  _onResponse(dom.HttpRequest request, _RunCoaleced runCoalesced, _CompleteResponse onComplete,
+              HttpResponseConfig config, cache, String url) {
+    // TODO: Uncomment after apps migrate off of this class.
+    // assert(request.status >= 200 && request.status < 300);
+
+    var response = new HttpResponse(
+        request.status, request.responseText, parseHeaders(request), config);
+
+    if (cache != null) cache.put(url, response);
+    _pendingRequests.remove(url);
+    return runCoalesced(() => onComplete(response));
+  }
+
+  _onError(error, _RunCoaleced runCoalesced, _CompleteResponse onError,
+           HttpResponseConfig config, String url) {
+    if (error is! dom.ProgressEvent) throw error;
+    dom.ProgressEvent event = error;
+    _pendingRequests.remove(url);
+    dom.HttpRequest request = event.currentTarget;
+    var response = new HttpResponse(
+        request.status, request.response, parseHeaders(request), config);
+    return runCoalesced(() => onError(new async.Future.error(response)));
+  }
+
+  _coalesce(fn()) {
+    _responseQueue.add(fn);
+    if (_responseQueueTimer == null) {
+      _responseQueueTimer = new async.Timer(_httpConfig.coalesceDuration, _flushResponseQueue);
+    }
+  }
+
+  _flushResponseQueue() => _zone.run(_flushResponseQueueSync);
+
+  _flushResponseQueueSync() {
+    _responseQueueTimer = null;
+    _responseQueue.forEach(_runNow);
+    _responseQueue.clear();
+  }
+
   /**
    * Parse raw headers into key-value object
    */
-  static Map<String, String> parseHeaders(dom.HttpRequest value) {
-    var headers = value.getAllResponseHeaders();
+  static Map<String, String> parseHeaders(dom.HttpRequest request) {
+    var headers = request.getAllResponseHeaders();
 
     var parsed = new HashMap();
 
@@ -704,3 +750,10 @@ class Http {
           .replaceAll('%2C', ',')
           .replaceAll('%20', pctEncodeSpaces ? '%20' : '+');
 }
+
+@Injectable()
+class HttpConfig {
+  final Duration coalesceDuration;
+
+  HttpConfig({this.coalesceDuration});
+}

lib/core_dom/module_internal.dart

@@ -86,6 +86,7 @@ class CoreDomModule extends Module {
     bind(HttpDefaultHeaders);
     bind(HttpDefaults);
     bind(HttpInterceptors);
+    bind(HttpConfig, toValue: new HttpConfig());
     bind(Animate);
     bind(ViewCache);
     bind(BrowserCookies);

test/angular_spec.dart

@@ -126,6 +126,7 @@ main() {
         "angular.core.dom_internal.EventHandler",
         "angular.core.dom_internal.Http",
         "angular.core.dom_internal.HttpBackend",
+        "angular.core.dom_internal.HttpConfig",
         "angular.core.dom_internal.HttpDefaultHeaders",
         "angular.core.dom_internal.HttpDefaults",
         "angular.core.dom_internal.HttpInterceptor",

test/core_dom/http_spec.dart

@@ -1423,6 +1423,39 @@ void main() {
         });
       });
     });
+
+    describe('coalesce', () {
+      beforeEachModule((Module module) {
+        var coalesceDuration = new Duration(milliseconds: 100);
+        module.bind(HttpConfig, toValue: new HttpConfig(coalesceDuration: coalesceDuration));
+      });
+
+      it('should coalesce requests', async((Http http) {
+        backend.expect('GET', '/foo').respond(200, 'foo');
+        backend.expect('GET', '/bar').respond(200, 'bar');
+
+        var fooResp, barResp;
+        http.get('/foo').then((HttpResponse resp) => fooResp = resp.data);
+        http.get('/bar').then((HttpResponse resp) => barResp = resp.data);
+
+        microLeap();
+        backend.flush();
+        microLeap();
+        expect(fooResp).toBeNull();
+        expect(barResp).toBeNull();
+
+        clockTick(milliseconds: 99);
+        microLeap();
+        expect(fooResp).toBeNull();
+        expect(barResp).toBeNull();
+
+        clockTick(milliseconds: 1);
+        microLeap();
+        expect(fooResp).toEqual('foo');
+        expect(barResp).toEqual('bar');
+      }));
+
+    });
   });
 }