fix(bindNodeCallback): properly set context of input function

spec/observables/bindNodeCallback-spec.ts

@@ -25,6 +25,23 @@ describe('Observable.bindNodeCallback', () => {
       expect(results).to.deep.equal([42, 'done']);
     });
 
+    it('should set context of callback to context of boundCallback', () => {
+      function callback(cb) {
+        cb(null, this.datum);
+      }
+      const boundCallback = Observable.bindNodeCallback(callback);
+      const results = [];
+
+      boundCallback.call({datum: 42})
+        .subscribe(
+          (x: number) => results.push(x),
+          null,
+          () => results.push('done')
+        );
+
+      expect(results).to.deep.equal([42, 'done']);
+    });
+
     it('should emit one value chosen by a selector', () => {
       function callback(datum, cb) {
         cb(null, datum);
@@ -128,6 +145,25 @@ describe('Observable.bindNodeCallback', () => {
       expect(results).to.deep.equal([42, 'done']);
     });
 
+    it('should set context of callback to context of boundCallback', () => {
+      function callback(cb) {
+        cb(null, this.datum);
+      }
+      const boundCallback = Observable.bindNodeCallback(callback, null, rxTestScheduler);
+      const results = [];
+
+      boundCallback.call({datum: 42})
+        .subscribe(
+          (x: number) => results.push(x),
+          null,
+          () => results.push('done')
+        );
+
+      rxTestScheduler.flush();
+
+      expect(results).to.deep.equal([42, 'done']);
+    });
+
     it('should error if callback throws', () => {
       const expected = new Error('haha no callback for you');
       function callback(datum, cb) {

src/observable/BoundNodeCallbackObservable.ts

@@ -39,11 +39,16 @@ export class BoundNodeCallbackObservable<T> extends Observable<T> {
    * last parameter must be a callback function that `func` calls when it is
    * done. The callback function is expected to follow Node.js conventions,
    * where the first argument to the callback is an error, while remaining
-   * arguments are the callback result. The output of `bindNodeCallback` is a
+   * arguments are the callback result.
+   *
+   * The output of `bindNodeCallback` is a
    * function that takes the same parameters as `func`, except the last one (the
    * callback). When the output function is called with arguments, it will
    * return an Observable where the results will be delivered to.
    *
+   * As in {@link bindCallback}, context (`this` property) of input function will be set to context
+   * of returned function, when it is called.
+   *
    * @example <caption>Read a file from the filesystem and get the data as an Observable</caption>
    * import * as fs from 'fs';
    * var readFileAsObservable = Rx.Observable.bindNodeCallback(fs.readFile);
@@ -69,14 +74,15 @@ export class BoundNodeCallbackObservable<T> extends Observable<T> {
   static create<T>(func: Function,
                    selector: Function | void = undefined,
                    scheduler?: IScheduler): (...args: any[]) => Observable<T> {
-    return (...args: any[]): Observable<T> => {
-      return new BoundNodeCallbackObservable<T>(func, <any>selector, args, scheduler);
+    return function(this: any, ...args: any[]): Observable<T> {
+      return new BoundNodeCallbackObservable<T>(func, <any>selector, args, this, scheduler);
     };
   }
 
   constructor(private callbackFunc: Function,
               private selector: Function,
               private args: any[],
+              private context: any,
               public scheduler: IScheduler) {
     super();
   }
@@ -113,26 +119,27 @@ export class BoundNodeCallbackObservable<T> extends Observable<T> {
         // use named function instance to avoid closure.
         (<any>handler).source = this;
 
-        const result = tryCatch(callbackFunc).apply(this, args.concat(handler));
+        const result = tryCatch(callbackFunc).apply(this.context, args.concat(handler));
         if (result === errorObject) {
           subject.error(errorObject.e);
         }
       }
       return subject.subscribe(subscriber);
     } else {
-      return scheduler.schedule(dispatch, 0, { source: this, subscriber });
+      return scheduler.schedule(dispatch, 0, { source: this, subscriber, context: this.context });
     }
   }
 }
 
 interface DispatchState<T> {
   source: BoundNodeCallbackObservable<T>;
   subscriber: Subscriber<T>;
+  context: any;
 }
 
 function dispatch<T>(this: Action<DispatchState<T>>, state: DispatchState<T>) {
   const self = (<Subscription> this);
-  const { source, subscriber } = state;
+  const { source, subscriber, context } = state;
   // XXX: cast to `any` to access to the private field in `source`.
   const { callbackFunc, args, scheduler } = source as any;
   let subject = source.subject;
@@ -162,7 +169,7 @@ function dispatch<T>(this: Action<DispatchState<T>>, state: DispatchState<T>) {
     // use named function to pass values in without closure
     (<any>handler).source = source;
 
-    const result = tryCatch(callbackFunc).apply(this, args.concat(handler));
+    const result = tryCatch(callbackFunc).apply(context, args.concat(handler));
     if (result === errorObject) {
       subject.error(errorObject.e);
     }