feat(store): create action stream that shows the action lifecycle (#255

)

* feat(store): create action stream that shows the action lifecycle

* feat(store): update unit test to show that all cbs are called correctly

* feat(store): update docs

* feat(store): remove ActionContext generic arg

* chore: misc cleanup

* feat(store): add ofActionErrored Operator

* feat(store): update changelog

* feat(store): update changelog

docs/advanced/action-handlers.md

@@ -1,19 +1,29 @@
 # Action Handlers
+
 Event sourcing involves modeling the state changes made by applications as an immutable sequence or “log” of events.
 Instead of focussing on current state, you focus on the changes that have occurred over time. It is the practice of
 modelling your system as a sequence of events. In NGXS, we called this Action Handlers.
 
 Typically actions directly correspond to state changes but it can be difficult to always make your component react
-based on state. As a side effect of this paradigm, we end up creating lots of intermediate state properrties
+based on state. As a side effect of this paradigm, we end up creating lots of intermediate state properties
 to do things like reset a form/etc. Action handlers let us drive our components based on state along with events
 that emit.
 
 For example, if we were to have a shopping cart and we were to delete an item out of it you might want to show
 a notification that it was successfully removed. In a pure state driven application, you might create some kind
-of message array to make the dialog show up. With Action Handlers, we can respond to the action directly. 
+of message array to make the dialog show up. With Action Handlers, we can respond to the action directly.
 
 The action handler is a observable that recieves all the actions dispatched before the state takes any action on it.
-Since its an observable, we can use pipes and we created a `ofAction` pipe to make filtering the actions easier.
+
+Actions in NGXS also have a lifecycle. Since any potential action can be async we tag actions showing whether they are "DISPATCHED" or "COMPLETED". This gives you the ability to react to actions at different points in their existence.
+
+Since its an observable, we can use pipes and created 4 of them.
+
+* `ofAction`: triggers when any of the below lifecycle events happen
+* `ofActionDispatched`: triggers when an action has been dispatched but NOT when it completes
+* `ofActionCompleted`: triggers when an action has been completed but NOT when it is dispatched
+* `ofActionErrored`: triggers when an action has caused an error to be thrown
+
 Below is a action handler that filters for `RouteNavigate` actions and then tells the router to navigate to that
 route.
 
@@ -24,7 +34,7 @@ import { Actions, ofAction } from '@ngxs/store';
 export class RouteHandler {
   constructor(private router: Router, private actions$: Actions) {
     this.actions$
-      .pipe(ofAction(RouteNavigate))
+      .pipe(ofActionDispatched(RouteNavigate))
       .subscribe(({ payload }) => this.router.navigate([payload]));
   }
 }
