RXJS Show accumulated result within a timerange

user_583666 0
Viewed 166

I need to make a function that takes in a RXJS stream and return an accumulated result for last x minute (or a given timerange) I think this is probably easier to explain if I give an example:

I'll try to illustrate what I want with marble diagrams:

I want to get the accumulated result for the last minute (60 seconds):


Stream:                  [---a---b---c---a--->
1 min in time:           [-------------------]
Result:  {a:2, b: 1, c:1}

After a couple of ms: 
Stream:                  [---a---b---c---a----a----a>
1 min in time:                  [-------------------]
Result:  {a:3, b: 1, c:1}

After a couple of ms: 
Stream:                  [---a---b---c---a----a----a----a-->
1 min in time:                         [-------------------]
Result:  {a:4, b: 0, c:0}

After a couple of ms: 
Stream:                  [---a---b---c---a----a----a----a---c---c>
1 min in time:                               [-------------------]
Result:  {a:3, b: 0, c:2}

After a couple of ms: 
Stream:                  [---a---b---c---a----a----a----a---c---c-----c-->
1 min in time:                                       [-------------------]
Result:  {a:1, b: 0, c:3}

It should emit a new result every time the accumulated result changes. (So when new value comes in to the result or when a old value pops out of the result)

rxjs javascript
2 Answers

Think of every value as having a minute-long lifetime - it "enters" when it's born, and it "exits" when it's finished.

So each value needs to be converted into an observable modeling the enter/exit events. Then all you need is a simple reducer to accumulate the events into the value/count map you want.

Creating a custom operator from this is straightforward. You can parameterize the "lifetime", and even the reducer that handles the enter/exit actions (tons of ways to accumulate and de-accumulate).

stream$.pipe(
  mergeMap((value) => {
    const add$ = of({ type: 'ENTER', value });
    const remove$ = of({ type: 'EXIT', value }).pipe(delay(60e3));
    return merge(add$, remove$);
  }),
  scan((state, { type, value }) => {
    switch (type) {
      case 'ENTER':
        return { ...state, [value]: (state[value] || 0) + 1 };
      case 'EXIT':
        return { ...state, [value]: state[value] - 1 };
      default:
        return state;
    }
  }, {})
);

Here is a stackblitz demonstrating the behavior with keypress events.

user_117510770

If I understand the problem right, you could try a solution using the scan operator with some map transformation.

The code could look like this, refer to the comments inline.

//  accumulator$ is the Observable that emits the values accumulated in the last minute
// source$ is the source Observable
const accumulator$ = source$.pipe(
  // first thing we add to each element notified by source$ the timestamp of when it is notified
  map((d) => ({ val: d, ts: Date.now() })),
  // then we use the scan operator to emit a new accumulator everytime a new
  // value is received from upstream
  scan((acc, val) => {
    const now = val.ts;
    // only the values which are not older than xxx time, in this case xxx is
    // 1000 milliseconds
    acc = acc.filter((o) => now - o.ts < 1000);
    // the last value received is added to the accumulator
    acc.push(val);
    return acc;
  }, []),
  map((accArray) =>
    // in this last transformation we transform the array in an object which contains
    // the occurrences of each value in the last minute using the reduce method
    // of javascript Array, not to be confused with the reduce operator of rxJs
    accArray.reduce((accObj, el) => {
      if (!accObj[el.val]) {
        accObj[el.val] = 0;
      }
      accObj[el.val] = accObj[el.val] + 1;
      return accObj;
    }, {})
  )
);

Here a stackblitz that demonstrate this approach.

user_56999930
Related