带有RxJS的可暂停缓冲区

Question

我正在尝试使用RxJS流实现可切换的自动保存功能。目标是：

• 启用自动保存时，将更改发送到服务器。
• 禁用自动保存时，缓冲更改并在重新启用自动保存时将其发送到服务器。

这是我遇到的：

autoSave$ = new BehaviorSubject(true);
change$ = new Subject();

change$.pipe(
  bufferToggle(
    autoSave$.pipe(filter(autoSave => autoSave === false)),
    () => autoSave$.pipe(filter(autoSave => autoSave === true)),
  ),
  concatMap(changes => changes),
  concatMap(change => apiService.patch(change)),
).subscribe(
  () => console.log('Change sent'),
  (error) => console.error(error),
);

感谢bufferToggle，我可以在autoSave关闭时缓冲更改并在重新启用时发送它们。

问题是当启用autoSave时，没有任何东西通过。我理解这是因为bufferToggle忽略了它的开放可观察量不会发出的流量。

我觉得我应该有条件绕过bufferToggle而启用autoSave，但我的所有尝试都失败了。

任何想法实现这一目标？

Answer 1

我们可以使用bufferToggle(on, off)缓冲启动和关闭自动保存之间的事件，并使用windowToggle(off, on)打开关闭和打开之间的过滤窗口。然后我们将它们合并在一起：

const on$ = autoSave$.filter(v=>v);
const off$ = autoSave$.filter(v=>!v);

const output$ =
  Observable.merge(
    changes$
      .bufferToggle(
        off$,
        ()=>on$
      )

    changes$
      .windowToggle(
        on$,
        ()=>off$
      )
  )
  .flatMap(x=>x) // < flattern buffer and window

在https://thinkrx.io/gist/3d5161fc29b8b48194f54032fb6d2363玩这个例子

*请注意，因为缓冲区包含数组中的值 - 我在示例中使用了另一个flatMap(v=>v)来解包缓冲值。您可能希望禁用此特定行以从混合了原始值的缓冲区中获取数组。

另外，请查看我的文章“Pausable Observables in RxJS”以查看更多示例。

希望这可以帮助