暂停/恢复计时器 Observable

Question

我正在用 angular/rxjs6 构建一个简单的秒表，我可以启动计时器，但无法暂停/恢复它。

  source: Observable<number>;
  subscribe: Subscription;

  start() {
    this.source = timer(0, 1000);
    this.subscribe = this.source
      .subscribe(number => {
        this.toSeconds = number % 60;
        this.toMinutes = Math.floor(number / 60);
        this.toHours = Math.floor(number / (60 * 60));

        this.seconds = (this.toSeconds < 10 ? '0' : '') + this.toSeconds;
        this.minutes = (this.toMinutes < 10 ? '0' : '') + this.toMinutes;
        this.hours = (this.toHours < 10 ? '0' : '') + this.toHours;
    });
  }

  pause() {
    this.subscribe.unsubscribe(); // not working
  }

经过大量搜索，我发现我应该使用switchMap 运算符来完成此操作，但我是 rxjs 新手，不知道如何正确操作。

任何帮助将不胜感激。

Answer 1

我今天也遇到了同样的问题（使用 Angular 实现俄罗斯方块克隆时）。这是我最终得到的结果：

import { Subject, timer } from 'rxjs';

export class Timer {
  private timeElapsed = 0;
  private timer = null;
  private subscription = null;

  private readonly step: number;

  update = new Subject<number>();

  constructor(step: number) {
    this.timeElapsed = 0;
    this.step = step;
  }

  start() {
    this.timer = timer(this.step, this.step);
    this.subscription = this.timer.subscribe(() => {
      this.timeElapsed = this.timeElapsed + this.step;
      this.update.next(this.timeElapsed);
    });
  }

  pause() {
    if (this.timer) {
      this.subscription.unsubscribe();
      this.timer = null;
    } else {
      this.start();
    }
  }

  stop() {
    if (this.timer) {
      this.subscription.unsubscribe();
      this.timer = null;
    }
  }
}

在我的游戏服务中，我是这样使用它的：

  init() {
    this.timer = new Timer(50);
    this.timer.start();
    this.timer.update.subscribe(timeElapsed => {
      if (timeElapsed % 1000 === 0) {
        this.step(); // step() runs one game iteration
      }
    });
  }

  togglePause() {
    this.timer.pause();
  }

注意：我是 Angular/RxJS 的新手，所以我不确定上面的代码是否良好。但它有效。

Answer 2

这是一个使用 rxjs 6 的 node.js sn-p。除非按下 p，否则将发出计时器事件。再次按下时，发射继续（ctrl-c 将退出）。

在内部，当暂停器发出false 时，实际上启动了一个新计时器。因此，我们在 (concat) 暂停器前面加上 false 发射来启动第一个计时器。 2 个 scan 运算符管理链的状态（暂停切换器 + 计数器）。

import { timer, concat, NEVER, of, fromEvent } from 'rxjs';
import { scan, tap, filter, switchMap } from 'rxjs/operators';
import { emitKeypressEvents } from 'readline';

process.stdin.setRawMode(true);
emitKeypressEvents(process.stdin);

const keypresses$ = fromEvent(process.stdin, 'keypress', (_, key) => key);
const pauser$ = keypresses$.pipe(
  tap(key => {
    if (key && key.ctrl && key.name == 'c') {
      process.exit(0);
    }
  }),
  filter(key => key.name === 'p'),
  scan(acc => !acc, false),
);

const starter$ = of(false);
concat(starter$, pauser$)
  .pipe(
    switchMap(stopped => (stopped ? NEVER : timer(0, 1000))),
    scan(acc => acc + 1, 0),
  )
  .subscribe(console.log);

Answer 3

我从未使用过timer() 函数，但您可以设置一个这样的标志。

  source: Observable<number>;
  subscribe: Subscription;
  timerPaused: boolean = false;


  start() {
    this.seconds = 0;
    this.minutes = 0;
    this.hours = 0;
    this.time = 0;

    this.source = timer(0, 1000);
    this.subscribe = this.source
      .subscribe(number => {
        if(!this.timerPaused) {
          this.toSeconds = this.time % 60;
          this.toMinutes = Math.floor(this.time / 60);
          this.toHours = Math.floor(this.time / (60 * 60));

          this.seconds = (this.toSeconds < 10 ? '0' : '') + this.toSeconds;
          this.minutes = (this.toMinutes < 10 ? '0' : '') + Math.floor(number / 60);
          this.hours = (this.toHours < 10 ? '0' : '') + Math.floor(number / (60 * 60));
          this.time += 1000
        }
      });
  }

  onPause() {
    this.timerPaused = true;
  }

  onResume() {
    this.timerPaused = false;
  }

  ngOnDestroy() {
    this.subscribe.unsubscribe();
  }

Answer 4

为了任何搜索的人的利益，这是我实现的可暂停 rxjs 计时器。它需要一个可观察的暂停，以便您可以随意暂停/取消暂停。

用法：

pausableTimer(1000, this.myPauseObservable)
  .subscribe(() => { /* do your thing */ };

代码：

import { BehaviorSubject, combineLatest, Observable, timer } from 'rxjs';
import { delayWhen, startWith } from 'rxjs/operators';

export function pausableTimer(
  delayMs: number = 0,
  pause: Observable<boolean>
): Observable<number> {
  return new Observable<number>((observer: any) => {
    // The sequence of delays we'd like to listen to.
    // Start off with zero & paused = false so that combineLatest always fires at least once.
    const delays = new BehaviorSubject(0);
    pause = pause.pipe(startWith(false));

    // Piped delays that emit at the time given.
    const temporalDelays = delays.pipe(delayWhen((val) => timer(val)));

    let startTime = Date.now();
    let totalPauseTimeMs = 0;
    let lastPaused: number | undefined;

    // This listens to pause/unpause events, as well as timers.
    const sub = combineLatest([pause, temporalDelays]).subscribe(
      ([paused, delay]) => {
        if (paused) {
          // If we're paused, we never want to complete, even if the timer expires.
          if (lastPaused === undefined) {
            lastPaused = Date.now();
          }

          return;
        }

        // If we've just un-paused, add on the paused time.
        if (lastPaused !== undefined) {
          totalPauseTimeMs += Date.now() - lastPaused;
          lastPaused = undefined;
        }

        // Look at how much time has expired.
        const totalElapsed = Date.now() - startTime;
        const remainingTime = delayMs + totalPauseTimeMs - totalElapsed;

        if (remainingTime <= 0) {
          // We're done!
          observer.next(totalElapsed);
          observer.complete();
        } else {
          // We're not done.  If there's not already a timer running, start a new one.
          const lastTimerFinished = delay === delays.value;
          if (lastTimerFinished) {
            delays.next(remainingTime);
          }
        }
      }
    );

    observer.add(() => sub.unsubscribe());
  });
}