使用来自另一个 observable 的值过滤和 debounce 一个 observable

Question

我有 2 个可观察对象：

• 源：每次对表单进行更改时都会发出一个事件
• “检查器”：发出一个事件来判断是否可以保存更改

我想做的是：

• 源发出值
• 一分钟的去抖动值
• 如果控制器在此期间没有发出“false”，则从源发出最新值

我已经看到withLatestFrom(inspector).filter(...) 解决了一些类似的问题，但它对我不起作用，因为我需要在去抖动期间观察到从检查器发出的所有值。

我也尝试了一个“合并”运算符，但我只关心源：如果检查器发出值但源没有，那么我尝试构建的 observable 也不应该。

有没有办法只使用 observables 来实现这一点？

Answer 1

分解这个问题很有帮助。我不太明白你在问什么，所以这是我最好的印象：

• 源发出一个值。
• 在初始发射后，我们开始听取检查器的真实值。
• 一旦经过去抖动时间，如果检查器仅发出真值，我们就会发出一个值。

我想说的第一个观察（请原谅双关语）是您不必使用 debounceTime 来获得类似 debounce 的效果。我发现内部带有 timer 的 switchMap 可以产生相同的结果：SwitchMap 将取消以前的发射，例如 debounce，定时器可以延迟发射。

我的建议是您使用源中的 switchMap，然后从 timer 和检查器的组合中创建一个 observable。使用 filter 运算符，以便仅在检查器最后发出的结果持续时间（计时器的持续时间为 true 时）从源发出。

this.source.pipe(
  switchMap(x => // switchMap will cancel any emissions if the timer hasn't emitted yet.
    // this combineLatest will only emit once - when the timer emits.
    combineLatest([
        timer(60000), 
        this.inspector.pipe(filter(x => !x), startWith(true))
    ]).pipe( 
      filter(([_, shouldEmit]) => shouldEmit), 
      mapTo(x) // emit source's value
    )
  )
)

• 注意：startWith 在检查器的管道中使用，以便至少发出一个结果。这保证了一旦 timer 发射，就会有一个发射。过滤器在检查器上，因为您只关心错误结果是否会阻止排放。
• 如果您不想强迫用户等待一分钟，您可以只使用 race 而不是 combineLatest。它将从第一个发出的 observable 发出结果。因此，您可以让计时器在一分钟后发出 true，而检查器发出任何错误的结果。

switchMap(x =>
  race(
    timer(6000).pipe(mapTo(true)), // emit true after a minute.
    this.inspector.pipe(filter(x => !x)) // only emit false
  ).pipe(
    take(1), // this might not be necessary.
    filter((shouldEmit) => shouldEmit), 
    mapTo(x) // emit source's value
  )
)

Answer 2

可以通过使用 buffer 运算符来解决，该运算符仅在经过所需间隔后才发出通知，以防阻塞流未提前取消它。

source$.pipe(
  buffer(source$.pipe(
    exhaustMap(() => timer(10).pipe(
      takeUntil(blocker$)
    ))
  ))
);

const {timer} = rxjs;
const {buffer, exhaustMap, takeUntil} = rxjs.operators;
const {TestScheduler} = rxjs.testing;
const {expect} = chai;

const test = (testName, testFn) => {
  try {
    testFn();
    console.log(`Test PASS "${testName}"`);
  } catch (error) {
    console.error(`Test FAIL "${testName}"`, error.message);
  }
}

const createTestScheduler = () => new TestScheduler((actual, expected) => {
  expect(actual).deep.equal(expected);
});

const createTestStream = (source$, blocker$) => {
  return source$.pipe(
    buffer(source$.pipe(
      exhaustMap(() => timer(10).pipe(
        takeUntil(blocker$)
      ))
    ))
  );
}

const testStream = ({ marbles, values}) => {
  const testScheduler = createTestScheduler();
  testScheduler.run((helpers) => {
    const { cold, hot, expectObservable } = helpers;
    const source$ = hot(marbles.source);
    const blocker$ = hot(marbles.blocker);
    const result$ = createTestStream(source$, blocker$)
    expectObservable(result$).toBe(marbles.result, values.result);
  });
}

test('should buffer changes with 10ms delay', () => {
  testStream({
    marbles: {
      source: ' ^-a-b 7ms ---c 9ms -----|   ',
      blocker: '^-   10ms --- 10ms -----|   ',
      result: ' --   10ms i-- 10ms j----(k|)',
    },
    values: {
      result: {
        i: ['a', 'b'],
        j: ['c'],
        k: [],
      },
    }
  });
});

test('should block buffer in progress and move values to next one', () => {
  testStream({
    marbles: {
      source: ' ^-a-b 7ms ---c 9ms -----|   ',
      blocker: '^-  8ms e---- 10ms -----|   ',
      result: ' --   10ms --- 10ms j----(k|)',
    },
    values: {
      result: {
        j: ['a', 'b', 'c'],
        k: [],
      },
    }
  });
});

<script src="https://cdnjs.cloudflare.com/ajax/libs/chai/4.1.2/chai.js"></script>
<script src="https://unpkg.com/rxjs@^7/dist/bundles/rxjs.umd.min.js"></script>