如何链接动作可观察对象以同步发出，然后发出最终值

Question

我有一个方法，它接受一个 observable 作为参数并返回另一个 observable（如果你对 Redux-Observable 有足够的了解，它有这个类型签名function (action$: Observable<Action>): Observable<Action>;

在该方法中，我需要 observable 在收到'POST_AND_GET_REQUEST' 操作时执行以下操作：

1. 使用操作的有效负载部分为数组中的每个对象同步执行 POST 请求
2. 这些同步调用完成后，执行 GET 请求
3. 当GET请求返回响应时，则返回'POST_AND_GET_SUCCESS'动作

我创建了一个 stackblitz 来说明我目前所拥有的：https://stackblitz.com/edit/rxjs-a2pwb8?file=index.ts

但这里的代码 sn-p 也是如此：

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));

我知道switchMap 的位置是错误的，但我不知道如何链接from observable 或mergeMap 运算符以获得我想要的结果。

Answer 1

在postPeriod 史诗中，我建议您通过返回单个可观察对象来结束操作，而不是返回多个可观察对象。

首先，您可以定义一个数组 (observableList)，它由一系列可观察对象组成。然后，您可以使用来自 delayed 的 observables 推送数组。最后但同样重要的是，您移动 switchMap() 逻辑，使其仅在返回 observableList 时执行。

这将确保'get success' 操作仅在请求完成后执行一次，并返回可观察对象。

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

如果 GET 请求不需要之前 post 请求的响应中的任何数据，您可以使用 concat 在末尾添加 get 操作

...
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'))
  )
)
...