使用rx-angular重新订阅一个observable

Question

我想使用angular-rx作为简单的刷新按钮来获得结果.如果用户单击刷新按钮,则重新加载结果.如果用户在1秒内单击刷新按钮100次,则仅加载最新结果.如果结果由于某种原因失败,那并不意味着刷新按钮应该停止工作.

为了达到最后一点,我想保留订阅(或重新订阅),即使它失败了,但我无法弄清楚如何做到这一点？

这不起作用,但这是一个简单的例子,我尝试重新订阅错误:

var refreshObs = $scope.$createObservableFunction('refresh');

var doSubscribe = function () {
  refreshObs
  .select(function (x, idx, obs) {
      // get the results.
      // in here might throw an exception
  })
  .switch()
  .subscribe(
  function (x) { /*show the results*/ }, // on next
  function (err) { // on error
      doSubscribe(); // re-subscribe
  },
  function () { } // on complete
  );
};
doSubscribe();

我觉得这很常见应该有一些标准的做法来实现这个目标吗？

UPDATE

使用建议的解决方案,这是我测试的:

// using angularjs and the rx.lite.js library
var testCount = 0;
var obsSubject = new rx.Subject(); // note. rx is injected but is really Rx
$scope.refreshButton = function () { // click runs this
  obsSubject.onNext();
};

obsSubject.map(function () {
  testCount++;
  if (testCount % 2 === 0) {
      throw new Error("something to catch");
  }
  return 1;
})
.catch(function (e) {
  return rx.Observable.return(1);
})
.subscribe(
    function (x) {
    // do something with results
});

这些是我的测试结果:

1. 点击刷新按钮
2. 调用obsSubject.onNext()
3. map函数返回1.
4. 订阅onNext被触发
5. 点击刷新按钮
6. 调用obsSubject.onNext()
7. map函数抛出错误
8. 进入捕捉功能
9. 订阅onNext被触发
10. 点击刷新按钮
11. 调用obsSubject.onNext()
12. 没有.我需要继续订阅

我的理解是catch应该保留订阅,但我的测试表明它没有.为什么？

Answer 1

根据评论中给出的上下文,您需要:

• 每个刷新按钮触发'获得结果'
• 每个错误都要显示给用户

你真的不需要重新订阅,它是一种反模式,因为Rx中的代码从不依赖于它,而额外的递归调用只会让读者感到困惑.它还提醒我们回调地狱.

在这种情况下,您应该:

• 删除doSubscribe()调用,因为您不需要它们.使用该代码,您已经拥有了每次刷新点击都会触发新的"获取结果"的行为.
• 替换select().switch()为.flatMap()(或.flatMapLatest()).当你这样做时select(),结果是一个元流(流的流),你switch()用来将元流展平成一个流.这就是flatMap所做的一切,但仅限于一次操作.您还可以了解.then()JS Promises中的flatMap .
• 包括.catch()将处理错误的运算符,如catch块中所示.发生错误后无法获得更多结果的原因是Observable始终在错误或"完整"事件上终止.使用catch()运算符,我们可以用Observable上的健全事件替换错误,以便它可以继续.

要改进您的代码:

var refreshObs = $scope.$createObservableFunction('refresh');

refreshObs
  .flatMapLatest(function (x, idx, obs) {
    // get the results.
    // in here might throw an exception
    // should return an Observable of the results
  })
  .catch(function(e) {
      // do something with the error
      return Rx.Observable.empty(); // replace the error with nothing
  })
  .subscribe(function (x) { 
      // on next
  });

另请注意,我删除了onError和onComplete处理程序,因为它们内部没有任何操作.

另外看看更多运营商.例如,retry()可以用于每次发生错误时自动"获得结果".请参阅https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/retry.md

使用retry()结合do()以处理错误(do),并且允许订户自动重新订阅源可观察到的(retry).

refreshObs
  .flatMapLatest(function (x, idx, obs) {
    // get the results.
    // in here might throw an exception
    // should return an Observable of the results
  })
  .do(function(){}, // noop for onNext
  function(e) {
      // do something with the error
  })
  .retry()
  .subscribe(function (x) { 
      // on next
  });

在这里查看一个工作示例:http://jsfiddle.net/staltz/9wd13gp9/9/