使用RxJS对业务事件进行复杂事件处理

Question

我有一个商业事件的数据库，我们的web应用程序产生。

我想编写JavaScript应用程序，它将接收这些事件(最多几百个)，使用复杂的事件处理将在其中找到一些模式。

例如，如果我们在15分钟内有登录失败事件而没有登录事件，我们想知道它。我们使用会话ID将事件关联起来。

我更喜欢使用exist库，据我所见，RxJS似乎是正确的工具。

我的问题是如何使用原始事件的时间戳而不是当前计算机时间来构建事件流？

我看到RxJS有时间运算符，但看起来它使用的是当前时间。

更新1

我已经找到了HistoricalScheduler类，它看起来是一个很好的方向，但是没有关于它的文档，我也不确定它是否存在于RxJS中。

Answer 1

检查rxmarble示例代码。

特别是：https://github.com/staltz/rxmarbles/blob/master/src/controllers/utils.coffee，它使用Rx.VirtualTimeScheduler。

根据我的理解，您应该能够调整它以适应您的用例。

Answer 2

想必，您应该能够根据发生时的时间戳对现有事件进行排序，并将其与未来的事件合并(因为它们显然已经被排序了)。之后，您可以以相同的方式处理这些事件。

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.

不幸的是，如果事件还没有时间戳，数据就会丢失，而且Rx没有特殊的方法来复制它。