refactor(async): fix ObservableWrapper.isObservable

Makes ObservableWrapper and AsyncPipe work with Observable, Subject, and EventEmitter

modules/angular2/src/core/facade/async.ts

@@ -25,13 +25,12 @@ 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: any, onNext: (value: T) => void,
+  static subscribe<T>(emitter: any, onNext: (value: T) => void, onError?: (exception: any) => void,
-                      onError: (exception: any) => void = null,
+                      onComplete?: () => void): Object {
-                      onComplete: () => void = null): Object {
     return emitter.subscribe({next: onNext, error: onError, complete: onComplete});
   }
 
-  static isObservable(obs: any): boolean { return obs instanceof EventEmitter; }
+  static isObservable(obs: any): boolean { return obs instanceof RxObservable; }
 
   /**
    * Returns whether `obs` has any subscribers listening to events.

modules/angular2/test/core/facade/async_spec.ts

@@ -12,7 +12,13 @@ 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', () => {
@@ -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", () => {