使用 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 忽略了即将到来的流量，而其开放的 observable 没有发出。

我觉得我应该有一个条件来绕过bufferToggle，同时启用autoSave，但我所有的尝试都惨遭失败。

有什么想法可以实现吗？

Answer 1

我们可以使用bufferToggle(on, off)在自动保存on和off之间缓冲事件，并在off和之间打开一个过滤窗口>on 使用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”以查看更多示例。

希望对你有帮助