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

问题描述

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

I have infinite stream of events that can emit some consecutive event portions and I want to take one event per 1000 each milliseconds.

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

I tried debounceTime / auditTime / throttleTime but they doesn't include all events I want - to demonstrate the behavior I created playground on stackblitz which fires events once per 300ms within portion of 10 events:

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

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

throttleTime(1000) will give events 1,5,9 but it will omit #10which is essential

auditTime(1000) 将给出事件 4,8

我想要的是获取事件 1,5,9,10(每 1000 毫秒间隔一个事件).我如何实现这一目标?

What I want here is to get events 1,5,9,10 (one event per 1000ms interval). How do I achieve this?

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 并通过区间 但没有运气

I also tried to play with zip / combineLatest and emitting values via interval but no luck with that

推荐答案

UPDATED

基于评论中的讨论

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

I hope that's what you wanted: 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;"
    )
);

这篇关于每 1000 毫秒间隔获取一个事件的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！