重新连接 websocket rxjs

Posted 2023-02-16

【中文标题】重新连接 websocket rxjs

【英文标题】：Reconnect websocket rxjs

【发布时间】：2019-07-31 03:14:46

【问题描述】：

我正在尝试让我的 websocket 代码自动尝试重新连接（无限期地），直到成功。通过每 x 秒发送一条“ping”消息，我可以检测到管道何时损坏，并调用 closeObserver。

但是，我不确定如何启动重新连接序列。

const notificationConnectionEpic: Epic<ActionTypes, any, RootState> = (
  action$,
  state$
) =>
  action$.pipe(
    filter(isActionOf(actions.connectNotificationPipeline.request)),
    switchMap(async action => 
      const resp = await requireValidToken(action$, state$, params =>
        AdminHubs.getHubNotificationsToken(
          ...params,
          id: action.payload.hubId
        )
      );

      return resp.pipe(
        switchMap(v => 
          if (isAction(v)) 
            return of(v);
          
          if (!v.ok) 
            return of(
              actions.connectNotificationPipeline.failure(
                hubId: action.payload.hubId,
                error: v.error
              )
            );
          

          const webSocketOpen$ = new Subject();
          const webSocketClose$ = new Subject();
          const webSocket$ = webSocket<AdminHubs.HubNotification>(
            url: v.value,
            openObserver: webSocketOpen$,
            closeObserver: webSocketClose$
          );

          const message$ = webSocket$.pipe(
            map(message => actions.receiveNotification( message )),
            takeUntil(action$.ofType(HubActionConsts.NOTIFICATION_PIPE_CLOSED))
          );

          const ping$ = interval(1000).pipe(
            map(_ => webSocket$.next("ping" as any)),
            ignoreElements()
          );

          const open$ = webSocketOpen$.pipe(
            take(1),
            map(_ =>
              actions.connectNotificationPipeline.success(
                hubId: action.payload.hubId
              )
            )
          );

          const close$ = webSocketClose$.pipe(
            // called when a network drop happens. handle reconnect?
          ); // also happens on net error
          return merge(message$, open$, ping$, close$);
        )
      );
    ),
    mergeMap(v => v)
  );

【问题讨论】：

为什么不使用 ondisconnect 事件

websocket 没有 onDisconnect 观察者，即使那样我将如何重新连接？

你在使用任何网络套接字库

rxjs 有一个我正在使用的 websocket 观察器实现。

看这里itnext.io/websocket-error-handling-with-rxjs-17125c6f2159

【参考方案1】：

当 WebSocket 连接关闭时，只需再次 dispatch actions.connectNotificationPipeline.request。这将重新运行此代码并创建一个新的 WebSocket 连接。