Angular : how to debounce an Observable?

Question

在我的应用程序中，我有一个返回这样的可观察对象的服务:

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

现在在另一个地方，我想这样调用我的函数

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

但是我没有成功实现debounceObservable()函数。

我基于 Promise 等价物 (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
            }
        });
    }
}

但这并没有像预期的那样工作。使用 Promises 时，返回 resolve(anotherPromise) 允许您调用:

genericServiceDebounce().then(response => {

})

当使用 Observables 时，返回 observer.next(anotherObservable) 返回一个嵌入的 observable，这意味着你应该调用:

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

    })
})

您将如何实现 debounceObservable() 函数？(以类似于 Promise 的方式)

澄清 1:我找到了 Observable.debounce() 函数，但这会去抖观察者而不是可观察对象本身。我想去除可观察的抖动

澄清 2:我将去抖动放在服务端，因为它是单例，并且它们是多个调用者。如果我将它放在调用方，每个调用方都会有一个不同的去抖动计时器。

编辑:这是我试图解释我的问题的片段。只需单击不同的按钮即可查看不同的行为(更多解释在 js 代码注释中)。

Observable.debounce 展示了 RxJs 中的 .debounce() 是如何工作的。它只输出“3”，但我想要“1”、“2”、“3”。

Observable.debounce x3 展示了如果我调用代码 3 次而没有将我的整个函数包装在 debounce 中会发生什么。

Observable wrapped x3 显示了我想要获取的内容。我的整个函数都被包装了，但是如果你看一下代码，订阅部分很挑剔。

Promise x3 展示了使用 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="rxjs@4.0.6" 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>

Answer 1

抱歉，我没有收到您对我的评论的回复通知。

针对此问题的更简洁的 Rx-only 解决方案是将您的服务调用视为事件流，如下所示:

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
}

虽然我看到了一些东西......您会接受哪些参数作为必须通过私有(private)_genericService的参数？

无论如何，你知道这里发生了什么吗？因此，每次有人调用 genericService() 时，它不会立即调用服务 - 相反，它会发出一个新的 _genericServiceCall$ 并返回 _genericServiceResult$ 流。如果我们看一下这个流是如何定义的，我们会发现它采用去抖动的 _genericServiceCall$ 然后将其映射到服务调用。理论上它应该可以工作 - 还没有尝试过。

编辑:现在我明白了——您可能需要发布 genericServiceResult 以使其成为热可观察对象，否则它会在任何观察者订阅它时立即返回:

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.
}