Understanding the inner workings of the NgRx ecosystem is a rewarding exercise. By the time you finish reading, you should have a clear picture of how the `ngrx/effects` package operates beneath the surface, as well as how it integrates with `ngrx/store`. A particularly intriguing aspect is the lifecycle of actions throughout the system. As you are likely aware, an action serves as an input to both reducers and effects. **The framework guarantees that reducers process actions first, and only afterwards are they picked up by the effects.** Several other insights will surface along the way. Let's get started.

Overview

How to provide effects

There are several ways to register your effects: you can use EffectsModule.forRoot([effectClass]), EffectsModule.forFeature([effectClass]), or the USER_PROVIDED_EFFECTS multi token. The forRoot method should only be invoked a single time because it is responsible for instantiating other crucial services, including EffectsRunner and EffectSources.

If you opt for the token-based approach, you are required to also call either forRoot or forFeature. Relying solely on the token will not work, as it is dependent on the presence of EffectsRootModule or EffectsFeatureModule.

After the effect classes are registered, an observable is constructed with the help of EffectSources, and a subscription is established through EffectRunner. The next section will detail this process.

The values emitted by this observable will be the actual instances of the supplied classes:

// EffectsModule.forRoot(rootEffects)
{
  return {
    ngModule: EffectsRootModule,
    providers: [
      {
        // Make sure the `forRoot` static method is called only once
        provide: _ROOT_EFFECTS_GUARD,
        useFactory: _provideForRootGuard,
        deps: [[EffectsRunner, new Optional(), new SkipSelf()]],
      },
      EffectsRunner,
      EffectSources,
      Actions,
      rootEffects, // The array of classes(effects)
      {
          // Dependency for `ROOT_EFFECTS`
          provide: _ROOT_EFFECTS,
          // Providing it as an array because of how `createEffects` is implemented
          useValue: [rootEffects],
        },
        {
          // This token would be provided by the user in its separate module
          provide: USER_PROVIDED_EFFECTS,
          multi: true,
          useValue: [], // [UserProvidedEffectsClass]
        },
        {
          provide: ROOT_EFFECTS,
          useFactory: createEffects,
          deps: [Injector, _ROOT_EFFECTS, USER_PROVIDED_EFFECTS],
        },
    ],
  };
}

export function createEffects(
  injector: Injector,
  effectGroups: Type<any>[][],
  userProvidedEffectGroups: Type<any>[][]
): any[] {
  const mergedEffects: Type<any>[] = [];

  for (let effectGroup of effectGroups) {
    mergedEffects.push(...effectGroup);
  }

  for (let userProvidedEffectGroup of userProvidedEffectGroups) {
    mergedEffects.push(...userProvidedEffectGroup);
  }

  return createEffectInstances(injector, mergedEffects);
}

// Here the instances are created
export function createEffectInstances(/* ... */): any[] {
  return effects.map(effect => injector.get(effect));
}

In this scenario, EffectsRootModule injects ROOT_EFFECTS, which holds all the required instances and inserts them into the shared effects stream:

@NgModule({})
export class EffectsRootModule {
  constructor(
    private sources: EffectSources,
    runner: EffectsRunner,
    store: Store<any>,
    @Inject(ROOT_EFFECTS) rootEffects: any[],
    /* ... */
  ) {
    // Subscribe to the `effects stream`
    // The `observer` is the Store entity 
    runner.start();

    rootEffects.forEach(effectSourceInstance =>
      // Push values into the stream
      sources.addEffects(effectSourceInstance)
    );

    store.dispatch({ type: ROOT_EFFECTS_INIT });
  }

  addEffects(effectSourceInstance: any) {
    this.sources.addEffects(effectSourceInstance);
  }
}

As an aside, you can react to the initialization of root effects by including the rootEffectsInit action in one of your reducers:

createReducer(
  initialState,
  on(rootEffectsInit, (s, a) = {/* ... */})
)

The EffectsFeatureModule which is the result of calling EffectsModule.forFeature() operates in a comparable fashion, with the key distinction being that it merely adds the effect instances to the stream:

