refactor: useStreaming

src/common/hook/useStreaming/index.js

@@ -1,9 +1,11 @@
-import { useState, useCallback, useEffect } from "react";
+import { useEffect, useState } from "react";
 import {
   fetchEventSource,
   EventStreamContentType,
 } from "@microsoft/fetch-event-source";
 import toast from "react-hot-toast";
+import dayjs from "dayjs";
+
 import BASE_URL from "../../../app/config";
 import { setReady } from "../../../app/slices/ui";
 import { useRenewMutation } from "../../../app/services/auth";
@@ -24,7 +26,7 @@ import {
 } from "../../../app/slices/contacts";
 import { resetAuthData } from "../../../app/slices/auth.data";
 import chatMessageHandler from "./chat.handler";
-
+import store from "../../../app/store";
 import { useDispatch, useSelector } from "react-redux";
 class RetriableError extends Error {}
 class FatalError extends Error {}
@@ -37,39 +39,52 @@ const getQueryString = (params = {}) => {
   });
   return sp.toString();
 };
-const StreamStatus = {
+// const StreamStatus = {
-  waiting: "waiting",
+//   waiting: "waiting",
-  initializing: "initializing",
+//   initialized: "initialized",
-  streaming: "streaming",
+//   streaming: "streaming",
-};
+// };
 let inter = null;
 export default function useStreaming() {
-  const store = useSelector((store) => store);
+  const [readyPullData, setReadyPullData] = useState(false);
-  const [
-    renewToken,
-    { data: tokenData, isSuccess: tokenRefreshSuccess },
-  ] = useRenewMutation();
-  const dispatch = useDispatch();
-  const [status, setStatus] = useState(StreamStatus.waiting);
-  const startStreaming = (token = "", refreshToken = "") => {
-    if (status !== StreamStatus.waiting) return;
-    const controller = new AbortController();
-    setStatus(StreamStatus.initializing);
   const {
-      authData: {
+    authData: { uid: loginUid },
-        token: reduxToken,
+    channels: { byId: channelData },
-        refreshToken: reduxRefreshToken,
+    ui: { ready, online },
-        uid: loginUid,
+    footprint: { afterMid, usersVersion, readUsers, readChannels },
-      },
+  } = useSelector((store) => store);
-      footprint: { afterMid, usersVersion },
+  const [renewToken] = useRenewMutation();
-    } = store;
+  const dispatch = useDispatch();
-    console.log("set uid use");
+  let initialized = false;
-    // 优先使用参数内的token
+  let initializing = false;
-    const finalToken = token || reduxToken;
+  let controller = new AbortController();
-    const finalRefreshToken = refreshToken || reduxRefreshToken;
+  const startStreaming = async () => {
-    fetchEventSource(
+    console.log("start streaming", initialized, initializing);
+    if (initialized || initializing) return;
+    // 如果token快要过期，先renew
+    const {
+      authData: { token, expireTime = new Date().getTime(), refreshToken },
+    } = store.getState();
+    let api_token = token;
+    const tokenAlmostExpire = dayjs().isAfter(new Date(expireTime - 20 * 1000));
+    console.log("check token expire time", tokenAlmostExpire);
+    if (tokenAlmostExpire) {
+      const {
+        data: { token: newToken },
+        isError,
+      } = await renewToken({
+        token,
+        refreshToken,
+      });
+      if (isError) return;
+      api_token = newToken;
+    }
+
+    // 开始初始化
+    initializing = true;
+    await fetchEventSource(
       `${BASE_URL}/user/events?${getQueryString({
-        "api-key": finalToken,
+        "api-key": api_token,
         users_version: usersVersion,
         after_mid: afterMid,
       })}`,
@@ -77,12 +92,13 @@ export default function useStreaming() {
         openWhenHidden: true,
         signal: controller.signal,
         async onopen(response) {
+          initializing = false;
           if (
             response.ok &&
             response.headers.get("content-type") === EventStreamContentType
           ) {
             console.log("sse everything ok");
-            setStatus(StreamStatus.streaming);
+            initialized = true;
             return; // everything's good
           } else if (
             response.status >= 400 &&
@@ -90,36 +106,29 @@ export default function useStreaming() {
             response.status !== 429
           ) {
             // 重新登录
-            if (response.status == 401) {
-              await renewToken({
-                token: finalToken,
-                refreshToken: finalRefreshToken,
-              });
-              return;
-            }
             // client-side errors are usually non-retriable:
+            console.log("sse debug: open fatal");
             throw new FatalError();
           } else {
+            // server error
+            console.log("sse debug: open retry");
             throw new RetriableError();
           }
         },
         onmessage(evt) {
+          initializing = false;
           console.log("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.log("sse debug: error message fatal");
             throw new FatalError(evt.data);
           }
-          const {
-            ui: { ready },
-            footprint: { readUsers, readChannels },
-            channels: { byId: channelData },
-          } = store;
           const data = JSON.parse(evt.data);
           const { type } = data;
           switch (type) {
             case "heartbeat":
-              console.log("heartbeat", store, loginUid);
+              console.log("heartbeat", loginUid);
               break;
             case "ready":
               console.log("streaming ready");
@@ -231,38 +240,63 @@ export default function useStreaming() {
         },
         onclose() {
           // if the server closes the connection unexpectedly, retry:
+          console.log("sse debug: closed");
+          initializing = false;
           throw new RetriableError();
         },
         onerror(err) {
+          initializing = false;
           if (err instanceof FatalError) {
+            console.log("sse debug: error fatal", err);
+            throw err; // rethrow to stop the operation
+          } else {
+            console.log("sse debug: error other", err);
+            stopStreaming();
             if (inter) {
               clearTimeout(inter);
             }
             // 重连
             inter = setTimeout(() => {
+              initialized = false;
               startStreaming();
-            }, 500);
+            }, 2000);
             throw err; // rethrow to stop the operation
-          } else {
             // do nothing to automatically retry. You can also
             // return a specific retry interval here.
           }
         },
       }
     );
+    initializing = false;
     // for controlling
     return controller;
   };
-  useEffect(() => {
+  const stopStreaming = () => {
-    if (tokenRefreshSuccess) {
+    console.log("stop st");
-      const { token, refresh_token } = tokenData;
+    if (controller && controller.abort) {
-      startStreaming(token, refresh_token);
+      controller.abort();
     }
-  }, [tokenData, tokenRefreshSuccess]);
+  };
+  const setStreamingReady = (ready) => {
+    setReadyPullData(ready);
+  };
+  useEffect(() => {
+    console.log("network changed", online, readyPullData);
+    if (readyPullData) {
+      if (online) {
+        startStreaming();
+      } else {
+        stopStreaming();
+      }
+    }
+    return () => {
+      stopStreaming();
+    };
+  }, [online, readyPullData]);
 
   return {
-    initializing: status == StreamStatus.initializing,
+    setStreamingReady,
-    streaming: status == StreamStatus.streaming,
     startStreaming,
+    stopStreaming,
   };
 }

src/routes/home/usePreload.js

@@ -8,11 +8,11 @@ import useStreaming from "../../common/hook/useStreaming";
 // const querySetting = {
 //   refetchOnMountOrArgChange: true,
 // };
-let request = null;
+// let request = null;
 export default function usePreload() {
   const { rehydrate, rehydrated } = useRehydrate();
   const loginUid = useSelector((store) => store.authData.uid);
-  const { startStreaming, streaming, initializing } = useStreaming();
+  const { setStreamingReady } = useStreaming();
   const [
     getContacts,
     {
@@ -35,9 +35,7 @@ export default function usePreload() {
     initCache();
     rehydrate();
     return () => {
-      if (request) {
+      setStreamingReady(false);
-        request.abort();
-      }
     };
   }, []);
 
@@ -47,10 +45,10 @@ export default function usePreload() {
       getServer();
     }
   }, [rehydrated]);
-  const canStreaming = loginUid && rehydrated && !initializing && !streaming;
+  const canStreaming = loginUid && rehydrated;
   useEffect(() => {
     if (canStreaming) {
-      request = startStreaming();
+      setStreamingReady(true);
     }
   }, [canStreaming]);