feat(throwIfEmpty): adds throwIfEmpty operator

spec/operators/throwIfEmpty-spec.ts

@@ -0,0 +1,134 @@
+import { expect } from 'chai';
+import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
+import { EMPTY, of } from '../../src';
+import { EmptyError } from '../../src/internal/util/EmptyError';
+import { throwIfEmpty } from '../../src/operators';
+
+/** @test {timeout} */
+describe('throwIfEmpty', () => {
+  describe('with errorFactory', () => {
+    it('should throw if empty', () => {
+      const error = new Error('So empty inside');
+      let thrown: any;
+
+      EMPTY.pipe(
+        throwIfEmpty(() => error),
+      )
+      .subscribe({
+        error(err) {
+          thrown = err;
+        }
+      });
+
+      expect(thrown).to.equal(error);
+    });
+
+    it('should NOT throw if NOT empty', () => {
+      const error = new Error('So empty inside');
+      let thrown: any;
+
+      of('test').pipe(
+        throwIfEmpty(() => error),
+      )
+      .subscribe({
+        error(err) {
+          thrown = err;
+        }
+      });
+
+      expect(thrown).to.be.undefined;
+    });
+
+    it('should pass values through', () => {
+      const source = cold('----a---b---c---|');
+      const sub1 =        '^               !';
+      const expected =    '----a---b---c---|';
+      expectObservable(
+        source.pipe(throwIfEmpty(() => new Error('test')))
+      ).toBe(expected);
+      expectSubscriptions(source.subscriptions).toBe([sub1]);
+    });
+
+    it('should never when never', () => {
+      const source = cold('-');
+      const sub1 =        '^';
+      const expected =    '-';
+      expectObservable(
+        source.pipe(throwIfEmpty(() => new Error('test')))
+      ).toBe(expected);
+      expectSubscriptions(source.subscriptions).toBe([sub1]);
+    });
+
+    it('should error when empty', () => {
+      const source = cold('----|');
+      const sub1 =        '^   !';
+      const expected =    '----#';
+      expectObservable(
+        source.pipe(throwIfEmpty(() => new Error('test')))
+      ).toBe(expected, undefined, new Error('test'));
+      expectSubscriptions(source.subscriptions).toBe([sub1]);
+    });
+  });
+
+  describe('without errorFactory', () => {
+    it('should throw EmptyError if empty', () => {
+      let thrown: any;
+
+      EMPTY.pipe(
+        throwIfEmpty(),
+      )
+      .subscribe({
+        error(err) {
+          thrown = err;
+        }
+      });
+
+      expect(thrown).to.be.instanceof(EmptyError);
+    });
+
+    it('should NOT throw if NOT empty', () => {
+      let thrown: any;
+
+      of('test').pipe(
+        throwIfEmpty(),
+      )
+      .subscribe({
+        error(err) {
+          thrown = err;
+        }
+      });
+
+      expect(thrown).to.be.undefined;
+    });
+
+    it('should pass values through', () => {
+      const source = cold('----a---b---c---|');
+      const sub1 =        '^               !';
+      const expected =    '----a---b---c---|';
+      expectObservable(
+        source.pipe(throwIfEmpty())
+      ).toBe(expected);
+      expectSubscriptions(source.subscriptions).toBe([sub1]);
+    });
+
+    it('should never when never', () => {
+      const source = cold('-');
+      const sub1 =        '^';
+      const expected =    '-';
+      expectObservable(
+        source.pipe(throwIfEmpty())
+      ).toBe(expected);
+      expectSubscriptions(source.subscriptions).toBe([sub1]);
+    });
+
+    it('should error when empty', () => {
+      const source = cold('----|');
+      const sub1 =        '^   !';
+      const expected =    '----#';
+      expectObservable(
+        source.pipe(throwIfEmpty())
+      ).toBe(expected, undefined, new EmptyError());
+      expectSubscriptions(source.subscriptions).toBe([sub1]);
+    });
+  });
+});

src/internal/operators/throwIfEmpty.ts

@@ -0,0 +1,41 @@
+import { tap } from './tap';
+import { EmptyError } from '../util/EmptyError';
+import { MonoTypeOperatorFunction } from '../types';
+
+/**
+ * If the source observable completes without emitting a value, it will emit
+ * an error. The error will be created at that time by the optional
+ * `errorFactory` argument, otherwise, the error will be {@link ErrorEmpty}.
+ *
+ * @example
+ *
+ * const click$ = fromEvent(button, 'click');
+ *
+ * clicks$.pipe(
+ *   takeUntil(timer(1000)),
+ *   throwIfEmpty(
+ *     () => new Error('the button was not clicked within 1 second')
+ *   ),
+ * )
+ * .subscribe({
+ *   next() { console.log('The button was clicked'); },
+ *   error(err) { console.error(err); },
+ * });
+ * @param {Function} [errorFactory] A factory function called to produce the
+ * error to be thrown when the source observable completes without emitting a
+ * value.
+ */
+export const throwIfEmpty =
+  <T>(errorFactory: (() => any) = defaultErrorFactory) => tap<T>({
+    hasValue: false,
+    next() { this.hasValue = true; },
+    complete() {
+      if (!this.hasValue) {
+        throw errorFactory();
+      }
+    }
+  } as any);
+
+function defaultErrorFactory() {
+  return new EmptyError();
+}

src/operators/index.ts

@@ -90,6 +90,7 @@ export { takeWhile } from '../internal/operators/takeWhile';
 export { tap } from '../internal/operators/tap';
 export { throttle } from '../internal/operators/throttle';
 export { throttleTime } from '../internal/operators/throttleTime';
+export { throwIfEmpty } from '../internal/operators/throwIfEmpty';
 export { timeInterval } from '../internal/operators/timeInterval';
 export { timeout } from '../internal/operators/timeout';
 export { timeoutWith } from '../internal/operators/timeoutWith';