refactor(http): use Observables in Http backends

BREAKING CHANGE: Http now returns Rx Observables directly, so calling .toRx() is no longer necessary. Additionally, Http calls are now cold, so backend requests will not fire unless .subscribe() is called.

closes angular#4043 and closes angular#2974

Closes angular#4376

Author: robwormald

modules/angular2/src/http/backends/jsonp_backend.ts

@@ -8,11 +8,12 @@ import {BrowserJsonp} from './browser_jsonp';
 import {EventEmitter, ObservableWrapper} from 'angular2/src/core/facade/async';
 import {makeTypeError} from 'angular2/src/core/facade/exceptions';
 import {StringWrapper, isPresent} from 'angular2/src/core/facade/lang';
+var Observable = require('@reactivex/rxjs/dist/cjs/Observable');
 
 export class JSONPConnection implements Connection {
   readyState: ReadyStates;
   request: Request;
-  response: EventEmitter;
+  response: any;
   private _id: string;
   private _script: Element;
   private _responseData: any;
@@ -27,50 +28,66 @@ export class JSONPConnection implements Connection {
       throw makeTypeError("JSONP requests must use GET request method.");
     }
     this.request = req;
-    this.response = new EventEmitter();
-    this.readyState = ReadyStates.Loading;
-    this._id = _dom.nextRequestID();
-
-    _dom.exposeConnection(this._id, this);
-
-    // Workaround Dart
-    // url = url.replace(/=JSONP_CALLBACK(&|$)/, `generated method`);
-    let callback = _dom.requestCallback(this._id);
-    let url: string = req.url;
-    if (url.indexOf('=JSONP_CALLBACK&') > -1) {
-      url = StringWrapper.replace(url, '=JSONP_CALLBACK&', `=${callback}&`);
-    } else if (url.lastIndexOf('=JSONP_CALLBACK') === url.length - '=JSONP_CALLBACK'.length) {
-      url = StringWrapper.substring(url, 0, url.length - '=JSONP_CALLBACK'.length) + `=${callback}`;
-    }
+    this.response = new Observable(responseObserver => {
 
-    let script = this._script = _dom.build(url);
+      this.readyState = ReadyStates.Loading;
+      let id = this._id = _dom.nextRequestID();
 
-    script.addEventListener('load', (event) => {
-      if (this.readyState === ReadyStates.Cancelled) return;
-      this.readyState = ReadyStates.Done;
-      _dom.cleanup(script);
-      if (!this._finished) {
-        ObservableWrapper.callThrow(
-            this.response, makeTypeError('JSONP injected script did not invoke callback.'));
-        return;
-      }
+      _dom.exposeConnection(id, this);
 
-      let responseOptions = new ResponseOptions({body: this._responseData});
-      if (isPresent(this.baseResponseOptions)) {
-        responseOptions = this.baseResponseOptions.merge(responseOptions);
+      // Workaround Dart
+      // url = url.replace(/=JSONP_CALLBACK(&|$)/, `generated method`);
+      let callback = _dom.requestCallback(this._id);
+      let url: string = req.url;
+      if (url.indexOf('=JSONP_CALLBACK&') > -1) {
+        url = StringWrapper.replace(url, '=JSONP_CALLBACK&', `=${callback}&`);
+      } else if (url.lastIndexOf('=JSONP_CALLBACK') === url.length - '=JSONP_CALLBACK'.length) {
+        url =
+            StringWrapper.substring(url, 0, url.length - '=JSONP_CALLBACK'.length) + `=${callback}`;
       }
 
-      ObservableWrapper.callNext(this.response, new Response(responseOptions));
-    });
+      let script = this._script = _dom.build(url);
 
-    script.addEventListener('error', (error) => {
-      if (this.readyState === ReadyStates.Cancelled) return;
-      this.readyState = ReadyStates.Done;
-      _dom.cleanup(script);
-      ObservableWrapper.callThrow(this.response, error);
-    });
+      let onLoad = event => {
+        if (this.readyState === ReadyStates.Cancelled) return;
+        this.readyState = ReadyStates.Done;
+        _dom.cleanup(script);
+        if (!this._finished) {
+          responseObserver.error(makeTypeError('JSONP injected script did not invoke callback.'));
+          return;
+        }
+
+        let responseOptions = new ResponseOptions({body: this._responseData});
+        if (isPresent(this.baseResponseOptions)) {
+          responseOptions = this.baseResponseOptions.merge(responseOptions);
+        }
+
+        responseObserver.next(new Response(responseOptions));
+        responseObserver.complete();
+      };
+
+      let onError = error => {
+        if (this.readyState === ReadyStates.Cancelled) return;
+        this.readyState = ReadyStates.Done;
+        _dom.cleanup(script);
+        responseObserver.error(error);
+      };
+
+      script.addEventListener('load', onLoad);
+      script.addEventListener('error', onError);
 
-    _dom.send(script);
+      _dom.send(script);
+
+      return () => {
+        this.readyState = ReadyStates.Cancelled;
+        script.removeEventListener('load', onLoad);
+        script.removeEventListener('error', onError);
+        if (isPresent(script)) {
+          this._dom.cleanup(script);
+        }
+
+      }
+    });
   }
 
   finished(data?: any) {
@@ -80,16 +97,6 @@ export class JSONPConnection implements Connection {
     if (this.readyState === ReadyStates.Cancelled) return;
     this._responseData = data;
   }
-
-  dispose(): void {
-    this.readyState = ReadyStates.Cancelled;
-    let script = this._script;
-    this._script = null;
-    if (isPresent(script)) {
-      this._dom.cleanup(script);
-    }
-    ObservableWrapper.callReturn(this.response);
-  }
 }
 
 @Injectable()

modules/angular2/src/http/backends/xhr_backend.ts

@@ -5,78 +5,78 @@ import {Response} from '../static_response';
 import {ResponseOptions, BaseResponseOptions} from '../base_response_options';
 import {Injectable} from 'angular2/src/core/di';
 import {BrowserXhr} from './browser_xhr';
-import {EventEmitter, ObservableWrapper} from 'angular2/src/core/facade/async';
 import {isPresent} from 'angular2/src/core/facade/lang';
-
+var Observable = require('@reactivex/rxjs/dist/cjs/Observable');
 /**
- * Creates connections using `XMLHttpRequest`. Given a fully-qualified
- * request, an `XHRConnection` will immediately create an `XMLHttpRequest` object and send the
- * request.
- *
- * This class would typically not be created or interacted with directly inside applications, though
- * the {@link MockConnection} may be interacted with in tests.
- */
+* Creates connections using `XMLHttpRequest`. Given a fully-qualified
+* request, an `XHRConnection` will immediately create an `XMLHttpRequest` object and send the
+* request.
+*
+* This class would typically not be created or interacted with directly inside applications, though
+* the {@link MockConnection} may be interacted with in tests.
+*/
 export class XHRConnection implements Connection {
   request: Request;
   /**
    * Response {@link EventEmitter} which emits a single {@link Response} value on load event of
    * `XMLHttpRequest`.
    */
-  response: EventEmitter;  // TODO: Make generic of <Response>;
+  response: any;  // TODO: Make generic of <Response>;
   readyState: ReadyStates;
-  private _xhr;  // TODO: make type XMLHttpRequest, pending resolution of
-                 // https://github.com/angular/ts2dart/issues/230
   constructor(req: Request, browserXHR: BrowserXhr, baseResponseOptions?: ResponseOptions) {
     this.request = req;
-    this.response = new EventEmitter();
-    this._xhr = browserXHR.build();
-    // TODO(jeffbcross): implement error listening/propagation
-    this._xhr.open(RequestMethods[req.method].toUpperCase(), req.url);
-    this._xhr.addEventListener('load', (_) => {
-      // responseText is the old-school way of retrieving response (supported by IE8 & 9)
-      // response/responseType properties were introduced in XHR Level2 spec (supported by IE10)
-      let response = isPresent(this._xhr.response) ? this._xhr.response : this._xhr.responseText;
+    this.response = new Observable(responseObserver => {
+      let _xhr: XMLHttpRequest = browserXHR.build();
+      _xhr.open(RequestMethods[req.method].toUpperCase(), req.url);
+      // load event handler
+      let onLoad = () => {
+        // responseText is the old-school way of retrieving response (supported by IE8 & 9)
+        // response/responseType properties were introduced in XHR Level2 spec (supported by
+        // IE10)
+        let response = isPresent(_xhr.response) ? _xhr.response : _xhr.responseText;
 
-      // normalize IE9 bug (http://bugs.jquery.com/ticket/1450)
-      let status = this._xhr.status === 1223 ? 204 : this._xhr.status;
+        // normalize IE9 bug (http://bugs.jquery.com/ticket/1450)
+        let status = _xhr.status === 1223 ? 204 : _xhr.status;
 
-      // fix status code when it is 0 (0 status is undocumented).
-      // Occurs when accessing file resources or on Android 4.1 stock browser
-      // while retrieving files from application cache.
-      if (status === 0) {
-        status = response ? 200 : 0;
-      }
+        // fix status code when it is 0 (0 status is undocumented).
+        // Occurs when accessing file resources or on Android 4.1 stock browser
+        // while retrieving files from application cache.
+        if (status === 0) {
+          status = response ? 200 : 0;
+        }
+        var responseOptions = new ResponseOptions({body: response, status: status});
+        if (isPresent(baseResponseOptions)) {
+          responseOptions = baseResponseOptions.merge(responseOptions);
+        }
+        responseObserver.next(new Response(responseOptions));
+        // TODO(gdi2290): defer complete if array buffer until done
+        responseObserver.complete();
+      };
+      // error event handler
+      let onError = (err) => {
+        var responseOptions = new ResponseOptions({body: err, type: ResponseTypes.Error});
+        if (isPresent(baseResponseOptions)) {
+          responseOptions = baseResponseOptions.merge(responseOptions);
+        }
+        responseObserver.error(new Response(responseOptions));
+      };
 
-      var responseOptions = new ResponseOptions({body: response, status: status});
-      if (isPresent(baseResponseOptions)) {
-        responseOptions = baseResponseOptions.merge(responseOptions);
+      if (isPresent(req.headers)) {
+        req.headers.forEach((value, name) => { _xhr.setRequestHeader(name, value); });
       }
 
-      ObservableWrapper.callNext(this.response, new Response(responseOptions));
-      // TODO(gdi2290): defer complete if array buffer until done
-      ObservableWrapper.callReturn(this.response);
-    });
+      _xhr.addEventListener('load', onLoad);
+      _xhr.addEventListener('error', onError);
 
-    this._xhr.addEventListener('error', (err) => {
-      var responseOptions = new ResponseOptions({body: err, type: ResponseTypes.Error});
-      if (isPresent(baseResponseOptions)) {
-        responseOptions = baseResponseOptions.merge(responseOptions);
-      }
-      ObservableWrapper.callThrow(this.response, new Response(responseOptions));
-    });
-    // TODO(jeffbcross): make this more dynamic based on body type
+      _xhr.send(this.request.text());
 
-    if (isPresent(req.headers)) {
-      req.headers.forEach((value, name) => { this._xhr.setRequestHeader(name, value); });
-    }
-
-    this._xhr.send(this.request.text());
+      return () => {
+        _xhr.removeEventListener('load', onLoad);
+        _xhr.removeEventListener('error', onError);
+        _xhr.abort();
+      };
+    });
   }
-
-  /**
-   * Calls abort on the underlying XMLHttpRequest.
-   */
-  dispose(): void { this._xhr.abort(); }
 }
 
 /**

modules/angular2/src/http/http.ts

@@ -34,19 +34,9 @@ function mergeOptions(defaultOpts, providedOpts, method, url): RequestOptions {
  * Performs http requests using `XMLHttpRequest` as the default backend.
  *
  * `Http` is available as an injectable class, with methods to perform http requests. Calling
- * `request` returns an {@link EventEmitter} which will emit a single {@link Response} when a
+ * `request` returns an {@link Observable} which will emit a single {@link Response} when a
  * response is received.
  *
- *
- * ## Breaking Change
- *
- * Previously, methods of `Http` would return an RxJS Observable directly. For now,
- * the `toRx()` method of {@link EventEmitter} needs to be called in order to get the RxJS
- * Subject. `EventEmitter` does not provide combinators like `map`, and has different semantics for
- * subscribing/observing. This is temporary; the result of all `Http` method calls will be either an
- * Observable
- * or Dart Stream when [issue #2794](https://github.com/angular/angular/issues/2794) is resolved.
- *
  * #Example
  *
  * ```
@@ -56,8 +46,6 @@ function mergeOptions(defaultOpts, providedOpts, method, url): RequestOptions {
  * class PeopleComponent {
  *   constructor(http: Http) {
  *     http.get('people.json')
- *       //Get the RxJS Subject
- *       .toRx()
  *       // Call map on the response observable to get the parsed people object
  *       .map(res => res.json())
  *       // Subscribe to the observable to get the parsed people object and attach it to the
@@ -67,9 +55,6 @@ function mergeOptions(defaultOpts, providedOpts, method, url): RequestOptions {
  * }
  * ```
  *
- * To use the {@link EventEmitter} returned by `Http`, simply pass a generator (See "interface
- *Generator" in the Async Generator spec: https://github.com/jhusain/asyncgenerator) to the
- *`observer` method of the returned emitter, with optional methods of `next`, `throw`, and `return`.
  *
  * #Example
  *
@@ -95,7 +80,7 @@ function mergeOptions(defaultOpts, providedOpts, method, url): RequestOptions {
  *       [MockBackend, BaseRequestOptions])
  * ]);
  * var http = injector.get(Http);
- * http.get('request-from-mock-backend.json').toRx().subscribe((res:Response) => doSomething(res));
+ * http.get('request-from-mock-backend.json').subscribe((res:Response) => doSomething(res));
  * ```
  *
  **/

modules/angular2/src/http/interfaces.ts

@@ -26,7 +26,6 @@ export class Connection {
   readyState: ReadyStates;
   request: Request;
   response: EventEmitter;  // TODO: generic of <Response>;
-  dispose(): void { throw new BaseException('Abstract!'); }
 }
 
 /**