feat(rx): add flattening operators with default error handling strategies

Author: michaelbe812

libs/rx/operators/README.md

@@ -10,6 +10,33 @@ A set of powerful RxJS operators for building reactive Angular applications.
 
 ## Operators
 
+### Flattening operators
+
+#### `rxSwitchMap`
+
+`switchMap`-operator from `RxJs` with a default error handling strategy.
+
+Strategies:
+tbd
+
+#### `rxMergeMap`
+`mergeMap`-operator from `RxJs` with a default error handling strategy.
+
+Strategies:
+tbd
+
+#### `rxConcatMap`
+`concatMap`-operator from `RxJs` with a default error handling strategy.
+
+Strategies:
+tbd
+#### `rxExhaustMap`
+`exhaustMap`-operator from `RxJs` with a default error handling strategy.
+
+Strategies:
+tbd
+
+
 ### Filter operators
 
 - `rxFilterNull`: Filters out nullish values

libs/rx/operators/src/index.ts

@@ -6,3 +6,9 @@ export {rxWrap} from './lib/rx-wrap.operator';
 
 export * from './lib/creational/rx-source';
 export * from './lib/rx-pluck';
+
+
+export * from './lib/flattening/rx-switchmap';
+export * from './lib/flattening/rx-concatmap';
+export * from './lib/flattening/rx-mergemap';
+export * from './lib/flattening/rx-exhaustmap';

libs/rx/operators/src/lib/flattening/rx-concatmap.spec.ts

@@ -0,0 +1,14 @@
+import {catchError, concatMap, Observable, of, OperatorFunction} from "rxjs";
+
+/**
+ * RxJs concatMap operator with error handling
+ */
+export function rxConcatmap<T, R>(project: (value: T, index: number) => Observable<R>): OperatorFunction<T, R> {
+  return (source: Observable<T>) => source.pipe(
+    concatMap((value, index) => project(value, index).pipe(
+      catchError(error => of(error))
+    )),
+  );
+
+}
+

libs/rx/operators/src/lib/flattening/rx-concatmap.ts

@@ -0,0 +1,14 @@
+import {catchError, exhaustMap, Observable, of, OperatorFunction} from "rxjs";
+
+/**
+ * RxJs exhaustMap operator with error handling
+ */
+export function rxExhaustMap<T, R>(project: (value: T, index: number) => Observable<R>): OperatorFunction<T, R> {
+  return (source: Observable<T>) => source.pipe(
+    exhaustMap((value, index) => project(value, index).pipe(
+      catchError(error => of(error))
+    )),
+  );
+
+}
+

libs/rx/operators/src/lib/flattening/rx-exhaustmap.spec.ts

@@ -0,0 +1,14 @@
+import {catchError, mergeMap, Observable, of, OperatorFunction} from "rxjs";
+
+/**
+ * RxJs mergeMap operator with error handling
+ */
+export function rxMergeMap<T, R>(project: (value: T, index: number) => Observable<R>): OperatorFunction<T, R> {
+  return (source: Observable<T>) => source.pipe(
+    mergeMap((value, index) => project(value, index).pipe(
+      catchError(error => of(error))
+    )),
+  );
+
+}
+

libs/rx/operators/src/lib/flattening/rx-exhaustmap.ts

@@ -0,0 +1,57 @@
+// create marble tests for rxSwitchMap operator
+
+
+import {rxSwitchMap} from "./rx-switchmap";
+import {TestScheduler} from "rxjs/internal/testing/TestScheduler";
+import {observableMatcher} from "@angular-kit/test-helpers";
+import {map} from "rxjs";
+
+describe('rxSwitchMap', () => {
+
+  let testScheduler: TestScheduler;
+
+  beforeEach(() => {
+    testScheduler = new TestScheduler(observableMatcher);
+  });
+
+  it('should map-and-flatten each item to an Observable', () => {
+    testScheduler.run(({ hot, cold, expectObservable, expectSubscriptions }) => {
+      const e1 = hot('   --1-----3--5-------|');
+      const e1subs = '   ^------------------!';
+      const e2 = cold('    x-x-x|            ', { x: 10 });
+      //                         x-x-x|
+      //                            x-x-x|
+      const expected = ' --x-x-x-y-yz-z-z---|';
+      const values = { x: 10, y: 30, z: 50 };
+
+      const result = e1.pipe(rxSwitchMap((x) => e2.pipe(map((i) => i * +x))));
+
+      expectObservable(result).toBe(expected, values);
+      expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    });
+  });
+
+  // todo
+  /*describe('default error handling', () => {
+    it('should catch and return the raised error', () => {
+      testScheduler.run(({ hot, cold, expectObservable }) => {
+        const e1 = hot('   --1-1--|');
+        const e2 = cold('    y-x--|            ', { x: 10, y: 1 });
+        //                         x-x-x|
+        //                            x-x-x|
+        const expected = ' --e-x--|';
+        const values = { x: 10, e: 'error' };
+
+        const result = e1.pipe(rxSwitchMap((x) => e2.pipe(map((i) => {
+          if(i === 2) {
+            throw ('error');
+          }
+          return i * +x
+        }))));
+
+        expectObservable(result).toBe(expected, values);
+      });
+    });
+  });*/
+
+})

libs/rx/operators/src/lib/flattening/rx-mergemap.spec.ts

@@ -0,0 +1,14 @@
+import {catchError, Observable, of, OperatorFunction, switchMap} from "rxjs";
+
+/**
+ * RxJs switchMap operator with error handling
+ */
+export function rxSwitchMap<T, R>(project: (value: T, index: number) => Observable<R>): OperatorFunction<T, R> {
+  return (source: Observable<T>) => source.pipe(
+    switchMap((value, index) => project(value, index).pipe(
+      catchError(error => of(error))
+    )),
+  );
+
+}
+