@NgModule({})
export class EffectsFeatureModule {
  constructor(
    // Make sure the essential services(EffectsRunner, EffectSources)
    // are initialized first
    root: EffectsRootModule,
    @Inject(FEATURE_EFFECTS) effectSourceGroups: any[][],
    /* ... */
  ) {
    effectSourceGroups.forEach(group =>
      group.forEach(effectSourceInstance =>
        root.addEffects(effectSourceInstance)
      )
    );
  }
}

A further distinction exists: the FEATURE_EFFECTS token is marked as a multi provider. Consequently, upon injection, you receive an array that aggregates all the classes that have been provided.

The combined effects stream

As previously indicated, the EffectsRootModule responsibilities extend beyond simple instantiation of effect classes. It also orchestrates the creation of a single, unified stream from all the individual effects belonging to the newly instantiated classes, and subscribes to it.

@NgModule({})
export class EffectsRootModule {
  constructor(
    private sources: EffectSources,
    runner: EffectsRunner,
    store: Store<any>,
    @Inject(ROOT_EFFECTS) rootEffects: any[],
    @Optional() storeRootModule: StoreRootModule,
    @Optional() storeFeatureModule: StoreFeatureModule,
    @Optional()
    @Inject(_ROOT_EFFECTS_GUARD)
    guard: any
  ) {
    runner.start(); // Creating the stream

    rootEffects.forEach(effectSourceInstance =>
      sources.addEffects(effectSourceInstance)
    );

    store.dispatch({ type: ROOT_EFFECTS_INIT });
  }

  addEffects(effectSourceInstance: any) {
    // Pushing values into the stream
    this.sources.addEffects(effectSourceInstance);
  }
}

The presence of @Optional() storeRootModule: StoreRootModule and @Optional() storeFeatureModule: StoreFeatureModule guarantees that effects are set up only after the core components of ngrx/store are ready. This preparatory phase for the store covers:

  • the construction of the reducers map: all registered reducers from both the root and feature modules are combined into one large object that defines the application's state shape
  • the State entity, which is the central repository for application data and the point where actions are processed by reducers, potentially leading to state updates
  • the Store entity serves as the intermediary between consumers of data, such as smart components, and the underlying state model
  • the ScannedActionsSubject, which is the observable stream that the effects subscribe to, albeit in an indirect manner; further details can be found in the section titled The actions observable.

By invoking runner.start(), a subscription is established to a stream formed by merging all of the registered effects. Irrespective of how they were created, such as with createEffect, all effects are merged into this single observable, from which the emitted items are actions.

// EffectsRunner

start() {
  if (!this.effectsSubscription) {
    this.effectsSubscription = this.effectSources
      .toActions()
      .subscribe(this.store);
  }
}

The observer for this stream is the Store itself. This is feasible because the store class conforms to the Observer interface:

export class Store<T = object> extends Observable<T>
  implements Observer<Action> {

    next(action: Action) {
      this.actionsObserver.next(action);
    }
  }

As a result, any action dispatched by the effects is channeled through the Store, which then forwards it to trigger any necessary state modifications.

Understanding EffectSources

This service acts as the central hub where all the registered effects are merged into a unified observable. The emitted values from this stream, which are actions, are then captured by the Store, which has the dispatch responsibility for updating the application's state. It also plays a pivotal role in enabling lifecycle hooks to be invoked.

The merging behavior is precisely what the EffectSources.toActions() method accomplishes:

export class EffectSources extends Subject<any> {
  constructor(
    /* ... */
    private store: Store<any>,
    /* ... */
  ) {
    super();
  }

  // Pushing an effect into the stream created by `toActions()`
  addEffects(effectSourceInstance: any): void {
    this.next(effectSourceInstance);
  }

