refactor: replace fetch SSE with original EventSource

package.json

@@ -1,12 +1,11 @@
 {
   "name": "vocechat-web",
-  "version": "0.3.11",
+  "version": "0.3.12",
   "private": true,
   "homepage": "https://voce.chat",
   "dependencies": {
     "@emoji-mart/data": "^1.0.6",
     "@metamask/onboarding": "^1.0.1",
-    "@microsoft/fetch-event-source": "^2.0.1",
     "@pmmmwh/react-refresh-webpack-plugin": "^0.5.8",
     "@react-oauth/google": "^0.2.8",
     "@reduxjs/toolkit": "^1.8.6",

src/common/hook/useStreaming/index.ts

@@ -1,5 +1,4 @@
 import { useEffect, useState } from "react";
-import { fetchEventSource, EventStreamContentType } from "@microsoft/fetch-event-source";
 import toast from "react-hot-toast";
 import BASE_URL from "../../../app/config";
 import { setReady } from "../../../app/slices/ui";
@@ -22,10 +21,6 @@ import chatMessageHandler from "./chat.handler";
 import store, { useAppDispatch, useAppSelector } from "../../../app/store";
 import { ServerEvent, UsersStateEvent } from "../../../types/sse";
 
-class RetryError extends Error { }
-
-class FatalError extends Error { }
-
 const getQueryString = (params: { [key: string]: string }) => {
   const sp = new URLSearchParams();
   Object.entries(params).forEach(([key, val]) => {
@@ -36,7 +31,7 @@ const getQueryString = (params: { [key: string]: string }) => {
   return sp.toString();
 };
 let inter: number | null = null;
-
+let SSE: EventSource | null = null;
 export default function useStreaming() {
   const [readyPullData, setReadyPullData] = useState(false);
   const {
@@ -48,9 +43,8 @@ export default function useStreaming() {
   const loginUid = authData.user?.uid || 0;
   let initialized = false;
   let initializing = false;
-  let controller = new AbortController();
 
-  const startStreaming = async () => {
+  const startStreaming = () => {
     console.info("sse start streaming", initialized, initializing);
     if (initialized || initializing) return;
     // 如果token快要过期，先renew
@@ -80,238 +74,202 @@ export default function useStreaming() {
     if (usersVersion !== 0) {
       params.users_version = `${usersVersion}`;
     }
-    await fetchEventSource(`${BASE_URL}/user/events?${getQueryString(params)}`, {
+    // 开始初始化推送
-      openWhenHidden: true,
+    SSE = new EventSource(`${BASE_URL}/user/events?${getQueryString(params)}`);
-      signal: controller.signal,
-      async onopen(response) {
-        initializing = false;
-        if (response.ok && response.headers.get("content-type") === EventStreamContentType) {
-          console.info("sse everything ok");
-          initialized = true;
-          return; // everything's good
-        } else if (response.status >= 400 && response.status < 500 && response.status !== 429) {
-          // 重新登录
-          // client-side errors are usually non-retriable:
-          console.info("sse debug: open fatal");
-          throw new FatalError();
-        } else {
-          // server error
-          console.info("sse debug: open retry");
-          throw new RetryError();
-        }
-      },
-      onmessage(evt) {
-        initializing = false;
-        console.info("sse message", evt.data);
-        // if the server emits an error message, throw an exception
-        // so it gets handled by the onerror callback below:
-        if (evt.event === "FatalError") {
-          console.info("sse debug: error message fatal");
-          throw new FatalError(evt.data);
-        }
-        const data: ServerEvent = JSON.parse(evt.data);
-        const { type } = data;
-        switch (type) {
-          case "heartbeat":
-            console.info("sse heartbeat", loginUid);
-            break;
-          case "ready":
-            console.info("sse streaming ready");
-            dispatch(setReady());
-            break;
-          case "users_snapshot":
-            {
-              console.info("sse users snapshot");
-              const { version } = data;
-              dispatch(updateUsersVersion(version));
-            }
-            break;
-          case "users_log":
-            {
-              const { logs } = data;
-              console.info("sse users change logs", logs);
-              dispatch(updateUsersByLogs(logs));
-            }
-            break;
-          case "user_settings":
-          case "user_settings_changed":
-            {
-              console.info("sse users settings");
-              Object.keys(data).forEach((key) => {
-                switch (key) {
-                  case "read_index_groups":
-                    dispatch(updateReadChannels(data[key]));
-                    break;
-                  case "read_index_users":
-                    dispatch(updateReadUsers(data[key]));
-                    break;
-                  case "add_mute_users":
-                  case "mute_users":
-                  case "add_mute_groups":
-                  case "mute_groups":
-                    {
-                      const arr = data[key];
-                      if (arr && arr.length) {
-                        const _key = key.endsWith("users") ? "add_users" : "add_groups";
-                        dispatch(updateMute({ [_key]: arr }));
-                      }
-                    }
-                    break;
-                  case "remove_mute_users":
-                  case "remove_mute_groups":
-                    {
-                      const arr = data[key];
-                      if (arr && arr.length) {
-                        const _key = key.endsWith("users") ? "remove_users" : "remove_groups";
-                        dispatch(updateMute({ [_key]: arr }));
-                      }
-                    }
-                    break;
 
-                  default:
+    SSE.onopen = () => {
-                    break;
+      initializing = false;
-                }
+      initialized = true;
-              });
+    };
-            }
+    SSE.onerror = (err) => {
-            break;
+      console.error("sse error", err);
-          case "users_state":
+
-          case "users_state_changed":
+      initializing = false;
-            {
+      stopStreaming();
-              let { type, ...rest } = data;
+      if (inter) {
-              const onlines =
+        clearTimeout(inter);
-                type == "users_state_changed" ? [rest] : (rest as UsersStateEvent).users;
+      }
-              dispatch(updateUsersStatus(onlines));
+      // 重连
-            }
+      inter = window.setTimeout(() => {
-            break;
+        initialized = false;
-          case "kick":
+        startStreaming();
-            {
+      }, 2000);
-              console.info("sse kicked");
+    };
-              switch (data.reason) {
+    SSE.onmessage = (evt) => {
-                case "login_from_other_device":
+      initializing = false;
-                  dispatch(resetAuthData());
+      console.info("sse message", evt.data);
-                  toast("kicked from the other device");
+      const data: ServerEvent = JSON.parse(evt.data);
+      const { type } = data;
+      switch (type) {
+        case "heartbeat":
+          console.info("sse heartbeat", loginUid);
+          break;
+        case "ready":
+          console.info("sse streaming ready");
+          dispatch(setReady());
+          break;
+        case "users_snapshot":
+          {
+            console.info("sse users snapshot");
+            const { version } = data;
+            dispatch(updateUsersVersion(version));
+          }
+          break;
+        case "users_log":
+          {
+            const { logs } = data;
+            console.info("sse users change logs", logs);
+            dispatch(updateUsersByLogs(logs));
+          }
+          break;
+        case "user_settings":
+        case "user_settings_changed":
+          {
+            console.info("sse users settings");
+            Object.keys(data).forEach((key) => {
+              switch (key) {
+                case "read_index_groups":
+                  dispatch(updateReadChannels(data[key]));
                   break;
-                case "delete_user":
+                case "read_index_users":
-                  dispatch(resetAuthData());
+                  dispatch(updateReadUsers(data[key]));
-                  toast("Your account has been deleted");
                   break;
+                case "add_mute_users":
+                case "mute_users":
+                case "add_mute_groups":
+                case "mute_groups":
+                  {
+                    const arr = data[key];
+                    if (arr && arr.length) {
+                      const _key = key.endsWith("users") ? "add_users" : "add_groups";
+                      dispatch(updateMute({ [_key]: arr }));
+                    }
+                  }
+                  break;
+                case "remove_mute_users":
+                case "remove_mute_groups":
+                  {
+                    const arr = data[key];
+                    if (arr && arr.length) {
+                      const _key = key.endsWith("users") ? "remove_users" : "remove_groups";
+                      dispatch(updateMute({ [_key]: arr }));
+                    }
+                  }
+                  break;
+
                 default:
                   break;
               }
+            });
+          }
+          break;
+        case "users_state":
+        case "users_state_changed":
+          {
+            let { type, ...rest } = data;
+            const onlines =
+              type == "users_state_changed" ? [rest] : (rest as UsersStateEvent).users;
+            dispatch(updateUsersStatus(onlines));
+          }
+          break;
+        case "kick":
+          {
+            console.info("sse kicked");
+            switch (data.reason) {
+              case "login_from_other_device":
+                dispatch(resetAuthData());
+                toast("kicked from the other device");
+                break;
+              case "delete_user":
+                dispatch(resetAuthData());
+                toast("Your account has been deleted");
+                break;
+              default:
+                break;
             }
-            break;
+          }
-          case "related_groups":
+          break;
-            console.info("sse fill channels from streaming", data);
+        case "related_groups":
-            dispatch(fillChannels(data.groups));
+          console.info("sse fill channels from streaming", data);
-            break;
+          dispatch(fillChannels(data.groups));
-          case "joined_group":
+          break;
-            console.info("sse joined group", data.group);
+        case "joined_group":
-            dispatch(addChannel(data.group));
+          console.info("sse joined group", data.group);
-            break;
+          dispatch(addChannel(data.group));
-          case "group_changed":
+          break;
-            {
+        case "group_changed":
-              const { gid, ...rest } = data;
+          {
+            const { gid, ...rest } = data;
+            dispatch(
+              updateChannel({
+                gid,
+                ...rest
+              })
+            );
+          }
+          break;
+        case "user_joined_group":
+          {
+            console.info("sse new user joined group", data.gid);
+            const { gid, uid: uids } = data;
+            // 去重
+            dispatch(
+              updateChannel({
+                operation: "add_member",
+                gid,
+                members: uids
+              })
+            );
+          }
+          break;
+        case "user_leaved_group":
+          {
+            const { gid, uid: uids } = data;
+            if (uids.findIndex((uid) => uid == loginUid) > -1) {
+              dispatch(removeChannel(gid));
+            } else {
               dispatch(
                 updateChannel({
-                  gid,
+                  operation: "remove_member",
-                  ...rest
-                })
-              );
-            }
-            break;
-          case "user_joined_group":
-            {
-              console.info("sse new user joined group", data.gid);
-              const { gid, uid: uids } = data;
-              // 去重
-              dispatch(
-                updateChannel({
-                  operation: "add_member",
                   gid,
                   members: uids
                 })
               );
             }
-            break;
-          case "user_leaved_group":
-            {
-              const { gid, uid: uids } = data;
-              if (uids.findIndex((uid) => uid == loginUid) > -1) {
-                dispatch(removeChannel(gid));
-              } else {
-                dispatch(
-                  updateChannel({
-                    operation: "remove_member",
-                    gid,
-                    members: uids
-                  })
-                );
-              }
-            }
-            break;
-          case "kick_from_group":
-            console.info("sse kicked from group", data.gid);
-            dispatch(removeChannel(data.gid));
-            break;
-          case "pinned_message_updated":
-            {
-              // const {gid,mid,msg}=data;
-              dispatch(updatePinMessage(data));
-            }
-            break;
-          case "chat":
-            {
-              chatMessageHandler(data, dispatch, {
-                ready,
-                loginUid,
-                readUsers,
-                readChannels
-              });
-            }
-            break;
-
-          default:
-            console.info("sse event data", data);
-            break;
-        }
-      },
-      onclose() {
-        // if the server closes the connection unexpectedly, retry:
-        console.info("sse debug: closed");
-        initializing = false;
-        throw new RetryError();
-      },
-      onerror(err) {
-        initializing = false;
-        if (err instanceof FatalError || err.toString().indexOf("network error") > -1) {
-          console.info("sse debug: error fatal", err);
-          throw err; // rethrow to stop the operation
-        } else {
-          console.info("sse debug: error other", err);
-          stopStreaming();
-          if (inter) {
-            clearTimeout(inter);
           }
-          // 重连
+          break;
-          inter = window.setTimeout(() => {
+        case "kick_from_group":
-            initialized = false;
+          console.info("sse kicked from group", data.gid);
-            startStreaming();
+          dispatch(removeChannel(data.gid));
-          }, 2000);
+          break;
-          throw err; // rethrow to stop the operation
+        case "pinned_message_updated":
-          // do nothing to automatically retry. You can also
+          {
-          // return a specific retry interval here.
+            // const {gid,mid,msg}=data;
-        }
+            dispatch(updatePinMessage(data));
+          }
+          break;
+        case "chat":
+          {
+            chatMessageHandler(data, dispatch, {
+              ready,
+              loginUid,
+              readUsers,
+              readChannels
+            });
+          }
+          break;
+
+        default:
+          console.info("sse event data", data);
+          break;
       }
-    });
+    };
-    initializing = false;
-    // for controlling
-    return controller;
   };
 
   const stopStreaming = () => {
     console.info("sse stop streaming");
-    if (controller && controller.abort) {
+    if (SSE) {
-      controller.abort();
+      SSE.close();
     }
   };
 

yarn.lock

@@ -2296,11 +2296,6 @@
   dependencies:
     bowser "^2.9.0"
 
-"@microsoft/fetch-event-source@^2.0.1":
-  version "2.0.1"
-  resolved "https://mirrors.cloud.tencent.com/npm/@microsoft%2ffetch-event-source/-/fetch-event-source-2.0.1.tgz#9ceecc94b49fbaa15666e38ae8587f64acce007d"
-  integrity sha512-W6CLUJ2eBMw3Rec70qrsEW0jOm/3twwJv21mrmj2yORiaVmVYGS4sSS5yUwvQc1ZlDLYGPnClVWmUUMagKNsfA==
-
 "@nicolo-ribaudo/eslint-scope-5-internals@5.1.1-v1":
   version "5.1.1-v1"
   resolved "https://mirrors.cloud.tencent.com/npm/@nicolo-ribaudo%2feslint-scope-5-internals/-/eslint-scope-5-internals-5.1.1-v1.tgz#dbf733a965ca47b1973177dc0bb6c889edcfb129"