Using RxJS for Complex Event Processing of business events

Question

I have a database of business events our web application produce.
I want to write JavaScript application that will take those events (several hundreds at most) and using Complex Event Processing will find some patterns in them.

For example, if we had login-failure event without login event in 15 minutes we want to know about it. We correlate between events using session ID.

I prefer to use exist library, from what I saw RxJS seem to be the right tool.
My question is how to build the stream of events using the timestamp of the original event and not the current computer time?

I saw that RxJS has operators of time but it looks like it is using the current time.

Update 1
I've found HistoricalScheduler class which looks like a good directions but there is no documentation about it and I'm not sure it is even exist in RxJS.

Answer 1

Check the rxmarbles example code.

Especifically: https://github.com/staltz/rxmarbles/blob/master/src/controllers/utils.coffee which uses the Rx.VirtualTimeScheduler.

From my understanding you should be able to adapt that to fit your use case

Answer 2

Presumably, you should be able to sort the existing events by the timestamp for when they occurred, and merge that with the future events (since they will obviously be sorted already). After which you can work with the events in the same manner.

var pastEvents = [
    { userId: 1, time: 100, status: 'failure' },
    { userId: 2, time: 400, status: 'success' },
    { userId: 1, time: 300, status: 'success' },
    { userId: 4, time: 200, status: 'success' },
  ]
  .sort(function (a, b) { return b.time - a.time; });

var futureEvents = getFutureEventObservable();

var events = Rx.Observable.fromArray(pastEvents).merge(futureEvents);

var groups = events.groupBy(function (x) { return x.userId; });

// do "complex event processing" here.

Unfortunately, if the events do not already have timestamps, that data is lost, and Rx has no special way of reproducing it.