如何在一个公共事件属性上加入两个事件流？答案

【问题标题】：How to join two event streams on a common event property?如何在一个公共事件属性上加入两个事件流？
【发布时间】：2016-04-26 18:31:09
【问题描述】：

考虑以下两个事件流。每个事件都有一个 timestamp/ts 和 value 属性。

我想将这两个事件具有相同时间戳的流组合成一个结果流，并应用值的转换。如果一个流缺少一个时间戳（例如下例中的黄色ts=3），则应忽略该时间戳。

希望使用响应式编程库（例如 xstream 或 rxjs）来解决问题。我对反应式编程的概念很陌生，但如果有人有其他建议，我会全神贯注。谢谢！

【问题讨论】：

我就是喜欢你的图表！！关于您的问题，一般来说很难，我的意思是您需要查看整个流来进行匹配，除非您可以保证排序（即时间戳随时间增加的函数，这意味着没有值出现乱序）。否则，它可能可以使用 join 运算符来完成
谢谢 :) 我们可以假设时间戳总是按递增顺序排列的。
如果所有值都有不同的时间戳也很好
我们也可以假设，每个输入流的所有时间戳都是唯一的。
好吧，现在没时间看，但我会尝试使用 s1.join(s2, s1Val => s1.skipUntil(s2.value = s1Val), timer(0)) 但就是这样只是我的想法。 join 函数的文档记录非常糟糕，但如果你没有得到任何答案，我可以仔细看看。

标签： javascript node.js reactive-programming rxjs xstream

【解决方案1】：

只需使用 combineLatest 并仅传递具有匹配时间戳的组合。其他组合映射到null，您稍后会过滤掉。

这是 xstream 中的解决方案：

var streamOut = xs.combine(
  (a, b) => {
    if (a.ts === b.ts) {
      return {ts: a.ts, value: a.value + b.value};
    } else {
      return null;
    }
  },
  streamA, streamB
).filter(x => x !== null);

检查它是否在 JSBin 中运行：https://jsbin.com/saxawatuza/edit?js,console。

【讨论】：

非常干净的解决方案。谢谢！
一个不相关的评论，安德烈，OP使用了xstream标签，与xtream流媒体库无关。最好有一个特定的标签。
@user3743222 我同意