如何编写使用带有redux-observable的observable的前一个返回值？

Question

我不知道我的问题标题是否适合这个问题。我想使用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做什么是聪明的方法？

Answer 1

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

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

1. Behaviour Subject

这是最受欢迎的方式，因为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. Use .map to propagate the value.

这是相当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))
        }) 

Answer 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捕获任何错误，你也必须使用这种模式，因为如果你让错误传播到顶级链，那么epic将停止监听未来的操作。

例如如果你的史诗看起来像这样：

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支持任意数量的参数。