angularjs - 使用 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();

我认为这很常见，应该有一些标准实践来实现这一点？

更新

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

// 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. 进入catch函数
9. 订阅 onNext 被触发
10. 点击了刷新按钮
11. 调用 obsSubject.onNext()
12. 没什么。我需要保持订阅

我的理解是 catch 应该保留订阅，但我的测试表明事实并非如此。为什么？

Answer 1

根据您评论中给出的上下文，您希望:

• 每个刷新按钮都会触发“获取结果”
• 向用户显示的每个错误

您确实不需要重新订阅，这是一种反模式，因为 Rx 中的代码从不依赖于此，并且额外的递归调用只会让读者感到困惑。它还让我们想起了回调 hell 。

在这种情况下，您应该:

• 删除 doSubscribe() 调用，因为您不需要它们。使用该代码，您已经具有每次刷新点击都会触发新的“获取结果”的行为。
• 将 select().switch() 替换为 .flatMap()(或 .flatMapLatest())。当您执行 select() 时，结果是元流(流的流)，并且您使用 switch() 将元流扁平化为流。这就是 flatMap 所做的全部事情，但仅在一项操作中。你也可以将flatMap理解为JS Promises的.then()。
• 包含运算符.catch()，它将处理您的错误，就像在catch block 中一样。发生错误后无法获得更多结果的原因是 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)，并允许订阅者自动重新订阅源可观察对象(重试)。

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/