rxjs 限制套接字重新连接的重试次数

Question

我目前正在关注 this 示例，我正在尝试用我自己的逻辑对其进行扩展。

我可以连接套接字，但我发现让客户端连续重新连接存在问题 - 因此我希望限制客户端可以执行的重试次数。

socket.pipe(
    tap((data => console.log(data))),
    retryWhen((errors) =>
        errors.pipe(
            take(this.retryCount),
            delayWhen(val =>
                timer(val * 1000)
            )
        )
    )
).subscribe()

我查阅了文档，发现我可以使用 rxjs 中的 take() 运算符。但是，鉴于我将 retryCount 设置为 5，可以假设以下情况：

• 客户端连接socket成功
• 客户端与套接字断开连接
• 客户端重试 4 次后重新连接
• 客户端断开连接
• 客户端重试 1 次，然后停止重试（已达到take(5)）

有没有一种方法可以“重置”take() 重试的次数？也就是说，每次客户端与套接字断开连接时，它总是有 5 次重试？

Answer 1

您示例中的retryWhen 逻辑没有考虑到连接/断开/重新连接的所有可能相互交织的流程。重新连接尝试的重置也不会发生。

除此之外，您还可以直接触摸内部 Observable 以打开和关闭 webSocket（这在技术上并没有错），而您可以将所有逻辑和处理程序放在主题的管道和订阅代码中。我已经从 StackBlitz 复制了您的示例并对其进行了调整，以便正确处理任何连接/断开和重新连接事件。

正如您在下面的屏幕截图中看到的，它尝试重新连接五次，然后才达到限制并退出。

可以通过这两个常量调整重试次数和每次尝试之间的延迟：

const maxReconnectAttempts = 5; // -1 stands for 'forever'
const reconnectAttemptDelay = 1000; // in ms

如评论中所述，如果您输入const maxReconnectAttempts = -1，则重新连接试验将永远进行。

您可以在下面找到从 index.js 或 here 中提取的完整代码，在分叉的 StackBlitz 存储库中获取完整参考。

import { iif, of, throwError } from 'rxjs';
import { concatMap, delay, retryWhen, tap } from 'rxjs/operators';
import { webSocket } from 'rxjs/webSocket';

// get elements
const btn1 = document.getElementById('btn1');
const btn2 = document.getElementById('btn2');

const display = document.getElementById('app');
const msg = document.getElementById('msg');

// bind click actions
btn1.onclick = connect;
btn2.onclick = disconnect;

// create a webSocketSubject
const wsUrl = 'wss://www.gasnow.org/ws/gasprice';
const maxReconnectAttempts = 5; // -1 stands for 'forever'
const reconnectAttemptDelay = 1000; // in ms

let wsSubject$;
let wsSubscription;
let disconnected = false;
let reconnectRetryCount = 0;

// display live data on view
const bindDataToView = ({ data }) => {
  console.log(data);
  display.innerText = Number(data.rapid);
};

// connect/subscribe to websocket url
function connect() {
  if (!this.wsSubject || this.wsSubject.closed) {
    wsSubject$ = webSocket(wsUrl); // create a fresh instance

    console.log(`Initializing WebSocket connection to ${wsUrl}...`);

    wsSubscription = wsSubject$
      .pipe(
        retryWhen((errors) =>
          errors.pipe(
            concatMap((error) =>
              iif(
                () =>
                  maxReconnectAttempts !== -1 &&
                  reconnectRetryCount++ >= maxReconnectAttempts,
                throwError('WebSocket reconnecting retry limit exceeded!'),
                of(error).pipe(
                  tap(() => {
                    disconnected = true;
                    console.warn('Trying to reconnect to WebSocket server...');
                  }),
                  delay(reconnectAttemptDelay)
                )
              )
            )
          )
        ),
        tap(() => {
          if (disconnected) {
            disconnected = false;
            reconnectRetryCount = 0;

            msg.innerText = 'Streaming ...';
            console.log('Successfully reconnected to the WebSocket server.');
          }
        })
      )
      .subscribe(
        (data) => {
          msg.innerText = 'Streaming ...';
          bindDataToView(data);
        },
        (err) => {
          reconnectRetryCount = 0;
          console.error(err);
        },
        () => {
          reconnectRetryCount = 0;
          msg.innerText = 'Connection closed';

          console.warn('Connection to the WebSocket server was closed!');
        }
      );
  }
}

// close websocket connection
function disconnect() {
  if (wsSubject$) {
    wsSubject$.complete(); // this will trigger closingObserver and closeObserver
    wsSubject$.unsubscribe();
    wsSubject$ = null;

    disconnected = true;
    reconnectRetryCount = 0;

    if (wsSubscription) {
      wsSubscription.unsubscribe();
      wsSubscription = null;
    }

    console.log('Disconnected from the WebSocket server.');
  }
}