@@ -38,7 +48,7 @@ export class CartComponent {
   constructor(private actions$: Actions) {}
 
   ngOnInit() {
-    this.actions$.pipe(ofAction(CartDeleteSuccess)).subscribe(() => alert('Item deleted'));
+    this.actions$.pipe(ofActionComplete(CartDelete)).subscribe(() => alert('Item deleted'));
   }
 }
 ```

packages/store/src/actions-stream.ts

@@ -2,11 +2,22 @@ import { Injectable } from '@angular/core';
 import { Subject } from 'rxjs/Subject';
 import { Observable } from 'rxjs/Observable';
 
+export enum ActionStatus {
+  Dispatched = 'DISPATCHED',
+  Completed = 'COMPLETED',
+  Errored = 'Errored'
+}
+
+export interface ActionContext {
+  status: ActionStatus;
+  action: any;
+}
+
 /**
  * Internal Action stream that is emitted anytime an action is dispatched.
  */
 @Injectable()
-export class InternalActions extends Subject<any> {}
+export class InternalActions extends Subject<ActionContext> {}
 
 /**
  * Action stream that is emitted anytime an action is dispatched.

packages/store/src/of-action.ts

@@ -1,6 +1,10 @@
 import { filter } from 'rxjs/operators';
 import { getActionTypeFromInstance } from './utils';
 
+import { ActionContext, ActionStatus } from './actions-stream';
+import { map } from 'rxjs/operators/map';
+import { Observable } from 'rxjs/Observable';
+
 // TODO: Fix when RXJS 6 is released
 // import { OperatorFunction } from 'rxjs/interfaces';
 
@@ -9,12 +13,65 @@ export function ofAction<T>(...allowedTypes);
 
 /**
  * RxJS operator for selecting out specific actions.
+ *
+ * This will grab actions that have just been dispatched as well as actions that have completed
  */
 export function ofAction(...allowedTypes: any[]) {
-  const allowedMap = allowedTypes.reduce((acc: any, klass: any) => {
-    acc[klass.type] = true;
+  return ofActionOperator(allowedTypes);
+}
+
+/**
+ * RxJS operator for selecting out specific actions.
+ *
+ * This will ONLY grab actions that have just been dispatched
+ */
+export function ofActionDispatched(...allowedTypes: any[]) {
+  return ofActionOperator(allowedTypes, ActionStatus.Dispatched);
+}
+
+/**
+ * RxJS operator for selecting out specific actions.
+ *
+ * This will ONLY grab actions that have just been completed
+ */
+export function ofActionComplete(...allowedTypes: any[]) {
+  return ofActionOperator(allowedTypes, ActionStatus.Completed);
+}
+
+/**
+ * RxJS operator for selecting out specific actions.
+ *
+ * This will ONLY grab actions that have thrown an error
+ */
+export function ofActionErrored(...allowedTypes: any[]) {
+  return ofActionOperator(allowedTypes, ActionStatus.Errored);
+}
+
+function ofActionOperator(allowedTypes: any[], status?: ActionStatus) {
+  const allowedMap = createAllowedMap(allowedTypes);
+
+  return function(o: Observable<any>) {
+    return o.pipe(filterStatus(allowedMap, status), mapAction());
+  };
+}
+
+function filterStatus(allowedTypes: { [key: string]: boolean }, status?: ActionStatus) {
+  return filter((ctx: ActionContext) => {
+    const actionType = getActionTypeFromInstance(ctx.action);
+    const type = allowedTypes[actionType];
+
+    return status ? type && ctx.status === status : type;
+  });
+}
+
+function mapAction() {
+  return map((ctx: ActionContext) => ctx.action);
+}
+
+function createAllowedMap(types: any[]): { [key: string]: boolean } {
+  return types.reduce((acc: any, klass: any) => {
+    acc[getActionTypeFromInstance(klass)] = true;
+
     return acc;
   }, {});
-
-  return filter(action => allowedMap[getActionTypeFromInstance(action)]);
 }

packages/store/src/public_api.ts

@@ -4,7 +4,7 @@ export { Store } from './store';
 export { State } from './state';
 export { Select } from './select';
 export { Actions } from './actions-stream';
-export { ofAction } from './of-action';
+export { ofAction, ofActionComplete, ofActionDispatched } from './of-action';
 export { NgxsPlugin, NgxsPluginFn, StateContext } from './symbols';
 export { Selector } from './selector';
 export { getActionTypeFromInstance, actionMatcher } from './utils';

packages/store/src/state-factory.ts

@@ -8,7 +8,7 @@ import { forkJoin } from 'rxjs/observable/forkJoin';
 import { META_KEY, StateContext, ActionOptions } from './symbols';
 import { topologicalSort, buildGraph, findFullParentPath, nameToState, MetaDataModel, isObject } from './internals';
 import { getActionTypeFromInstance, setValue, getValue } from './utils';
-import { ofAction } from './of-action';
+import { ofActionDispatched } from './of-action';
 
 @Injectable()
 export class StateFactory {
@@ -117,7 +117,7 @@ export class StateFactory {
           if (result instanceof Observable) {
             result = result.pipe(
               (<ActionOptions>actionMeta.options).cancelUncompleted
-                ? takeUntil(actions$.pipe(ofAction(action.constructor)))
+                ? takeUntil(actions$.pipe(ofActionDispatched(action)))
                 : map(r => r)
             ); // act like a noop
           } else {

packages/store/src/state-stream.ts

@@ -1,20 +1,12 @@
-import { Injectable, Optional, SkipSelf } from '@angular/core';
+import { Injectable } from '@angular/core';
 import { BehaviorSubject } from 'rxjs/BehaviorSubject';
 
 /**
  * BehaviorSubject of the entire state.
  */
 @Injectable()
 export class StateStream extends BehaviorSubject<any> {
-  constructor(
-    @Optional()
-    @SkipSelf()
-    parent: StateStream
-  ) {
+  constructor() {
     super({});
-
-    if (parent) {
-      Object.assign(this, parent);
-    }
   }
 }

packages/store/src/store.ts

@@ -5,9 +5,10 @@ import { distinctUntilChanged, catchError, take, shareReplay } from 'rxjs/operat
 import { forkJoin } from 'rxjs/observable/forkJoin';
 import { map } from 'rxjs/operators/map';
 import { of } from 'rxjs/observable/of';
+import { tap } from 'rxjs/operators/tap';
 
 import { compose } from './compose';
-import { InternalActions } from './actions-stream';
+import { InternalActions, ActionStatus } from './actions-stream';
 import { StateFactory } from './state-factory';
 import { StateStream } from './state-stream';
 import { PluginManager } from './plugin-manager';
@@ -105,7 +106,7 @@ export class Store {
     return this._stateStream.getValue();
   }
 
-  private _dispatch(action): Observable<any> {
+  private _dispatch(action: any): Observable<any> {
     const prevState = this._stateStream.getValue();
     const plugins = this._pluginManager.plugins;
 
@@ -116,7 +117,7 @@ export class Store {
           this._stateStream.next(nextState);
         }
 
-        this._actions.next(nextAction);
+        this._actions.next({ action, status: ActionStatus.Dispatched });
 
         return this._storeFactory
           .invokeActions(
@@ -126,7 +127,19 @@ export class Store {
             this._actions,
             action
           )
-          .pipe(map(() => this._stateStream.getValue()));
+          .pipe(
+            tap(() => {
+              this._actions.next({ action, status: ActionStatus.Completed });
+            }),
+            map(() => {
+              return this._stateStream.getValue();
+            }),
+            catchError(err => {
+              this._actions.next({ action, status: ActionStatus.Errored });
+
+              return of(err);
+            })
+          );
       }
     ])(prevState, action) as Observable<any>).pipe(shareReplay());
   }

packages/store/tests/action.spec.ts

@@ -1,65 +1,100 @@
+import { TestBed } from '@angular/core/testing';
+
 import { Action } from '../src/action';
 import { State } from '../src/state';
 import { META_KEY } from '../src/symbols';
-import { timer } from 'rxjs/observable/timer';
-import { TestBed } from '@angular/core/testing';
 import { NgxsModule } from '../src/module';
 import { Store } from '../src/store';
 import { Actions } from '../src/actions-stream';
-import { tap } from 'rxjs/operators';
+import { ofActionComplete, ofActionDispatched, ofAction, ofActionErrored } from '../src/of-action';
+import { _throw } from 'rxjs/observable/throw';
 
 describe('Action', () => {
-  it('supports multiple actions', () => {
-    class Action1 {
-      static type = 'ACTION 1';
-    }
+  let store: Store;
+  let actions: Actions;
 
-    class Action2 {
-      static type = 'ACTION 2';
-    }
+  class Action1 {
+    static type = 'ACTION 1';
+  }
+
+  class Action2 {
+    static type = 'ACTION 2';
+  }
+
+  class ErrorAction {
+    static type = 'ErrorAction';
+  }
+
+  @State({
+    name: 'bar'
+  })
+  class BarStore {
+    @Action([Action1, Action2])
+    foo() {}
 
-    @State({
-      name: 'bar'
-    })
-    class BarStore {
-      @Action([Action1, Action2])
-      foo() {}
+    @Action(ErrorAction)
+    onError() {
+      return _throw(new Error('this is a test error'));
     }
+  }
 
+  beforeEach(() => {
+    TestBed.configureTestingModule({
+      imports: [NgxsModule.forRoot([BarStore])]
+    });
+
+    store = TestBed.get(Store);
+    actions = TestBed.get(Actions);
+  });
+
+  it('supports multiple actions', () => {
     const meta = BarStore[META_KEY];
 
     expect(meta.actions[Action1.type]).toBeDefined();
     expect(meta.actions[Action2.type]).toBeDefined();
   });
-});
 
-describe('Actions', () => {
-  it('basic', () => {
-    let happened = false;
+  it('calls actions on dispatch and on complete', () => {
+    let callbackCalledCount = 0;
 
-    class Action1 {
-      static type = 'ACTION 1';
-    }
-
-    @State({ name: 'Bar' })
-    class BarStore {
-      @Action(Action1)
-      bar() {
-        return timer(10).pipe(tap(() => (happened = true)));
-      }
-    }
+    actions.pipe(ofAction(Action1)).subscribe(action => {
+      callbackCalledCount++;
+    });
 
-    TestBed.configureTestingModule({
-      imports: [NgxsModule.forRoot([BarStore])]
+    actions.pipe(ofActionDispatched(Action1)).subscribe(action => {
+      callbackCalledCount++;
     });
 
-    const store = TestBed.get(Store);
-    const actions = TestBed.get(Actions);
+    actions.pipe(ofActionComplete(Action1)).subscribe(action => {
+      callbackCalledCount++;
 
-    actions.subscribe(action => {
-      expect(happened).toBeFalsy();
+      expect(callbackCalledCount).toBe(4);
     });
 
     store.dispatch(new Action1());
   });
+
+  it('calls only the dispatched and error action', () => {
+    let callbackCalledCount = 0;
+
+    actions.pipe(ofAction(Action1)).subscribe(action => {
+      callbackCalledCount++;
+    });
+
+    actions.pipe(ofActionDispatched(ErrorAction)).subscribe(action => {
+      callbackCalledCount++;
+    });
+
+    actions.pipe(ofActionComplete(ErrorAction)).subscribe(action => {
+      callbackCalledCount++;
+    });
+
+    actions.pipe(ofActionErrored(ErrorAction)).subscribe(action => {
+      callbackCalledCount++;
+
+      expect(callbackCalledCount).toBe(2);
+    });
+
+    store.dispatch(new ErrorAction());
+  });
 });