每 1000 毫秒间隔获取一个事件

Question

我有无限的事件流，可以发出一些连续的事件部分，我想每毫秒每 1000 个事件发生一个。

我尝试了debounceTime/auditTime/throttleTime，但它们不包括我想要的所有事件 - 以演示我创建的行为 playground on stackblitz，它在 10 个事件的一部分中每 300 毫秒触发一次事件：

• debounceTime(1000) 将只给出事件#10

• throttleTime(1000) 将给出事件 1,5,9 但它会省略 #10 这是必不可少的

• auditTime(1000) 将提供事件4,8

我在这里想要的是获取事件1,5,9,10（每 1000 毫秒间隔一个事件）。我如何做到这一点？

const events$ = interval(300).pipe(
  map(num => `Firing event ${num + 1}`)
);

const source = events$.pipe(
  tap(console.log),
  take(10),
  // make some debouncing??
  map(x => `Received ${x}!`)
);

source.subscribe(x =>
  console.log(
    "%c" + x,
    "background: purple; color: white; padding: 3px 5px; border-radius: 3px;"
  )
);

我还尝试使用zip / combineLatest 并通过间隔but no luck with that 发出值

Answer 1

更新

基于cmets中的讨论

const events$ = timer(0, 6000).pipe(
  take(3),
  switchMap(x =>
    timer(0, 300).pipe(
      map(num => `event #${num + 1}`),
      take(x > 1 ? 9 : 10)
    )
  )
);

const source = merge(
  events$.pipe(
    tap(e => console.log(`%cStream: ${e}`, "color: blue;")),
    debounceTime(1000),
    tap(x => console.log(`%cdebounceTime captured: ${x}`, "color: red;"))
  ),
  events$.pipe(
    throttleTime(1000),
    tap(x => console.log(`%cthrottleTime captured: ${x}`, "color: green;"))
  ),
).pipe(
  // we need to avoid duplicates (like case with 9).
  // if all events aren't unique you need to use the original solution below.
  distinctUntilChanged(), // <-- if all events are unique.
  map(x => `Received ${x}!`)
);

source.subscribe(x =>
  console.log(
    "%c" + x,
    "background: purple; color: white; padding: 3px 5px; border-radius: 3px;"
  )
);

原创

我希望这就是你想要的：https://take.ms/VP7tA

const events$ = interval(300).pipe(
    map(num => `Firing event ${num + 1}`)
);

const source = concat(events$.pipe(
    tap(console.log),
    take(10),
), timer(1000).pipe(switchMapTo(EMPTY)), events$.pipe(
    tap(console.log),
    take(10),
));

let lastTimer = 0;
const last$ = new Subject<number>();
merge(
    source.pipe(
      scan((state, event) => {
        state[1] = null;
        const stamp = new Date().getTime();
        clearTimeout(lastTimer);
        if (stamp - state[0] < 1000) {
          lastTimer = setTimeout(() => last$.next(event), (stamp - state[0]) + 50);
          return state;
        }
        state[0] = stamp;
        state[1] = event;
        return state;
      }, [0, null]),
      filter(([, event]) => event !== null),
      map(([, event]) => event || 0),
    ),
    last$,
).pipe(
    map(x => `Received ${JSON.stringify(x)}!`)
).subscribe(x =>
    console.log(
        "%c" + x,
        "background: purple; color: white; padding: 3px 5px; border-radius: 3px;"
    )
);

Answer 2

这是解决此问题的一种方法：

const events$ = interval(300).pipe(
  map(num => `Firing event ${num + 1}`),
  share(),
);

const source$ = events$.pipe(
  take(10),
);

const lastValue$ = source$.pipe(last());

merge(source$.pipe(throttleTime(1000)), lastValue$).subscribe(console.log);

share() 将确保源（由 interval 生成的 observable）只订阅一次

last() 将在其源（本例中为source$）由于take(10) 而完成时返回最新的nexted 值

编辑

根据我对问题的理解，这将是另一种选择：

const events$ = merge(
  timer(300), // 1
  timer(600), // 2
  timer(900), // 3
  timer(1301), // 4
  timer(1500), // 5
  timer(1800), // 6
  timer(2302), // 7
).pipe(
  map((_, num) => `Firing event ${num + 1}`),
  share(),
);

const lastValue$ = events$.pipe(debounceTime(1000));

const source$ = events$.pipe(
  throttleTime(1000),
  buffer(lastValue$)
)
.subscribe(console.log)

• debounceTime(1000) - 如果 1 秒过去而没有任何通知

• buffer(lastValue$) - 当lastValue$ 发出时，发送收集的值

StackBlitz