Angular : how to debounce an Observable?

Question

In my app, I have a service that returns an observable like this :

public genericService(params) {
    //Do some stuff
    //...

    return this.http.post('http://foo.com', params)
        .map((response) => {
            //Do some generic stuff
            //...

            return someData;
        })
        .catch((error: any) => {
            //Manage error in a generic way + do some generic stuff
            //...

            return Observable.throw(error);
        });
}

let debouncePointer = debounceObservable(genericService, 200);

public genericServiceDebounce(params) {
    return debouncePointer(params);
}

Now in another place, I would like to call my function like this

genericServiceDebounce(params)
    .subscribe((response) => {
        //Do some non-generic stuff
    }, (error) => {
        //Manage error in a non-generic way + do some non-generic stuff
    });

But I didn't succeed to implement the debounceObservable() function.

I tried this implementation based on a Promise equivalent ( https://github.com/moszeed/es6-promise-debounce/blob/master/src/es6-promise-debounce.js ) :

debounceObservable(callback, delay, immediate?) {
    let timeout;
    return function () {
        let context = this, args = arguments;

        return Observable.create((observer) => {
            let later = function () {
                timeout = null;

                if(!immediate) {
                    observer.next(callback.apply(context, args));
                    //observer.onCompleted(); // don't know if this is needed
                }
            };
            let callNow = immediate && !timeout;
            clearTimeout(timeout);
            timeout = setTimeout(later, delay);

            if(callNow) {
                observer.next(callback.apply(context, args));
                //observer.onCompleted(); // don't know if this is needed
            }
        });
    }
}

But this don't work as expected. When using Promises, returning resolve(anotherPromise) allows you to call :

genericServiceDebounce().then(response => {

})

When using Observables, returning observer.next(anotherObservable) return an embedded observable, which means you should call :

genericServiceDebounce().subscribe(obs => {
    obs.subscribe(response => {

    })
})

How would you implement the debounceObservable() function? (in a Promise like way)

Clarification 1 : I found the Observable.debounce() function but this debounces the observer and not the observable itself. And I want to debounce the observable

Clarification 2 : I placed the debounce on the service side because it is a singleton, and their are multiple callers. If I placed it on caller side, there would be a different debounce timer for each caller.

EDIT : Here is a snippet where I try to explain my problem. Just click the different buttons to see the different behaviors (more explanation in js code comments).

Observable.debounce shows how .debounce() from RxJs works. It outputs only '3' but I want '1', '2', '3'.

Observable.debounce x3 shows what happens if I call the code 3 times without wrapping my entire function in a debounce.

Observable wrapped x3 shows what I want to obtain. My entire function is wrapped, but if you look at the code, the subscribe part is fastidious.

Promise x3 shows how simple it is when using Promises.

let log = (logValue) => {
    const list = document.querySelector('#logs');
    const li = document.createElement('li');
    li.innerHTML = logValue;
    list.appendChild(li);
}

/* ************************ */
/* WITH OBSERVABLE.DEBOUNCE */
/* ************************ */

let doStuffObservable = () => {
    Rx.Observable.create((observer) => {
        log('this should be called only one time (observable.debounce)');
        setTimeout(() => {
            observer.next('observable.debounce 1');
            observer.next('observable.debounce 2');
            observer.next('observable.debounce 3');
        }, 1000);
    })
        .debounce(500)
        .subscribe((response) => {
            log(response);
        }, (error) => {
            log(error);
        });
}

/* *********************************** */
/* WITH OBSERVABLE WRAPPED IN DEBOUNCE */
/* *********************************** */

let doStuffObservable2 = (param) => {
    return Rx.Observable.create((observer) => {
        log('this should be called only one time (observable wrapped)');
        setTimeout(() => {
            observer.next('observable wrapped ' + param);
        }, 1000);
    })
}

let debounceObservable = (callback, delay, immediate) => {
    let timeout;
    return function () {
        let context = this, args = arguments;

        return Rx.Observable.create((observer) => {
            let later = function () {
                timeout = null;

                if(!immediate) {
                    observer.next(callback.apply(context, args));
                }
            };
            let callNow = immediate && !timeout;
            clearTimeout(timeout);
            timeout = setTimeout(later, delay);

            if(callNow) {
                observer.next(callback.apply(context, args));
            }
        });
    }
}

let doStuffObservable2Debounced = debounceObservable(doStuffObservable2);

  
/* ************* */
/* WITH PROMISES */
/* ************* */

let doStuffPromise = (param) => {
    return new Promise((resolve, reject) => {
        log('this should be called only one time (promise)');
        setTimeout(() => {
            resolve('promise ' + param);
        }, 1000);
    });
}

let debouncePromise = (callback, delay, immediate) => {
    let timeout;
    return function () {
        let context = this, args = arguments;
        return new Promise(function (resolve) {
            let later = function () {
                timeout = null;
                
                if (!immediate) {
                    resolve(callback.apply(context, args));
                }
            };
            let callNow = immediate && !timeout;
            clearTimeout(timeout);
            timeout = setTimeout(later, delay);
            
            if (callNow) {
                resolve(callback.apply(context, args));
            }
        });
    }
}

  
/* ******* */
/* SAMPLES */
/* ******* */

function doObservableDebounce() {
  doStuffObservable();
  
  // result :
  
      // this should be called only one time (observable.debounce)
      // observable.debounce 3
  
  // this is not what i want, i want all three values in output
}

function doObservableDebounce3Times() {
  doStuffObservable();
  doStuffObservable();
  doStuffObservable();
  
  // result :
  
      // this should be called only one time (observable.debounce)
      // this should be called only one time (observable.debounce)
      // this should be called only one time (observable.debounce)
      // observable.debounce 3
      // observable.debounce 3
      // observable.debounce 3
  
  // this is bad
}

function doObservableWrappedDebounce3Times() {
  doStuffObservable2Debounced(1)
      .subscribe((response) => {
          log(response);
          response.subscribe((response2) => {
              log(response2);
          }, (error) => {
              log(error);
          })
      }, (error) => {
          log(error);
      });
  doStuffObservable2Debounced(2)
      .subscribe((response) => {
          log(response);
          response.subscribe((response2) => {
              log(response2);
          }, (error) => {
              log(error);
          })
      }, (error) => {
          log(error);
      });
  doStuffObservable2Debounced(3)
      .subscribe((response) => {
          log(response);
          response.subscribe((response2) => {
              log(response2);
          }, (error) => {
              log(error);
          })
      }, (error) => {
          log(error);
      });
  
  
  // result :
  
      // AnonymousObservable { source: undefined, __subscribe: [Function] }
      // this should be called only one time (observable wrapped)
      // observable wrapped 3
  
  // this is good but there are 2 embedded subscribe
}

function doPromiseDebounce3Times() {
  let doStuffPromiseDebounced = debouncePromise(doStuffPromise);
  
  doStuffPromiseDebounced(1).then(response => {
      log(response);
  })
  doStuffPromiseDebounced(2).then(response => {
      log(response);
  })
  doStuffPromiseDebounced(3).then(response => {
      log(response);
  })
  
  // result :
  
      // this should be called only one time (promise)
      // promise 3
  
  // this is perfect
}

<!DOCTYPE html>
<html>

  <head>
    <script data-require="[email protected]" data-semver="4.0.6" src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/4.0.6/rx.all.js"></script>
  </head>

  <body>
    <button onclick='doObservableDebounce()'>Observable.debounce</button>
    <button onclick='doObservableDebounce3Times()'>Observable.debounce x3</button>
    <button onclick='doObservableWrappedDebounce3Times()'>Observable wrapped x3</button>
    <button onclick='doPromiseDebounce3Times()'>Promise x3</button>
    <ul id="logs"></ul>
  </body>

</html>

Why are you reinventing the wheel with debounceObservable? RxJS already provides debouncing operators. That's why it is a library and not just observable implementation. — Estus Flask
What does debounce the observable instead of the observer mean? .debounce() debounces the items going through the stream, observable and observer have nothing to do in there. — olivarra1
Maybe you mean that you need to use debounceTime? reactivex.io/rxjs/class/es6/… — olivarra1
@estus and olivarra1 : As I understand the documentation .debounce() and .debounceTime() both execute the observable's code directly. Then if the Observable emits observer.next() multiple times, the subscribe() callback event will be called only one time. The problem here is that the http.post (in my example) is still executed each time I call the observable code. What I want is to execute the code inside the observable only one time even if called multiple times in a small limit of time. — Ldoppea
@Ldoppea what do you mean by .debounce() and .debounceTime() both execute the observable's code directly? What I want is to execute the code inside the observable only one time even if called multiple times in a small limit of time this is what debounceTime() does. — Igor Soloydenko

olivarra1 olivarra1 · Accepted Answer · 2017-12-01T08:23:06

Sorry, I didn't get any notification from your reply to my comment.

A cleaner Rx-only solution for this problem would be to think of your service call as a stream of events, like so:

constructor() {
    this._genericServiceCall$ = new ReplaySubject(1);
    this._genericServiceResult$ = this._genericServiceCall$
        .asObservable()
        .debounceTime(1000)
        .switchMap(params => this._genericService(params));
}

private _genericService(params) {
    //Do some stuff
    //...

    return this.http.post('http://foo.com', params)
        .map((response) => {
            //Do some generic stuff
            //...

            return someData;
        })
        .catch((error: any) => {
            //Manage error in a generic way + do some generic stuff
            //...

            return Observable.throw(error);
        });
}

public genericService(params) {
    this._genericServiceCall$.next(params);
    return this._genericServiceResult$; // Optionally add `.take(1)` so the observer has the expected behaviour of only getting 1 answer back
}

I see something in this though... which params will you accept as the ones that have to get through the private _genericService?

Anyway, do you follow what's going on in here? So every time someone calls genericService() it won't call the service straight away - instead, it will emit a new _genericServiceCall$ and return the _genericServiceResult$ stream. If we take a look on how is this stream defined, we see that's it takes a debounced _genericServiceCall$ and then maps it to the service call. Theoretically it should work - haven't tried.

Edit: Now I see - You may need to publish the genericServiceResult to make it a hot observable, else it will return as soon as any observer subscribes to it:

constructor() {
    this._genericServiceCall$ = new ReplaySubject(1);
    this._genericServiceResult$ = this._genericServiceCall$
        .asObservable()
        .debounceTime(1000)
        .switchMap(params => this._genericService(params))
        .publish();
    const subscription = this._genericServiceResult$.connect();
    // You must store subscription somewhere and dispose it when this object is destroyed - If it's a singleton service this might not be needed.
}

Angular : how to debounce an Observable?

2 Answers