How to join two streams of events in a common property of an event?

Question

Consider the following two streams of events. Each event has a timestamp / ts and value property.

I want to combine these two streams, where events have the same timestamps, into a resulting stream with a value conversion attached. If one thread does not have one timestamp (for example, yellow ts=3 in the example below), this timestamp should be ignored.

I would like to solve the problem using a reactive programming library such as xstream or rxjs . I am completely new to reactive programming concepts, but if anyone has a different suggestion, I'm all ears. Thanks!

Answer 1

Just use combLatest and only pass combinations that have the appropriate timestamp. Other combinations are mapped to null , which you filter out later.

Here is 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 if it works in JSBin: https://jsbin.com/saxawatuza/edit?js,console .