javascript - Rx.Observable.groupBy 会清理空流吗？

Question

在 Node 应用程序中，我尝试使用 RxJS 处理事件流。事件流是许多文档的更改列表。我正在使用 groupBy 按 documentId 将流划分为新的流。但我想知道，一旦客户端上的文档关闭并且没有新事件添加到该 documentId 的流中，groupBy 是否会在该文档的流为空时处理它？如果没有，我将如何手动执行此操作？我想避免由创建但从未销毁的新文档流引起的内存泄漏。

Answer 1

我的建议是:

不是只有 documentChanges 可观察，有 documentEvents 可观察。

客户端在打开文档时发送documentOpened事件，在更改文档时发送documentChanged事件，在关闭文档时发送documentClosed事件.

通过同一个可观察对象发送所有 3 种类型的事件，您可以建立并保证顺序。如果客户端按此顺序发送 documentOpened、documentChanged、documentClosed 事件，则您的服务器将按此顺序查看它们。请注意，对于 2 个不同的客户端发送的事件的顺序没有任何保证。这只会让您确保特定客户端发送的事件是有序的。

然后，这就是您使用 groupByUntil 的方式:

documentEvents
    .groupByUntil(
        function (e) { return e.documentId; }, // key
        null, // element
        function (group) { // duration selector
            var documentId = group.key;
            return group.filter(function (e) { return e.eventType === 'documentClosed'; });
      })
    .flatMap(function (eventsForDocument) {
        var documentId = eventsForDocument.key;
        return eventsForDocument.whatever(...);
    })
    .subscribe(...);

另一个简单得多的选项:您可以在空闲期后让组过期。根据您对事件的处理方式，这可能绰绰有余。如果文档在 5 分钟内未被编辑，则此示例会使组过期。如果进行更多编辑，则会启动一个新组。

var idleTime = 5 * 60 * 1000;
events
    .groupByUntil(
        function(e) { return e.documentId; },
        null,
        function(g) { return g.debounce(idleTime); })
    .flatMap...