根据流中的值暂停 RxJS 流

Question

我有一个简单的组件，带有一个按钮，可以启动和暂停由 RxJS 计时器生成的数字流。

import { Component, OnInit } from '@angular/core';

import { BehaviorSubject, Observable, timer, merge } from 'rxjs';

import { filter, bufferToggle, windowToggle, mergeMap, mergeAll, share } from 'rxjs/operators';

@Component({

  selector: 'my-app',

  template: `<button (click)="toggle()">{{ (active$ | async) ? 'Pause' : 'Play' }}</button>`,

  styleUrls: [ './app.component.css' ]

})

export class AppComponent implements OnInit {

  active$ = new BehaviorSubject<boolean>(true);

  ngOnInit(): void {

    const on$ = this.active$.pipe(filter(v => v));

    const off$ = this.active$.pipe(filter(v => !v));

    const stream$ = timer(500, 500).pipe(share());

    const out$ = merge(

      stream$.pipe(

        bufferToggle(off$, () => on$),

        mergeAll(),

      ),

      stream$.pipe(

        windowToggle(on$, () => off$),

        mergeAll(),

      ),

    );

    out$.subscribe(v => console.log(v));

  }

  toggle(): void {

    this.active$.next(!this.active$.value);

  }

这非常有效，但我需要再添加一项功能！

我需要根据满足条件的流中的值自动暂停流。

例如，如果最新值是 5 的倍数，则暂停流。

---

你有什么想法如何做到这一点吗？