diff --git a/package.json b/package.json index 4ade2e72..e139c75a 100644 --- a/package.json +++ b/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", diff --git a/src/common/hook/useStreaming/index.ts b/src/common/hook/useStreaming/index.ts index 03575457..939c5491 100644 --- a/src/common/hook/useStreaming/index.ts +++ b/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, - 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; + // 开始初始化推送 + SSE = new EventSource(`${BASE_URL}/user/events?${getQueryString(params)}`); - 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"); + SSE.onopen = () => { + initializing = false; + initialized = true; + }; + SSE.onerror = (err) => { + console.error("sse error", err); + + initializing = false; + stopStreaming(); + if (inter) { + clearTimeout(inter); + } + // 重连 + inter = window.setTimeout(() => { + initialized = false; + startStreaming(); + }, 2000); + }; + SSE.onmessage = (evt) => { + initializing = false; + console.info("sse message", 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 "delete_user": - dispatch(resetAuthData()); - toast("Your account has been deleted"); + 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 "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; + } + 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 { 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", + operation: "remove_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); } - // 重连 - inter = window.setTimeout(() => { - initialized = false; - startStreaming(); - }, 2000); - throw err; // rethrow to stop the operation - // do nothing to automatically retry. You can also - // return a specific retry interval here. - } + 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; } - }); - initializing = false; - // for controlling - return controller; + }; }; const stopStreaming = () => { console.info("sse stop streaming"); - if (controller && controller.abort) { - controller.abort(); + if (SSE) { + SSE.close(); } }; diff --git a/yarn.lock b/yarn.lock index 2fdb945b..4a830c91 100644 --- a/yarn.lock +++ b/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"