Filter and debounce an observable using values from another observable

Question

I have 2 observables :

a source : emits an event each time a change is made in a form
a "inspector" : emits an event to tell if the change can be saved or not

What I'm trying to do is :

the source emits values
debounce values for a minute
if the controller didn't emit "false" during that time then emit latest value from source

I have seen some similar problems resolved with withLatestFrom(inspector).filter(...) but it does not work for me as I need all values emitted from the inspector observable during the debounce time.

I also tried a 'merge' operator but I only care about the source : if the inspector emits values but the source doesn't, the observable i'm trying to build shouldn't either.

Is there a way to achieve this using only observables ?

user_2497335 · Answer

It is helpful to break this problem down. I don't quite understand what you're asking for, so here's my best impression:

Source emits a value.
After that initial emission we start listening to the inspector for true values.
Once the debounce time is passed we emit a value if the inspector only emitted true.

The first observation I'd like to make (pardon the pun), is that you don't have to use debounceTime to have a debounce-like effect. I find that switchMap with a timer on the inside can produce the same results: SwitchMap will cancel previous emissions like debounce, and the timer can act to delay the emission.

My proposal is that you use switchMap from your source and then create an observable from the combination of a timer and your inspector. Use a filter operator so you only emit from the source if the inspector's last emitted result duration the duration of the timer was 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
    )
  )
)

Note: startWith is used in the inspector's pipe so that at least one result is emitted. This guarantees that there will be a single emission once timer emits. The filter is on the inspector since all you care about is if a false result prevents emission.
If you don't want to force the user to wait a minute, you could just use race instead of combineLatest. It will emit the result from the first observable that emits. So you can have the timer emit true after a minute, and the inspector emit any false result.

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
  )
)

user_17254077 · Answer

It could be solved by using the buffer operator that raises notification only after required interval elapsed in case the blocker stream does not cancel it in advance.

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: [],
      },
    }
  });
});

Filter and debounce an observable using values from another observable

2 Answers