chore(http): Use Observables in Http

- Remove ObservableWrapper/EventEmitter from Http.
- Temporarily use complete Rx build w/ all operators.

karma-js.conf.js

@@ -24,7 +24,7 @@ module.exports = function(config) {
       // Including systemjs because it defines `__eval`, which produces correct stack traces.
       'modules/angular2/src/test_lib/shims_for_IE.js',
       'node_modules/systemjs/dist/system.src.js',
-      {pattern: 'node_modules/@reactivex/rxjs/dist/cjs/**', included: false, watched: false, served: true},
+      {pattern: 'node_modules/@reactivex/rxjs/**', included: false, watched: false, served: true},
       'node_modules/reflect-metadata/Reflect.js',
       'tools/build/file2modulename.js',
       'test-main.js',

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

@@ -8,8 +8,9 @@ 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');
+// todo(robwormald): temporary until https://github.com/angular/angular/issues/4390 decided
-
+var Rx = require('@reactivex/rxjs/dist/cjs/Rx');
+var {Observable} = Rx;
 export class JSONPConnection implements Connection {
   readyState: ReadyStates;
   request: Request;

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

@@ -3,9 +3,10 @@ import {Request} from '../static_request';
 import {Response} from '../static_response';
 import {ReadyStates} from '../enums';
 import {Connection, ConnectionBackend} from '../interfaces';
-import {ObservableWrapper, EventEmitter} from 'angular2/src/core/facade/async';
 import {isPresent} from 'angular2/src/core/facade/lang';
 import {BaseException, WrappedException} from 'angular2/src/core/facade/exceptions';
+var Rx = require('@reactivex/rxjs/dist/cjs/Rx');
+let{Subject, ReplaySubject} = Rx;
 
 /**
  *
@@ -30,23 +31,14 @@ export class MockConnection implements Connection {
    * {@link EventEmitter} of {@link Response}. Can be subscribed to in order to be notified when a
    * response is available.
    */
-  response: EventEmitter;
+  response: any;  // Subject<Response>
 
   constructor(req: Request) {
-    this.response = new EventEmitter();
+    this.response = new ReplaySubject(1).take(1);
     this.readyState = ReadyStates.Open;
     this.request = req;
   }
 
-  /**
-   * Changes the `readyState` of the connection to a custom state of 5 (cancelled).
-   */
-  dispose() {
-    if (this.readyState !== ReadyStates.Done) {
-      this.readyState = ReadyStates.Cancelled;
-    }
-  }
-
   /**
    * Sends a mock response to the connection. This response is the value that is emitted to the
    * {@link EventEmitter} returned by {@link Http}.
@@ -66,8 +58,8 @@ export class MockConnection implements Connection {
       throw new BaseException('Connection has already been resolved');
     }
     this.readyState = ReadyStates.Done;
-    ObservableWrapper.callNext(this.response, res);
+    this.response.next(res);
-    ObservableWrapper.callReturn(this.response);
+    this.response.complete();
   }
 
   /**
@@ -92,8 +84,7 @@ export class MockConnection implements Connection {
   mockError(err?: Error) {
     // Matches XHR semantics
     this.readyState = ReadyStates.Done;
-    ObservableWrapper.callThrow(this.response, err);
+    this.response.error(err);
-    ObservableWrapper.callReturn(this.response);
   }
 }
 
@@ -162,7 +153,7 @@ export class MockBackend implements ConnectionBackend {
    *
    * This property only exists in the mock implementation, not in real Backends.
    */
-  connections: EventEmitter;  //<MockConnection>
+  connections: any;  //<MockConnection>
 
   /**
    * An array representation of `connections`. This array will be updated with each connection that
@@ -179,13 +170,12 @@ export class MockBackend implements ConnectionBackend {
    *
    * This property only exists in the mock implementation, not in real Backends.
    */
-  pendingConnections: EventEmitter;  //<MockConnection>
+  pendingConnections: any;  // Subject<MockConnection>
   constructor() {
     this.connectionsArray = [];
-    this.connections = new EventEmitter();
+    this.connections = new Subject();
-    ObservableWrapper.subscribe<MockConnection>(
+    this.connections.subscribe(connection => this.connectionsArray.push(connection));
-        this.connections, connection => this.connectionsArray.push(connection));
+    this.pendingConnections = new Subject();
-    this.pendingConnections = new EventEmitter();
   }
 
   /**
@@ -195,7 +185,7 @@ export class MockBackend implements ConnectionBackend {
    */
   verifyNoPendingRequests() {
     let pending = 0;
-    ObservableWrapper.subscribe(this.pendingConnections, c => pending++);
+    this.pendingConnections.subscribe(c => pending++);
     if (pending > 0) throw new BaseException(`${pending} pending connections to be resolved`);
   }
 
@@ -205,9 +195,7 @@ export class MockBackend implements ConnectionBackend {
    *
    * This method only exists in the mock implementation, not in real Backends.
    */
-  resolveAllConnections() {
+  resolveAllConnections() { this.connections.subscribe(c => c.readyState = 4); }
-    ObservableWrapper.subscribe<MockConnection>(this.connections, c => c.readyState = 4);
-  }
 
   /**
    * Creates a new {@link MockConnection}. This is equivalent to calling `new
@@ -220,7 +208,7 @@ export class MockBackend implements ConnectionBackend {
       throw new BaseException(`createConnection requires an instance of Request, got ${req}`);
     }
     let connection = new MockConnection(req);
-    ObservableWrapper.callNext(this.connections, connection);
+    this.connections.next(connection);
     return connection;
   }
 }

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

@@ -6,7 +6,9 @@ import {ResponseOptions, BaseResponseOptions} from '../base_response_options';
 import {Injectable} from 'angular2/src/core/di';
 import {BrowserXhr} from './browser_xhr';
 import {isPresent} from 'angular2/src/core/facade/lang';
-var Observable = require('@reactivex/rxjs/dist/cjs/Observable');
+// todo(robwormald): temporary until https://github.com/angular/angular/issues/4390 decided
+var Rx = require('@reactivex/rxjs/dist/cjs/Rx');
+var {Observable} = Rx;
 /**
 * Creates connections using `XMLHttpRequest`. Given a fully-qualified
 * request, an `XHRConnection` will immediately create an `XMLHttpRequest` object and send the

modules/angular2/src/http/http.ts

@@ -5,9 +5,8 @@ import {RequestOptionsArgs, Connection, ConnectionBackend} from './interfaces';
 import {Request} from './static_request';
 import {BaseRequestOptions, RequestOptions} from './base_request_options';
 import {RequestMethods} from './enums';
-import {EventEmitter} from 'angular2/src/core/facade/async';
 
-function httpRequest(backend: ConnectionBackend, request: Request): EventEmitter {
+function httpRequest(backend: ConnectionBackend, request: Request): any {
   return backend.createConnection(request).response;
 }
 
@@ -94,8 +93,8 @@ export class Http {
    * object can be provided as the 2nd argument. The options object will be merged with the values
    * of {@link BaseRequestOptions} before performing the request.
    */
-  request(url: string | Request, options?: RequestOptionsArgs): EventEmitter {
+  request(url: string | Request, options?: RequestOptionsArgs): any {
-    var responseObservable: EventEmitter;
+    var responseObservable: any;
     if (isString(url)) {
       responseObservable = httpRequest(
           this._backend,
@@ -111,7 +110,7 @@ export class Http {
   /**
    * Performs a request with `get` http method.
    */
-  get(url: string, options?: RequestOptionsArgs): EventEmitter {
+  get(url: string, options?: RequestOptionsArgs): any {
     return httpRequest(this._backend, new Request(mergeOptions(this._defaultOptions, options,
                                                                RequestMethods.Get, url)));
   }
@@ -119,7 +118,7 @@ export class Http {
   /**
    * Performs a request with `post` http method.
    */
-  post(url: string, body: string, options?: RequestOptionsArgs): EventEmitter {
+  post(url: string, body: string, options?: RequestOptionsArgs): any {
     return httpRequest(
         this._backend,
         new Request(mergeOptions(this._defaultOptions.merge(new RequestOptions({body: body})),
@@ -129,7 +128,7 @@ export class Http {
   /**
    * Performs a request with `put` http method.
    */
-  put(url: string, body: string, options?: RequestOptionsArgs): EventEmitter {
+  put(url: string, body: string, options?: RequestOptionsArgs): any {
     return httpRequest(
         this._backend,
         new Request(mergeOptions(this._defaultOptions.merge(new RequestOptions({body: body})),
@@ -139,7 +138,7 @@ export class Http {
   /**
    * Performs a request with `delete` http method.
    */
-  delete (url: string, options?: RequestOptionsArgs): EventEmitter {
+  delete (url: string, options?: RequestOptionsArgs): any {
     return httpRequest(this._backend, new Request(mergeOptions(this._defaultOptions, options,
                                                                RequestMethods.Delete, url)));
   }
@@ -147,7 +146,7 @@ export class Http {
   /**
    * Performs a request with `patch` http method.
    */
-  patch(url: string, body: string, options?: RequestOptionsArgs): EventEmitter {
+  patch(url: string, body: string, options?: RequestOptionsArgs): any {
     return httpRequest(
         this._backend,
         new Request(mergeOptions(this._defaultOptions.merge(new RequestOptions({body: body})),
@@ -157,7 +156,7 @@ export class Http {
   /**
    * Performs a request with `head` http method.
    */
-  head(url: string, options?: RequestOptionsArgs): EventEmitter {
+  head(url: string, options?: RequestOptionsArgs): any {
     return httpRequest(this._backend, new Request(mergeOptions(this._defaultOptions, options,
                                                                RequestMethods.Head, url)));
   }
@@ -175,8 +174,8 @@ export class Jsonp extends Http {
    * object can be provided as the 2nd argument. The options object will be merged with the values
    * of {@link BaseRequestOptions} before performing the request.
    */
-  request(url: string | Request, options?: RequestOptionsArgs): EventEmitter {
+  request(url: string | Request, options?: RequestOptionsArgs): any {
-    var responseObservable: EventEmitter;
+    var responseObservable: any;
     if (isString(url)) {
       url = new Request(mergeOptions(this._defaultOptions, options, RequestMethods.Get, url));
     }

modules/angular2/src/http/interfaces.ts

@@ -22,7 +22,7 @@ export abstract class ConnectionBackend {
 export abstract class Connection {
   readyState: ReadyStates;
   request: Request;
-  response: EventEmitter;  // TODO: generic of <Response>;
+  response: any;  // TODO: generic of <Response>;
 }
 
 /**

modules/angular2/test/http/backends/mock_backend_spec.ts

@@ -0,0 +1,120 @@
+import {
+  AsyncTestCompleter,
+  afterEach,
+  beforeEach,
+  ddescribe,
+  describe,
+  expect,
+  iit,
+  inject,
+  it,
+  xit,
+  SpyObject
+} from 'angular2/test_lib';
+import {ObservableWrapper} from 'angular2/src/core/facade/async';
+import {BrowserXhr} from 'angular2/src/http/backends/browser_xhr';
+import {MockConnection, MockBackend} from 'angular2/src/http/backends/mock_backend';
+import {bind, Injector} from 'angular2/core';
+import {Request} from 'angular2/src/http/static_request';
+import {Response} from 'angular2/src/http/static_response';
+import {Headers} from 'angular2/src/http/headers';
+import {Map} from 'angular2/src/core/facade/collection';
+import {RequestOptions, BaseRequestOptions} from 'angular2/src/http/base_request_options';
+import {BaseResponseOptions, ResponseOptions} from 'angular2/src/http/base_response_options';
+import {ResponseTypes} from 'angular2/src/http/enums';
+
+export function main() {
+  describe('MockBackend', () => {
+
+    var backend;
+    var sampleRequest1;
+    var sampleResponse1;
+    var sampleRequest2;
+    var sampleResponse2;
+    var connection;
+
+    beforeEach(() => {
+      var injector = Injector.resolveAndCreate(
+          [bind(ResponseOptions).toClass(BaseResponseOptions), MockBackend]);
+      backend = injector.get(MockBackend);
+      var base = new BaseRequestOptions();
+      sampleRequest1 = new Request(base.merge(new RequestOptions({url: 'https://google.com'})));
+      sampleResponse1 = new Response(new ResponseOptions({body: 'response1'}));
+      sampleRequest2 = new Request(base.merge(new RequestOptions({url: 'https://google.com'})));
+      sampleResponse2 = new Response(new ResponseOptions({body: 'response2'}));
+    });
+
+    it('should create a new MockBackend', () => {expect(backend).toBeAnInstanceOf(MockBackend)});
+
+    it('should create a new MockConnection',
+       () => {expect(backend.createConnection(sampleRequest1)).toBeAnInstanceOf(MockConnection)});
+
+    it('should create a new connection and allow subscription', () => {
+      let connection = backend.createConnection(sampleRequest1);
+      connection.response.subscribe(() => {});
+    });
+
+    it('should allow responding after subscription', inject([AsyncTestCompleter], async => {
+         let connection = backend.createConnection(sampleRequest1);
+         connection.response.subscribe((res) => { async.done(); });
+         connection.mockRespond(sampleResponse1);
+       }));
+
+    it('should allow subscribing after responding', inject([AsyncTestCompleter], async => {
+         let connection = backend.createConnection(sampleRequest1);
+         connection.mockRespond(sampleResponse1);
+         connection.response.subscribe((res) => { async.done(); });
+       }));
+
+    it('should allow responding after subscription with an error',
+       inject([AsyncTestCompleter], async => {
+         let connection = backend.createConnection(sampleRequest1);
+         connection.response.subscribe(null, () => { async.done(); });
+         connection.mockError(new Error('nope'));
+       }));
+
+    it('should not throw when there are no unresolved requests',
+       inject([AsyncTestCompleter], async => {
+         let connection = backend.createConnection(sampleRequest1);
+         connection.response.subscribe(() => { async.done(); });
+         connection.mockRespond(sampleResponse1);
+         backend.verifyNoPendingRequests();
+       }));
+
+    xit('should throw when there are unresolved requests', inject([AsyncTestCompleter], async => {
+          let connection = backend.createConnection(sampleRequest1);
+          connection.response.subscribe(() => { async.done(); });
+          backend.verifyNoPendingRequests();
+        }));
+
+    it('should work when requests are resolved out of order',
+       inject([AsyncTestCompleter], async => {
+         let connection1 = backend.createConnection(sampleRequest1);
+         let connection2 = backend.createConnection(sampleRequest1);
+         connection1.response.subscribe(() => { async.done(); });
+         connection2.response.subscribe(() => {});
+         connection2.mockRespond(sampleResponse1);
+         connection1.mockRespond(sampleResponse1);
+         backend.verifyNoPendingRequests();
+       }));
+
+    xit('should allow double subscribing', inject([AsyncTestCompleter], async => {
+          let responses = [sampleResponse1, sampleResponse2];
+          backend.connections.subscribe(c => c.mockRespond(responses.shift()));
+          let responseObservable = backend.createConnection(sampleRequest1).response;
+          responseObservable.subscribe(res => expect(res.text()).toBe('response1'));
+          responseObservable.subscribe(res => expect(res.text()).toBe('response2'), null,
+                                       async.done);
+        }));
+
+    // TODO(robwormald): readyStates are leaving?
+    it('should allow resolution of requests manually', () => {
+      let connection1: MockConnection = backend.createConnection(sampleRequest1);
+      let connection2: MockConnection = backend.createConnection(sampleRequest1);
+      connection1.response.subscribe(() => {});
+      connection2.response.subscribe(() => {});
+      backend.resolveAllConnections();
+      backend.verifyNoPendingRequests();
+    });
+  });
+}

modules/angular2/test/http/http_spec.ts

@@ -13,7 +13,6 @@ import {
 } from 'angular2/test_lib';
 import {Injector, bind} from 'angular2/core';
 import {MockBackend, MockConnection} from 'angular2/src/http/backends/mock_backend';
-import {EventEmitter, ObservableWrapper} from 'angular2/src/core/facade/async';
 import {
   BaseRequestOptions,
   ConnectionBackend,
@@ -31,6 +30,9 @@ import {
   Jsonp
 } from 'angular2/http';
 
+var Rx = require('@reactivex/rxjs/dist/cjs/Rx');
+let{Observable, Subject} = Rx;
+
 class SpyObserver extends SpyObject {
   onNext: Function;
   onError: Function;
@@ -75,7 +77,7 @@ export function main() {
          var jsonpCreatedConnections = 0;
 
 
-         ObservableWrapper.subscribe(xhrBackend.connections, () => {
+         xhrBackend.connections.subscribe(() => {
            xhrCreatedConnections++;
            expect(xhrCreatedConnections).toEqual(1);
            if (jsonpCreatedConnections) {
@@ -83,9 +85,9 @@ export function main() {
            }
          });
 
-         ObservableWrapper.subscribe(http.get(url), () => {});
+         http.get(url).subscribe(() => {});
 
-         ObservableWrapper.subscribe(jsonpBackend.connections, () => {
+         jsonpBackend.connections.subscribe(() => {
            jsonpCreatedConnections++;
            expect(jsonpCreatedConnections).toEqual(1);
            if (xhrCreatedConnections) {
@@ -93,7 +95,7 @@ export function main() {
            }
          });
 
-         ObservableWrapper.subscribe(jsonp.request(url), () => {});
+         jsonp.request(url).subscribe(() => {});
        }));
   });
 
@@ -130,32 +132,49 @@ export function main() {
     describe('Http', () => {
       describe('.request()', () => {
         it('should return an Observable',
-           () => { expect(ObservableWrapper.isObservable(http.request(url))).toBe(true); });
+           () => { expect(http.request(url)).toBeAnInstanceOf(Observable); });
 
 
         it('should accept a fully-qualified request as its only parameter',
            inject([AsyncTestCompleter], (async) => {
-             ObservableWrapper.subscribe<MockConnection>(backend.connections, c => {
+             backend.connections.subscribe(c => {
                expect(c.request.url).toBe('https://google.com');
                c.mockRespond(new Response(new ResponseOptions({body: 'Thank you'})));
                async.done();
              });
-             ObservableWrapper.subscribe(
+             http.request(new Request(new RequestOptions({url: 'https://google.com'})))
-                 http.request(new Request(new RequestOptions({url: 'https://google.com'}))),
+                 .subscribe((res) => {});
-                 (res) => {});
            }));
 
 
         it('should perform a get request for given url if only passed a string',
            inject([AsyncTestCompleter], (async) => {
-             ObservableWrapper.subscribe<MockConnection>(backend.connections,
+             backend.connections.subscribe(c => c.mockRespond(baseResponse));
-                                                         c => c.mockRespond(baseResponse));
+             http.request('http://basic.connection')
-             ObservableWrapper.subscribe<Response>(http.request('http://basic.connection'), res => {
+                 .subscribe(res => {
-               expect(res.text()).toBe('base response');
+                   expect(res.text()).toBe('base response');
-               async.done();
+                   async.done();
-             });
+                 });
            }));
 
+        it('should perform a get request and complete the response',
+           inject([AsyncTestCompleter], (async) => {
+             backend.connections.subscribe(c => c.mockRespond(baseResponse));
+             http.request('http://basic.connection')
+                 .subscribe(res => { expect(res.text()).toBe('base response'); }, null,
+                            () => { async.done(); });
+           }));
+
+        it('should perform multiple get requests and complete the responses',
+           inject([AsyncTestCompleter], (async) => {
+             backend.connections.subscribe(c => c.mockRespond(baseResponse));
+
+             http.request('http://basic.connection')
+                 .subscribe(res => { expect(res.text()).toBe('base response'); });
+             http.request('http://basic.connection')
+                 .subscribe(res => { expect(res.text()).toBe('base response'); }, null,
+                            () => { async.done(); });
+           }));
         // TODO: make dart not complain about "argument type 'Map' cannot be assigned to the
         // parameter type 'IRequestOptions'"
         // xit('should perform a get request for given url if passed a dictionary',
@@ -179,103 +198,103 @@ export function main() {
 
       describe('.get()', () => {
         it('should perform a get request for given url', inject([AsyncTestCompleter], async => {
-             ObservableWrapper.subscribe<MockConnection>(backend.connections, c => {
+             backend.connections.subscribe(c => {
                expect(c.request.method).toBe(RequestMethods.Get);
                backend.resolveAllConnections();
                async.done();
              });
-             ObservableWrapper.subscribe(http.get(url), res => {});
+             http.get(url).subscribe(res => {});
            }));
       });
 
 
       describe('.post()', () => {
         it('should perform a post request for given url', inject([AsyncTestCompleter], async => {
-             ObservableWrapper.subscribe<MockConnection>(backend.connections, c => {
+             backend.connections.subscribe(c => {
                expect(c.request.method).toBe(RequestMethods.Post);
                backend.resolveAllConnections();
                async.done();
              });
-             ObservableWrapper.subscribe(http.post(url, 'post me'), res => {});
+             http.post(url, 'post me').subscribe(res => {});
            }));
 
 
         it('should attach the provided body to the request', inject([AsyncTestCompleter], async => {
              var body = 'this is my post body';
-             ObservableWrapper.subscribe<MockConnection>(backend.connections, c => {
+             backend.connections.subscribe(c => {
                expect(c.request.text()).toBe(body);
                backend.resolveAllConnections();
                async.done();
              });
-             ObservableWrapper.subscribe(http.post(url, body), res => {});
+             http.post(url, body).subscribe(res => {});
            }));
       });
 
 
       describe('.put()', () => {
         it('should perform a put request for given url', inject([AsyncTestCompleter], async => {
-             ObservableWrapper.subscribe<MockConnection>(backend.connections, c => {
+             backend.connections.subscribe(c => {
                expect(c.request.method).toBe(RequestMethods.Put);
                backend.resolveAllConnections();
                async.done();
              });
-             ObservableWrapper.subscribe(http.put(url, 'put me'), res => {});
+             http.put(url, 'put me').subscribe(res => {});
            }));
 
         it('should attach the provided body to the request', inject([AsyncTestCompleter], async => {
              var body = 'this is my put body';
-             ObservableWrapper.subscribe<MockConnection>(backend.connections, c => {
+             backend.connections.subscribe(c => {
                expect(c.request.text()).toBe(body);
                backend.resolveAllConnections();
                async.done();
              });
-             ObservableWrapper.subscribe(http.put(url, body), res => {});
+             http.put(url, body).subscribe(res => {});
            }));
       });
 
 
       describe('.delete()', () => {
         it('should perform a delete request for given url', inject([AsyncTestCompleter], async => {
-             ObservableWrapper.subscribe<MockConnection>(backend.connections, c => {
+             backend.connections.subscribe(c => {
                expect(c.request.method).toBe(RequestMethods.Delete);
                backend.resolveAllConnections();
                async.done();
              });
-             ObservableWrapper.subscribe(http.delete(url), res => {});
+             http.delete(url).subscribe(res => {});
            }));
       });
 
 
       describe('.patch()', () => {
         it('should perform a patch request for given url', inject([AsyncTestCompleter], async => {
-             ObservableWrapper.subscribe<MockConnection>(backend.connections, c => {
+             backend.connections.subscribe(c => {
                expect(c.request.method).toBe(RequestMethods.Patch);
                backend.resolveAllConnections();
                async.done();
              });
-             ObservableWrapper.subscribe(http.patch(url, 'this is my patch body'), res => {});
+             http.patch(url, 'this is my patch body').subscribe(res => {});
            }));
 
         it('should attach the provided body to the request', inject([AsyncTestCompleter], async => {
              var body = 'this is my patch body';
-             ObservableWrapper.subscribe<MockConnection>(backend.connections, c => {
+             backend.connections.subscribe(c => {
                expect(c.request.text()).toBe(body);
                backend.resolveAllConnections();
                async.done();
              });
-             ObservableWrapper.subscribe(http.patch(url, body), res => {});
+             http.patch(url, body).subscribe(res => {});
            }));
       });
 
 
       describe('.head()', () => {
         it('should perform a head request for given url', inject([AsyncTestCompleter], async => {
-             ObservableWrapper.subscribe<MockConnection>(backend.connections, c => {
+             backend.connections.subscribe(c => {
                expect(c.request.method).toBe(RequestMethods.Head);
                backend.resolveAllConnections();
                async.done();
              });
-             ObservableWrapper.subscribe(http.head(url), res => {});
+             http.head(url).subscribe(res => {});
            }));
       });
 
@@ -284,54 +303,51 @@ export function main() {
         it('should append search params to url', inject([AsyncTestCompleter], async => {
              var params = new URLSearchParams();
              params.append('q', 'puppies');
-             ObservableWrapper.subscribe<MockConnection>(backend.connections, c => {
+             backend.connections.subscribe(c => {
                expect(c.request.url).toEqual('https://www.google.com?q=puppies');
                backend.resolveAllConnections();
                async.done();
              });
-             ObservableWrapper.subscribe(
+             http.get('https://www.google.com', new RequestOptions({search: params}))
-                 http.get('https://www.google.com', new RequestOptions({search: params})),
+                 .subscribe(res => {});
-                 res => {});
            }));
 
 
         it('should append string search params to url', inject([AsyncTestCompleter], async => {
-             ObservableWrapper.subscribe<MockConnection>(backend.connections, c => {
+             backend.connections.subscribe(c => {
                expect(c.request.url).toEqual('https://www.google.com?q=piggies');
                backend.resolveAllConnections();
                async.done();
              });
-             ObservableWrapper.subscribe(
+             http.get('https://www.google.com', new RequestOptions({search: 'q=piggies'}))
-                 http.get('https://www.google.com', new RequestOptions({search: 'q=piggies'})),
+                 .subscribe(res => {});
-                 res => {});
            }));
 
 
         it('should produce valid url when url already contains a query',
            inject([AsyncTestCompleter], async => {
-             ObservableWrapper.subscribe<MockConnection>(backend.connections, c => {
+             backend.connections.subscribe(c => {
                expect(c.request.url).toEqual('https://www.google.com?q=angular&as_eq=1.x');
                backend.resolveAllConnections();
                async.done();
              });
-             ObservableWrapper.subscribe(http.get('https://www.google.com?q=angular',
+             http.get('https://www.google.com?q=angular', new RequestOptions({search: 'as_eq=1.x'}))
-                                                  new RequestOptions({search: 'as_eq=1.x'})),
+                 .subscribe(res => {});
-                                         res => {});
            }));
       });
 
       describe('string method names', () => {
         it('should allow case insensitive strings for method names', () => {
           inject([AsyncTestCompleter], (async) => {
-            ObservableWrapper.subscribe<MockConnection>(backend.connections, c => {
+            backend.connections.subscribe(c => {
               expect(c.request.method)
                   .toBe(RequestMethods.Post)
                       c.mockRespond(new Response(new ResponseOptions({body: 'Thank you'})));
               async.done();
             });
-            ObservableWrapper.subscribe(http.request(new Request(new RequestOptions(
+            http.request(
-                                            {url: 'https://google.com', method: 'PosT'}))),
+                    new Request(new RequestOptions({url: 'https://google.com', method: 'PosT'})))
-                                        (res) => {});
+                .subscribe((res) => {});
           });
         });
 

modules/examples/src/http/http_comp.ts

@@ -15,5 +15,7 @@ import {Http} from 'angular2/http';
 })
 export class HttpCmp {
   people: Object;
-  constructor(http: Http) { http.get('./people.json').subscribe(res => this.people = res.json()); }
+  constructor(http: Http) {
+    http.get('./people.json').map(res => res.json()).subscribe(res => this.people = res);
+  }
 }