開始イベントと終了イベントを無限ストリームからどのように関連付けるのですか？

Question

私はrabbitMQのイベントのストリームを
Event: {id, type, timestamp}のように表示しています。
値は次のとおりです。
id：いくつかのユニークな文字列
type：（a）のrrive /（d）は、私が（出発予定イベントに私が到着したイベントと一致するイベントの新しいストリームを生成したい

をepart連続する）。

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 

はIタイプCorrelated:{id, duration}の新しいストリーム生成する：同じIDを持つイベントがイベントのストリーム所与例えばので を表示することができdurationは、2つの相関イベントからのタイムスタンプの差がある を

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

受信ストリームをIDでグループ化できましたが、イベントの関連付けに関するドキュメントを見つけられませんでした。私はこのように、期間を計算するために、groupBy後、あなただけの逸脱を考慮し、連続したイベントを組み合わせることfilterをpairwiseを使用してmapでき到着は出発の前にあると仮定するとRxJS

Answer 1

を使用しています：

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>