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

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));
+ });
+}