diff --git a/gulpfile.js b/gulpfile.js
index fef8bbb50244..8df33c0ff725 100644
--- a/gulpfile.js
+++ b/gulpfile.js
@@ -787,7 +787,7 @@ gulp.task('static-checks', ['!build.tools'], function(done) {
 var tmpdir = path.join(os.tmpdir(), 'test.typings',  new Date().getTime().toString());
 gulp.task('!pre.test.typings.layoutNodeModule', ['build.js.cjs'], function() {
   return gulp
-    .src(['dist/js/cjs/angular2/**/*'], {base: 'dist/js/cjs'})
+    .src(['dist/js/cjs/angular2/**/*', 'node_modules/@reactivex/rxjs/dist/cjs/**'], {base: 'dist/js/cjs'})
     .pipe(gulp.dest(path.join(tmpdir, 'node_modules')));
 });
 gulp.task('!pre.test.typings.copyTypingsSpec', function() {
@@ -803,6 +803,7 @@ gulp.task('test.typings', [
     .pipe(tsc({target: 'ES5', module: 'commonjs',
       experimentalDecorators: true,
       noImplicitAny: true,
+      moduleResolution: 'node',
       typescript: require('typescript')}));
 });
 
diff --git a/modules/angular2/src/core/facade.ts b/modules/angular2/src/core/facade.ts
index b6918338d9f4..cf04c74f7de8 100644
--- a/modules/angular2/src/core/facade.ts
+++ b/modules/angular2/src/core/facade.ts
@@ -1,5 +1,5 @@
 // Public API for Facade
 export {ConcreteType, Type} from './facade/lang';
-export {Observable, EventEmitter} from './facade/async';
+export {Observable, EventEmitter, Subject} from './facade/async';
 export {Predicate} from './facade/collection';
 export {WrappedException} from './facade/exceptions';
diff --git a/modules/angular2/src/core/facade/async.dart b/modules/angular2/src/core/facade/async.dart
index bb7d7fdab4d7..7d95e8911098 100644
--- a/modules/angular2/src/core/facade/async.dart
+++ b/modules/angular2/src/core/facade/async.dart
@@ -50,16 +50,16 @@ class ObservableWrapper {
     emitter.add(value);
   }
 
-  static void callThrow(EventEmitter emitter, error) {
+  static void callError(EventEmitter emitter, error) {
     emitter.addError(error);
   }
 
-  static void callReturn(EventEmitter emitter) {
+  static void callComplete(EventEmitter emitter) {
     emitter.close();
   }
 }
 
-class EventEmitter extends Stream {
+class EventEmitter<T> extends Stream<T> {
   StreamController<dynamic> _controller;
 
   /// Creates an instance of [EventEmitter], which depending on [isAsync],
@@ -86,3 +86,30 @@ class EventEmitter extends Stream {
     _controller.close();
   }
 }
+
+//todo(robwormald): maybe fix in ts2dart?
+class Subject<T> extends Stream<T> {
+  StreamController<dynamic> _controller;
+
+  Subject([bool isAsync = true]) {
+    _controller = new StreamController.broadcast(sync: !isAsync);
+  }
+
+  StreamSubscription listen(void onData(dynamic line),
+      {void onError(Error error), void onDone(), bool cancelOnError}) {
+    return _controller.stream.listen(onData,
+        onError: onError, onDone: onDone, cancelOnError: cancelOnError);
+  }
+
+  void add(value) {
+    _controller.add(value);
+  }
+
+  void addError(error) {
+    _controller.addError(error);
+  }
+
+  void close() {
+    _controller.close();
+  }
+}
diff --git a/modules/angular2/src/core/facade/async.ts b/modules/angular2/src/core/facade/async.ts
index 6107ee520a8a..a82776094e37 100644
--- a/modules/angular2/src/core/facade/async.ts
+++ b/modules/angular2/src/core/facade/async.ts
@@ -3,8 +3,9 @@ import {global, isPresent} from 'angular2/src/core/facade/lang';
 // without depending on rxjs.
 import {PromiseWrapper, Promise, PromiseCompleter} from 'angular2/src/core/facade/promise';
 export {PromiseWrapper, Promise, PromiseCompleter} from 'angular2/src/core/facade/promise';
-// TODO(jeffbcross): use ES6 import once typings are available
-var Subject = require('@reactivex/rxjs/dist/cjs/Subject');
+import {Subject, Subscription, Observable as RxObservable} from '@reactivex/rxjs/dist/cjs/Rx';
+export {Subject} from '@reactivex/rxjs/dist/cjs/Rx';
+import Operator from '@reactivex/rxjs/dist/cjs/Operator';
 
 export namespace NodeJS {
   export interface Timer {}
@@ -24,31 +25,25 @@ export class TimerWrapper {
 
 export class ObservableWrapper {
   // TODO(vsavkin): when we use rxnext, try inferring the generic type from the first arg
-  static subscribe<T>(emitter: Observable, onNext: (value: T) => void,
-                      onThrow: (exception: any) => void = null,
-                      onReturn: () => void = null): Object {
-    return emitter.observer({next: onNext, throw: onThrow, return: onReturn});
+  static subscribe<T>(emitter: any, onNext: (value: T) => void, onError?: (exception: any) => void,
+                      onComplete?: () => void): Object {
+    return emitter.subscribe({next: onNext, error: onError, complete: onComplete});
   }
 
-  static isObservable(obs: any): boolean { return obs instanceof Observable; }
+  static isObservable(obs: any): boolean { return obs instanceof RxObservable; }
 
   /**
    * Returns whether `obs` has any subscribers listening to events.
    */
-  static hasSubscribers(obs: EventEmitter): boolean { return obs._subject.observers.length > 0; }
+  static hasSubscribers(obs: EventEmitter<any>): boolean { return obs.observers.length > 0; }
 
   static dispose(subscription: any) { subscription.unsubscribe(); }
 
-  static callNext(emitter: EventEmitter, value: any) { emitter.next(value); }
+  static callNext(emitter: EventEmitter<any>, value: any) { emitter.next(value); }
 
-  static callThrow(emitter: EventEmitter, error: any) { emitter.throw(error); }
+  static callError(emitter: EventEmitter<any>, error: any) { emitter.error(error); }
 
-  static callReturn(emitter: EventEmitter) { emitter.return (null); }
-}
-
-// TODO: vsavkin change to interface
-export class Observable {
-  observer(generator: any): Object { return null; }
+  static callComplete(emitter: EventEmitter<any>) { emitter.complete(); }
 }
 
 /**
@@ -90,9 +85,7 @@ export class Observable {
  *
  * Once a reference implementation of the spec is available, switch to it.
  */
-export class EventEmitter extends Observable {
-  /** @internal */
-  _subject = new Subject();
+export class EventEmitter<T> extends Subject<T> {
   /** @internal */
   _isAsync: boolean;
 
@@ -105,19 +98,30 @@ export class EventEmitter extends Observable {
     this._isAsync = isAsync;
   }
 
-  observer(generator: any): any {
-    var schedulerFn = this._isAsync ? (value) => { setTimeout(() => generator.next(value)); } :
-                                      (value) => { generator.next(value); };
-    return this._subject.subscribe(schedulerFn,
-                                   (error) => generator.throw ? generator.throw(error) : null,
-                                   () => generator.return ? generator.return () : null);
+  subscribe(generatorOrNext?: any, error?: any, complete?: any): any {
+    if (generatorOrNext && typeof generatorOrNext === 'object') {
+      let schedulerFn = this._isAsync ?
+                            (value) => { setTimeout(() => generatorOrNext.next(value)); } :
+                            (value) => { generatorOrNext.next(value); };
+      return super.subscribe(schedulerFn,
+                             (err) => generatorOrNext.error ? generatorOrNext.error(err) : null,
+                             () => generatorOrNext.complete ? generatorOrNext.complete() : null);
+    } else {
+      let schedulerFn = this._isAsync ? (value) => { setTimeout(() => generatorOrNext(value)); } :
+                                        (value) => { generatorOrNext(value); };
+
+      return super.subscribe(schedulerFn, (err) => error ? error(err) : null,
+                             () => complete ? complete() : null);
+    }
   }
+}
 
-  toRx(): any { return this._subject; }
-
-  next(value: any) { this._subject.next(value); }
-
-  throw(error: any) { this._subject.error(error); }
-
-  return (value?: any) { this._subject.complete(); }
+// todo(robwormald): ts2dart should handle this properly
+export class Observable<T> extends RxObservable<T> {
+  lift<T, R>(operator: Operator<T, R>): Observable<T> {
+    const observable = new Observable();
+    observable.source = this;
+    observable.operator = operator;
+    return observable;
+  }
 }
diff --git a/modules/angular2/src/core/forms/model.ts b/modules/angular2/src/core/forms/model.ts
index 23a7b7a1d3e6..6023c918f3cf 100644
--- a/modules/angular2/src/core/forms/model.ts
+++ b/modules/angular2/src/core/forms/model.ts
@@ -51,7 +51,7 @@ export abstract class AbstractControl {
   _value: any;
 
   /** @internal */
-  _valueChanges: EventEmitter;
+  _valueChanges: EventEmitter<any>;
 
   private _status: string;
   private _errors: {[key: string]: any};
@@ -86,7 +86,8 @@ export abstract class AbstractControl {
 
   get untouched(): boolean { return !this._touched; }
 
-  get valueChanges(): Observable { return this._valueChanges; }
+  get valueChanges(): Observable<any> { return this._valueChanges; }
+
   get pending(): boolean { return this._status == PENDING; }
 
   markAsTouched(): void { this._touched = true; }
diff --git a/modules/angular2/src/core/linker/query_list.ts b/modules/angular2/src/core/linker/query_list.ts
index cc80d2c8bc15..9f96f761e91d 100644
--- a/modules/angular2/src/core/linker/query_list.ts
+++ b/modules/angular2/src/core/linker/query_list.ts
@@ -31,7 +31,7 @@ export class QueryList<T> {
   private _results: Array<T> = [];
   private _emitter = new EventEmitter();
 
-  get changes(): Observable { return this._emitter; }
+  get changes(): Observable<any> { return this._emitter; }
   get length(): number { return this._results.length; }
   get first(): T { return ListWrapper.first(this._results); }
   get last(): T { return ListWrapper.last(this._results); }
diff --git a/modules/angular2/src/core/pipes/async_pipe.ts b/modules/angular2/src/core/pipes/async_pipe.ts
index 6ec7bda5cb12..cf38f43b32ec 100644
--- a/modules/angular2/src/core/pipes/async_pipe.ts
+++ b/modules/angular2/src/core/pipes/async_pipe.ts
@@ -1,5 +1,5 @@
 import {isBlank, isPresent, isPromise, CONST} from 'angular2/src/core/facade/lang';
-import {Promise, ObservableWrapper, Observable} from 'angular2/src/core/facade/async';
+import {Promise, ObservableWrapper, Observable, EventEmitter} from 'angular2/src/core/facade/async';
 import {Pipe} from 'angular2/src/core/metadata';
 import {Injectable} from 'angular2/src/core/di';
 import {
@@ -69,7 +69,7 @@ export class AsyncPipe implements PipeTransform, PipeOnDestroy {
   /** @internal */
   _subscription: Object = null;
   /** @internal */
-  _obj: Observable | Promise<any> = null;
+  _obj: Observable<any>| Promise<any>| EventEmitter<any> = null;
   private _strategy: any = null;
   /** @internal */
   public _ref: ChangeDetectorRef;
@@ -81,7 +81,7 @@ export class AsyncPipe implements PipeTransform, PipeOnDestroy {
     }
   }
 
-  transform(obj: Observable | Promise<any>, args?: any[]): any {
+  transform(obj: Observable<any>| Promise<any>| EventEmitter<any>, args?: any[]): any {
     if (isBlank(this._obj)) {
       if (isPresent(obj)) {
         this._subscribe(obj);
@@ -103,7 +103,7 @@ export class AsyncPipe implements PipeTransform, PipeOnDestroy {
   }
 
   /** @internal */
-  _subscribe(obj: Observable | Promise<any>): void {
+  _subscribe(obj: Observable<any>| Promise<any>| EventEmitter<any>): void {
     this._obj = obj;
     this._strategy = this._selectStrategy(obj);
     this._subscription =
@@ -111,7 +111,7 @@ export class AsyncPipe implements PipeTransform, PipeOnDestroy {
   }
 
   /** @internal */
-  _selectStrategy(obj: Observable | Promise<any>): any {
+  _selectStrategy(obj: Observable<any>| Promise<any>| EventEmitter<any>): any {
     if (isPromise(obj)) {
       return _promiseStrategy;
     } else if (ObservableWrapper.isObservable(obj)) {
diff --git a/modules/angular2/src/core/zone/ng_zone.ts b/modules/angular2/src/core/zone/ng_zone.ts
index 2ed00380721c..b75026f3f6d1 100644
--- a/modules/angular2/src/core/zone/ng_zone.ts
+++ b/modules/angular2/src/core/zone/ng_zone.ts
@@ -113,13 +113,13 @@ export class NgZone {
   _onErrorHandler: ErrorHandlingFn;
 
   /** @internal */
-  _onTurnStartEvents: EventEmitter;
+  _onTurnStartEvents: EventEmitter<any>;
   /** @internal */
-  _onTurnDoneEvents: EventEmitter;
+  _onTurnDoneEvents: EventEmitter<any>;
   /** @internal */
-  _onEventDoneEvents: EventEmitter;
+  _onEventDoneEvents: EventEmitter<any>;
   /** @internal */
-  _onErrorEvents: EventEmitter;
+  _onErrorEvents: EventEmitter<any>;
 
   // Number of microtasks pending from _innerZone (& descendants)
   /** @internal */
diff --git a/modules/angular2/src/mock/location_mock.ts b/modules/angular2/src/mock/location_mock.ts
index f9365c07ec02..28defeebbe15 100644
--- a/modules/angular2/src/mock/location_mock.ts
+++ b/modules/angular2/src/mock/location_mock.ts
@@ -9,7 +9,7 @@ export class SpyLocation implements Location {
   /** @internal */
   _query: string = '';
   /** @internal */
-  _subject: EventEmitter = new EventEmitter();
+  _subject: EventEmitter<any> = new EventEmitter();
   /** @internal */
   _baseHref: string = '';
 
diff --git a/modules/angular2/src/mock/mock_location_strategy.ts b/modules/angular2/src/mock/mock_location_strategy.ts
index c18459b66781..b52c99ff212a 100644
--- a/modules/angular2/src/mock/mock_location_strategy.ts
+++ b/modules/angular2/src/mock/mock_location_strategy.ts
@@ -8,7 +8,7 @@ export class MockLocationStrategy extends LocationStrategy {
   internalTitle: string = '';
   urlChanges: string[] = [];
   /** @internal */
-  _subject: EventEmitter = new EventEmitter();
+  _subject: EventEmitter<any> = new EventEmitter();
   constructor() { super(); }
 
   simulatePopState(url: string): void {
diff --git a/modules/angular2/src/router/location.ts b/modules/angular2/src/router/location.ts
index 7ccab928c161..00860f221789 100644
--- a/modules/angular2/src/router/location.ts
+++ b/modules/angular2/src/router/location.ts
@@ -79,7 +79,7 @@ export const APP_BASE_HREF: OpaqueToken = CONST_EXPR(new OpaqueToken('appBaseHre
 @Injectable()
 export class Location {
   /** @internal */
-  _subject: EventEmitter = new EventEmitter();
+  _subject: EventEmitter<any> = new EventEmitter();
   /** @internal */
   _baseHref: string;
 
diff --git a/modules/angular2/src/router/router.ts b/modules/angular2/src/router/router.ts
index e4724792b7fd..5d7ec2015946 100644
--- a/modules/angular2/src/router/router.ts
+++ b/modules/angular2/src/router/router.ts
@@ -59,7 +59,7 @@ export class Router {
   private _auxRouters = new Map<string, Router>();
   private _childRouter: Router;
 
-  private _subject: EventEmitter = new EventEmitter();
+  private _subject: EventEmitter<any> = new EventEmitter();
 
 
   constructor(public registry: RouteRegistry, public parent: Router, public hostComponent: any) {}
diff --git a/modules/angular2/src/web_workers/shared/client_message_broker.ts b/modules/angular2/src/web_workers/shared/client_message_broker.ts
index 07cc31778027..d014ba303abe 100644
--- a/modules/angular2/src/web_workers/shared/client_message_broker.ts
+++ b/modules/angular2/src/web_workers/shared/client_message_broker.ts
@@ -44,7 +44,7 @@ export abstract class ClientMessageBroker {
 
 export class ClientMessageBroker_ extends ClientMessageBroker {
   private _pending: Map<string, PromiseCompleter<any>> = new Map<string, PromiseCompleter<any>>();
-  private _sink: EventEmitter;
+  private _sink: EventEmitter<any>;
   /** @internal */
   public _serializer: Serializer;
 
diff --git a/modules/angular2/src/web_workers/shared/message_bus.ts b/modules/angular2/src/web_workers/shared/message_bus.ts
index 387d5362f55a..2843e6a45de1 100644
--- a/modules/angular2/src/web_workers/shared/message_bus.ts
+++ b/modules/angular2/src/web_workers/shared/message_bus.ts
@@ -30,14 +30,14 @@ export abstract class MessageBus implements MessageBusSource, MessageBusSink {
    * Returns an {@link EventEmitter} that emits every time a message
    * is received on the given channel.
    */
-  abstract from(channel: string): EventEmitter;
+  abstract from(channel: string): EventEmitter<any>;
 
 
   /**
    * Returns an {@link EventEmitter} for the given channel
    * To publish methods to that channel just call next (or add in dart) on the returned emitter
    */
-  abstract to(channel: string): EventEmitter;
+  abstract to(channel: string): EventEmitter<any>;
 }
 
 export interface MessageBusSource {
@@ -60,7 +60,7 @@ export interface MessageBusSource {
    * Returns an {@link EventEmitter} that emits every time a message
    * is received on the given channel.
    */
-  from(channel: string): EventEmitter;
+  from(channel: string): EventEmitter<any>;
 }
 
 export interface MessageBusSink {
@@ -83,5 +83,5 @@ export interface MessageBusSink {
    * Returns an {@link EventEmitter} for the given channel
    * To publish methods to that channel just call next (or add in dart) on the returned emitter
    */
-  to(channel: string): EventEmitter;
+  to(channel: string): EventEmitter<any>;
 }
diff --git a/modules/angular2/src/web_workers/shared/post_message_bus.ts b/modules/angular2/src/web_workers/shared/post_message_bus.ts
index 0d528d2a30cc..448d87a7a739 100644
--- a/modules/angular2/src/web_workers/shared/post_message_bus.ts
+++ b/modules/angular2/src/web_workers/shared/post_message_bus.ts
@@ -27,9 +27,9 @@ export class PostMessageBus implements MessageBus {
     this.sink.initChannel(channel, runInZone);
   }
 
-  from(channel: string): EventEmitter { return this.source.from(channel); }
+  from(channel: string): EventEmitter<any> { return this.source.from(channel); }
 
-  to(channel: string): EventEmitter { return this.sink.to(channel); }
+  to(channel: string): EventEmitter<any> { return this.sink.to(channel); }
 }
 
 export class PostMessageBusSink implements MessageBusSink {
@@ -52,19 +52,17 @@ export class PostMessageBusSink implements MessageBusSink {
     var emitter = new EventEmitter();
     var channelInfo = new _Channel(emitter, runInZone);
     this._channels[channel] = channelInfo;
-    emitter.observer({
-      next: (data: Object) => {
-        var message = {channel: channel, message: data};
-        if (runInZone) {
-          this._messageBuffer.push(message);
-        } else {
-          this._sendMessages([message]);
-        }
+    emitter.subscribe((data: Object) => {
+      var message = {channel: channel, message: data};
+      if (runInZone) {
+        this._messageBuffer.push(message);
+      } else {
+        this._sendMessages([message]);
       }
     });
   }
 
-  to(channel: string): EventEmitter {
+  to(channel: string): EventEmitter<any> {
     if (StringMapWrapper.contains(this._channels, channel)) {
       return this._channels[channel].emitter;
     } else {
@@ -107,7 +105,7 @@ export class PostMessageBusSource implements MessageBusSource {
     this._channels[channel] = channelInfo;
   }
 
-  from(channel: string): EventEmitter {
+  from(channel: string): EventEmitter<any> {
     if (StringMapWrapper.contains(this._channels, channel)) {
       return this._channels[channel].emitter;
     } else {
@@ -140,7 +138,7 @@ export class PostMessageBusSource implements MessageBusSource {
  * keeps track of if it should run in the zone.
  */
 class _Channel {
-  constructor(public emitter: EventEmitter, public runInZone: boolean) {}
+  constructor(public emitter: EventEmitter<any>, public runInZone: boolean) {}
 }
 
 // TODO(jteplitz602) Replace this with the definition in lib.webworker.d.ts(#3492)
diff --git a/modules/angular2/src/web_workers/shared/service_message_broker.ts b/modules/angular2/src/web_workers/shared/service_message_broker.ts
index ac6199db569a..a54f15cde8e1 100644
--- a/modules/angular2/src/web_workers/shared/service_message_broker.ts
+++ b/modules/angular2/src/web_workers/shared/service_message_broker.ts
@@ -45,7 +45,7 @@ export abstract class ServiceMessageBroker {
  * If that method returns a promise, the UIMessageBroker returns the result to the worker.
  */
 export class ServiceMessageBroker_ extends ServiceMessageBroker {
-  private _sink: EventEmitter;
+  private _sink: EventEmitter<any>;
   private _methods: Map<string, Function> = new Map<string, Function>();
 
   constructor(messageBus: MessageBus, private _serializer: Serializer, public channel) {
diff --git a/modules/angular2/src/web_workers/ui/event_dispatcher.ts b/modules/angular2/src/web_workers/ui/event_dispatcher.ts
index 2630c7a054eb..3c66de7b2155 100644
--- a/modules/angular2/src/web_workers/ui/event_dispatcher.ts
+++ b/modules/angular2/src/web_workers/ui/event_dispatcher.ts
@@ -14,7 +14,7 @@ import {StringMapWrapper} from 'angular2/src/core/facade/collection';
 import {EventEmitter, ObservableWrapper} from 'angular2/src/core/facade/async';
 
 export class EventDispatcher implements RenderEventDispatcher {
-  constructor(private _viewRef: RenderViewRef, private _sink: EventEmitter,
+  constructor(private _viewRef: RenderViewRef, private _sink: EventEmitter<any>,
               private _serializer: Serializer) {}
 
   dispatchRenderEvent(elementIndex: number, eventName: string, locals: Map<string, any>): boolean {
diff --git a/modules/angular2/test/core/debug/debug_element_spec.ts b/modules/angular2/test/core/debug/debug_element_spec.ts
index 0c331ac3fef4..cc69f725a4e0 100644
--- a/modules/angular2/test/core/debug/debug_element_spec.ts
+++ b/modules/angular2/test/core/debug/debug_element_spec.ts
@@ -89,7 +89,7 @@ class ParentComp {
 @Directive({selector: 'custom-emitter', outputs: ['myevent']})
 @Injectable()
 class CustomEmitter {
-  myevent: EventEmitter;
+  myevent: EventEmitter<any>;
 
   constructor() { this.myevent = new EventEmitter(); }
 }
diff --git a/modules/angular2/test/core/facade/async_spec.ts b/modules/angular2/test/core/facade/async_spec.ts
index 0415c26cba5c..926ec74b96a5 100644
--- a/modules/angular2/test/core/facade/async_spec.ts
+++ b/modules/angular2/test/core/facade/async_spec.ts
@@ -12,11 +12,17 @@ import {
   inject
 } from 'angular2/testing_internal';
 
-import {ObservableWrapper, EventEmitter, PromiseWrapper} from 'angular2/src/core/facade/async';
+import {
+  ObservableWrapper,
+  Observable,
+  Subject,
+  EventEmitter,
+  PromiseWrapper
+} from 'angular2/src/core/facade/async';
 
 export function main() {
   describe('EventEmitter', () => {
-    var emitter: EventEmitter;
+    var emitter: EventEmitter<any>;
 
     beforeEach(() => { emitter = new EventEmitter(); });
 
@@ -34,18 +40,18 @@ export function main() {
            expect(error).toEqual("Boom");
            async.done();
          });
-         ObservableWrapper.callThrow(emitter, "Boom");
+         ObservableWrapper.callError(emitter, "Boom");
        }));
 
     it("should work when no throw callback is provided", inject([AsyncTestCompleter], (async) => {
          ObservableWrapper.subscribe(emitter, (_) => {}, (_) => { async.done(); });
-         ObservableWrapper.callThrow(emitter, "Boom");
+         ObservableWrapper.callError(emitter, "Boom");
        }));
 
     it("should call the return callback", inject([AsyncTestCompleter], (async) => {
          ObservableWrapper.subscribe(emitter, (_) => {}, (_) => {}, () => { async.done(); });
 
-         ObservableWrapper.callReturn(emitter);
+         ObservableWrapper.callComplete(emitter);
        }));
 
     it("should subscribe to the wrapper asynchronously", () => {
@@ -92,6 +98,20 @@ export function main() {
     // should call dispose on the subscription on return
   });
 
+  describe("ObservableWrapper", () => {
+
+    it('should correctly check isObservable for EventEmitter', () => {
+      var e = new EventEmitter(false);
+      expect(ObservableWrapper.isObservable(e)).toBe(true);
+    });
+
+    it('should correctly check isObservable for Subject', () => {
+      var e = new Subject();
+      expect(ObservableWrapper.isObservable(e)).toBe(true);
+    });
+
+  });
+
   // See ECMAScript 6 Spec 25.4.4.1
   describe("PromiseWrapper", () => {
     describe("#all", () => {
diff --git a/modules/angular2/test/core/forms/integration_spec.ts b/modules/angular2/test/core/forms/integration_spec.ts
index 26505bf47903..533fa3a3af57 100644
--- a/modules/angular2/test/core/forms/integration_spec.ts
+++ b/modules/angular2/test/core/forms/integration_spec.ts
@@ -907,7 +907,7 @@ class WrappedValue implements ControlValueAccessor {
 
 @Component({selector: "my-input", template: ''})
 class MyInput implements ControlValueAccessor {
-  @Output('change') onChange: EventEmitter = new EventEmitter();
+  @Output('change') onChange: EventEmitter<any> = new EventEmitter();
   value: string;
 
   constructor(cd: NgControl) { cd.valueAccessor = this; }
diff --git a/modules/angular2/test/core/forms/model_spec.ts b/modules/angular2/test/core/forms/model_spec.ts
index 810f30544604..dee17c62fc0c 100644
--- a/modules/angular2/test/core/forms/model_spec.ts
+++ b/modules/angular2/test/core/forms/model_spec.ts
@@ -133,7 +133,7 @@ export function main() {
         if (!IS_DART) {
           it("should update set errors and status before emitting an event",
              inject([AsyncTestCompleter], (async) => {
-               c.valueChanges.toRx().subscribe(value => {
+               c.valueChanges.subscribe(value => {
                  expect(c.valid).toEqual(false);
                  expect(c.errors).toEqual({"required": true});
                  async.done();
diff --git a/modules/angular2/test/core/linker/integration_spec.ts b/modules/angular2/test/core/linker/integration_spec.ts
index b801211df420..e26331f51b64 100644
--- a/modules/angular2/test/core/linker/integration_spec.ts
+++ b/modules/angular2/test/core/linker/integration_spec.ts
@@ -1976,7 +1976,7 @@ class DoublePipe implements PipeTransform {
 @Injectable()
 class DirectiveEmitingEvent {
   msg: string;
-  event: EventEmitter;
+  event: EventEmitter<any>;
 
   constructor() {
     this.msg = '';
@@ -2002,7 +2002,7 @@ class DirectiveUpdatingHostProperties {
 @Directive({selector: '[update-host-actions]', host: {'@setAttr': 'setAttribute'}})
 @Injectable()
 class DirectiveUpdatingHostActions {
-  setAttr: EventEmitter;
+  setAttr: EventEmitter<any>;
 
   constructor() { this.setAttr = new EventEmitter(); }
 
diff --git a/modules/angular2/test/public_api_spec.ts b/modules/angular2/test/public_api_spec.ts
index 640bb92024dc..3bfc4580d641 100644
--- a/modules/angular2/test/public_api_spec.ts
+++ b/modules/angular2/test/public_api_spec.ts
@@ -485,7 +485,12 @@ var NG_API = [
   'ErrorHandlingFn:dart',
   'Output',
   'Output.bindingPropertyName',
+
   'EventEmitter',
+
+  /*
+   Dart Stream/EventEmitter
+  */
   'EventEmitter.add():dart',
   'EventEmitter.addError():dart',
   'EventEmitter.any():dart',
@@ -512,24 +517,216 @@ var NG_API = [
   'EventEmitter.length:dart',
   'EventEmitter.listen():dart',
   'EventEmitter.map():dart',
-  'EventEmitter.next():js',
-  'EventEmitter.observer():js',
   'EventEmitter.pipe():dart',
   'EventEmitter.reduce():dart',
-  'EventEmitter.return():js',
   'EventEmitter.single:dart',
   'EventEmitter.singleWhere():dart',
   'EventEmitter.skip():dart',
   'EventEmitter.skipWhile():dart',
   'EventEmitter.take():dart',
   'EventEmitter.takeWhile():dart',
-  'EventEmitter.throw():js',
   'EventEmitter.timeout():dart',
   'EventEmitter.toList():dart',
-  'EventEmitter.toRx():js',
   'EventEmitter.toSet():dart',
   'EventEmitter.transform():dart',
   'EventEmitter.where():dart',
+
+  /*
+   RxJS API - may need to maintain as RxJS evolves
+  */
+  'EventEmitter.mapTo():js',
+  'EventEmitter.next():js',
+  'EventEmitter.materialize():js',
+  'EventEmitter.merge():js',
+  'EventEmitter.mergeAll():js',
+  'EventEmitter.mergeMap():js',
+  'EventEmitter.mergeMapTo():js',
+  'EventEmitter.multicast():js',
+  'EventEmitter.observeOn():js',
+  'EventEmitter.remove():js',
+  'EventEmitter.repeat():js',
+  'EventEmitter.retry():js',
+  'EventEmitter.retryWhen():js',
+  'EventEmitter.throttle():js',
+  'EventEmitter.toPromise():js',
+  'EventEmitter.window():js',
+  'EventEmitter.windowCount():js',
+  'EventEmitter.windowTime():js',
+  'EventEmitter.windowToggle():js',
+  'EventEmitter.windowWhen():js',
+  'EventEmitter.withLatestFrom():js',
+  'EventEmitter.zip():js',
+  'EventEmitter.zipAll():js',
+
+  'Observable:js',
+  'Observable#combineLatest():js',
+  'Observable#concat():js',
+  'Observable#create():js',
+  'Observable#defer():js',
+  'Observable#empty():js',
+  'Observable#forkJoin():js',
+  'Observable#from():js',
+  'Observable#fromArray():js',
+  'Observable#fromEvent():js',
+  'Observable#fromEventPattern():js',
+  'Observable#fromPromise():js',
+  'Observable#interval():js',
+  'Observable#merge():js',
+  'Observable#never():js',
+  'Observable#of():js',
+  'Observable#range():js',
+  'Observable#throw():js',
+  'Observable#timer():js',
+  'Observable#zip():js',
+  'Observable.buffer():js',
+  'Observable.bufferCount():js',
+  'Observable.bufferTime():js',
+  'Observable.bufferToggle():js',
+  'Observable.bufferWhen():js',
+  'Observable.catch():js',
+  'Observable.combineAll():js',
+  'Observable.combineLatest():js',
+  'Observable.concat():js',
+  'Observable.concatAll():js',
+  'Observable.concatMap():js',
+  'Observable.concatMapTo():js',
+  'Observable.count():js',
+  'Observable.debounce():js',
+  'Observable.debounceTime():js',
+  'Observable.defaultIfEmpty():js',
+  'Observable.delay():js',
+  'Observable.dematerialize():js',
+  'Observable.distinctUntilChanged():js',
+  'Observable.do():js',
+  'Observable.every():js',
+  'Observable.expand():js',
+  'Observable.filter():js',
+  'Observable.finally():js',
+  'Observable.first():js',
+  'Observable.flatMap():js',
+  'Observable.flatMapTo():js',
+  'Observable.forEach():js',
+  'Observable.groupBy():js',
+  'Observable.ignoreElements():js',
+  'Observable.last():js',
+  'Observable.lift():js',
+  'Observable.map():js',
+  'Observable.mapTo():js',
+  'Observable.materialize():js',
+  'Observable.merge():js',
+  'Observable.mergeAll():js',
+  'Observable.mergeMap():js',
+  'Observable.mergeMapTo():js',
+  'Observable.multicast():js',
+  'Observable.observeOn():js',
+
+  'Subject',
+  'Subject#combineLatest():js',
+  'Subject#concat():js',
+  'Subject#create():js',
+  'Subject#defer():js',
+  'Subject#empty():js',
+  'Subject#forkJoin():js',
+  'Subject#from():js',
+  'Subject#fromArray():js',
+  'Subject#fromEvent():js',
+  'Subject#fromEventPattern():js',
+  'Subject#fromPromise():js',
+  'Subject#interval():js',
+  'Subject#merge():js',
+  'Subject#never():js',
+  'Subject#of():js',
+  'Subject#range():js',
+  'Subject#throw():js',
+  'Subject#timer():js',
+  'Subject#zip():js',
+  'Subject.add():js',
+  'Subject.buffer():js',
+  'Subject.bufferCount():js',
+  'Subject.bufferTime():js',
+  'Subject.bufferToggle():js',
+  'Subject.bufferWhen():js',
+  'Subject.catch():js',
+  'Subject.combineAll():js',
+  'Subject.combineLatest():js',
+  'Subject.complete():js',
+  'Subject.concat():js',
+  'Subject.concatAll():js',
+  'Subject.concatMap():js',
+  'Subject.concatMapTo():js',
+  'Subject.count():js',
+  'Subject.debounce():js',
+  'Subject.debounceTime():js',
+  'Subject.defaultIfEmpty():js',
+  'Subject.delay():js',
+  'Subject.dematerialize():js',
+  'Subject.distinctUntilChanged():js',
+  'Subject.do():js',
+  'Subject.error():js',
+  'Subject.every():js',
+  'Subject.expand():js',
+  'Subject.filter():js',
+  'Subject.finally():js',
+  'Subject.first():js',
+  'Subject.flatMap():js',
+  'Subject.flatMapTo():js',
+  'Subject.forEach():js',
+  'Subject.groupBy():js',
+  'Subject.ignoreElements():js',
+  'Subject.last():js',
+  'Subject.lift():js',
+  'Subject.map():js',
+  'Subject.mapTo():js',
+  'Subject.materialize():js',
+  'Subject.merge():js',
+  'Subject.mergeAll():js',
+  'Subject.mergeMap():js',
+  'Subject.mergeMapTo():js',
+  'Subject.multicast():js',
+  'Subject.next():js',
+  'Subject.observeOn():js',
+  'Subject.partition():js',
+  'Subject.publish():js',
+  'Subject.publishBehavior():js',
+  'Subject.publishReplay():js',
+  'Subject.reduce():js',
+  'Subject.remove():js',
+  'Subject.repeat():js',
+  'Subject.retry():js',
+  'Subject.retryWhen():js',
+  'Subject.sample():js',
+  'Subject.sampleTime():js',
+  'Subject.scan():js',
+  'Subject.share():js',
+  'Subject.shareBehavior():js',
+  'Subject.shareReplay():js',
+  'Subject.single():js',
+  'Subject.skip():js',
+  'Subject.skipUntil():js',
+  'Subject.startWith():js',
+  'Subject.subscribe():js',
+  'Subject.subscribeOn():js',
+  'Subject.switch():js',
+  'Subject.switchMap():js',
+  'Subject.switchMapTo():js',
+  'Subject.take():js',
+  'Subject.takeUntil():js',
+  'Subject.throttle():js',
+  'Subject.timeout():js',
+  'Subject.timeoutWith():js',
+  'Subject.toArray():js',
+  'Subject.toPromise():js',
+  'Subject.unsubscribe():js',
+  'Subject.window():js',
+  'Subject.windowCount():js',
+  'Subject.windowTime():js',
+  'Subject.windowToggle():js',
+  'Subject.windowWhen():js',
+  'Subject.withLatestFrom():js',
+  'Subject.zip():js',
+  'Subject.zipAll():js',
+
+
   'OutputMetadata',
   'OutputMetadata.bindingPropertyName',
   'ExpressionChangedAfterItHasBeenCheckedException',
@@ -844,8 +1041,6 @@ var NG_API = [
   'NoProviderError.message=',
   'NoProviderError.stackTrace',
   'NumberPipe',
-  'Observable.observer():js',
-  'Observable:js',
   'ObservableListDiff.check():dart',
   'ObservableListDiff.collection:dart',
   'ObservableListDiff.diff():dart',
diff --git a/modules/angular2/test/router/integration/lifecycle_hook_spec.ts b/modules/angular2/test/router/integration/lifecycle_hook_spec.ts
index bb73b2519d17..7892164e6554 100644
--- a/modules/angular2/test/router/integration/lifecycle_hook_spec.ts
+++ b/modules/angular2/test/router/integration/lifecycle_hook_spec.ts
@@ -51,7 +51,7 @@ import {DirectiveResolver} from 'angular2/src/core/linker/directive_resolver';
 
 var cmpInstanceCount;
 var log: string[];
-var eventBus: EventEmitter;
+var eventBus: EventEmitter<any>;
 var completer: PromiseCompleter<any>;
 
 export function main() {
diff --git a/modules/angular2/test/web_workers/shared/mock_event_emitter.dart b/modules/angular2/test/web_workers/shared/mock_event_emitter.dart
index ee7546b0d8ca..2a55d5e8d9ef 100644
--- a/modules/angular2/test/web_workers/shared/mock_event_emitter.dart
+++ b/modules/angular2/test/web_workers/shared/mock_event_emitter.dart
@@ -4,7 +4,7 @@ import 'dart:core';
 import 'dart:async';
 import "package:angular2/src/core/facade/async.dart";
 
-class MockEventEmitter extends EventEmitter {
+class MockEventEmitter<T> extends EventEmitter<T> {
   final controller = new StreamController.broadcast(sync: true);
 
   @override
diff --git a/modules/angular2/test/web_workers/shared/mock_event_emitter.ts b/modules/angular2/test/web_workers/shared/mock_event_emitter.ts
index ed5d43f026ac..7cbd2435565e 100644
--- a/modules/angular2/test/web_workers/shared/mock_event_emitter.ts
+++ b/modules/angular2/test/web_workers/shared/mock_event_emitter.ts
@@ -1,11 +1,11 @@
 import {EventEmitter} from 'angular2/src/core/facade/async';
 
-export class MockEventEmitter extends EventEmitter {
+export class MockEventEmitter<T> extends EventEmitter<T> {
   private _nextFns: Function[] = [];
 
   constructor() { super(); }
 
-  observer(generator: any): any {
+  subscribe(generator: any): any {
     this._nextFns.push(generator.next);
     return new MockDisposable();
   }
diff --git a/modules/angular2/test/web_workers/shared/web_worker_test_util.ts b/modules/angular2/test/web_workers/shared/web_worker_test_util.ts
index a52b95e6181e..7dae3f3eae3c 100644
--- a/modules/angular2/test/web_workers/shared/web_worker_test_util.ts
+++ b/modules/angular2/test/web_workers/shared/web_worker_test_util.ts
@@ -13,11 +13,11 @@ import {NgZone} from 'angular2/src/core/zone/ng_zone';
  * Such that whatever goes into one's sink comes out the others source.
  */
 export function createPairedMessageBuses(): PairedMessageBuses {
-  var firstChannels: {[key: string]: MockEventEmitter} = {};
+  var firstChannels: {[key: string]: MockEventEmitter<any>} = {};
   var workerMessageBusSink = new MockMessageBusSink(firstChannels);
   var uiMessageBusSource = new MockMessageBusSource(firstChannels);
 
-  var secondChannels: {[key: string]: MockEventEmitter} = {};
+  var secondChannels: {[key: string]: MockEventEmitter<any>} = {};
   var uiMessageBusSink = new MockMessageBusSink(secondChannels);
   var workerMessageBusSource = new MockMessageBusSource(secondChannels);
 
@@ -30,7 +30,7 @@ export class PairedMessageBuses {
 }
 
 export class MockMessageBusSource implements MessageBusSource {
-  constructor(private _channels: {[key: string]: MockEventEmitter}) {}
+  constructor(private _channels: {[key: string]: MockEventEmitter<any>}) {}
 
   initChannel(channel: string, runInZone = true) {
     if (!StringMapWrapper.contains(this._channels, channel)) {
@@ -38,7 +38,7 @@ export class MockMessageBusSource implements MessageBusSource {
     }
   }
 
-  from(channel: string): MockEventEmitter {
+  from(channel: string): MockEventEmitter<any> {
     if (!StringMapWrapper.contains(this._channels, channel)) {
       throw new BaseException(`${channel} is not set up. Did you forget to call initChannel?`);
     }
@@ -49,7 +49,7 @@ export class MockMessageBusSource implements MessageBusSource {
 }
 
 export class MockMessageBusSink implements MessageBusSink {
-  constructor(private _channels: {[key: string]: MockEventEmitter}) {}
+  constructor(private _channels: {[key: string]: MockEventEmitter<any>}) {}
 
   initChannel(channel: string, runInZone = true) {
     if (!StringMapWrapper.contains(this._channels, channel)) {
@@ -57,7 +57,7 @@ export class MockMessageBusSink implements MessageBusSink {
     }
   }
 
-  to(channel: string): MockEventEmitter {
+  to(channel: string): MockEventEmitter<any> {
     if (!StringMapWrapper.contains(this._channels, channel)) {
       this._channels[channel] = new MockEventEmitter();
     }
@@ -79,9 +79,9 @@ export class MockMessageBus extends MessageBus {
     this.source.initChannel(channel, runInZone);
   }
 
-  to(channel: string): MockEventEmitter { return this.sink.to(channel); }
+  to(channel: string): MockEventEmitter<any> { return this.sink.to(channel); }
 
-  from(channel: string): MockEventEmitter { return this.source.from(channel); }
+  from(channel: string): MockEventEmitter<any> { return this.source.from(channel); }
 
   attachToZone(zone: NgZone) {}
 }
diff --git a/modules/angular2_material/src/components/input/input.ts b/modules/angular2_material/src/components/input/input.ts
index d9b4f8da46c8..4541d8c40d57 100644
--- a/modules/angular2_material/src/components/input/input.ts
+++ b/modules/angular2_material/src/components/input/input.ts
@@ -71,8 +71,8 @@ export class MdInput {
 
   // Events emitted by this directive. We use these special 'md-' events to communicate
   // to the parent MdInputContainer.
-  mdChange: EventEmitter;
-  mdFocusChange: EventEmitter;
+  mdChange: EventEmitter<any>;
+  mdFocusChange: EventEmitter<any>;
 
   constructor(@Attribute('value') value: string, @SkipSelf() @Host() container: MdInputContainer,
               @Attribute('id') id: string) {
diff --git a/modules/angular2_material/src/components/radio/radio_button.ts b/modules/angular2_material/src/components/radio/radio_button.ts
index 2d852cb2a878..fbdc5499b618 100644
--- a/modules/angular2_material/src/components/radio/radio_button.ts
+++ b/modules/angular2_material/src/components/radio/radio_button.ts
@@ -68,7 +68,7 @@ export class MdRadioGroup implements OnChanges {
   /** The ID of the selected radio button. */
   selectedRadioId: string;
 
-  change: EventEmitter;
+  change: EventEmitter<any>;
 
   tabindex: number;
 
diff --git a/modules/playground/src/zippy_component/zippy.ts b/modules/playground/src/zippy_component/zippy.ts
index 1f3a0b721870..cca3d0332496 100644
--- a/modules/playground/src/zippy_component/zippy.ts
+++ b/modules/playground/src/zippy_component/zippy.ts
@@ -7,8 +7,8 @@ import {ObservableWrapper} from 'angular2/src/core/facade/async';
 export class Zippy {
   visible: boolean = true;
   title: string = '';
-  openHandler: EventEmitter = new EventEmitter();
-  closeHandler: EventEmitter = new EventEmitter();
+  openHandler: EventEmitter<any> = new EventEmitter();
+  closeHandler: EventEmitter<any> = new EventEmitter();
 
   toggle() {
     this.visible = !this.visible;
diff --git a/modules/upgrade/src/downgrade_ng2_adapter.ts b/modules/upgrade/src/downgrade_ng2_adapter.ts
index 7632b126aded..50324e88a609 100644
--- a/modules/upgrade/src/downgrade_ng2_adapter.ts
+++ b/modules/upgrade/src/downgrade_ng2_adapter.ts
@@ -149,7 +149,7 @@ export class DowngradeNg2ComponentAdapter {
         }
         var emitter = this.component[output.prop];
         if (emitter) {
-          emitter.observer({
+          emitter.subscribe({
             next: assignExpr ? ((setter) => (value) => setter(this.scope, value))(setter) :
                                ((getter) => (value) => getter(this.scope, {$event: value}))(getter)
           });
diff --git a/modules/upgrade/src/upgrade_ng1_adapter.ts b/modules/upgrade/src/upgrade_ng1_adapter.ts
index 07fc0b3ae57b..e9924fa82017 100644
--- a/modules/upgrade/src/upgrade_ng1_adapter.ts
+++ b/modules/upgrade/src/upgrade_ng1_adapter.ts
@@ -244,7 +244,7 @@ class UpgradeNg1ComponentAdapter implements OnChanges, DoCheck {
         if (typeof value == 'number' && isNaN(value) && typeof last == 'number' && isNaN(last)) {
           // ignore because NaN != NaN
         } else {
-          var eventEmitter: EventEmitter = this[this.propOuts[i]];
+          var eventEmitter: EventEmitter<any> = this[this.propOuts[i]];
           eventEmitter.next(lastValues[i] = value);
         }
       }
diff --git a/npm-shrinkwrap.clean.json b/npm-shrinkwrap.clean.json
index cd47b13bf96b..dec5e2b2c0dd 100644
--- a/npm-shrinkwrap.clean.json
+++ b/npm-shrinkwrap.clean.json
@@ -1,7 +1,7 @@
 {
   "dependencies": {
     "@reactivex/rxjs": {
-      "version": "5.0.0-alpha.4"
+      "version": "5.0.0-alpha.7"
     },
     "angular": {
       "version": "1.4.7"
diff --git a/npm-shrinkwrap.json b/npm-shrinkwrap.json
index c2ae85849308..3336d1e1b1e5 100644
--- a/npm-shrinkwrap.json
+++ b/npm-shrinkwrap.json
@@ -3,9 +3,9 @@
   "version": "2.0.0-alpha.44",
   "dependencies": {
     "@reactivex/rxjs": {
-      "version": "5.0.0-alpha.4",
-      "from": "https://registry.npmjs.org/@reactivex/rxjs/-/rxjs-5.0.0-alpha.4.tgz",
-      "resolved": "https://registry.npmjs.org/@reactivex/rxjs/-/rxjs-5.0.0-alpha.4.tgz"
+      "version": "5.0.0-alpha.7",
+      "from": "http://registry.npmjs.org/@reactivex/rxjs/-/rxjs-5.0.0-alpha.7.tgz",
+      "resolved": "http://registry.npmjs.org/@reactivex/rxjs/-/rxjs-5.0.0-alpha.7.tgz"
     },
     "angular": {
       "version": "1.4.7",
diff --git a/package.json b/package.json
index 91fbacdf0f68..2936c6d50571 100644
--- a/package.json
+++ b/package.json
@@ -28,7 +28,7 @@
     "test": "gulp test.all.js && gulp test.all.dart"
   },
   "dependencies": {
-    "@reactivex/rxjs": "5.0.0-alpha.4",
+    "@reactivex/rxjs": "5.0.0-alpha.7",
     "reflect-metadata": "0.1.2",
     "zone.js": "0.5.8"
   },
diff --git a/tools/broccoli/broccoli-typescript.ts b/tools/broccoli/broccoli-typescript.ts
index 06d6089f6d21..5156422e8ff6 100644
--- a/tools/broccoli/broccoli-typescript.ts
+++ b/tools/broccoli/broccoli-typescript.ts
@@ -230,6 +230,8 @@ class CustomLanguageServiceHost implements ts.LanguageServiceHost {
     } else if (this.compilerOptions.moduleResolution === ts.ModuleResolutionKind.NodeJs &&
                tsFilePath.match(/^node_modules/)) {
       absoluteTsFilePath = path.resolve(tsFilePath);
+    } else if (tsFilePath.match(/^@reactivex/)) {
+      absoluteTsFilePath = path.resolve('node_modules', tsFilePath);
     } else {
       absoluteTsFilePath = path.join(this.treeInputPath, tsFilePath);
     }