Redux-observable: Add epics lazily through "replaceEpic" / epics called multiple times

[actra-gschuster]

Do you want to request a feature or report a bug?

Possible bug / documentation issue

What is the current behavior?

Epics run multiple times instead of once when implemented as seen below AND when implemented as per the migration docs for "custom replaceEpics implementation".

If the current behavior is a bug, please provide the steps to reproduce and a minimal demo of the problem using JSBin, StackBlitz, or similar.

this.epics$: BehaviorSubject<Array<any>> = (new BehaviorSubject([])).pipe(scan((epics: Array<any>, epic: any): Array<any> => {
    epics.push(epic);
    return epics;
})) as BehaviorSubject<Array<any>>;

// ...

const epicMiddleware: EpicMiddleware<any, any, any, any> = createEpicMiddleware();

this.epics$
    .subscribe((epics: Array<any>): void => {
        if(epics.length) {
            epicMiddleware.run(combineEpics(...epics));
        }
    });

// ...

this.epics$.next((action$: ActionsObservable, state$: StateObservable): any => {
    // ... do something...
    return EMPTY;
});

What is the expected behavior?

Epics should run once but as currently implemented, the middleware.run() pipes the new root epic to a subject that runs the root epic immediately instead of replacing it and waiting for the next action to be dispatched by the store.

Which versions of redux-observable, and which browser and OS are affected by this issue? Did this work in previous versions of redux-observable?
1.0.0-beta.1

[actra-gschuster]

I checked the migration docs again and still had no luck with the solution provided there (takeUntil and END action is dispatched), the app went to a loop freezing the browser.

Some time later I cam up with a - working - solution:

class StoreService {
    protected epics$: BehaviorSubject<Array<any>> = (new BehaviorSubject([])).pipe(scan((epics: Array<any>, epic: any): Array<any> => {
        epics.push(epic);

        return epics;
    })) as BehaviorSubject<Array<any>>;

    constructor() {
        const middlewares: Middleware[] = [];
        const epicMiddleware: EpicMiddleware<any, any, any, any> = createEpicMiddleware();
        middlewares.push(epicMiddleware);

        this.store = createStore<Map<string, any>, Action, any, any>(
            // ...,
            // ...,
            applyMiddleware(...middlewares)
        );

        // Run the root epic on the middleware
        const cancelEpicMiddleware$: Subject<void> = new Subject<void>();
        const createRootEpic: (epics: Array<Epic<any, any, any, any>>) => Epic<any, any, any, any> = (epics: Array<Epic<any, any, any, any>>): Epic<any, any, any, any> => {
            return (action$: ActionsObservable<any>, state$: StateObservable<any>, dependencies: any): any => {
                cancelEpicMiddleware$.next();

                return combineEpics(...epics)(action$, state$, dependencies).pipe(takeUntil(cancelEpicMiddleware$.asObservable()));
            }
        };

        // Subscribe on epics to create a new root epic whenever new epics are added
        this.epics$
            .pipe(takeUntilDestroy(this))
            .subscribe((epics: Array<Epic<any, any, any, any>>): void => {
                epicMiddleware.run(createRootEpic(epics));
            });
    }

    public addEpic(epic: Epic<any>): void {
        this.epics$.next(epic);
    }
}

[jayphelps]

Is the goal to add new epics lazily or to _replace_ the root epic lazily?

[jayphelps]

• Here's lazy loading
• Here's replacing the root epic without a notifying END action (i.e. just using switchMap instead of mergeMap
  Here's replace the root epic by utilizing a takeUntil(action$.pipe(ofType('END'))

[actra-gschuster]

Thanks for your help, @jayphelps
The main goal was to be able to register epics on the (abbreviated) StoreService when they get available, e.g. by lazy loaded angular modules, so to me it was equal if I add epics or replace the root epic, the result should be the same.
I've tested the implementation from my comment above and am happy so far, with the cancelEpicMiddleware$-Subject I'll also be able to cancel the epics from outside the store should I ever need to do so.

[DariaYarohovich]

I'm working on an app where I need to be able to both add epics lazily and remove them one by one at some point.

Here is what I came up with:

// epic1 and epic2 - are the epics that should run always no matter what other epics are added dynamically
const getBaseEpic = () => new BehaviorSubject(combineEpics(epic1, epic2)); 

const rootEpic = (baseEpic$) => (action$, state$, dependencies) =>
  baseEpic$.pipe(
    mergeMap(
      (epic) => epic(action$, state$, dependencies)
        .pipe(takeUntil(action$.pipe(ofType('END'))))
    )
  );


let store;

let epicMiddleware;
let baseEpic$;
let injectedEpics = {};

const configureStore = () => {
  store = {}; //  connect reducer

  epicMiddleware = createEpicMiddleware();

  baseEpic$ = getBaseEpic();
  epicMiddleware.run(rootEpic(baseEpic$));

  return store;
}

const injectEpic = (epic, key) => {
  baseEpic$.next(epic);

  // when injecting dynamic epics I want to keep track of them so that I can re-inject them when needed
  injectedEpics[key] = epic;
}

// this utility removes one epic by its key
const ejectEpic = (key) => {
  store.dispatch({ type:  'END'});

  // filter out epics which need to be re-injected
  const epicsToInject = Object.entries(injectedEpics).filter(([epicKey, _]) => epicKey !== key);
  store.injectedEpics = {};

  // re-run middleware with a new root epic 
  baseEpic$ = getBaseEpic();
  epicMiddleware.run(rootEpic(baseEpic$));

  // inject remaining epics one by one
  epicsToInject.forEach(([epicKey, epic]) => {
      injectEpic(epic, epicKey);
   });
}

So, my approach is to remove one epic by stopping all the added epics and re-injecting all of them by one anew. The question is whether this approach is good. What happens with the previously created BehaviorSubject when a new one is created? If the process of ejecting and injecting happens often won't it cause memory leaks?