javascript - RXJS retryWhen 重置等待间隔

Question

我想以增加的时间间隔触发 retrywhen()，

   socketResponse.retryWhen(attempts => {
    return attempts.zip(Observable.range(1, 4)).mergeMap(([error, i]) => {
        console.log(`[socket] Wait ${i} seconds, then retry!`);
        if (i === 4) {
            console.log(`[socket] maxReconnectAttempts ${i} reached!`);
        }
        return Observable.timer(i * 1000);
    });
});

上面的代码工作正常。当前实现输出:

连接错误(第一次)

---

• [socket] 等待 1 秒，然后重试!//等待 1 秒
• [socket] 等待 2 秒，然后重试!//等待 2 秒

连接成功

---

//连接成功。

关于连接错误(第二次)

---

• [socket] 等待 3 秒，然后重试!//等待 3 秒
• [socket] 等待 4 秒，然后重试!//等待 4 秒

现在我想重置套接字连接成功时的等待时间。

期望的输出:

连接错误(第一次)

---

• [socket] 等待 1 秒，然后重试!//等待 1 秒
• [socket] 等待 2 秒，然后重试!//等待 2 秒

连接成功

---

//连接成功。

关于连接错误(第二次)

---

• [socket] 等待 1 秒，然后重试!//等待 1 秒

• [socket] 等待 2 秒，然后重试!//等待 2 秒

但我不知道如何重置 retrywhen() 时间间隔。

Answer 1

我也遇到了同样的问题。我的目标是用可观察对象包装套接字连接。我希望网络 observable 永远不会完成(除非特别要求)并在出现任何错误时无限重试连接。

下面是我如何完成的(使用 rxjs 版本 6+)。这是我的 retryWhen 运算符，无限重试，同时添加缩放等待持续时间 scalingDuration，最高为 maxDuration。

import { Observable, timer } from 'rxjs';
import { mergeMap } from 'rxjs/operators';

export const infiniteRetryStrategy = (opts: { scalingDuration: number, maxDuration: number }, attempt?: () => number) => (attempts: Observable<{}>) => {
    return attempts.pipe(
        mergeMap((error, i) => {
            // use attempt if we have it, otherwise use the number of errors received
            const retryAttempt = attempt ? attempt() : i + 1;
            const waitDuration = Math.min(opts.maxDuration, retryAttempt * opts.scalingDuration);

            console.error(`Got error: ${error}. Retrying attempt ${retryAttempt} in ${waitDuration}ms`);
            return timer(waitDuration);
        })
    );
};

用法非常简单。如果您提供 attempt 函数，您可以在外部管理重试次数，因此您可以在下一次报价时重置它。

import { retryWhen, tap } from 'rxjs/operators';

class Connection {
    private reconnectionAttempt = 0;

    public connect() {
        // replace myConnectionObservable$ by your own observale
        myConnectionObservable$.pipe(
            retryWhen(
                infiniteRetryStrategy({
                    scalingDuration: 1000,
                    maxDuration: 10000
                }, () => ++this.reconnectionAttempt)
            ),
            tap(() => this.reconnectionAttempt = 0)
        ).subscribe(
            (cmd) => console.log(cmd),
            (err) => console.error(err)
        );
    }
}

这并不完美，因为您需要将状态保持在流之外，但这是我设法做到的唯一方法。欢迎任何更清洁的解决方案。