feat(fromEventPattern): support pass signal from addHandler to remove…

…Handler

- relates to #1900

src/observable/FromEventPatternObservable.ts

@@ -45,9 +45,10 @@ export class FromEventPatternObservable<T> extends Observable<T> {
    * @param {function(handler: Function): any} addHandler A function that takes
    * a `handler` function as argument and attaches it somehow to the actual
    * source of events.
-   * @param {function(handler: Function): void} removeHandler A function that
+   * @param {function(handler: Function, signal?: any): void} removeHandler A function that
    * takes a `handler` function as argument and removes it in case it was
-   * previously attached using `addHandler`.
+   * previously attached using `addHandler`. if addHandler returns signal to teardown when remove,
+   * removeHandler function will forward it.
    * @param {function(...args: any): T} [selector] An optional function to
    * post-process results. It takes the arguments from the event handler and
    * should return a single value.
@@ -57,13 +58,13 @@ export class FromEventPatternObservable<T> extends Observable<T> {
    * @owner Observable
    */
   static create<T>(addHandler: (handler: Function) => any,
-                   removeHandler: (handler: Function) => void,
+                   removeHandler: (handler: Function, signal?: any) => void,
                    selector?: (...args: Array<any>) => T) {
     return new FromEventPatternObservable(addHandler, removeHandler, selector);
   }
 
   constructor(private addHandler: (handler: Function) => any,
-              private removeHandler: (handler: Function) => void,
+              private removeHandler: (handler: Function, signal?: any) => void,
               private selector?: (...args: Array<any>) => T) {
     super();
   }
@@ -75,10 +76,11 @@ export class FromEventPatternObservable<T> extends Observable<T> {
       this._callSelector(subscriber, args);
     } : function(e: any) { subscriber.next(e); };
 
-    this._callAddHandler(handler, subscriber);
+    const retValue = this._callAddHandler(handler, subscriber);
+
     subscriber.add(new Subscription(() => {
       //TODO: determine whether or not to forward to error handler
-      removeHandler(handler);
+      removeHandler(handler, retValue) ;
     }));
   }
 
@@ -92,9 +94,9 @@ export class FromEventPatternObservable<T> extends Observable<T> {
     }
   }
 
-  private _callAddHandler(handler: (e: any) => void, errorSubscriber: Subscriber<T>): void {
+  private _callAddHandler(handler: (e: any) => void, errorSubscriber: Subscriber<T>): any | null {
     try {
-      this.addHandler(handler);
+      return this.addHandler(handler) || null;
     }
     catch (e) {
       errorSubscriber.error(e);