feat(operator): add defaultIfEmpty

spec/operators/defaultIfEmpty-spec.js

@@ -0,0 +1,50 @@
+/* globals describe, it, expect */
+var Rx = require('../../dist/cjs/Rx');
+var Observable = Rx.Observable;
+
+describe('Observable.prototype.defaultIfEmpty()', function () {
+ it('should return the argument if Observable is empty', function (done) {
+ var emptyObservable = Observable.empty();
+ emptyObservable.defaultIfEmpty(2)
+ .subscribe(function(x) {
+ expect(x).toBe(2);
+ }, null, done);
+ });
+
+ it('should return null if the Observable is empty and no arguments', function(done) {
+ var emptyObservable = Observable.empty();
+ emptyObservable.defaultIfEmpty()
+ .subscribe(function(x) {
+ expect(x).toBe(null);
+ }, null, done);
+ });
+
+ it('should return the Observable if not empty with a default value', function(done) {
+ var expected = [1,2,3];
+ var observable = Observable.of(1,2,3);
+ observable.defaultIfEmpty(2)
+ .subscribe(function(x) {
+ expect(x).toBe(expected.shift());
+ }, null, done);
+ });
+
+ it('should return the Observable if not empty with no default value', function(done) {
+ var expected = [1,2,3];
+ var observable = Observable.of(1,2,3);
+ observable.defaultIfEmpty()
+ .subscribe(function(x) {
+ expect(x).toBe(expected.shift());
+ }, null, done);
+ });
+
+ it('should error if the Observable errors', function(done) {
+ var observable = Observable.throw("candy");
+ observable.defaultIfEmpty(2)
+ .subscribe(function(x) {
+ throw "this should not be called";
+ }, function(err) {
+ expect(err).toBe("candy");
+ done();
+ });
+ });
+});

src/Observable.ts

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

src/Rx.ts

@@ -123,9 +123,11 @@ observableProto.subscribeOn = subscribeOn;
 
 import partition from './operators/partition';
 import toPromise from './operators/toPromise';
+import defaultIfEmpty from './operators/defaultIfEmpty';
 
 observableProto.partition = partition;
 observableProto.toPromise = toPromise;
+observableProto.defaultIfEmpty = defaultIfEmpty;
 
 import _catch from './operators/catch';
 import retryWhen from './operators/retryWhen';

src/operators/defaultIfEmpty.ts

@@ -0,0 +1,44 @@
+import Operator from '../Operator';
+import Observer from '../Observer';
+import Observable from '../Observable';
+import Subscriber from '../Subscriber';
+
+import tryCatch from '../util/tryCatch';
+import {errorObject} from '../util/errorObject';
+import bindCallback from '../util/bindCallback';
+
+export default function defaultIfEmpty<T,R>(defaultValue: R = null) : Observable<T>|Observable<R> {
+ return this.lift(new DefaultIfEmptyOperator(defaultValue));
+}
+
+export class DefaultIfEmptyOperator<T, R> extends Operator<T, R> {
+
+ constructor(private defaultValue: R) {
+ super();
+ }
+
+ call(observer: Observer<T>): Observer<T> {
+ return new DefaultIfEmptySubscriber(observer, this.defaultValue);
+ }
+}
+
+export class DefaultIfEmptySubscriber<T, R> extends Subscriber<T> {
+
+ isEmpty: boolean = true;
+
+ constructor(destination: Observer<T>, private defaultValue: R) {
+ super(destination);
+ }
+
+ _next(x) {
+ this.isEmpty = false;
+ this.destination.next(x);
+ }
+
+ _complete() {
+ if(this.isEmpty) {
+ this.destination.next(this.defaultValue);
+ }
+ this.destination.complete();
+ }
+}