RxJs：如何在预定义的时间发出事件？

Question

我将一些预定义事件设置为在特定时间发生。我有一个计时器，像这样：

const timer = Rx.Observable.interval(100).timeInterval()
    .map(x => x.interval)
    .scan((ms, total) => total + ms, 0)

计时器发出接近 100,200,300,400,500 的值（尽管实际上它更像是 101,200,302,401,500...完全没问题）我也有一些我想在特定时间做的事情。例如，假设我想在以下时间做一些事情：

const stuff = Rx.Observable.from([1000, 2000, 2250, 3000, 5000]);

我想要的是将 "stuff" 和 "timer" 组合在一起，使得结果流每次在 "stuff" 中定义的时间发出一个值（或者稍晚一点）。在这种情况下，t=1000 ms、2000 ms、2250 ms、3000 ms 和 5000 ms。注意：由于间隔大小，2250 的家伙应该在 2300 左右发出。没关系。他们就是不能早点来，也不能来不止一次。

我有一个解决方案，但不是很好。它重新启动 "stuff" 每一步（在这种情况下每 100 毫秒）并对其进行过滤并取 1。我更希望，一旦从 "stuff" 发出事件，它就会消失，所以它的后续过滤器没有这些值。

在实际应用中，会有 stuff 和 stuff2，也许还有 stuff3...（但我会给他们起别的名字！）

提前致谢！我希望这很清楚。

Answer 1

如果我正确理解了您的要求，那么这应该可以通过简单的投影来实现：

const times$ = stuff.flatMap(x => Rx.Observable.timer(x));

这是一个工作示例：https://jsbin.com/negiyizibu/edit?html,js,console,output

编辑

对于第二个要求，尝试这样的事情：

const times$ = Rx.Observable
                 .from([{"val":"jeff", "t": 1000}, {"val":"fred", "t": 2500}])
                 .flatMap(x => Rx.Observable.timer(x.t).map(y => x.val));

https://jsbin.com/cegijudoci/edit?js,console,output

Answer 2

这是我根据 Matt 的解决方案编写的打字稿函数。

import {from, timer} from 'rxjs';
import {flatMap, map} from 'rxjs/operators';

export interface ActionQueueEntry {
    action: string;
    payload?: any;
    delay: number;
}

export function actionQueue(entries: ActionQueueEntry[]) {
    return from(entries).pipe(flatMap((x: any) => {
        return timer(x.delay).pipe(map(y => x));
    }));
}

const q = actionQueue([
  {action: 'say: hi', delay: 500},
  {action: 'ask: how you are', delay: 2500},
  {action: 'say: im fine', delay: 5000},
]);
q.subscribe(console.log);

RxJs：如何在预定义的时间发出事件？

RxJs: How to emit events at predefined times?

events

rxjs

rxjs5

timed