如何使用 redux-observable 编写使用 observable 的先前返回值？答案

【问题标题】：How can I write use previous returned value of an observable with redux-observable?如何使用 redux-observable 编写使用 observable 的先前返回值？
【发布时间】：2018-01-02 19:23:02
【问题描述】：

我不知道我的问题标题是否适合这个问题。我想使用 someAsync1（与 v2 相同）的返回值作为 flatMap 内部 action1 的参数。

const anEpic = ($action: ActionsObservable<MyAction>, store: Store<MyRootStore>) => {
return $action.ofType(ActionTypes.AN_ASYNC_ACTION)
    .switchMap((v1) => someAsync1(v1)) 
    .switchMap((v2) => someAsync2(v2))
    .map((v) => applyToUI(v))
    .flatMap((v) => Observable.concat(Observable.of(action1(v)), Observable.of(action2(true)))) // 
}

我想我可以通过将 v2 注入 someAsync2 的返回值来使用该值。但是那个代码看起来很恶心。使用 redux-observable 有什么聪明的方法来做到这一点？

【问题讨论】：

标签： redux rxjs redux-observable

【解决方案1】：

switchMap 在技术上意味着，切换到另一个 Observable 流。这意味着您无法保留该值，因为您的观察者现在正在观察不同的来源。

有几种方法可以将值从一个流“保留”到另一个流，具体取决于您喜欢哪一种。

1。行为主体

这是最优选的方式，因为BehaviourSubject的目的是保留Observable的值：

//initialize a BehaviourSubject
let v2BSubject = new BehaviourSubject<any>(null);

const anEpic = ($action: ActionsObservable<MyAction>, store: Store<MyRootStore>) => {
    return $action.ofType(ActionTypes.AN_ASYNC_ACTION)
        .switchMap((v1) => someAsync1(v1))
        .switchMap((v2) => {
            //store your v2 value here!
            v2BSubject.next(v2);
            return someAsync2(v2)
        })
        .map((v) => applyToUI(v))
        .flatMap((v) => {
            //get your v2 values here
            let v2Value = v2BSubject.value;
            return Observable.concat(Observable.of(action1(v)), Observable.of(action2(true)))
        }) //
}

或者您可以将其用作Observable。这样您就可以将其视为可观察的并使用 rxjs 运算符可以提供的任何内容：

.flatMap((v) => {
    return Observable.concat(Observable.of(action1(v)), v2BSubject.asObservable())
})

2。使用`.map` 传播值。

这是相当hacky，但可以完成工作。但是请注意它正在修改流源。如果您在管道中有许多操作，它可能会迅速爆发且难以管理：

 const anEpic = ($action: ActionsObservable<MyAction>, store: Store<MyRootStore>) => {
    return $action.ofType(ActionTypes.AN_ASYNC_ACTION)
        .switchMap((v1) => someAsync1(v1))
        .switchMap((v2) => {
            someAsync2(v2)
                .map(afterSomeAsync2 => {
                    return {
                        v1Value: v2,
                        v2Value: afterSomeAsync2
                    }
                })
        })
        .map(({v1Value, v2Value}) => {
            return applyToUI(v1Value).map(v1 => {
                return {
                    v1Value: v1,
                    v2Value: v2Value
                }
            })
        })
        .flatMap(({v1Value, v2Value}) => {
            return Observable.concat(Observable.of(action1(v1Value)), Observable.of(v2Value))
        })

【讨论】：

感谢您的回答。您的回答消除了我对 rxjs 的误解。谢谢。

【解决方案2】：

最简单的解决方案是将运算符直接应用于返回的内部 Observable，而不是折叠的外链。然后您可以访问发出的值，因为它们是闭包的一部分。

这可能令人困惑，但希望这段代码能清楚地说明：

const anEpic = ($action: ActionsObservable<MyAction>, store: Store<MyRootStore>) => {
  return $action.ofType(ActionTypes.AN_ASYNC_ACTION)
    .switchMap((v1) =>
      someAsync1(v1)
        .switchMap((v2) =>
          someAsync2(v2)
            .map((v) => applyToUI(v))
            .flatMap((v) => Observable.of(action1(v, v1, v2), action2(true))
        )
    ) 
  }

如果您想通过someAsync1 或someAsync2 捕获任何错误，也必须使用此模式，因为如果您让错误传播到顶级链，史诗将停止侦听未来的操作.

例如如果你的史诗是这样的：

const somethingEpic = (action$, store) => {
  return action$.ofType(SOMETHING)
    .switchMap(action => someAsync1(v1))
    .map(() => ({ type: SOMETHING_FULFILLED }))
    .catch(error => Observable.of({
      type: SOMETHING_REJECTED,
      error
    }));
  }

当错误到达catch 运算符时为时已晚，您的史诗不再监听未来的操作。您可以“重新启动”它，但这可能会产生意想不到的后果，因此最好避免这种模式。

相反，在错误传播出去之前捕获它

const somethingEpic = (action$, store) => {
  return action$.ofType(SOMETHING)
    .switchMap(action =>
      someAsync1(v1)
        .map(() => ({ type: SOMETHING_FULFILLED }))
        .catch(error => Observable.of({
          type: SOMETHING_REJECTED,
          error
        }))
    );
  }

有些人将此称为“隔离观察者链”。

还要注意我不需要将concat 与多个of 一起使用，因为of 支持任意数量的参数。

【讨论】：

1。行为主体

2。使用.map 传播值。

2。使用`.map` 传播值。