ColdBreeze/ColdBreeze-chat-web
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: streaming

Browse Source
This commit is contained in:
Tristan Yang
2022-12-11 17:39:31 +08:00
parent dac1665994
commit ad8c05e8cc
1 changed files with 254 additions and 240 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/hook/useStreaming/index.ts
+254 -240
View File
@@ -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 {