feat(config): add global configuration of Promise capability

closes #115

spec/helpers/test-helper.js

@@ -77,4 +77,30 @@ afterEach(function () {
 
 function assertDeepEqual(actual, expected) {
   expect(actual).toDeepEqual(expected);
-}
+}
+
+
+(function (){
+  var objectTypes = {
+      'boolean': false,
+      'function': true,
+      'object': true,
+      'number': false,
+      'string': false,
+      'undefined': false
+  };
+
+  var _root = (objectTypes[typeof self] && self) || (objectTypes[typeof window] && window);
+
+  var freeExports = objectTypes[typeof exports] && exports && !exports.nodeType && exports;
+  var freeModule = objectTypes[typeof module] && module && !module.nodeType && module;
+  var freeGlobal = objectTypes[typeof global] && global;
+
+  if (freeGlobal && (freeGlobal.global === freeGlobal || freeGlobal.window === freeGlobal)) {
+      _root = freeGlobal;
+  }
+
+
+
+  global.__root__ = _root;
+}());

spec/observable-spec.js

@@ -1,6 +1,6 @@
 /* globals describe, it, expect */
 var Rx = require('../dist/cjs/Rx');
-
+var Promise = require('promise');
 var Observable = Rx.Observable;
 
 describe('Observable', function () {
@@ -14,6 +14,47 @@ describe('Observable', function () {
     source.subscribe(function (x) { expect(x).toBe(1); }, null, done);
   });
 
+  describe('forEach', function(){
+    it('should iterate and return a Promise', function (done){
+      var expected = [1,2,3];
+      var result = Observable.of(1,2,3).forEach(function(x) {
+        expect(x).toBe(expected.shift());
+      }, Promise)
+      .then(done);
+
+      expect(typeof result.then).toBe('function');
+    });
+
+    it('should reject promise when in error', function(done){
+      Observable.throw('bad').forEach(function(x) {
+        throw 'should not be called';
+      }).then(function() {
+        throw 'should not complete';
+      }, function(err) {
+        expect(err).toBe('bad');
+        done();
+      }, Promise);
+    });
+
+    it('should allow Promise to be globally configured', function (done) {
+      var wasCalled = false;
+      __root__.Rx = {};
+      __root__.Rx.config = {};
+      __root__.Rx.config.Promise = function MyPromise(callback) {
+        wasCalled = true;
+        return new Promise(callback);
+      };
+
+
+      Observable.of(42).forEach(function(x) {
+        expect(x).toBe(42);
+      }).then(function(){
+        expect(wasCalled).toBe(true);
+        done();
+      });
+    });
+  });
+
   describe('subscribe', function () {
     it('should be synchronous', function () {
       var subscribed = false;

spec/operators/toPromise-spec.js

@@ -1,22 +1,38 @@
 /* globals describe, it, expect */
 var Rx = require('../../dist/cjs/Rx');
-var promise = require('promise');
+var Promise = require('promise');
 var Observable = Rx.Observable;
 
 describe('Observable.prototype.toPromise()', function () {
   it('should convert an Observable to a promise of its last value', function (done) {
-    Observable.of(1, 2, 3).toPromise(promise).then(function (x) {
+    Observable.of(1, 2, 3).toPromise(Promise).then(function (x) {
       expect(x).toBe(3);
       done();
     });
   });
 
   it('should handle errors properly', function (done) {
-    Observable.throw('bad').toPromise(promise).then(function () {
+    Observable.throw('bad').toPromise(Promise).then(function () {
       throw 'should not be called';
     }, function (err) {
       expect(err).toBe('bad');
       done();
     });
   });
+
+  it('should allow for global config via Rx.config.Promise', function(done){
+    var wasCalled = false;
+    __root__.Rx = {};
+    __root__.Rx.config = {};
+    __root__.Rx.config.Promise = function MyPromise(callback) {
+      wasCalled = true;
+      return new Promise(callback);
+    };
+
+    Observable.of(42).toPromise().then(function(x) {
+      expect(wasCalled).toBe(true);
+      expect(x).toBe(42);
+      done();
+    });
+  });
 });

src/Observable.ts

@@ -6,7 +6,7 @@ import Subscriber from './Subscriber';
 import Subscription from './Subscription';
 import ConnectableObservable from './observables/ConnectableObservable';
 import GroupSubject from './subjects/GroupSubject';
-
+import {root} from './util/root';
 
 import $$observable from './util/Symbol_observable';
 
@@ -112,8 +112,20 @@ export default class Observable<T>  {
    * @returns {Promise} a promise that either resolves on observable completion or
    *  rejects with the handled error
    */
-  forEach(next:(value:T) => void) {
-    return new Promise((resolve, reject) => {
+  forEach(next:(value:T) => void, PromiseCtor?: PromiseConstructor): Promise<void> {
+    if(!PromiseCtor) {
+      if(root.Rx && root.Rx.config && root.Rx.config.Promise) {
+        PromiseCtor = root.Rx.config.Promise;
+      } else if (root.Promise) {
+        PromiseCtor = root.Promise;
+      }
+    }
+
+    if(!PromiseCtor) {
+      throw new Error('no Promise impl found');
+    }
+
+    return new PromiseCtor<void>((resolve, reject) => {
       this.subscribe(next, reject, resolve);
     });
   }

src/Rx.ts

@@ -241,5 +241,5 @@ export {
     VirtualTimeScheduler,
     TestScheduler,
     EmptyError,
-    ArgumentOutOfRangeError
+    ArgumentOutOfRangeError,
 };

src/operators/toPromise.ts

@@ -1,6 +1,19 @@
 import Subscriber from '../Subscriber';
+import { root } from '../util/root';
 
-export default function toPromise<T>(PromiseCtor: PromiseConstructor = Promise): Promise<T> {
+export default function toPromise<T>(PromiseCtor?: PromiseConstructor): Promise<T> {
+  if(!PromiseCtor) {
+    if(root.Rx && root.Rx.config && root.Rx.config.Promise) {
+      PromiseCtor = root.Rx.config.Promise;
+    } else if (root.Promise) {
+      PromiseCtor = root.Promise;
+    }
+  }
+
+  if(!PromiseCtor) {
+    throw new Error('no Promise impl found');
+  }
+
   return new PromiseCtor((resolve, reject) => {
     let value: any;
     this.subscribe(x => value = x, err => reject(err), () => resolve(value));