RxJS: Debounce with regular sampling

Question

I have an Observable that emits a stream of values from user input (offset values of a slider).

I want to debounce that stream, so while the user is busy sliding, I only emit a value if nothing has come through for, say 100ms, to avoid being flooded with values. But then I also want to emit a value every 1 second if it is just endlessly debouncing (user is sliding back and forth continuously). Once the user stops sliding though, I just want the final value from the debounced stream.

So I want to combine the debounce with a regular "sampling" of the stream. Right now my setup is something like this:

const debounce$ = slider$.debounceTime(100),
      sampler$ = slider$.auditTime(1000);

debounce$
    .merge(sampler$)
    .subscribe((value) => console.log(value));

Assuming the user moves the slider for 2.4 seconds, this emits values as follows:

 start                      end
  (x)---------|---------|---(x)|----|
              |         |      |    |
             1.0       2.0    2.5  3.0  <-- unwanted value at the end
              ^         ^      ^
            sample   sample   debounce  <-- these are all good

I don't want that extra value emitted at 3 seconds (from the sampler$ stream).

Obviously merge is the wrong way to combine these two streams, but I can't figure out what combination of switch, race, window or whatever to use here.

Answer 1

You can solve the problem by composing an observable that serves as a signal, indicating whether or not the user is currently sliding. This should do it:

const sliding$ = slider$.mapTo(true).merge(debounce$.mapTo(false));

And you can use that to control whether or not the sampler$ emits a value.

A working example:

const since = Date.now();
const slider$ = new Rx.Subject();

const debounce$ = slider$.debounceTime(100);
const sliding$ = slider$.mapTo(true).merge(debounce$.mapTo(false));

const sampler$ = slider$
  .auditTime(1000)
  .withLatestFrom(sliding$)
  .filter(([value, sliding]) => sliding)
  .map(([value]) => value);

debounce$
  .merge(sampler$)
  .subscribe(value => console.log(`${time()}: ${value}`));

// Simulate sliding:

let value = 0;
for (let i = 0; i <= 2400; i += 10) {
  value += Math.random() > 0.5 ? 1 : -1;
  slide(value, i);
}

function slide(value, at) {
  setTimeout(() => slider$.next(value), at);
}

function time() {
  return `T+${((Date.now() - since) / 1000).toFixed(3)}`;
}

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>

Answer 2

For those who are interested, this is the approach I took, inspired by @cartant's answer.

const slider$ = new Rx.Subject();
const nothing$ = Rx.Observable.never();

const debounce$ = slider$.debounceTime(100);
const sliding$ = slider$.mapTo(true)
  .merge(debounce$.mapTo(false))
  .distinctUntilChanged();

const sampler$ = sliding$
  .switchMap((active) => active ? slider$.auditTime(1000) : nothing$);

debounce$
  .merge(sampler$)
  .subscribe(value => console.log(`${time()}: ${value}`));

The difference is adding distinctUntilChanged on the sliding$ stream to only get the on/off changes, and then doing a switchMap on that to either have the sampler return values or not.