Add apply operator

The apply operator allows us to carry some context to multiple operators
at once. In most cases, `carry` would be preferred, but we have not
found any other way to provide context to `catchError`.

src/operators/index.ts

@@ -3,5 +3,6 @@ export {
   extractPayload,
   withNamespace,
   carry,
+  apply,
 } from 'rxbeach/operators/operators';
 export { mergeOperators } from 'rxbeach/operators/mergeOperators';

src/operators/operators.spec.ts

@@ -10,9 +10,11 @@ import {
   withNamespace,
   ofType,
   carry,
+  apply,
 } from 'rxbeach/operators';
 import { mockAction } from 'rxbeach/internal/testUtils';
 import { map } from 'rxjs/operators';
+import { pipe } from 'rxjs';
 
 const extractsPayload: Macro<[any]> = (t, payload) =>
   marbles(m => {
@@ -167,7 +169,7 @@ test(
 test(
   'carry should combine initial payload with the results of the operator',
   marbles(m => {
-    const source = m.hot('f', { f });
+    const source = m.hot(' f', { f });
     const expected = m.hot('1', {
       '1': [f.payload, f.payload.foo] as [FooPayload, number],
     });
@@ -177,3 +179,23 @@ test(
     ).toBeObservable(expected);
   })
 );
+
+test(
+  'apply should provide context to operators',
+  marbles(m => {
+    const source = m.hot('  f', { f });
+    const expected = m.hot('n', { n: f.payload.foo });
+
+    m.expect(
+      source.pipe(
+        extractPayload(),
+        apply(({ foo }) =>
+          pipe(
+            map(() => undefined),
+            map(() => foo)
+          )
+        )
+      )
+    ).toBeObservable(expected);
+  })
+);

src/operators/operators.ts

@@ -1,5 +1,5 @@
-import { OperatorFunction, MonoTypeOperatorFunction } from 'rxjs';
-import { map, filter, withLatestFrom } from 'rxjs/operators';
+import { OperatorFunction, MonoTypeOperatorFunction, of } from 'rxjs';
+import { map, filter, withLatestFrom, flatMap } from 'rxjs/operators';
 import { ActionWithPayload, ActionWithoutPayload } from 'rxbeach';
 import {
   UnknownActionCreatorWithPayload,
@@ -104,7 +104,7 @@ export const withNamespace = (
  * from the operator parameter
  *
  * ```
- * action$.pipe(
+ * routine(
  *   extractPayload(),
  *   carry(map(payload => payload.foo))
  *   tap(([payload, foo]) => {
@@ -122,3 +122,31 @@ export const carry = <Carried, Emitted>(
     operator,
     withLatestFrom(observable, (emitted, carried) => [carried, emitted])
   );
+
+/**
+ * A utility operator for using pipes which need a value to be present
+ * throughout the pipe.
+ *
+ * The main use for this operator is to provide context to `catchError`. `carry`
+ * should be preferred where possible.
+ *
+ * Example:
+ * ```ts
+ * routine(
+ *   ofType(myAction),
+ *   apply(action => pipe(
+ *     map(action => ...),
+ *     tap(mapped => ...),
+ *     catchError(() => {
+ *       console.log('Error from action:', action);
+ *     })
+ *   ))
+ * )
+ * ```
+ *
+ * @param operator A function that returns an operator function
+ */
+export const apply = <P, R>(
+  operator: (payload: P) => OperatorFunction<P, R>
+): OperatorFunction<P, R> =>
+  flatMap(payload => of(payload).pipe(operator(payload)));