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

Question

[我有一个方法，它采用一个observable作为参数，并返回另一个observable（如果您希望在Redux-Observable上尽快使用，则此类型签名为function (action$: Observable<Action>): Observable<Action>;

在该方法中，当它收到'POST_AND_GET_REQUEST'动作时，我需要观察到的内容才能执行以下操作：

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

我创建了一个堆栈闪电来说明到目前为止的内容：https://stackblitz.com/edit/rxjs-a2pwb8?file=index.ts

但是这里也是代码片段：

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可观察值或mergeMap运算符中获得所需的结果。

Answer 1

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

首先，您可以定义一个数组（observableList），该数组将包含一个可观察对象列表。然后，您可以将阵列与delayed中的观察值一起推入。最后但并非最不重要的一点是，您可以移动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请求之前不需要发布请求的响应中的任何数据，则可以使用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'))
  )
)
...