feat(forEach): add thisArg

- adds thisArg as second argument to have symmetry with native Array forEach
- eliminates closures by leveraging named function instances
- adds test for thisArg passing
- updates tests to accommodate thisArg

BREAKING CHANGE: Observable.prototype.forEach argument order changed to accommodate thisArg. Optional PromiseCtor argument moved to third arg from second
closes #878

Author: benlesh

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 () {

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 {