2

我正在使用 Redux Observable 并且需要在从史诗中触发动作时解决时间问题。

我有一组要循环的项目，以便对每个项目进行 AJAX 调用。收到 AJAX 响应后，我想立即执行一些操作。在原始数组中的每个项目的所有 AJAX 响应都返回后，我想触发更多操作。

timer即使原始数组尚未完成循环，如何在过期后立即触发这些操作？

const someEpic = action$ => (
    action$
    .pipe(
        ofType(SOME_ACTION),
        switchMap(({ payload }) => (
            from(payload) // This is an array of items
            .pipe(
                mergeMap(() => (
                    timer(5000) // This is my AJAX call
                    .pipe(
                        map(loadedAction),
                    )
                )),
                toArray(),
                map(anotherAction),
            )
        ))
    )
)
4

2 回答 2

1

可能最简单的方法实际上是tap在任何你想要的地方发出动作。这是假设您可以访问商店。例如：

tap(result => this.store.dispatch(...))

然而，一种更“Rx”的方式是使用拆分链multicast，然后立即重新发送一个部分（即加载进度），另一半链接toArray()以收集所有结果，然后将这些结果转换为另一个指示加载完成的动作。

import { range, Subject, of } from 'rxjs';
import { multicast, delay, merge, concatMap, map, toArray } from 'rxjs/operators';

const process = v => of(v).pipe(
  delay(1000),
  map(p => `processed: ${p}`),
);

range(1, 5)
  .pipe(
    concatMap(v => process(v)),
    multicast(
      () => new Subject(), 
      s => s.pipe(
        merge(s.pipe(toArray()))
      )
    ),
  )
  .subscribe(console.log);

现场演示：https ://stackblitz.com/edit/rxjs6-demo-k9hwtu?file=index.ts

于 2018-09-27T08:22:51.800 回答
0

这需要2个史诗。

const ajaxCallerEpic = action$ => (
    action$
    .pipe(
        ofType(AJAX_ACTION),
        switchMap(({
            payload,
            payloadId,
        }) => (
            merge(
                from(payload) // This is an array of items
                .pipe(
                    mergeMap(() => (
                        timer(5000) // This is my AJAX call
                        .pipe(
                            switchMap(() => (
                                merge(
                                    of(loadedAction),
                                    of(
                                        sentData(
                                            payloadId
                                        )
                                    ),
                                )
                            )),
                        )
                    )),
                ),
            )
        ))
    )
)

const ajaxResponsesEpic = action$ => (
    action$
    .pipe(
        ofType(AJAX_ACTION),
        switchMap(({
            payload,
            payloadId,
        }) => (
            action$
            .pipe(
                ofType(SENT_DATA_ACTION),
                filter(({ id }) => (
                    id === payloadId
                )),
                bufferCount(
                    payload
                    .length
                ),
                map(anotherAction),
            )
        ))
    )
)

重要的部分是第二部分SENT_DATA_ACTION。我在调用它时传递了一个唯一的 ID，以确保您正在收听正确的 ID。如果您不发送所有这些，只要浏览器打开，它就会一直在监听。您始终可以在内部侦听器上添加超时以确保它完成。另一个问题是如果在运行 AJAX 调用之后ajaxResponsesEpic设置action$侦听器。ajaxCallerEpic

它很可能最终成为一个race条件。为了解决这些问题，您需要先执行ajaxResponsesEpic，以便设置动作侦听器，同时在其侦听后启动 AJAX 调用。

像这样：

const ajaxCallerEpic = action$ => (
    action$
    .pipe(
        ofType(AJAX_READY_ACTION),
        switchMap(({
            payload,
            payloadId,
        }) => (
            merge(
                from(payload) // This is an array of items
                .pipe(
                    mergeMap(() => (
                        timer(5000) // This is my AJAX call
                        .pipe(
                            switchMap(() => (
                                merge(
                                    of(loadedAction),
                                    of(
                                        sentAjaxData(
                                            payloadId
                                        )
                                    ),
                                )
                            )),
                        )
                    )),
                ),
            )
        ))
    )
)

const ajaxResponsesEpic = action$ => (
    action$
    .pipe(
        ofType(AJAX_ACTION),
        switchMap(({
            payload,
            payloadId,
        }) => (
            merge(
                (
                    action$
                    .pipe(
                        ofType(SENT_AJAX_DATA_ACTION),
                        filter(({ id }) => (
                            id === payloadId
                        )),
                        bufferCount(
                            payload
                            .length
                        ),
                        map(anotherAction),
                    )
                ),
                (
                    of(ajaxReadyAction)
                ),
            )
        ))
    )
)
于 2018-09-27T23:03:11.977 回答