Action stream consolidation

packages/store/src/actions-stream.ts

@@ -7,19 +7,54 @@ import { Observable, Subject } from 'rxjs';
 export enum ActionStatus {
   Dispatched = 'DISPATCHED',
   Completed = 'COMPLETED',
-  Errored = 'Errored'
+  Canceled = 'CANCELED',
+  Errored = 'ERRORED'
 }
 
 export interface ActionContext {
   status: ActionStatus;
   action: any;
 }
 
+/**
+ * Custom Subject that ensures that subscribers are notified of values in the order that they arrived.
+ * A standard Subject does not have this guarantee.
+ * For example, given the following code:
+ * ```typescript
+ *   const subject = new Subject<string>();
+     subject.subscribe(value => {
+       if (value === 'start') subject.next('end');
+     });
+     subject.subscribe(value => { });
+     subject.next('start');
+ * ```
+ * When `subject` is a standard `Subject<T>` the second subscriber would recieve `end` and then `start`.
+ * When `subject` is a `OrderedSubject<T>` the second subscriber would recieve `start` and then `end`.
+ */
+export class OrderedSubject<T> extends Subject<T> {
+  private _itemQueue: T[] = [];
+  private _busyPushingNext = false;
+
+  next(value?: T): void {
+    if (this._busyPushingNext) {
+      this._itemQueue.unshift(value);
+      return;
+    }
+    this._busyPushingNext = true;
+    super.next(value);
+    while (this._itemQueue.length > 0) {
+      const nextValue = this._itemQueue.pop();
+      super.next(nextValue);
+    }
+    this._busyPushingNext = false;
+  }
+}
+
 /**
  * Internal Action stream that is emitted anytime an action is dispatched.
  */
 @Injectable()
-export class InternalActions extends Subject<ActionContext> {}
+export class InternalActions extends OrderedSubject<ActionContext> {}
 
 /**
  * Action stream that is emitted anytime an action is dispatched.

packages/store/src/dispatcher.ts

@@ -0,0 +1,101 @@
+import { Injectable, ErrorHandler } from '@angular/core';
+import { Observable, of, forkJoin, empty, Subject } from 'rxjs';
+import { catchError, shareReplay, filter, exhaustMap, take } from 'rxjs/operators';
+
+import { compose } from './compose';
+import { InternalActions, ActionStatus, ActionContext } from './actions-stream';
+import { StateStream } from './state-stream';
+import { PluginManager } from './plugin-manager';
+
+/**
+ * Internal Action result stream that is emitted when an action is completed.
+ * This is used as a method of returning the action result to the dispatcher
+ * for the observable returned by the dispatch(...) call.
+ * The dispatcher then asynchronously pushes the result from this stream onto the main action stream as a result.
+ */
+@Injectable()
+export class InternalDispatchedActionResults extends Subject<ActionContext> {}
+
+@Injectable()
+export class InternalDispatcher {
+  constructor(
+    private _errorHandler: ErrorHandler,
+    private _actions: InternalActions,
+    private _actionResults: InternalDispatchedActionResults,
+    private _pluginManager: PluginManager,
+    private _stateStream: StateStream
+  ) {}
+
+  /**
+   * Dispatches event(s).
+   */
+  dispatch(event: any | any[]): Observable<any> {
+    let result: Observable<any>;
+
+    if (Array.isArray(event)) {
+      result = forkJoin(event.map(a => this.dispatchSingle(a)));
+    } else {
+      result = this.dispatchSingle(event);
+    }
+
+    result.pipe(
+      catchError(err => {
+        // handle error through angular error system
+        this._errorHandler.handleError(err);
+        return of(err);
+      })
+    );
+
+    result.subscribe();
+
+    return result;
+  }
+
+  private dispatchSingle(action: any): Observable<any> {
+    const prevState = this._stateStream.getValue();
+    const plugins = this._pluginManager.plugins;
+
+    return compose([
+      ...plugins,
+      (nextState, nextAction) => {
+        if (nextState !== prevState) {
+          this._stateStream.next(nextState);
+        }
+        const actionResult$ = this.getActionResultStream(nextAction);
+        actionResult$.subscribe(ctx => this._actions.next(ctx));
+        this._actions.next({ action: nextAction, status: ActionStatus.Dispatched });
+        return this.createDispatchObservable(actionResult$);
+      }
+    ])(prevState, action) as Observable<any>;
+  }
+
+  private getActionResultStream(action: any): Observable<ActionContext> {
+    const actionResult$ = this._actionResults.pipe(
+      filter((ctx: ActionContext) => {
+        return ctx.action === action && ctx.status !== ActionStatus.Dispatched;
+      }),
+      take(1),
+      shareReplay()
+    );
+    return actionResult$;
+  }
+
+  private createDispatchObservable(actionResult$: Observable<ActionContext>): Observable<any> {
+    return actionResult$
+      .pipe(
+        exhaustMap((ctx: ActionContext) => {
+          switch (ctx.status) {
+            case ActionStatus.Completed:
+              return of(this._stateStream.getValue());
+            case ActionStatus.Errored:
+              return of(this._stateStream.getValue()); // This was previously the error value
+            // I think that this should rather
+            // return throwError(new Error('the error goes here'))
+            default:
+              return empty();
+          }
+        })
+      )
+      .pipe(shareReplay());
+  }
+}

