javascript - Observable Race Condition，如何正确计时两个 Observables

Question

我有一个可观察的序列，每次发生调用释放事件时，我都会将该事件映射到返回 calllog json 数组的 http 请求。

我需要将从调用事件发出的值与 json 数组结合起来。然而，当 json 数组返回调用事件时，Observable 已经发出了一个新值。

const agentCallEventStream = Rx.Observable.fromEvent(call realease event)
const agentCallLogStream = agentCallEventStream.flatMap( (agentObj) => {
       return Rx.Observable.fromPromise(callLogHelper.getUserCallLogs(agentObj.agentId));
      }

        // I tried this 
    const callLogMerged = agentCallLogStream.combineLatest(agentCallEventStream)

// but the event data returned is newer than the call log data returned

我正在寻找运算符(operator)或某种方法来保持数据同步

Answer 1

如果你想合并生成promise的同一个agentObj，你可以直接在选择器函数中携带它。你会从SO的问题中找到一个很好的例子:RxJs avoid external state but still access previous values

agentCallEventStream.flatMap( (agentObj) => 
                {       
                  Rx.Observable.fromPromise(callLogHelper.getUserCallLogs(agentObj.agentId))
                    .map(function (promiseValue){return {
                         promise : promiseValue, 
                         agentObj : agentObj
                        }});
                 }

如果这不是您想要的，您应该在弹珠图中清楚地指定输入和预期输出是什么。

更新:根据 paulpadniels 的评论，这里是使用带有结果选择器功能的 flatMap 的较短版本:

agentCallEventStream.flatMap( 
    (agentObj) => Rx.Observable.fromPromise(callLogHelper.getUserCallLogs(agentObj.agentId)),
    (promiseValue) => {promise : promiseValue, agentObj : agentObj}
)

真相是 flatMap 的签名很少使用，但应该尽可能使用，因为它性能更高(避免创建额外的内部可观察值和订阅)并且更具表现力。