  toActions(): Observable<Action> {
    return this.pipe(
      groupBy(getSourceForInstance),
      mergeMap(source$ => {
        return source$.pipe(groupBy(effectsInstance));
      }),
      mergeMap(source$ => {
        const effect$ = source$.pipe(
          exhaustMap(sourceInstance => {
            return resolveEffectSource(
              this.errorHandler,
              this.effectsErrorHandler
            )(sourceInstance);
          }),
          map(output => {
            reportInvalidActions(output, this.errorHandler);
            return output.notification;
          }),
          filter(
            (notification): notification is Notification<Action> =>
              notification.kind === 'N'
          ),
          dematerialize()
        );

        // start the stream with an INIT action
        // do this only for the first Effect instance
        const init$ = source$.pipe(
          take(1),
          filter(isOnInitEffects),
          map(instance => instance.ngrxOnInitEffects())
        );

        return merge(effect$, init$);
      })
    );
  }
}

To fully grasp how this works, let’s dissect the fundamental logic step by step:

  • data is grouped based on the origin of each instance, which is the prototype class.
    As you know, groupBy generates an observable for every new key it encounters. When another value with an existing key arrives, the operator directs it into the already-spawned observable for that key.
    This specific behavior occurs in (groupBy(getSourceForInstance)) where keys represent the classes responsible for the instances. With 3 distinct effect classes, groupBy will produce 3 separate observables.
  • the freshly grouped observables, each containing effect instances, are then grouped again, this time by their identifiers.
mergeMap(source$ => { // <- `source$` an observable that is resulted from the previous `groupBy`
  return source$.pipe(groupBy(effectsInstance));
}),

The mergeMap operator is chosen here because many observables may be produced by the first groupBy, and we want to process them all concurrently. In the scenario where a particular effect class is loaded multiple times, only a single instance will be utilized. This is possible because the default identifier for effect classes is the same, and the logic confines it to one class per identifier.

function effectsInstance(sourceInstance: any) {
  if (isOnIdentifyEffects(sourceInstance)) {
    return sourceInstance.ngrxOnIdentifyEffects();
  }

  return '';
}

The ngrxOnIdentifyEffects method, which is a requirement of the OnIdentityEffects interface, gives you the ability to assign a custom unique identifier to an effect class.

This, together with the second groupBy(effectsInstance) and the subsequent exhaustMap, will guarantee that only the very first unique instance is kept.
In a case where EffectsModule.forRoot([A, A, A]) is used and all three share the same identifier, like the default '' string, the second groupBy produces just one observable, containing all 3 items in sequence.

By relying on the exhaustMap operator,

mergeMap(source$ => {
    const effect$ = source$.pipe(
      exhaustMap(sourceInstance => { /* ... */ }),
    )
    /* ... */
  }
)

only one of the potential 3 items, which would be the first instance of A, will actually be processed.

Under the hood, exhaustMap tracks whether an inner subscription is active using a flag called hasSubscription. To verify this, you can look at the source code located here.

  • all the effects are then combined into one seamless stream
