How to chain action observables to emit synchronously and then emit a final value

Question

I have a method that takes an observable as a parameter and returns another observable (if you're up to speed on Redux-Observable, it's has this type signature function (action$: Observable<Action>): Observable<Action>;

Within that method, I need the observable to do the following when it receives a 'POST_AND_GET_REQUEST' action:

1. Uses the payload part of the action to synchronously perform a POST request for each object in an array
2. Once those synchronous calls are complete, perform a GET request
3. When the GET request returns a response, then return a 'POST_AND_GET_SUCCESS' action

I've created a stackblitz to illustrate what I have so far: https://stackblitz.com/edit/rxjs-a2pwb8?file=index.ts

But here's the code snippet as well:

import { from, of, Observable } from 'rxjs';
import { filter, delay, mapTo, mergeMap, concatMap, tap, switchMap } from 'rxjs/operators';

// GOAL
// synchronously post each period, when that's finished perform a get request
// postPeriods should only console the final POST_AND_GET_SUCCESS string that all was successful

interface ActionModel {
  type: string;
  payload: any;
}

const postPeriods = (action$: Observable<ActionModel>) => action$.pipe(
  filter(action => action.type === 'POST_AND_GET_REQUEST'),
  mergeMap(action =>
    from(action.payload).pipe(
      concatMap(period => of(`${period} post success`).pipe(
        delay(1000),
        tap(() => console.log(`${period} post success`)),
        // this is wrong - we only want this to emit once
        switchMap(response => of('get success').pipe(map => of('POST_AND_GET_SUCCESS')))
      ))
    ),
  )
);

const postRequestAction = from([{ type: 'POST_AND_GET_REQUEST', payload: ['period 1', 'period 2'] }]);

// this should be a single 'POST_AND_GET_SUCCESS'
postPeriods(postRequestAction).subscribe(val => console.log(val));

I know that where I have the switchMap is wrong, but I can't figure out how to chain off of the from observable or the mergeMap operator to get the result I want.

Answer 1

Within the postPeriod epic, I would recommend you to end the action by returning a single observable, rather than returning multiple observables.

First, you may define an array (observableList), which will consists of a list of observables. Then, you may push the array with the observables from the delayed. Last but not least, you move the switchMap() logic such that it will only be carried out when observableList is returned.

This will ensure that the 'get success' action will only be carried out once, after the requests are completed, and the observables are returned.

import { from, of, Observable } from 'rxjs';
import { filter, delay, mapTo, mergeMap, concatMap, tap, switchMap } from 'rxjs/operators';

// GOAL
// synchronously post each period, when that's finished perform a get request
// postPeriods should only console the final POST_AND_GET_SUCCESS string that all was successful

interface ActionModel {
  type: string;
  payload: any;
}

const postPeriods = (action$: Observable<ActionModel>) => action$.pipe(
  filter(action => action.type === 'POST_AND_GET_REQUEST'),
  mergeMap(action => {
    // replace the generics with a suitable typing
    const observableList: Observable<any>[] = [];

    const delayed = from(action.payload).pipe(
      concatMap(period => of(`${period} post success`)
        .pipe(
          delay(1000),
          tap(() => console.log(`${period} post success`)),
        )),
    )
    observableList.push(delayed);

    return observableList;
  }),
  switchMap(response => of('get success').pipe(map => of('POST_AND_GET_SUCCESS')))
);

const postRequestAction = from([{ type: 'POST_AND_GET_REQUEST', payload: ['period 1', 'period 2'] }]);

// this should be a single 'POST_AND_GET_SUCCESS'
postPeriods(postRequestAction).subscribe(val => {
  console.log('end:', val);
});

Answer 2

If the GET request doesn't need any data from the responses of the post requests before, you could use concat to add the get action at the end

...
mergeMap(action =>
  concat(
    from(action.payload).pipe(
      concatMap(period => of(`${period} post success`).pipe(
        delay(1000),
        tap(() => console.log(`${period} post success`)),
      ))
    ),
    of('get success').pipe(map => of('POST_AND_GET_SUCCESS'))
  )
)
...