feat(operator): add fromEventPattern creator function

spec/observables/fromEventPattern-spec.js

@@ -0,0 +1,100 @@
+/* globals describe, it, expect, jasmine */
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+var Promise = require('promise');
+
+describe('Observable.fromEventPattern', function(){
+  it('should call addHandler on subscription', function () {
+    var addHandlerCalledWith;
+    var addHandler = function (h) {
+      addHandlerCalledWith = h;
+    };
+
+    var removeHandler = function () { };
+
+    Observable.fromEventPattern(addHandler, removeHandler)
+      .subscribe(function () { });
+
+    expect(typeof addHandlerCalledWith).toBe('function');
+  });
+
+  it('should call removeHandler on unsubscription', function () {
+    var removeHandlerCalledWith;
+    var addHandler = function () { };
+    var removeHandler = function (h) {
+      removeHandlerCalledWith = h;  
+    };
+
+    var subscription = Observable.fromEventPattern(addHandler, removeHandler)
+      .subscribe(function () { });
+
+    subscription.unsubscribe();
+
+    expect(typeof removeHandlerCalledWith).toBe('function');
+  });
+
+  it('should send errors in addHandler down the error path', function () {
+    Observable.fromEventPattern(function (handler) {
+      throw 'bad';
+    }, function () { })
+      .subscribe(function () { },
+        function (err) {
+          expect(err).toBe('bad');
+        });
+  });
+
+  it('should accept a selector that maps outgoing values', function (done) {
+    var target;
+    var trigger = function () {
+      if (target) {
+        target.apply(null, arguments);
+      }
+    };
+
+    var addHandler = function (handler) {
+      target = handler;
+    };
+    var removeHandler = function (handler) {
+      target = null;
+    };
+    var selector = function (a, b) {
+      return a + b + '!';
+    };
+
+    Observable.fromEventPattern(addHandler, removeHandler, selector)
+      .subscribe(function (x) {
+        expect(x).toBe('testme!');
+        done();
+      });
+
+    trigger('test', 'me');
+  });
+
+  it('should send errors in the selector down the error path', function (done) {
+    var target;
+    var trigger = function (value) {
+      if (target) {
+        target(value);
+      }
+    };
+
+    var addHandler = function (handler) {
+      target = handler;
+    };
+    var removeHandler = function (handler) {
+      target = null;
+    };
+    var selector = function (x) {
+      throw 'bad';
+    };
+
+    Observable.fromEventPattern(addHandler, removeHandler, selector)
+      .subscribe(function () { },
+      function (err) {
+        expect(err).toBe('bad');
+        done();
+      });
+
+    trigger('test');
+  });
+});

src/Observable.ts

@@ -72,7 +72,7 @@ export default class Observable<T> {
   static from: <T>(iterable: any, project?: (x?: any, i?: number) => T, thisArg?: any, scheduler?: Scheduler) => Observable<T>;
   static fromArray: <T>(array: T[], scheduler?: Scheduler) => Observable<T>;
   // static fromEvent: <T, R>(element: any, eventName: string, selector: (event: R) => T) => Observable<T>;
-  // static fromEventPattern: <T, R>(addHandler: Function, removeHandler: Function, selector: (event: R) => T) => Observable<T>;
+  static fromEventPattern: <T>(addHandler: (handler:Function)=>void, removeHandler: (handler:Function) => void, selector?: (...args:Array<any>) => T) => Observable<T>;
   static throw: <T>(error: T) => Observable<T>;
   static empty: <T>() => Observable<T>;
   static never: <T>() => Observable<T>;

src/Rx.ts

@@ -18,13 +18,15 @@ import PromiseObservable from './observables/PromiseObservable';
 import RangeObservable from './observables/RangeObservable';
 import ScalarObservable from './observables/ScalarObservable';
 import TimerObservable from './observables/TimerObservable';
+import FromEventPatternObservable from './observables/FromEventPatternObservable';
 
 Observable.defer = DeferObservable.create;
 Observable.from = IteratorObservable.create;
 Observable.fromArray = ArrayObservable.create;
 Observable.fromPromise = PromiseObservable.create;
 Observable.of = ArrayObservable.of;
 Observable.range = RangeObservable.create;
+Observable.fromEventPattern = FromEventPatternObservable.create;
 
 Observable.just = ScalarObservable.create;
 Observable.return = ScalarObservable.create;

src/Scheduler.ts

@@ -173,7 +173,6 @@ export class FutureAction<T> extends Action<T> {
   }
 
   unsubscribe() {
-    debugger;
     const id = this.id;
     if (id != null) {
       this.id = void 0;

src/observables/FromEventPatternObservable.ts

@@ -0,0 +1,40 @@
+import Scheduler from '../Scheduler';
+import Observable from '../Observable';
+import Subscription from '../Subscription';
+import tryCatch from '../util/tryCatch';
+import {errorObject} from '../util/errorObject';
+
+export default class FromEventPatternObservable<T, R> extends Observable<T> {
+
+  static create<T>(addHandler: (handler: Function) => any, removeHandler: (handler: Function) => void, selector?: (...args:Array<any>) => T) {
+    return new FromEventPatternObservable(addHandler, removeHandler, selector);;
+  }
+
+  constructor(private addHandler: (handler:Function) => any, private removeHandler: (handler:Function) => void, private selector?: (...args:Array<any>) => T) {
+    super();
+  }
+
+  _subscribe(subscriber) {
+    const addHandler = this.addHandler;
+    const removeHandler = this.removeHandler;
+    const selector = this.selector;
+
+    const handler = selector ? function(e) {
+      let result = tryCatch(selector).apply(null, arguments);
+      if (result === errorObject) {
+        subscriber.error(result.e);
+      } else {
+        subscriber.next(result);
+      }
+    } : function(e) { subscriber.next(e); }
+
+    let result = tryCatch(addHandler)(handler);
+    if (result === errorObject) {
+      subscriber.error(result.e);
+    }
+    subscriber.add(new Subscription(() => {
+      //TODO: determine whether or not to forward to error handler
+      removeHandler(handler)
+    }));
+  }
+}