rxjs - 带有 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) 缓冲打开和关闭自动保存之间的事件，并使用 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”以查看更多示例。

希望这可以帮助