feat(operator): add toPromise operator. closes #159

Author: benlesh

spec/operators/toPromise-spec.js

@@ -0,0 +1,22 @@
+/* globals describe, it, expect */
+var Rx = require('../../dist/cjs/Rx');
+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) {
+      expect(x).toBe(3);
+      done();
+    });
+  });
+  
+  it('should handle errors properly', function (done) {
+    Observable.throw('bad').toPromise(promise).then(function () {
+      throw 'should not be called';
+    }, function (err) {
+      expect(err).toBe('bad');
+      done();
+    });
+  });
+});

src/Observable.ts

@@ -129,7 +129,8 @@ export default class Observable<T> {
   take: (count: number) => Observable<T>;
   takeUntil: (observable: Observable<any>) => Observable<T>;
   partition: (predicate: (x: T) => boolean) => Observable<T>[];
-
+  toPromise: (PromiseCtor: PromiseConstructor) => Promise<T>;
+  
   observeOn: (scheduler: Scheduler, delay?: number) => Observable<T>;
   subscribeOn: (scheduler: Scheduler, delay?: number) => Observable<T>;
 

src/Rx.ts

@@ -122,8 +122,10 @@ observableProto.observeOn = observeOn;
 observableProto.subscribeOn = subscribeOn;
 
 import partition from './operators/partition';
+import toPromise from './operators/toPromise';
 
 observableProto.partition = partition;
+observableProto.toPromise = toPromise;
 
 import _catch from './operators/catch';
 import retryWhen from './operators/retryWhen';

src/operators/toPromise.ts

@@ -0,0 +1,8 @@
+import Subscriber from '../Subscriber';
+
+export default function toPromise<T>(PromiseCtor: PromiseConstructor = Promise): Promise<T> {
+  return new PromiseCtor((resolve, reject) => {
+    let value: any;
+    this.subscribe(x => value = x, err => reject(err), () => resolve(value));
+  });
+}