Updates for thisArgs issue

spec/observable-spec.js

@@ -27,7 +27,7 @@ describe('Observable', function () {
       var expected = [1,2,3];
       var result = Observable.of(1,2,3).forEach(function (x) {
         expect(x).toBe(expected.shift());
-      }, Promise)
+      }, null, Promise)
       .then(done);
 
       expect(typeof result.then).toBe('function');
@@ -41,7 +41,7 @@ describe('Observable', function () {
       }, function (err) {
         expect(err).toBe('bad');
         done();
-      }, Promise);
+      }, null, Promise);
     });
 
     it('should allow Promise to be globally configured', function (done) {
@@ -60,6 +60,18 @@ describe('Observable', function () {
         done();
       });
     });
+
+    it('should accept a thisArg argument', function (done) {
+      var expected = [1,2,3];
+      var thisArg = {};
+      var result = Observable.of(1,2,3).forEach(function (x) {
+        expect(this).toBe(thisArg);
+        expect(x).toBe(expected.shift());
+      }, thisArg, Promise)
+      .then(done);
+
+      expect(typeof result.then).toBe('function');
+    });
   });
 
   describe('subscribe', function () {

spec/operators/skipWhile-spec.js

@@ -160,23 +160,6 @@ describe('Observable.prototype.skipWhile()', function () {
     expectSubscriptions(source.subscriptions).toBe(sourceSubs);
   });
 
-  it('should accept a thisArg', function () {
-    var source = hot('-1-^--2--3--4--5--6--|');
-    var sourceSubs =    '^                 !';
-    var expected =      '---------4--5--6--|';
-
-    function Skiper() {
-      this.doSkip = function (v) { return +v < 4; };
-    }
-
-    var skiper = new Skiper();
-
-    expectObservable(
-      source.skipWhile(function (v) { return this.doSkip(v); }, skiper)
-    ).toBe(expected);
-    expectSubscriptions(source.subscriptions).toBe(sourceSubs);
-  });
-
   it('should handle Observable.empty', function () {
     var source = cold('|');
     var subs =        '(^!)';

src/CoreOperators.ts

@@ -30,7 +30,7 @@ export interface CoreOperators<T> {
   filter?: (predicate: (x: T) => boolean, ix?: number, thisArg?: any) => Observable<T>;
   finally?: (finallySelector: () => void) => Observable<T>;
   first?: <R>(predicate?: (value: T, index: number, source: Observable<T>) => boolean,
-              resultSelector?: (value: T, index: number) => R, thisArg?: any, defaultValue?: any) => Observable<T> | Observable<R>;
+              resultSelector?: (value: T, index: number) => R, defaultValue?: any) => Observable<T> | Observable<R>;
   flatMap?: <R>(project: ((x: T, ix: number) => Observable<any>),
                 projectResult?: (x: T, y: any, ix: number, iy: number) => R,
                 concurrent?: number) => Observable<R>;
@@ -41,7 +41,7 @@ export interface CoreOperators<T> {
   ignoreElements?: () => Observable<T>;
   last?: <R>(predicate?: (value: T, index: number) => boolean,
              resultSelector?: (value: T, index: number) => R,
-             thisArg?: any, defaultValue?: any) => Observable<T> | Observable<R>;
+             defaultValue?: any) => Observable<T> | Observable<R>;
   every?: (predicate: (value: T, index: number) => boolean, thisArg?: any) => Observable<T>;
   map?: <R>(project: (x: T, ix?: number) => R, thisArg?: any) => Observable<R>;
   mapTo?: <R>(value: R) => Observable<R>;
@@ -69,7 +69,7 @@ export interface CoreOperators<T> {
   single?: (predicate?: (value: T, index: number) => boolean) => Observable<T>;
   skip?: (count: number) => Observable<T>;
   skipUntil?: (notifier: Observable<any>) => Observable<T>;
-  skipWhile?: (predicate: (x: T, index: number) => boolean, thisArg?: any) => Observable<T>;
+  skipWhile?: (predicate: (x: T, index: number) => boolean) => Observable<T>;
   startWith?: (x: T) => Observable<T>;
   subscribeOn?: (scheduler: Scheduler, delay?: number) => Observable<T>;
   switch?: () => Observable<T>;

src/Observable.ts

@@ -110,11 +110,12 @@ export class Observable<T> implements CoreOperators<T>  {
   /**
    * @method forEach
    * @param {Function} next a handler for each value emitted by the observable
-   * @param {PromiseConstructor} PromiseCtor? a constructor function used to instantiate the Promise
+   * @param {any} [thisArg] a `this` context for the `next` handler function
+   * @param {PromiseConstructor} [PromiseCtor] a constructor function used to instantiate the Promise
    * @returns {Promise} a promise that either resolves on observable completion or
    *  rejects with the handled error
    */
-  forEach(next: (value: T) => void, PromiseCtor?: PromiseConstructor): Promise<void> {
+  forEach(next: (value: T) => void, thisArg: any, PromiseCtor?: PromiseConstructor): Promise<void> {
     if (!PromiseCtor) {
       if (root.Rx && root.Rx.config && root.Rx.config.Promise) {
         PromiseCtor = root.Rx.config.Promise;
@@ -127,9 +128,27 @@ export class Observable<T> implements CoreOperators<T>  {
       throw new Error('no Promise impl found');
     }
 
-    return new PromiseCtor<void>((resolve, reject) => {
-      this.subscribe(next, reject, resolve);
-    });
+    let nextHandler;
+
+    if (thisArg) {
+      nextHandler = function nextHandlerFn(value: any): void {
+        const { thisArg, next } = <any>nextHandlerFn;
+        return next.call(thisArg, value);
+      };
+      nextHandler.thisArg = thisArg;
+      nextHandler.next = next;
+    } else {
+      nextHandler = next;
+    }
+
+    const promiseCallback = function promiseCallbackFn(resolve, reject) {
+      const { source, nextHandler } = <any>promiseCallbackFn;
+      source.subscribe(nextHandler, reject, resolve);
+    };
+    (<any>promiseCallback).source = this;
+    (<any>promiseCallback).nextHandler = nextHandler;
+
+    return new PromiseCtor<void>(promiseCallback);
   }
 
   _subscribe(subscriber: Subscriber<any>): Subscription<T> | Function | void {

src/operator/skipWhile.ts

@@ -3,17 +3,13 @@ import {Operator} from '../Operator';
 import {Subscriber} from '../Subscriber';
 import {tryCatch} from '../util/tryCatch';
 import {errorObject} from '../util/errorObject';
-import {bindCallback} from '../util/bindCallback';
 
-export function skipWhile<T>(predicate: (x: T, index: number) => boolean, thisArg?: any): Observable<T> {
-  return this.lift(new SkipWhileOperator(predicate, thisArg));
+export function skipWhile<T>(predicate: (x: T, index: number) => boolean): Observable<T> {
+  return this.lift(new SkipWhileOperator(predicate));
 }
 
 class SkipWhileOperator<T, R> implements Operator<T, R> {
-  private predicate: (x: T, index: number) => boolean;
-
-  constructor(predicate: (x: T, index: number) => boolean, thisArg?: any) {
-    this.predicate = <(x: T, index: number) => boolean>bindCallback(predicate, thisArg, 2);
+  constructor(private predicate: (x: T, index: number) => boolean) {
   }
 
   call(subscriber: Subscriber<T>): Subscriber<T> {