如何关联来自无限流的开始/结束事件？

Question

我有一个来自 rabbitMQ 的事件流，看起来像
Event: {id, type, timestamp}。
这些值是：
id: 一些唯一的字符串
type: (a)rrive/(d)epart

我想生成一个新的事件流，其中我将每个 ID 的到达事件与离开事件（连续）匹配。可以出现相同id的事件 例如，给定一个事件流：

id | type | time  
1  |  a   | 0  
1  |  d   | 1   
2  |  a   | 2  
3  |  a   | 3  
3  |  d   | 4  
1  |  a   | 5  
2  |  d   | 6
1  |  d   | 7  

我会生成一个Correlated:{id, duration} 类型的新流： 其中duration 是两个相关事件的时间戳差异

id | duration  
{1, 1}  
{3, 1}  
{2, 4}
{1, 1}

我已经能够按 id 对传入流进行分组，但无法找到任何有关将事件与另一个事件关联的文档。我正在使用 RxJS

Answer 1

假设到达先于出发，在groupBy之后，您可以使用pairwise合并连续事件，filter仅考虑出发，map计算持续时间，如下所示：

const source = Rx.Observable.of(
  { id: 1, type: "a", time: 0 },
  { id: 1, type: "d", time: 1 },
  { id: 2, type: "a", time: 2 },
  { id: 3, type: "a", time: 3 },
  { id: 3, type: "d", time: 4 },
  { id: 1, type: "a", time: 5 },
  { id: 2, type: "d", time: 6 },
  { id: 1, type: "d", time: 7 },
  Rx.Scheduler.async
);

const grouped = source
  .groupBy(event => event.id)
  .mergeMap(group => group
    .pairwise()
    .filter(([, last]) => last.type === "d")
    .map(([prev, last]) => ({
      id: last.id,
      duration: last.time - prev.time
    }))
  );

grouped.subscribe(value => console.log(value));

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>