packages/store/src/internals.ts

@@ -17,9 +17,11 @@ export interface ActionHandlerMetaData {
   type: string;
 }
 
-export type GetStateFn<T> = () => T;
-export type SetStateFn<T> = (newState: T) => void;
-export type DispatchFn = (actions: any | any[]) => Observable<any>;
+export interface InternalStateOperations<T> {
+  getState(): T;
+  setState(val: T);
+  dispatch(actions: any | any[]): Observable<void>;
+}
 
 export interface MetaDataModel {
   name: string;

packages/store/src/module.ts

@@ -3,6 +3,7 @@ import { NgModule, ModuleWithProviders, Optional, Inject } from '@angular/core';
 import { ROOT_STATE_TOKEN, FEATURE_STATE_TOKEN } from './symbols';
 import { StateFactory } from './state-factory';
 import { Actions, InternalActions } from './actions-stream';
+import { InternalDispatcher, InternalDispatchedActionResults } from './dispatcher';
 import { Store } from './store';
 import { SelectFactory } from './select';
 import { StateStream } from './state-stream';
@@ -35,16 +36,13 @@ export class NgxsRootModule {
       stateStream.next({ ...cur, ...results.defaults });
     }
 
-    store.dispatch(new InitState());
+    factory.connectActionHandlers(results.states);
 
-    if (results) {
-      factory.invokeInit(
-        () => stateStream.getValue(),
-        newState => stateStream.next(newState),
-        actions => store.dispatch(actions),
-        results.states
-      );
-    }
+    store.dispatch(new InitState()).subscribe(() => {
+      if (results) {
+        factory.invokeInit(results.states);
+      }
+    });
   }
 }
 
@@ -76,16 +74,13 @@ export class NgxsFeatureModule {
       stateStream.next({ ...cur, ...results.defaults });
     }
 
-    store.dispatch(new UpdateState());
+    factory.connectActionHandlers(results.states);
 
-    if (results) {
-      factory.invokeInit(
-        () => stateStream.getValue(),
-        newState => stateStream.next(newState),
-        actions => store.dispatch(actions),
-        results.states
-      );
-    }
+    store.dispatch(new UpdateState()).subscribe(() => {
+      if (results) {
+        factory.invokeInit(results.states);
+      }
+    });
   }
 }
 
@@ -104,6 +99,8 @@ export class NgxsModule {
         StateFactory,
         Actions,
         InternalActions,
+        InternalDispatcher,
+        InternalDispatchedActionResults,
         Store,
         StateStream,
         SelectFactory,

packages/store/src/of-action.ts

@@ -33,6 +33,15 @@ export function ofActionCompleted(...allowedTypes: any[]) {
   return ofActionOperator(allowedTypes, ActionStatus.Completed);
 }
 
+/**
+ * RxJS operator for selecting out specific actions.
+ *
+ * This will ONLY grab actions that have just been completed
+ */
+export function ofActionCanceled(...allowedTypes: any[]) {
+  return ofActionOperator(allowedTypes, ActionStatus.Canceled);
+}
+
 /**
  * RxJS operator for selecting out specific actions.
  *