refactor(Notification): Notification.observe/accpet should take a par…

…tially implemented observer

src/Notification.ts

@@ -1,21 +1,53 @@
-import {Observer} from './Observer';
 import {Observable} from './Observable';
 
+export type NextObserver<T> = {
+  next(value: T): void;
+};
+function isNextObserver<T>(v: any): v is NextObserver<T> {
+  return !!v && typeof v.next === 'function';
+}
+
+export type ErrorObserver = {
+  error(error: any): void;
+};
+function isErrorObserver<T>(v: any): v is ErrorObserver {
+  return !!v && typeof v.error === 'function';
+}
+
+export type CompleteObserver = {
+  complete(): void;
+};
+function isCompleteObserver(v: any): v is CompleteObserver {
+  return !!v && typeof v.complete === 'function';
+}
+
 export class Notification<T> {
   hasValue: boolean;
 
   constructor(public kind: string, public value?: T, public exception?: any) {
     this.hasValue = kind === 'N';
   }
 
-  observe(observer: Observer<T>): any {
+  observe(observer: NextObserver<T> | ErrorObserver | CompleteObserver): any {
     switch (this.kind) {
       case 'N':
-        return observer.next(this.value);
+        if (isNextObserver(observer)) {
+          return observer.next(this.value);
+        } else {
+          throw new Error('`observer` should implement NextObserver<T>');
+        }
       case 'E':
-        return observer.error(this.exception);
+        if (isErrorObserver(observer)) {
+          return observer.error(this.exception);
+        } else {
+          throw new Error('`observer` should implement ErrorObserver');
+        }
       case 'C':
-        return observer.complete();
+        if (isCompleteObserver(observer)) {
+          return observer.complete();
+        } else {
+          throw new Error('`observer` should implement CompleteObserver');
+        }
     }
   }
 
@@ -31,9 +63,9 @@ export class Notification<T> {
     }
   }
 
-  accept(nextOrObserver: Observer<T> | ((value: T) => void), error?: (err: any) => void, complete?: () => void) {
-    if (nextOrObserver && typeof (<Observer<T>>nextOrObserver).next === 'function') {
-      return this.observe(<Observer<T>>nextOrObserver);
+  accept(nextOrObserver: NextObserver<T> | ((value: T) => void), error?: (err: any) => void, complete?: () => void) {
+    if (isNextObserver(nextOrObserver)) {
+      return this.observe(nextOrObserver);
     } else {
       return this.do(<(value: T) => void>nextOrObserver, error, complete);
     }