使用 RxJS 重放带有时间戳的一系列事件 [英] Replay series of events with timestamps using RxJS

问题描述

如果我有一个包含 utc 时间戳和事件数据的事件数组，如下所示:[{utcts: , data: , ... ];

If I have an array of events that include a utc timestamp and event data like as follows: [{utcts: , data: , ... ];

您将如何使用 RxJS 以正确的时间差在数组中的每个项目之间重放"这些事件?假设数组按 utcts 字段排序，因此第一项具有最低值.

how would you use RxJS to "replay" those events with the correct time differentials between each item in the array? Assume the array is ordered by the utcts field so the first item has the lowest value.

这里有一组非常基本的数据可以开始使用:

here is a very basic set of data to get started:

var testdata = [
  {utcts: 1, data: 'a'},
  {utcts: 4, data: 'b'},
  {utcts: 6, data: 'c'},
  {utcts: 10, data: 'd'}
];

假设 utcts 只是从重放事件开始的秒数，该事件从 0 秒开始.

Assume the utcts is just the number of seconds from the start of replaying the event which starts at 0 seconds.

推荐答案

使用delayWhen给你定时重播.

由于给出的 utcts 是相对(不是绝对)时间，所以不需要刷新数据对象内部的时间戳.

Since utcts given is relative (not absolute) time, don't need to refresh the timestamp inside the data object.

我在控制台日志中添加了时间戳，以便我们可以看到经过的输出时间.

I have added a timestamp to the console log so we can see the elapsed output time.

请注意，额外的几毫秒是 rxjs 处理时间的典型特征.

Note the extra few milliseconds is typical of rxjs process time.

console.clear()

const testdata = [
  {utcts: 1, data: 'a'},
  {utcts: 4, data: 'b'},
  {utcts: 6, data: 'c'},
  {utcts: 10, data: 'd'}
];

const replayData = (data) => Rx.Observable.from(data)
  .delayWhen(event => Rx.Observable.of(event).delay(event.utcts * 1000))

// Show replay items with output time (in milliseconds)
const start = new Date()
replayData(testdata)
  .timestamp()
  .subscribe(x => console.log(x.value, 'at', x.timestamp - start, 'ms'))
  

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.6/Rx.js"></script>

参考 delayWhen,时间戳

这也有效，可以说更简单，但不确定哪个最好.

This also works, arguably simpler, not sure which is best.

mergeMap() 展平内部 observable，这是应用延迟所必需的.

mergeMap() flattens the inner observable, which is necessary to apply the delay.

const replayData = (data) => Rx.Observable.from(data)
  .mergeMap(event => Rx.Observable.of(event).delay(event.utcts * 1000))

这篇关于使用 RxJS 重放带有时间戳的一系列事件的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！