// `mergeMap` - for each emitted observable(emitted by the second `groupBy`)
// perform the same logic: merge all the effects(class properties)
// into one single observable
mergeMap(source$ => {
  const effect$ = source$.pipe(
    exhaustMap(sourceInstance => {
      return resolveEffectSource(
        this.errorHandler,
        this.effectsErrorHandler
    )(sourceInstance);
  }),
  /* ... */
)

The utility resolveEffectSource gathers all the instance properties that are observables, typically generated by createEffect(), and combines them. If the instance implements the OnRunEffects interface, the ngrxOnRunEffects lifecycle method is invoked on this new observable.

function resolveEffectSource(/* ... */): (sourceInstance: any) => Observable<EffectNotification> {
  return sourceInstance => {
    const mergedEffects$ = mergeEffects(
      sourceInstance,
      errorHandler,
      effectsErrorHandler
    );

    if (isOnRunEffects(sourceInstance)) {
      return sourceInstance.ngrxOnRunEffects(mergedEffects$);
    }

    return mergedEffects$;
  };
}

Through the ngrxOnRunEffects hook, you have the power to reshape the observable that results from the merging of all the effects belonging to that particular instance.

export function mergeEffects(
  sourceInstance: any,
  globalErrorHandler: ErrorHandler,
  effectsErrorHandler: EffectsErrorHandler
): Observable<EffectNotification> {
  const sourceName = getSourceForInstance(sourceInstance).constructor.name;

  // `getSourceMetadata(sourceInstance)` - getting all the effect class' properties
  const observables$: Observable<any>[] = getSourceMetadata(sourceInstance).map(
    ({
      propertyName,
      dispatch,
      useEffectsErrorHandler,
    }): Observable<EffectNotification> => {
      const observable$: Observable<any> =
        typeof sourceInstance[propertyName] === 'function'
          ? sourceInstance[propertyName]()
          : sourceInstance[propertyName];

      // Whether it should re-subscribe if errors occur
      const effectAction$ = useEffectsErrorHandler
        ? effectsErrorHandler(observable$, globalErrorHandler)
        : observable$;

      // You might not want the `Store` to intercept the action
      // and trigger state changes based on it
      if (dispatch === false) {
        return effectAction$.pipe(ignoreElements());
      }

      const materialized$ = effectAction$.pipe(materialize());

      return materialized$.pipe(
        map(/* ... */)
      );
    }
  );

  return merge(...observables$);
}

You can explore a more compact demonstration of this merging process in this StackBlitz example.

The line const materialized$ = effectAction$.pipe(materialize()) serves a protective purpose. It catches and effectively neutralizes any errors that are emitted if a re-subscription doesn't happen, therefore ensuring the composite stream of all merged effects remains active.

Once all the properties from an effect class have been merged into a single observable, the framework will call the ngrxOnInitEffects lifecycle method for each class, assuming it has been defined. This allows an action to be dispatched right away and exactly once:

mergeMap(source$ => {
    // Merged effects
    const effect$ = source$.pipe(
      exhaustMap(/* ... */),
      /* ... */
    );

    // `source$`'s value is an effect class instance
    const init$ = source$.pipe(
      // `take(1)` -> make sure the `exhaustMap`'s behavior is `replicated`
      // as there is only one effect class per identifier
      take(1),  
      filter(isOnInitEffects),
      // `instance.ngrxOnInitEffects()` -> Action
      map(instance => instance.ngrxOnInitEffects())
    );

    return merge(effect$, init$);
})

A visual representation of this entire sequence is shown below:

Setting Up Effects

To define an effect, you can use the createEffect() function:

type DispatchType<T> = T extends { dispatch: infer U } ? U : true;
type ObservableType<T, OriginalType> = T extends false ? OriginalType : Action;

export function createEffect<
  C extends EffectConfig,
  DT extends DispatchType<C>,
  OT extends ObservableType<DT, OT>,
  R extends Observable<OT> | ((...args: any[]) => Observable<OT>)
>(source: () => R, config?: Partial<C>): R & CreateEffectMetadata {
  const effect = source();
  const value: EffectConfig = {
    ...DEFAULT_EFFECT_CONFIG,
    ...config, // Overrides any defaults if values are provided
  };
  Object.defineProperty(effect, CREATE_EFFECT_METADATA_KEY, {
    value,
  });
  return effect as typeof effect & CreateEffectMetadata;
}

When invoked, createEffect() produces an observable that carries a property named CREATE_EFFECT_METADATA_KEY. This property holds the configuration object for that effect. During the process of combining all properties from an effects class into a single observable, each property is transformed according to its configuration. This object supports two settings:

  • dispatch: boolean – decides whether the resulting action should be forwarded to the store;
const observable$: Observable<any> =
  typeof sourceInstance[propertyName] === 'function'
    ? sourceInstance[propertyName]()
    : sourceInstance[propertyName];

const effectAction$ = useEffectsErrorHandler
  ? effectsErrorHandler(observable$, globalErrorHandler)
  : observable$;

if (dispatch === false) {
  return effectAction$.pipe(ignoreElements());
}

The ignoreElements operator filters out all notifications, except those for errors or completion.

  • useEffectsErrorHandler: boolean – determines whether errors thrown by effects (for instance, from external API requests) should be managed;
const observable$: Observable<any> =
  typeof sourceInstance[propertyName] === 'function'
    ? sourceInstance[propertyName]()
    : sourceInstance[propertyName];

const effectAction$ = useEffectsErrorHandler
  ? effectsErrorHandler(observable$, globalErrorHandler)
  : observable$;

The effectsErrorHandler option corresponds to a value supplied by EFFECTS_ERROR_HANDLER. By default, it points to defaultEffectsErrorHandler, the standard error handler for effects:

export function defaultEffectsErrorHandler<T extends Action>(
  observable$: Observable<T>,
  errorHandler: ErrorHandler,
  retryAttemptLeft: number = MAX_NUMBER_OF_RETRY_ATTEMPTS
): Observable<T> {
  return observable$.pipe(
    catchError(error => {
      if (errorHandler) errorHandler.handleError(error);
      if (retryAttemptLeft <= 1) {
        return observable$; // last attempt
      }
      // Return observable that produces this particular effect
      return defaultEffectsErrorHandler(
        observable$,
        errorHandler,
        retryAttemptLeft - 1
      );
    })
  );
}

If you're curious about why the number of retries must be capped, this issue discusses the reasoning behind it.

Whenever an error surfaces, the observable gets unsubscribed. The role of defaultEffectsErrorHandler is to re-subscribe to the _just-terminated_ observable, provided the maximum number of allowed attempts hasn't been surpassed.

For example, consider this effect:

addUser$ = createEffect(
  () => this.actions$.pipe(
    ofType(UserAction.add),
    exhaustMap(u => this.userService.add(u)),
    map(/* Map to action */)
  ),
)

If calling userService.add() results in an error that remains unhandled elsewhere, like this:

// `this.userService.add(u)` is a cold observable
exhaustMap(
  u => this.userService.add(u).pipe(catchError(err => /* Action */))
),

addUser$ will detach from the actions$ stream. The defaultEffectsErrorHandler will then re-subscribe to actions$. There's an important nuance: actions$ is actually a Subject, so upon re-subscription, no previously emitted values are replayed—only future emissions are received.

You also have the option to supply custom error handlers for your effects:

{
  provide: EFFECTS_ERROR_HANDLER,
  useValue: customErrHandler,
},

function customErrHandler (obs$, handler) {
  return obs$.pipe(
    catchError((err, caught$) => {
      console.log('caught!')
      
      // Only re-subscribe once
      // return obs$;

      // Re-subscribe every time an error occurs
      return caught$;
    }),
  )
}

In this case, customErrHandler must be a function that takes an observable$ (built from the action$ observable) and an errHandler object.

TypeScript's Contribution

Now, we'll examine how TypeScript plays a pivotal role in defining effects with createEffect.

Take this effect:

addUser$ = createEffect(
  () => this.actions$.pipe(/* ... */),
)

The inferred type for addUser$ is Observable<Action> & CreateEffectMetadata. The CreateEffectMetadata type serves as a marker to identify properties produced by createEffect(). This becomes especially handy when merging all properties into one stream.

Before we dive into why Observable<Action> appears, let's attempt the same effect with dispatch: false specified in the configuration:

addUser$ = createEffect(
  () => of(1),
  { dispatch: false }
)

Here, addUser$ will have the type Observable<number> & CreateEffectMetadata.

However, if we write:

addUser$ = createEffect(
  () => of(1),
  // { dispatch: false }
)

we'll encounter the error: Type 'number' is not assignable to type 'Action'. This indicates that the dispatch setting has a direct impact on the effect's type.

Here's how this magic is accomplished:

// U defaults to `undefined` by default
type DispatchType<T> = T extends { dispatch: infer U } ? U : true;
// `OriginalType` will be used only if the `dispatch` is explicitly set to `false`
type ObservableType<T, OriginalType> = T extends false ? OriginalType : Action;

export function createEffect<
  C extends EffectConfig, // { dispatch?: boolean, useEffectsErrorHandler?: boolean; }
  DT extends DispatchType<C>, // U(undefined | boolean) || true
  OT extends ObservableType<DT, OT>, // If `DT` is false(`dispatch` explicitly set to `false`), use the original type
  R extends Observable<OT> | ((...args: any[]) => Observable<OT>) // Use `OT` to infer the Observable's type
>(source: () => R, config?: Partial<C>) { }

With that in mind, in this snippet

addUser$ = createEffect(
  () => of(1),
  // { dispatch: false }
)

we have (reading from the bottom up):

  • R – Observable<number>
  • OT — starts out as type number
  • DT — set to false, because dispatch is explicitly configured as false
  • OT extends ObservableType<DT, OT> determines the final type of OT: false extends false ? number : Action — leads to number

On the flip side, in this snippet

addUser$ = createEffect(
  () => of(1),
  // { dispatch: false }
)

the expression OT extends ObservableType<DT, OT> translates to undefined extends false ? number : Action. As a result, OT resolves to Action, and we run into the error because addUser$ is typed as Observable<number>, whereas it should be Observable<Action>.

The actions$ stream

You've likely stumbled upon actions$ multiple times throughout this article. Here, we'll explore what it is, how it functions, and how TypeScript once again ensures a smooth developer experience.

Typically, you'd inject actions$ into an effects class like so:

constructor(private actions$: Actions) {}

export interface Action {
  type: string;
}

@Injectable()
export class Actions<V = Action> extends Observable<V> {
  constructor(@Inject(ScannedActionsSubject) source?: Observable<V>) {
    super();

    if (source) {
      this.source = source;
    }
  }

  lift<R>(operator: Operator<V, R>): Observable<R> {
    const observable = new Actions<R>();
    observable.source = this;
    observable.operator = operator;
    return observable;
  }
}

ScannedActionsSubject originates from @ngrx/store and is a Subject (hence an Observable) that emits whenever actions are dispatched—but only once the state changes have been processed. Here's the flow: when an action is dispatched via Store.dispatch(), the State entity first updates the application state by running the reducers with the current state and that action, and only then pushes the action into an actions stream generated by ScannedActionsSubject.

By assigning the Actions' source to ScannedActionsSubject, any code like this.actions$.pipe().subscribe(observer) registers that observer in ScannedActionsSubject's list. So, when the subject emits an action (e.g., subject.next(action)), all registered observers get that same action. This clarifies why every effect receives the same stream of actions, though ofType helps sieve out the relevant ones.

Here's how ScannedActionsSubject pushes notifications to its active observers:

// State
/* ... */
const stateAndAction$: Observable<{
  state: any;
  action?: Action;
}> = withLatestReducer$.pipe(
  scan<[Action, ActionReducer<T, Action>], StateActionPair<T>>(
    reduceState, // Handling state changes
    seed
  )
);

this.stateSubscription = stateAndAction$.subscribe(({ state, action }) => {
  this.next(state); // `state` -> the new state, after reducers have been invoked 
  scannedActions.next(action);
});

ofType

To decide which actions should trigger which effects, the ofType custom operator steps in:

export function ofType(
  ...allowedTypes: Array<string | ActionCreator<string, Creator>>
): OperatorFunction<Action, Action> {
  return filter((action: Action) =>
    allowedTypes.some(typeOrActionCreator => {
      if (typeof typeOrActionCreator === 'string') {
        // Comparing the string to type
        return typeOrActionCreator === action.type;
      }

      // We are filtering by ActionCreator
      return typeOrActionCreator.type === action.type;
    })
  );
}

As shown, it relies internally on the RxJS filter operator, where the predicate function's outcome depends on whether the currently emitted action is among the values supplied to ofType.

What's genuinely impressive here is the way TypeScript's capabilities are harnessed.

Regarding ofType's type inference, there are two scenarios:

  • you can pass actions generated via createAction(), which conform to the ActionCreator type;
export type ActionCreator<
  T extends string = string,
  C extends Creator = Creator // `Creator` -> a function that returns an object
> = C & TypedAction<T>; // A function that has a readonly property `type`, which also returns an object

When using ofType(action1, action2, ...), the return type becomes a union of the return types of action1, action2, and so on up to actionN:

export function ofType<
  AC extends ActionCreator<string, Creator>[],
  U extends Action = Action, // A created action
  V = ReturnType<AC[number]>

  // `U` - the type of the incoming observable
  // `V` - the type of the returned observable
>(...allowedTypes: AC): OperatorFunction<U, V>;

Our focus is on V = ReturnType<AC[number]>. Here, AC is an array of ActionCreator types (the results of createAction). The expression AC[number] yields a union of all elements within AC. For instance:

type Action<T extends string = string> = { readonly type: T; }

function createAction<P extends object, T extends string>(t: T, payload: P): P & Action<T> {
  return {
    ...payload,
    type: t,
  };
}

const actions = [createAction('type1', { name: 'andrei' }), createAction('type2', { age: 123 })];

// `(typeof actions)[number]` -> a union of types
const action: (typeof actions)[number] = {
  // We can discriminate unions with the help of the `type` property
  // because `createAction` returns an object with one `readonly` property,
  // namely `type`
  type: 'type2',
  age: 123,
  // name: 'John' -> ? error
}

Likewise, the union produced by AC[number] can be discriminated via the type property.

Now, ReturnType<Union> is equivalent to ReturnType<Union_M1 | Union_M2 | ...> (with Union_Mn denoting the n-th member of the union). It computes the return type for each individual action. The resulting union becomes the type of the observable returned by ofType.

  • you can provide strings that correspond to action types;

But since ofType only receives a collection of strings in this case, you'll need to explicitly specify a union of types expected to align with those action types to get proper inference.

Let's examine one of the overloads of Observable.pipe:

export class Observable<T> implements Subscribable<T> {
  /* ... */
  
  pipe<A, B>(op1: OperatorFunction<T, A>, op2: OperatorFunction<A, B>): Observable<B>;
  
  /* ... */
}

Here, OperatorFunction<T,A> defines the type of a function that takes an observable as its argument and yields another observable:

export interface UnaryFunction<T, R> { (source: T): R; }
export interface OperatorFunction<T, R> extends UnaryFunction<Observable<T>, Observable<R>> {}

From the snippets above, we can see that the first operator passed to pipe is a function whose sole parameter type is an observable of type T (with T being the type parameter of Observable).

Equipped with that, let's inspect the Actions class, which offers a stream of actions that effects respond to:

@Injectable()
export class Actions<V = Action> extends Observable<V> {
  constructor(@Inject(ScannedActionsSubject) source?: Observable<V>) { }
}

Notably, Actions<V = Action> itself extends Observable<V>, meaning the first operator's parameter type in pipe will be of type V.

Let's also review the other overloads of ofType:

export function ofType<
  E extends Extract<U, { type: T1 }>,
  AC extends ActionCreator<string, Creator>,
  T1 extends string | AC,
  U extends Action = Action,
  V = T1 extends string ? E : ReturnType<Extract<T1, AC>>
>(t1: T1): OperatorFunction<U, V>;

For the moment, disregard the content inside <> and pay attention to ofType's return type as well as the first operator's type in Observable.pipe:

ofType(): OperatorFunction<U, V> <---> pipe<A>(op1: OperatorFunction<T, A>)

What this tells us is that the U type parameter of ofType aligns with T, which is essentially V (derived from Actions<V extends Action> extends Observable<V>).

This is why you must supply a union of actions when injecting the Actions observable into your effects class—without it, inferring the return types of the actions an effect cares about would be impossible. Providing this union lets TypeScript do its job effectively.

export function ofType<
  E extends Extract<U, { type: T1 }>,
  AC extends ActionCreator<string, Creator>,
  T1 extends string | AC,
  U extends Action = Action,
  V = T1 extends string ? E : ReturnType<Extract<T1, AC>>
>(t1: T1): OperatorFunction<U, V>;

So, we've confirmed that U corresponds to V (from Actions<V = Action>), which, upon injection, equals the provided union of actions.
E represents the extracted action, determined by the singleton type (the type property). An action created via createAction is a function featuring a readonly type property. This allows us to pinpoint the exact action within the union V of the injected Actions<V>, since each action extends <{ type: aSingletonType }>.
Finally, the return type is V (the V in ofType), determined by a binary choice:

  • E (the inferred created action), since T1 is a singleton type, allowing TypeScript to deduce the actual action;

Here's an illustrative example that replicates this behavior:

// Can be thought of as actions
type A = { type: 'andrei' };
type J = { type: 'john' };
type JA = { type: 'jane' };

// E extends Extract<U, { type: 'andrei' | 'john' | 'jane' }>,
type Names = { type: 'andrei' | 'john' | 'jane' };

// === `createAction('john', props<{ age: number }>())`
type JSub = J & { age: number };
// === `createAction('john', props<{ city: string }>())`
type ASub = A & { city: string };

type R = Extract<ASub | JSub, A | JA | J>;
type R2 = Extract<ASub | JSub, Names>;

// After choosing the value of the `type` property
// the unions will be discriminated
const o: R = { type: 'andrei', city: 'city', };
const o2: R2 = { type: 'john', age: 18 };

TypeScript Playground.

  • ReturnType<Extract<T1, AC>>, because when T1 isn't a string subtype, it must be an action creator, so we only need its return type.

Bridging ngrx/effects and ngrx/store

With the insights gathered here and in Understanding the magic behind StoreModule of NgRx (@ngrx/store), we can now piece together the internal flow.

  • Store.dispatch()

Calling this method signals that an event demanding state updates has arrived from the UI layer (e.g. a smart component). The dispatch pushes the action (event) into an actions stream—one that is distinct from the stream dedicated to the effects:

// Store
dispatch(action) {
  this.actionsObserver.next(action);
}

Another important detail is that the Store class behaves as an observable, with the State acting as its source. The State itself is an observable, specifically a BehaviorSubject:

// Store
constructor(
  state$: StateObservable, // The `State` class
  private actionsObserver: ActionsSubject,
) {
  super();

  this.source = state$;
}

This design matters because UI components can subscribe to the Store directly to receive notifications whenever the state changes.

  • Intercept the action within the State class
// State
constructor(
  actions$: ActionsSubject, // Receive the actions dispatched from `Store`
  reducer$: ReducerObservable,
  scannedActions: ScannedActionsSubject, // The `actions stream` that belong to effects
  @Inject(INITIAL_STATE) initialState: any
) {
  const actionsOnQueue$: Observable<Action> = actions$.pipe(
    observeOn(queueScheduler)
  );
  const withLatestReducer$: Observable<
    [Action, ActionReducer<any, Action>]
  > = actionsOnQueue$.pipe(withLatestFrom(reducer$));

  const seed: StateActionPair<T> = { state: initialState };
  const stateAndAction$: Observable<{
    state: any;
    action?: Action;
  }> = withLatestReducer$.pipe(
    scan<[Action, ActionReducer<T, Action>], StateActionPair<T>>(
      // a)
      reduceState, // Invoke the reducers -> the result will be a new state
      // =====
      seed
    )
  );

  this.stateSubscription = stateAndAction$.subscribe(({ state, action }) => {
    // b)
    this.next(state); // Send the new state to the data consumer(e.g: a smart component)
    // =====

    // c)
    scannedActions.next(action); // Notify effects that an action ocurred
    // =====
  });
}
  • a): invoke the reducers using the current action and state, which yields a fresh state
  • b) deliver the updated state to the data consumers;
    Keep in mind that State serves as the source for Store. Consequently, this.next(state) makes the new state available inside the Store class, where it can be accessed via subscriptions like Store.select() or Store.pipe(select())
  • c): once the state updates have been processed and delivered to consumers, forward the action to the effects subsystem;
    If any registered effect is listening for that action, it will produce a new action. That action then re-enters the Store, restarting the cycle through steps a), b), c):
this.effectSources
  .toActions() // The action resulted from all the merged effects 
  .subscribe(this.store);

This mechanism relies on the fact that the Store can also function as a subscriber:

// Store
next(action: Action) {
  this.actionsObserver.next(action);
}

That's a wrap—thanks for following along!