上一个ajax承诺解析后，如何构建一个等待一定时间间隔的rx轮询器? [英] How to build an rx poller that waits some interval AFTER the previous ajax promise resolves?

问题描述

致力于此方面的几种方法.基本上，我不希望有一个轮询器从轮询开始起每30秒启动一个ajax；我希望有一个轮询器在前一个请求返回后的30秒内启动请求.另外，我想针对故障的指数补偿制定某种策略.

Been working on a few approaches to this. Basically, I don't want a poller that kicks off an ajax every 30 seconds from the start of polling -- I want a poller that kicks off requests 30 seconds AFTER the previous request returns. Plus, I want to work in some strategy around exponential back-off for failures.

这就是我到目前为止(Rx4):

Here's what I have so far (Rx4):

rx.Observable.create(function(observer) {
    var nextPoll = function(obs) {
      // the action function invoked below is what i'm passing in
      // to my poller service as any function which returns a promise
      // and must be invoked each loop due to fromPromise caching
      rx.Observable.fromPromise(action())
        .map(function (x){ return x.data; })
        .subscribe(function(d) {       
            // pass promise up to parent observable  
            observer.onNext(d);

            // reset interval in case previous call was an error
            interval = initInterval;   
            setTimeout(function(){ nextPoll(obs); }, interval);
        }, function(e) {
          // push interval higher (exponential backoff)
          interval = interval < maxInterval ? interval * 2 : maxInterval;
          setTimeout(function(){ nextPoll(obs); }, interval);

        });
    };
    nextPoll(observer);
});

在大多数情况下，这就是我想要的.我不喜欢使用setTimeout，但似乎无法找到更好的Observable方法(除了一个固定的时间间隔/定时器和另一个订阅).

For the most part, this does what I want. I don't like the use of setTimeout, but I can't seem to find a better Observable approach to this (other than a one-off interval/timer with another subscribe).

我无法进行的另一件事是能够控制轮询器在最初启动时是否可以延迟启动或立即启动.对于某些用途，我将在开始轮询之前获取数据，因此我可以让它等待时间间隔后再进行首次触发.到目前为止，我只在第一个Ajax之前或在Ajax之间并将其提供给订户之间的计时器/延迟方面很幸运，这对我不起作用.

The other thing that I haven't been able to work into this is the ability to control whether the poller, when initially started, can start with a delay or fire immediately. For some uses, I will have just fetched the data prior to starting to poll, so I can let it wait the interval before firing for the first time. So far, I've only had luck with timer/delay that happens before the first ajax or between the ajax and providing it to subscribers, which doesn't work for me.

无论是从总体上还是从摆脱setTimeout的角度考虑，都希望对清理此内容有任何想法.而且，如果有人可以选择延迟启动此轮询器，那就太好了！谢谢大家！

Would appreciate any thoughts on cleaning this in, both generally and in terms of getting rid of the setTimeout. And, if anyone has a way to kick off this poller with an optional delay, that would be tremendous! Thanks all!!

更新:终于按照我的设想进行了工作.看起来像这样:

UPDATE: Finally got this working the way I envisioned. Here's what that looks like:

function computeInterval(error) {
  if (error) {
    // double until maximum interval on errors
    interval = interval < maxInterval ? interval * 2 : maxInterval;
  } else {
    // anytime the poller succeeds, make sure we've reset to
    // default interval.. this also allows the initInterval to 
    // change while the poller is running
    interval = initInterval;
  }
  return interval;
}

poller$ = rx.Observable.fromPromise(function(){ return _this.action(); })
  .retryWhen(function(errors){
    return errors.scan(function(acc, x) { return acc + x; }, 0)
      .flatMap(function(x){ 
        return rx.Observable.timer(computeInterval(true));
      });
  })
  .repeatWhen(function(notification){
    return notification
      .scan(function(acc, x) { return acc + x; }, 0)
      .flatMap(function(x){ 
        return rx.Observable.timer(computeInterval());
      });
  });

推荐答案

我使用另一种可能有用和/或更简单的技术添加了另一个答案.它基于 repeatWhen 运算符.正如文档所述，它使您可以重复订阅可观察的对象，直到您发出重复结束的信号以指示正常完成或错误.

I add another answer using another technique which can be useful and/or maybe simpler. It is based on the repeatWhen operator. As the documentation explains, it allows you to repeat subscription to an observable till you signal the end of the repetition indicating normal completion or an error.

var action; // your action function
Rx.Observable.create(function (observer) {
    function executeAction(action) {
        return Rx.Observable.fromPromise(action());
    }

    function computeDelay(){
        // put your exponential delaying logic here
    }

executeAction()
  .repeatWhen(function(notification){
    return Rx.Observable.return({}).delay(computeDelay());
  })
  .subscribe(function(x){
    observer.onNext(x.data);
  })
});

有很棒的文章在这里，对 repeatWhen 和 retryWhen 的解释要比官方文档更好(它适用于RxJava，但适用于Rxjs，但做了一些小的修改).它还给出了您想要实现的目标的一个示例(指数延迟重复).

There is a terrific article here explaining repeatWhen and retryWhen better than the official documentation does (it is for RxJava but it applies to Rxjs with some minor modifications). It also gives an example of exactly what you want to achieve (exponential delayed repeating).

这篇关于上一个ajax承诺解析后，如何构建一个等待一定时间间隔的rx轮询器?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！