链式 redux-observable 史诗只正确触发一次

Dan*_*isp 2 rxjs rxjs5 redux-observable

我已经设置了一个等待另一部史诗完成的史诗,就像@jayphelps 在这里的回答一样:从其他史诗中调用史诗

但是我发现它似乎只运行一次。之后,我可以CART_CONFIG_READY在控制台中看到该操作,DO_THE_NEXT_THING但未触发该操作。

我尝试了mergeMap和的各种组合switchMap,有和没有,take但似乎没有任何帮助。

这是(某种)我的代码的样子。

import { NgRedux } from '@angular-redux/store';
import { Observable } from 'rxjs/Observable';
import { ActionsObservable } from 'redux-observable';

export class CartEpicsService {

checkCart = (action$: ActionsObservable<any>, store: NgRedux<any>) => {

    return action$.ofType('CHECK_CART')
        .switchMap(() => {

            console.log('___LISTENING___');

            return action$.ofType('CART_CONFIG_READY')
                .take(1) // removing this doesn't help
                .mergeMap(() => {

                    console.log('___RECEIVED___');

                    // do stuff here

                    return Observable.of({
                        type: 'DO_THE_NEXT_THING'
                    });

                })
                .startWith({
                    type: 'GET_CART_CONFIG'
                });

        });

}

getCartConfig = (action$: ActionsObservable<any>, store: NgRedux<any>) => {

    return action$.ofType('GET_CART_CONFIG')
        .switchMap(() => {

            const config = store.getState().config;

            // we already have the config
            if (config) {
                return Observable.of({
                    type: 'CART_CONFIG_READY'
                });
            }

            // otherwise load it from the server using out HTTP service
            return this.http.get('/cart/config')
                .switchMap((response) => {
                    return Observable.concat(
                        Observable.of({
                            type: 'CART_CONFIG_SUCCESS'
                        }),
                        Observable.of({
                            type: 'CART_CONFIG_READY'
                        })
                    );
                })
                .catch(error => Observable.of({
                    type: 'CART_CONFIG_ERROR',
                    error
                }));


        });

    }

}
Run Code Online (Sandbox Code Playgroud)

对于上下文,我需要来自 /cart/config 端点的响应来检查购物车的有效性。我只需要下载一次配置。

这是一个关于 JS Bin 的可运行示例:

https://jsbin.com/vovejibuwi/1/edit?js,console

jay*_*lps 5

这绝对是一个棘手的问题!

原因

state.config === true你返回一个CART_CONFIG_READY同步发出的 Observable 时,而在第一次 http 请求(或延迟,在 jsbin 中)意味着它总是异步的。

为什么这使得一个区别是在checkCart史诗返回侦听可观察到的链CART_CONFIG_READYaction$.ofType('CART_CONFIG_READY'),但也适用于一个.startWith({ type: 'GET_CART_CONFIG' })。认为它的意思GET_CART_CONFIG是要synconously被发射之前 action$.ofType('CART_CONFIG_READY')被预订,因为startWith基本简写为CONCAT,这可能会使问题,如果你熟悉它更清晰。这与执行此操作几乎完全相同:

Observable.concat(
  Observable.of({
    type: 'GET_CART_CONFIG'
  }),
  action$.ofType('CART_CONFIG_READY') // not subscribed until prior complete()s
    .take(1)
    .mergeMap(() => {
      // stuff
    })
);
Run Code Online (Sandbox Code Playgroud)

总而言之,第二次发生的事情GET_CART_CONFIG是同步调度的,getCartConfig接收它并看到配置已经在商店中,所以它同步调度CART_CONFIG_READY。但是我们还没有听到它,checkCart所以它没有得到答复。然后调用栈返回,concat 中的下一个 Observable,我们的action$.ofType('CART_CONFIG_READY')链,被订阅。但是来不及了,它监听的动作已经发出了!

解决方案

解决这个问题的一种方法是让CART_CONFIG_READY总是异步的发射,或者我们 dispatch之前开始在另一个史诗中监听它GET_CART_CONFIG

1. 发出 CART_CONFIG_READY 异步

Observable.of接受一个调度程序作为它的最后一个参数,而RxJS 支持其中的几个

在这种情况下,您可以使用(macrotask)AsyncSchedulerAsapScheduler(microtask)。在这种情况下,两者都可以工作,但它们在 JavaScript 事件循环中安排在不同的时间。如果您不熟悉事件循环任务,请查看此

我个人建议AsyncSheduler在这种情况下使用 ,因为它将提供与发出 http 请求最接近的异步行为。

import { async } from 'rxjs/scheduler/async';

// later inside your epic...

return Observable.of({
  type: 'CART_CONFIG_READY'
}, async);
Run Code Online (Sandbox Code Playgroud)

2.CART_CONFIG_READY发射前监听GET_CART_CONFIG

因为startWith是 a 的简写concat(我们不想这样做),所以我们需要使用某种形式的mergeofType首先使用我们的链,以便我们在发出之前先听。

action$.ofType('CART_CONFIG_READY')
  .take(1)
  .mergeMap(() => {
    // stuff
  })
  .merge(
    Observable.of({ type: 'GET_CART_CONFIG' })
  )

// or

Observable.merge(
  action$.ofType('CART_CONFIG_READY')
    .take(1)
    .mergeMap(() => {
      // stuff
    }),
  Observable.of({ type: 'GET_CART_CONFIG' })
)

// both are exactly the same, pick personal preference on appearance
Run Code Online (Sandbox Code Playgroud)

您只需要执行这些解决方案中的一个,但同时执行这两个解决方案也无妨。我可能会建议同时使用两者,以便使事情保持一致和预期,即使它们有点冗长。


您可能也很高兴知道它Observable.of接受任意数量的项目,这些项目将按顺序发出。所以你不需要使用concat

// before

Observable.concat(
  Observable.of({
    type: 'CART_CONFIG_SUCCESS'
  }),
  Observable.of({
    type: 'CART_CONFIG_READY'
  })
)

// after

Observable.of({
  type: 'CART_CONFIG_SUCCESS'
}, {
  type: 'CART_CONFIG_READY'
})
Run Code Online (Sandbox Code Playgroud)

非常感谢 jsbin 顺便说一句,它使调试变得更加容易。


根据您的评论进行编辑:

出于好奇,您是通过经验还是调试来解决这个问题的?

两者的结合。我已经处理了大量的异步/调度代码,排序通常是问题的根源。我扫描了代码,在脑海中勾勒出执行过程,注意到 async 与 sync 之间的差异取决于代码路径,然后我做了一个快速操作符,让我可以轻松确认订阅任何 Observable 链的顺序。

Observable.prototype.logOnSubscribe = function (msg) {
  // defer is a pretty useful Observable to learn if you haven't yet
  return Observable.defer(() => {
    console.log(msg);
    return this; // the original source
  });
};
Run Code Online (Sandbox Code Playgroud)

我把它应用到了几个地方,但最重要的是这两个:

action$.ofType('CART_CONFIG_READY')
  .take(1)
  .mergeMap(() => {
    // stuff
  })
  .logOnSubscribe('listening for CART_CONFIG_READY') // <--- here
  .startWith({
    type: 'GET_CART_CONFIG'
  });

  //  and in the other epic...

  if (hasConfig) {
    return Observable.of({
      type: 'CART_CONFIG_READY'
    })
    .logOnSubscribe('emitting CART_CONFIG_READY');  // <--- and here
  }
Run Code Online (Sandbox Code Playgroud)

它确认在第二个代码路径CART_CONFIG_READY中,在另一个史诗正在侦听它之前发出。