From ad8c05e8cc54d7c4b46744f45219fffdf05e051f Mon Sep 17 00:00:00 2001 From: Tristan Yang Date: Sun, 11 Dec 2022 17:39:31 +0800 Subject: [PATCH] refactor: streaming --- src/common/hook/useStreaming/index.ts | 494 +++++++++++++------------- 1 file changed, 254 insertions(+), 240 deletions(-) diff --git a/src/common/hook/useStreaming/index.ts b/src/common/hook/useStreaming/index.ts index 41d76a68..fc9060cb 100644 --- a/src/common/hook/useStreaming/index.ts +++ b/src/common/hook/useStreaming/index.ts @@ -1,6 +1,6 @@ -import { useEffect, useState } from "react"; +import { useEffect, useState, useCallback, useRef } from "react"; import toast from "react-hot-toast"; -import BASE_URL from "../../../app/config"; +import BASE_URL, { KEY_EXPIRE, KEY_REFRESH_TOKEN, KEY_TOKEN } from "../../../app/config"; import { setReady } from "../../../app/slices/ui"; import { fillChannels, @@ -33,287 +33,301 @@ const getQueryString = (params: { [key: string]: string }) => { }); return sp.toString(); }; -let inter: number | null = null; +const getLocalAuthData = () => { + return { + token: localStorage.getItem(KEY_TOKEN) || "", + refreshToken: localStorage.getItem(KEY_REFRESH_TOKEN) || "", + expireTime: Number(localStorage.getItem(KEY_EXPIRE) || +new Date()) + }; +}; let SSE: EventSource | undefined = undefined; -let opened = false; - +let connectionIsOpen = false; +let aliveInter: number = 0; export default function useStreaming() { + const opened = useRef(false); const [renewToken] = useRenewMutation(); - const [readyPullData, setReadyPullData] = useState(false); + const [streamingReady, setStreamingReady] = useState(false); const { - authData, + authData: { user }, ui: { ready }, footprint: { afterMid, usersVersion, readUsers, readChannels } } = useAppSelector((store) => store); const dispatch = useAppDispatch(); - const loginUid = authData.user?.uid || 0; - let aliveInter: number = 0; - const keepAlive = () => { - clearTimeout(aliveInter); - // 比15秒多2秒 + const loginUid = user?.uid || 0; + + const keepAlive = (timeout?: number) => { + window.clearTimeout(aliveInter); + // 比15秒多5秒 aliveInter = window.setTimeout(() => { // 重启连接 stopStreaming(); startStreaming(); - }, 17000); + }, timeout ?? 20000); }; - const startStreaming = async () => { - console.log("start streaming", SSE, SSE?.readyState); - - if (SSE && (SSE.readyState === EventSource.OPEN || SSE.readyState === EventSource.CONNECTING)) return; - const { token = "", refreshToken, expireTime = +new Date() } = authData; - // token 非空 - if (!token) { - console.info("sse start streaming no token", token); - return; - } - // 如果token快要过期，先renew - if (dayjs().isAfter(new Date(expireTime - 20 * 1000))) { - renewToken({ token, refresh_token: refreshToken }); - return; - } - // 开始初始化 - const params: { - "api-key": string; - after_mid?: string; - users_version?: string; - } = { - "api-key": token, - }; - // 如果afterMid是0，则不传该参数 - if (afterMid !== 0) { - params.after_mid = `${afterMid}`; - } - // 如果usersVersion是0，则不传该参数 - if (usersVersion !== 0) { - params.users_version = `${usersVersion}`; - } - // 开始初始化推送 - SSE = new EventSource(`${BASE_URL}/user/events?${getQueryString(params)}`); - - SSE.onopen = () => { - //todo - opened = true; - }; - SSE.onerror = (err) => { - const { readyState } = err.target as EventSource; - console.error("sse error", readyState, err); - // 连接还处于开启状态 - if (readyState === EventSource.OPEN) { + const startStreaming = useCallback( + async () => { + console.log("start streaming", SSE, SSE?.readyState); + if (connectionIsOpen) return; + window.clearTimeout(aliveInter); + if (SSE && (SSE.readyState === EventSource.OPEN || SSE.readyState === EventSource.CONNECTING)) return; + const { token, refreshToken, expireTime } = getLocalAuthData(); + // token 非空 + if (!token) { + console.info("sse start streaming no token", token); return; } - if (inter) { - clearTimeout(inter); - } - // // 重连 - inter = window.setTimeout(() => { - startStreaming(); - }, 1000); - }; - SSE.onmessage = (evt) => { - console.info("sse message", evt.data); - const data: ServerEvent = JSON.parse(evt.data); - const { type } = data; - switch (type) { - case "heartbeat": { - keepAlive(); - console.info("sse heartbeat", loginUid); + let _token = token; + // 如果token快要过期，先renew + if (dayjs().isAfter(new Date(expireTime - 20 * 1000))) { + const resp = await renewToken({ token, refresh_token: refreshToken }); + if ("error" in resp) { + return; + } else { + _token = resp.data.token; } - 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)); - // 特殊处理当前登录用户的更新 - logs.forEach((log) => { - const { uid, action, log_id, ...rest } = log; - if (uid === loginUid && action === 'update') { - dispatch(updateLoginUser(omitBy(rest, isNull))); - } - }); - } - 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; + // return; + } + // 开始初始化 + const params: { + "api-key": string; + after_mid?: string; + users_version?: string; + } = { + "api-key": _token, + }; + // 如果afterMid是0，则不传该参数 + if (afterMid !== 0) { + params.after_mid = `${afterMid}`; + } + // 如果usersVersion是0，则不传该参数 + if (usersVersion !== 0) { + params.users_version = `${usersVersion}`; + } + // 开始初始化推送 + SSE = new EventSource(`${BASE_URL}/user/events?${getQueryString(params)}`); + SSE.onopen = () => { + //todo + opened.current = true; + connectionIsOpen = true; + }; + SSE.onerror = (err) => { + const { readyState } = err.target as EventSource; + console.error("sse error", readyState, err); + // 连接还处于开启状态 + if (readyState === EventSource.OPEN || readyState === EventSource.CONNECTING) { + return; + } + // 重连 + connectionIsOpen = false; + keepAlive(2000); + }; + SSE.onmessage = (evt) => { + console.info("sse message", evt.data); + const data: ServerEvent = JSON.parse(evt.data); + const { type } = data; + switch (type) { + case "heartbeat": { + keepAlive(); + console.info("sse heartbeat", loginUid); + } + break; + case "ready": + console.info("sse streaming ready"); + // 有时候，heartbeat不会发？ + keepAlive(); + 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)); + // 特殊处理当前登录用户的更新 + logs.forEach((log) => { + const { uid, action, log_id, ...rest } = log; + if (uid === loginUid && action === 'update') { + dispatch(updateLoginUser(omitBy(rest, isNull))); + } + }); + } + 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: + 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 "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": - console.info("sse fill channels from streaming", data); - dispatch(fillChannels(data.groups)); - break; - case "joined_group": - console.info("sse joined group", data.group); - dispatch(addChannel(data.group)); - break; - case "group_changed": - { - 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 { + break; + case "related_groups": + console.info("sse fill channels from streaming", data); + dispatch(fillChannels(data.groups)); + break; + case "joined_group": + console.info("sse joined group", data.group); + dispatch(addChannel(data.group)); + break; + case "group_changed": + { + const { gid, ...rest } = data; dispatch( updateChannel({ - operation: "remove_member", + 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 "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; + 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; + } + }; + }, + [afterMid, usersVersion], + ); - default: - console.info("sse event data", data); - break; - } - }; - }; const stopStreaming = () => { console.info("sse stop streaming"); if (SSE) { SSE.close(); SSE = undefined; + connectionIsOpen = false; } }; - const setStreamingReady = (ready: boolean) => { - setReadyPullData(ready); - }; - useEffect(() => { - if (readyPullData) { - // if (online) { + // 确保只执行一次 + const hasOpened = opened.current; + if (streamingReady && !hasOpened) { startStreaming(); - // } else { - // stopStreaming(); - // } } return () => { - console.log("stop from readyPullData"); - stopStreaming(); + if (streamingReady && !hasOpened) { + console.log("stop from streamingReady"); + stopStreaming(); + } }; - }, [readyPullData]); + }, [streamingReady]); return {