RXJS：组合两个可观察的物品，因此仅在第一个可观察的值为true的值时，可观察的可观察到仅发射

Question

我目前正在尝试使用可观察到的简单离线/在线同步机制。

基本上，我有两个可观察到的东西：

1. 可观察的连接：第一个可观察的可观察到我是否有互联网连接的信息。当网络状态更改
2. 数据可观察到的数据时，它会发出：第二个可观察的数据具有需要同步的数据。当要同步

我要实现的新数据是结合上述观测值时，它会发出：

• 只要连接状态为 false ，可观察到的组合不应发出。在这种情况下，只要连接状态为，可观察到的数据应保留其状态
• 可以观察到的合并可观察到的可观察到的数据应发出。
• ，如果连接状态从切换到，则 false <代码> true ，它应该在可观察到的数据上的每个值发射

一个小示例，当前使用过滤器和commineLatest可以在此处找到： https://codesandbox.io/s/offline-sync-sync-sync-sync-s5lv49?file=/src/index.js

不幸的是，这根本没有预期。

是否有运营商可以实现所需的行为？作为替代方案，我可以当然可以轮询连接状态，并每x秒发出一次。但是理想情况下，我希望可以清洁可观察到的东西，尽管我对操作员最有意义的东西有些迷失。

要清除这个想法：我需要同步所有数据，而不仅仅是最新数据。因此，可观察到的数据应缓解数据。

Answer 1

看起来好像是正确的道路，可以做以下操作：

combineLatest([
  offlineOnlineSubject,
  dataSubject
])
  .pipe(
    filter(([online, counter]) => online),
  )
  .subscribe(([online, counter]) => {
    syncedIndicator.textContent = counter;
  });

Answer 2

因此，当我们处于离线模式时，我将所有发射的数据存储在缓冲区中，一旦我们从 offline 更改为 在线，我简单地在缓冲区中发射所有内容：

let dataBuffer = [];
let isPreviouslyOnline = false;

combineLatest([dataSubject, offlineOnlineSubject])
  .pipe(
    filter(([data, isOnline]) => {
      if (!isOnline) {
        if (!isPreviouslyOnline) {
          dataBuffer.push(data);
        }

        isPreviouslyOnline = false;
        return false;
      }

      return true;
    }),
    switchMap(([data]) => {
      isPreviouslyOnline = true;
      if (dataBuffer.length > 0) {
        const tempData = [...dataBuffer];
        dataBuffer = [];

        return from(tempData);
      } else {
        return of(data);
      }
    })
  )
  .subscribe((data) => {
    console.log("Data is: ", data);
  });

代码沙盒：

我认为它有效，但感觉不佳，试图使用本机 rxjs 缓冲区操作员实现它，但无法弄清楚如何实现。很想看看是否有人有更好/清洁的解决方案。

Answer 3

在搜索“门”一词之后，我找到了以下堆栈溢出问题并发表：
>条件延迟，以rxjs

基本上，该答案是基本的，该答案是基本的，该答案是基本的。所需的结果。

我在此处更新了一个示例： https://codesandbox.io/s/offline-sync-experiments-nimoox?file=/src/index.js：0-1357

关键部分是：

const offlineOnlineSubject = new BehaviorSubject(false);
const dataSubject = new Subject();

const triggerFn = (_) => offlineOnlineSubject.pipe(filter((v) => v));

dataSubject.pipe(delayWhen(triggerFn)).subscribe((counter) => {
  console.log("Syncing data", {
    counter
  });

  syncedIndicator.innerHTML += `<li>${counter}</li>`;
});

包裹在自定义类型的操作员中：

import { MonoTypeOperatorFunction, Observable } from 'rxjs';
import { delayWhen, filter } from 'rxjs/operators';

export function gate<T>(gateTrigger: Observable<boolean>): MonoTypeOperatorFunction<T> {
  const gateTriggerFn = () => gateTrigger.pipe(
    filter((v) => v)
  );

  return (source: Observable<T | null | undefined>) => source.pipe(
    delayWhen(gateTriggerFn)
  );
}

看起来如此这个解决方案迄今正在做我打算做的事情。