How to join two event streams on a common event property?

Question

Consider the following two event streams. Every event has a timestamp/ts and value property.

I want to combine these two streams where the events have the same timestamps, into a resulting stream with an applied transformation of the value. If one stream is missing one timestamp (such as the yellow ts=3 in the example below), that timestamp should be disregarded.

Would like to solve the problem using a reactive programming library such as xstream or rxjs. I'm quite new to the concepts of reactive programming, but if someone has another suggestion I'm all ears. Thanks!

Answer 1

Just use combineLatest and pass only those combinations that have a matching timestamp. The other combinations are mapped to null, which you filter out later.

Here's the solution in 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);

Check it running in JSBin: https://jsbin.com/saxawatuza/edit?js,console.