feat(if): add static Observable.if creation operator.

spec/observables/if-spec.js

@@ -0,0 +1,32 @@
+/* globals describe, it, expect, spyOn */
+var Rx = require('../../dist/cjs/Rx.KitchenSink');
+var Observable = Rx.Observable;
+
+describe('Observable.if', function () {
+ it('should subscribe to thenSource when the conditional returns true', function () {
+ var e1 = Observable.if(function () {
+ return true;
+ }, Observable.of('a'));
+ var expected = '(a|)';
+
+ expectObservable(e1).toBe(expected);
+ });
+
+ it('should subscribe to elseSource when the conditional returns false', function () {
+ var e1 = Observable.if(function () {
+ return false;
+ }, Observable.of('a'), Observable.of('b'));
+ var expected = '(b|)';
+
+ expectObservable(e1).toBe(expected);
+ });
+
+ it('should complete without an elseSource when the conditional returns false', function () {
+ var e1 = Observable.if(function () {
+ return false;
+ }, Observable.of('a'));
+ var expected = '|';
+
+ expectObservable(e1).toBe(expected);
+ });
+});

src/Observable.ts

@@ -29,6 +29,7 @@ import {raceStatic} from './operator/race';
 import {RangeObservable} from './observable/RangeObservable';
 import {NeverObservable} from './observable/NeverObservable';
 import {ErrorObservable} from './observable/ErrorObservable';
+import {IfObservable} from './observable/IfObservable';
 import {AjaxCreationMethod} from './observable/dom/AjaxObservable';
 import {WebSocketSubject} from './observable/dom/WebSocketSubject';
 
@@ -259,6 +260,7 @@ export class Observable<T> implements CoreOperators<T> {
  static fromEvent: typeof FromEventObservable.create;
  static fromEventPattern: typeof FromEventPatternObservable.create;
  static fromPromise: typeof PromiseObservable.create;
+ static if: typeof IfObservable.create;
  static interval: typeof IntervalObservable.create;
  static merge: typeof mergeStatic;
  static never: typeof NeverObservable.create;

src/Rx.KitchenSink.ts

@@ -51,8 +51,10 @@ import './add/observable/fromArray';
 import './add/observable/fromEvent';
 import './add/observable/fromEventPattern';
 import './add/observable/fromPromise';
+import './add/observable/if';
 import './add/observable/interval';
 import './add/observable/never';
+import './add/observable/of';
 import './add/observable/range';
 import './add/observable/throw';
 import './add/observable/timer';

src/add/observable/if.ts

@@ -0,0 +1,6 @@
+import {Observable} from '../../Observable';
+import {IfObservable} from '../../observable/IfObservable';
+
+Observable.if = IfObservable.create;
+
+export var _void: void;

src/observable/IfObservable.ts

@@ -0,0 +1,42 @@
+import {Observable} from '../Observable';
+import {Subscriber} from '../Subscriber';
+import {Subscription} from '../Subscription';
+
+export class IfObservable<T, R> extends Observable<T> {
+
+ static create<T, R>(condition: () => boolean,
+ thenSource?: Observable<T>,
+ elseSource?: Observable<R>): Observable<T|R> {
+ return new IfObservable(condition, thenSource, elseSource);
+ }
+
+ constructor(private condition: () => boolean,
+ private thenSource?: Observable<T>,
+ private elseSource?: Observable<R>) {
+ super();
+ }
+
+ protected _subscribe(subscriber: Subscriber<T|R>): Subscription | Function | void {
+
+ const { condition, thenSource, elseSource } = this;
+
+ let result: boolean, error: any, errorHappened = false;
+
+ try {
+ result = condition();
+ } catch (e) {
+ error = e;
+ errorHappened = true;
+ }
+
+ if (errorHappened) {
+ subscriber.error(error);
+ } else if (result && thenSource) {
+ return thenSource.subscribe(subscriber);
+ } else if (elseSource) {
+ return elseSource.subscribe(subscriber);
+ } else {
+ subscriber.complete();
+ }
+ }
+}