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

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

我正在寻找操作员或某种方式来保持数据同步

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 的这种签名很少使用，但应该尽可能使用它，因为它更高效（避免创建额外的内部可观察对象和订阅）并且更具表现力。