How to combine ngrx entities using effects (which are returned periodically as a result of call from the frontend)?

Question

[angular] [javascript] [ngrx/entities] [ngrx/effects] [rxjs]

Following is my effects. This is used to fetch data from the backend.

//effects.ts

  loadCourierItems$ = createEffect(() =>
    this.actions$.pipe(
      ofType(CourierItemActions.loadCourierItems),
      mergeMap(action =>
        this.defaultService.getCourierItems(
        ).pipe(
          map(CourierItems => CourierItemActions.loadCourierItemsSuccess({ CourierItems })),
          catchError(error =>
            of(CourierItemActions.loadCourierItemsFailure({ error }))
          )
        )
      )
    )
  )

Following is my selector

//selector.ts

  export const selectCourierItemState = createFeatureSelector<CourierItemState>(
  CourierItemsFeatureKey
);

export const selectCourierItems = createSelector(selectCourierItemState, selectAll);

This is the component where I dispatch action to get data for the first time:

//app.component.ts

  constructor(private store: Store<CourierItemsState>) {
    this.store.dispatch(loadCourierItems())
  }

ngOnInit() {
this.courierItems$ = this.store.pipe(select(selectCourierItems))
}

//template.html


<div *ngFor="let item of courierItems$ | async as courier>
<p>{{courier.name}}</p>
<p>{{courier.loc_cur}}</p>

etc...
</div>

...

How can I do this?

What I want to achieve is that I want to make a new request every 1s with the effect (backend will send back an array of 10 items) I want to combine that items in to a single entity collection.

Updated

loadCourierItems$ = createEffect(() =>
this.actions$.pipe(
  ofType(CourierItemActions.loadCourierItems),
  exhaustMap(action => interval(1000).pipe(
    takeUntil(this.actions$.pipe(ofType(CourierItemActions.stopCourierItems))),
    exhaustMap(() =>
      this.defaultService.getCourierItems(
        action.limit,
        action.start,
        action.end,
      ).pipe(
        map(courierItems => CourierItemActions.loadCourierItemsSuccess({ courierItems })),
        catchError(error =>
          of(CourierItemActions.loadCourierItemsFailure({ error }))
        )
      )
    )
  ))
)
)

ANOTHER UPDATE If I use something like this I am being able to fetch every second and the entity collection is growing. But through this approach I am not being able to control start / stop.

    let date = 1587513626000; // date is required because the backend only sends data with a start and end date
    interval(1000).pipe(tap(_ => {
      this.store.dispatch(loadStoreItems({ limit: 10, start: date, end: date + 1000 }))
      date += 1000
    }))
      .subscribe()

REDUCER UPDATE

export const reducer = createReducer(
    initialState,
    on(CourierItemActions.loadCourierItemsSuccess,
      (state, action) => adapter.addMany(action.courierItems, state)
    ),
    on(CourierItemActions.loadCourierItemsFailure,
      (state, action) => {
        return {
          ...state,
          error: action.error
        }
      }
    ),

I've tried addMany & addAll... Both are not working. Only the items from the first call are getting in to the entity collection. But in the background the request is still continuing every 1s.

UPDATE This is the selector.

export const selectCourierItemState = createFeatureSelector<CourierItemState>(
    CourierItemsFeatureKey
  );

  export const selectCourierItems = createSelector(selectCourierItemState, selectAll);

Answer 1

If you need your effect to send a request every second - you need the interval operator.

//effects.ts

loadCourierItems$ = createEffect(() =>
  this.actions$.pipe(
    ofType(CourierItemActions.loadCourierItems),
    exhaustMap(action => interval(1000).pipe(
      startWith(0),
      exhaustMap(step =>
        this.defaultService.getCourierItems(
          action.limit,
          action.start + step * 1000,
          action.end + step * 1000,
        ).pipe(
          map(courierItems => CourierItemActions.loadCourierItemsSuccess({ courierItems })),
          catchError(error =>
            of(CourierItemActions.loadCourierItemsFailure({ error }))
          )
        )
      ),
    ),
    takeUntil(this.actions$.pipe(ofType(CourierItemActions.pauseStreamingCourierItems))),
    repeat(),
  )
)

//app.component.ts

  constructor(private store: Store<CourierItemsState>) {
  }

  ngOnInit() {
    this.store.dispatch(loadCourierItems());
    this.courierItems$ = this.store.pipe(select(selectCourierItems));
  }

  ngOnDestroy() {
    this.store.dispatch(stopCourierItems());
  }