rxjs - 从 Promise 接收并重试

Question

为什么必须flatMap fromPromise 才能使重试正常工作？
(“/error”返回状态 500)

var responseStream = Rx.Observable.just('/error').
        flatMap(requestUrl => {
            return Rx.Observable.fromPromise(_this.$http.get(requestUrl));
    });

    responseStream.
        retry(3).
        subscribe(
        function (x) {
            console.log('Next: ' + x);
        },
        function (err) {
            console.log('Error: ' + err);
        },
        function () {
            console.log('Completed');
        });

我想剪掉仪式，然后:

Rx.Observable.fromPromise(_this.$http.get('error')).
        retry(3).
        subscribe(
        function (x) {
            console.log('Next: ' + x);
        },
        function (err) {
            console.log('Error: ' + err);
        },
        function () {
            console.log('Completed');
        }); 

在第二种情况下，xhr 调用仅执行一次。

Answer 1

因为Promises没有用于重试的API。 fromPromise只是包装 Promise 并使其通过 Observable 发出。当您调用 _this.$http.get('error') 时，您已经创建了一个 Promise 并且它已经在运行中，因此无法重试该 Promise无需再次调用该方法。

它在用 flatMap 包装时起作用的原因是，当您重试 Observable 时，它实际上是在重新执行生成 Promise 的方法。

如果冗长的内容确实让您感到困扰，请意识到许多运算符隐式支持 Promises，而无需调用 fromPromise 方法。

所以你的例子可以减少到

var responseStream = Rx.Observable.just('/error')
        .flatMap(requestUrl => _this.$http.get(requestUrl));

    responseStream
        .retry(3)
        .subscribe(
          (x) => console.log('Next: ' + x),
          (err) => console.log('Error: ' + err),
          () => console.log('Completed'));

或者更简单地使用defer:

Observable.defer(() => _this.$http.get('/error'))
   .retry(3)
   .subscribe(
      (x) => console.log('Next: ' + x),
      (err) => console.log('Error: ' + err),
      () => console.log('Completed'));