如何暂停效果的 concatMap 处理，但不取消完整队列？

Question

我将 NGRX 和 Effects 用于应用程序中的基本内容。该应用程序进行蓝牙文件通信并不断刷新/写入各种参数。 但有时需要暂停刷新。

我在暂停执行使用 concatMap 将操作排队到串行队列中的 NGRX 效果时遇到问题。在给出某种“继续”信号或 bluetoothService.paused 属性再次变为 false 后，仍应处理排队的操作。 concatMap 中可以有许多 ReadFromDevice 操作排队。

蓝牙服务有一个布尔属性this.bluetoothService.pauseCommunication，但是我不知道如何将它集成到效果中。我尝试了各种（可能是愚蠢的）事情，但到目前为止都失败了。不幸的是，我目前无法更改 bluetoothService 代码。

我知道我可以通过抛出错误来取消完整的 concatMap，但这不是我需要的。我只需要暂停处理，直到布尔标志变为 false。

这是我正在使用的简化示例效果

@Effect()
  readParameterFromDevice$: Observable<Action> = this.actions$.pipe(
    ofType<ReadFromDevice>(CommunicationActionTypes.ReadFromDevice),
    map(action => action.payload),
    concatMap(async request => {
        try {
          const result = await this.bluetoothService.readFromDevice(
            request
          );
          return new ReadSuccess({
             result
          });
        } catch (error) {
          return new ReadError({
             result
          });
        }
    })
  );

如果有人能指出我正确的方向，那就太好了。

Answer 1

我认为您正在寻找 buffer 运算符。

缓冲源 Observable 值直到 closeNotifier 发出。

请参阅docs。

import { fromEvent, interval } from 'rxjs';
import { buffer } from 'rxjs/operators';

const clicks = fromEvent(document, 'click');
const interval = interval(1000);
const buffered = interval.pipe(buffer(clicks));
buffered.subscribe(x => console.log(x));

或bufferToggle 运算符。

缓冲源 Observable 值，从开头的发射开始，到 closeSelector 的输出发射时结束。

见docs

import { fromEvent, interval, empty } from 'rxjs';
import { bufferToggle } from 'rxjs/operators';

const clicks = fromEvent(document, 'click');
const openings = interval(1000);
const buffered = clicks.pipe(bufferToggle(openings, i =>
  i % 2 ? interval(500) : empty()
));
buffered.subscribe(x => console.log(x));