Add async epic helper

.size-snapshot.json

@@ -1,18 +1,18 @@
 {
   "dist/typesafe-actions.cjs.development.js": {
-    "bundled": 9154,
-    "minified": 5296,
-    "gzipped": 1299
+    "bundled": 14036,
+    "minified": 7070,
+    "gzipped": 1842
   },
   "dist/typesafe-actions.cjs.production.js": {
-    "bundled": 9154,
-    "minified": 5296,
-    "gzipped": 1299
+    "bundled": 14036,
+    "minified": 7070,
+    "gzipped": 1842
   },
   "dist/typesafe-actions.es.production.js": {
-    "bundled": 8964,
-    "minified": 5123,
-    "gzipped": 1262,
+    "bundled": 13820,
+    "minified": 6873,
+    "gzipped": 1802,
     "treeshaked": {
       "rollup": {
         "code": 0,
@@ -24,13 +24,13 @@
     }
   },
   "dist/typesafe-actions.umd.development.js": {
-    "bundled": 10584,
-    "minified": 4086,
-    "gzipped": 1270
+    "bundled": 16098,
+    "minified": 5675,
+    "gzipped": 1802
   },
   "dist/typesafe-actions.umd.production.js": {
-    "bundled": 10584,
-    "minified": 4086,
-    "gzipped": 1270
+    "bundled": 16098,
+    "minified": 5675,
+    "gzipped": 1802
   }
 }

README.md

@@ -95,6 +95,8 @@ _Found it useful? Want more updates?_
     - [`createStandardAction`](#createstandardaction)
     - [`createCustomAction`](#createcustomaction)
     - [`createAsyncAction`](#createasyncaction)
+  - [Async-Flow Helpers API](#async-flow-helpers-api)
+    - [`createAsyncEpic`](#createasyncepic)
   - [Reducer-Creators API](#reducer-creators-api)
     - [`createReducer`](#createreducer)
   - [Action-Helpers API](#action-helpers-api)
@@ -450,6 +452,24 @@ const fetchTodosFlow: Epic<RootAction, RootAction, RootState, Services> = (actio
   );
 ```
 
+The same flow can be expressed with `createAsyncEpic`, which wires the `request`, `success`, `failure` and optional `cancel` action-creators automatically:
+
+```ts
+import { Epic, StateObservable } from 'redux-observable';
+import { from } from 'rxjs';
+import { createAsyncEpic } from 'typesafe-actions';
+
+const fetchTodosFlow: Epic<RootAction, RootAction, RootState, Services> =
+  createAsyncEpic<
+    typeof fetchTodosAsync,
+    RootAction,
+    StateObservable<RootState>,
+    Services
+  >(fetchTodosAsync, (action, _state$, { todosApi }) =>
+    from(todosApi.getAll(action.payload))
+  );
+```
+
 #### With `redux-saga` sagas
 With sagas it's not possible to achieve the same degree of type-safety as with epics because of limitations coming from `redux-saga` API design.
 
@@ -746,6 +766,58 @@ fn(fetchUsersAsync);
 
 ---
 
+### Async-Flow Helpers API
+
+#### `createAsyncEpic`
+
+_Create a redux-observable-compatible epic from an async action object._
+
+```ts
+createAsyncEpic(asyncAction, handler, options?)
+```
+
+The returned epic listens for `asyncAction.request`, calls `handler` with the request action, `state$` and services, maps emitted values to `asyncAction.success`, catches errors as `asyncAction.failure`, and stops the active request when `asyncAction.cancel` is emitted.
+
+`createAsyncEpic` requires `rxjs` as a peer dependency.
+
+Examples:
+[> Advanced Usage Examples](src/create-async-epic.spec.ts)
+
+```ts
+import { Epic, StateObservable } from 'redux-observable';
+import { from } from 'rxjs';
+import { createAsyncAction, createAsyncEpic } from 'typesafe-actions';
+
+const fetchUserAsync = createAsyncAction(
+  'FETCH_USER_REQUEST',
+  'FETCH_USER_SUCCESS',
+  'FETCH_USER_FAILURE',
+  'FETCH_USER_CANCEL'
+)<string, User, Error, string>();
+
+const fetchUserEpic: Epic<RootAction, RootAction, RootState, Services> =
+  createAsyncEpic<
+    typeof fetchUserAsync,
+    RootAction,
+    StateObservable<RootState>,
+    Services
+  >(fetchUserAsync, (action, state$, { userApi }) =>
+    from(userApi.fetch(action.payload, state$.value.auth.token))
+  );
+
+const fetchUserWithErrorMapperEpic = createAsyncEpic(
+  fetchUserAsync,
+  (action, _state$, { userApi }: Services) => from(userApi.fetch(action.payload)),
+  {
+    mapError: error => error as Error,
+  }
+);
+```
+
+[⇧ back to top](#table-of-contents)
+
+---
+
 ### Reducer-Creators API
 
 #### `createReducer`

package.json

@@ -58,10 +58,14 @@
     "rollup-plugin-size-snapshot": "0.10.0",
     "rollup-plugin-sourcemaps": "0.4.2",
     "rollup-plugin-terser": "5.1.2",
+    "rxjs": "^6.5.3",
     "ts-jest": "24.1.0",
     "tslint": "5.20.1",
     "typescript": "3.7.2"
   },
+  "peerDependencies": {
+    "rxjs": "^6.0.0"
+  },
   "keywords": [
     "typescript",
     "typesafe",

src/__snapshots__/create-async-epic.spec.ts.snap

@@ -0,0 +1,11 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`createAsyncEpic testType<Observable<FetchUserOutputAction>>(
+      epicCheck(of(fetchUserAsync.request('42')), state$, services)
+    ) (type) should match snapshot 1`] = `"Observable<FetchUserOutputAction>"`;
+
+exports[`createAsyncEpic testType<Services>(service) (type) should match snapshot 1`] = `"Services"`;
+
+exports[`createAsyncEpic testType<string>(action.payload) (type) should match snapshot 1`] = `"string"`;
+
+exports[`createAsyncEpic testType<string>(state.value.token) (type) should match snapshot 1`] = `"string"`;

src/create-async-epic.spec.snap.ts

@@ -0,0 +1,193 @@
+// tslint:disable:import-blacklist
+import { Observable, of, Subject, throwError } from 'rxjs';
+
+import { createAsyncAction } from './create-async-action';
+import { createAsyncEpic } from './create-async-epic';
+import * as T from './type-helpers';
+import { testType } from './utils/testing';
+
+type User = {
+  id: string;
+  name: string;
+};
+
+type State$ = {
+  value: {
+    token: string;
+  };
+};
+
+type Services = {
+  getUser: (id: string, token: string) => Observable<User>;
+};
+
+const user: User = {
+  id: '42',
+  name: 'Piotr',
+};
+
+const state$: State$ = {
+  value: {
+    token: 'secret',
+  },
+};
+
+const fetchUserAsync = createAsyncAction(
+  'FETCH_USER_REQUEST',
+  'FETCH_USER_SUCCESS',
+  'FETCH_USER_FAILURE',
+  'FETCH_USER_CANCEL'
+)<string, User, Error, string>();
+
+const actions = {
+  fetchUserAsync,
+};
+
+type RootAction = T.ActionType<typeof actions>;
+type FetchUserOutputAction =
+  | ReturnType<typeof fetchUserAsync.success>
+  | ReturnType<typeof fetchUserAsync.failure>;
+
+const services: Services = {
+  getUser: () => of(user),
+};
+
+// @dts-jest:group createAsyncEpic
+{
+  const epic = createAsyncEpic<
+    typeof fetchUserAsync,
+    RootAction,
+    State$,
+    Services
+  >(fetchUserAsync, (action, state, service) => {
+    // @dts-jest:pass:snap -> string
+    testType<string>(action.payload);
+    // @dts-jest:pass:snap -> string
+    testType<string>(state.value.token);
+    // @dts-jest:pass:snap -> Services
+    testType<Services>(service);
+
+    return service.getUser(action.payload, state.value.token);
+  });
+
+  // @dts-jest:pass
+  const epicCheck: (
+    action$: Observable<RootAction>,
+    state$: State$,
+    services: Services
+  ) => Observable<FetchUserOutputAction> = epic;
+
+  // @dts-jest:pass:snap -> Observable<FetchUserOutputAction>
+  testType<Observable<FetchUserOutputAction>>(
+    epicCheck(of(fetchUserAsync.request('42')), state$, services)
+  );
+}
+
+describe('createAsyncEpic', () => {
+  it('maps request handler results to success actions', () => {
+    const epic = createAsyncEpic<
+      typeof fetchUserAsync,
+      RootAction,
+      State$,
+      Services
+    >(fetchUserAsync, (action, state, service) =>
+      service.getUser(action.payload, state.value.token)
+    );
+
+    const emitted: RootAction[] = [];
+
+    epic(of(fetchUserAsync.request('42')), state$, services).subscribe(action =>
+      emitted.push(action)
+    );
+
+    expect(emitted).toEqual([fetchUserAsync.success(user)]);
+  });
+
+  it('waits for active promise handlers after the action stream completes', async () => {
+    const epic = createAsyncEpic<
+      typeof fetchUserAsync,
+      RootAction,
+      State$,
+      Services
+    >(fetchUserAsync, () => Promise.resolve(user));
+
+    const emitted: RootAction[] = [];
+
+    await new Promise(resolve => {
+      epic(of(fetchUserAsync.request('42')), state$, services).subscribe({
+        next: action => emitted.push(action),
+        complete: resolve,
+      });
+    });
+
+    expect(emitted).toEqual([fetchUserAsync.success(user)]);
+  });
+
+  it('maps request handler errors to failure actions', () => {
+    const error = new Error('not found');
+    const epic = createAsyncEpic<
+      typeof fetchUserAsync,
+      RootAction,
+      State$,
+      Services
+    >(fetchUserAsync, () => throwError(error));
+
+    const emitted: RootAction[] = [];
+
+    epic(of(fetchUserAsync.request('42')), state$, services).subscribe(action =>
+      emitted.push(action)
+    );
+
+    expect(emitted).toEqual([fetchUserAsync.failure(error)]);
+  });
+
+  it('supports mapping thrown errors before creating failure actions', () => {
+    const fetchUserWithMessageAsync = createAsyncAction(
+      'FETCH_USER_WITH_MESSAGE_REQUEST',
+      'FETCH_USER_WITH_MESSAGE_SUCCESS',
+      'FETCH_USER_WITH_MESSAGE_FAILURE'
+    )<string, User, string>();
+
+    const error = new Error('not found');
+    const epic = createAsyncEpic(
+      fetchUserWithMessageAsync,
+      () => throwError(error),
+      {
+        mapError: value => (value as Error).message,
+      }
+    );
+
+    const emitted: Array<T.ActionType<typeof fetchUserWithMessageAsync>> = [];
+
+    epic(
+      of(fetchUserWithMessageAsync.request('42')),
+      state$,
+      services
+    ).subscribe(action => emitted.push(action));
+
+    expect(emitted).toEqual([fetchUserWithMessageAsync.failure(error.message)]);
+  });
+
+  it('cancels pending request handlers when the cancel action is emitted', () => {
+    const action$ = new Subject<RootAction>();
+    const response$ = new Subject<User>();
+    const epic = createAsyncEpic<
+      typeof fetchUserAsync,
+      RootAction,
+      State$,
+      Services
+    >(fetchUserAsync, () => response$);
+    const emitted: RootAction[] = [];
+    const subscription = epic(action$, state$, services).subscribe(action =>
+      emitted.push(action)
+    );
+
+    action$.next(fetchUserAsync.request('42'));
+    action$.next(fetchUserAsync.cancel('user cancelled'));
+    response$.next(user);
+
+    expect(emitted).toEqual([]);
+
+    subscription.unsubscribe();
+  });
+});