Rxjs

Question

我开始使用 RxJs 5 做函数链并且非常喜欢它。我注意到一个我经常遇到的模式，但是找不到使用 RxJs 的方法解决它的方法。

我使用 value 获得了 一些异步数据 并且 将它们 与值本身合并。

让我用代码告诉你：

Rx.Observable.fromPromise(somePromise) // doesn't really matter
  .flatMap(value => {  // this is the original value I have
    const promise = aFunctionThatReturnsPromise(value);
    return Promise.all([promise, value])); // I created a promise using this value, but I still want to keep value as well
  })
  .map(([promiseValue, originalValue]) => ({ // I merge new produced values with the value itself and get rid of array.
    value1: promiseValue.value1,
    value2: promiseValue.value2
    originalValue: originalValue
  }))
  // ...rest of the chain...

正如所见，这意味着创建一个与 Promise.all 的组合承诺，只是为了保持 value 和一个与 map 的清理阶段，以使以下部分的事情变得更好链条。

是否有针对这种模式的内置 RxJs 结构？或者是否有另一种方法可以产生更清晰、更易于理解的解决方案？

Answer 1

使用纯 RxJS，嵌套可观察链以在上下文中保留先前的值是很常见的：

Rx.Observable.fromPromise(somePromise)
  .flatMap(value =>
    Rx.Observable.fromPromise(aFunctionThatReturnsPromise(value))
      .map((promiseValue) => ({
        value1: promiseValue.value1,
        value2: promiseValue.value2,
        originalValue: value
      }))
  )
// ...rest of the chain...

如果使用过多会很快失控，所以您通常会尽快尝试减少数据（并返回到外部流）。

Answer 2

当您想结合输入值和输出值来做某事时，人们倾向于这样写：

Rx.Observable.of('foo')
  .mergeMap(i => aFunctionThatReturnsPromise(i)
    .map(asyncResult => ({ foo: i, result: asyncResult})
  )

可以使用 mergeMap resultSelector（flatMap 如果 rxjs4）将此模式简化为以下内容：

Rx.Observable.of('foo')
  .mergeMap(
    i => aFunctionThatReturnsPromise(i),
    (i, asyncResult) => ({ syncValue: i, asyncResult })
  )
// ... rest of the chain

这使您可以访问原始输入值和收到的每个输出值。额外的好处是 mergeMap 将在不使用 Rx.Observable.fromPromise() 的情况下将您的承诺包装到可观察对象中。

Rxjs - 将异步响应与同步响应合并

Rxjs - Merging an async response with a sync response

rxjs5