fix connection event push

This commit is contained in:
undefined committed 2025-06-10 00:24:55 +08:00
1 parent 067ca3c6bb
commit 2f72600872
3 files changed
+28 -26

No files matched your search

@@ -45,12 +45,6 @@ class WebsocketEventsConnectionManagerHandler extends ConnectionHandler {
} }
async message(payload: any) { async message(payload: any) {
try {
payload = JSON.parse(payload);
} catch (e) {
console.error('Invalid payload', e);
return;
}
const success = []; const success = [];
const failed = []; const failed = [];
const session = payload?.credential const session = payload?.credential
@@ -59,14 +59,14 @@ const onmessage = (msg) => {
return true; return true;
}; };
const initWorkerMode = (endpoint) => { const initWorkerMode = () => {
console.log('Messages: using SharedWorker'); console.log('Messages: using SharedWorker');
const worker = new SharedWorker(new URL('./worker.ts', import.meta.url), { name: 'HydroMessagesWorker' }); const worker = new SharedWorker(new URL('./worker.ts', import.meta.url), { name: 'HydroMessagesWorker' });
worker.port.start(); worker.port.start();
window.addEventListener('beforeunload', () => { window.addEventListener('beforeunload', () => {
worker.port.postMessage({ type: 'unload' }); worker.port.postMessage({ type: 'unload' });
}); });
worker.port.postMessage({ type: 'conn', path: endpoint, cookie: document.cookie }); worker.port.postMessage({ type: 'conn', cookie: document.cookie });
worker.port.onmessage = async (message) => { worker.port.onmessage = async (message) => {
if (process.env.NODE_ENV !== 'production') console.log('onmessage: ', message); if (process.env.NODE_ENV !== 'production') console.log('onmessage: ', message);
const { payload, type } = message.data; const { payload, type } = message.data;
@@ -89,13 +89,11 @@ const messagePage = new AutoloadPage('messagePage', (pagename) => {
action: () => window.open('/home/messages', '_blank'), action: () => window.open('/home/messages', '_blank'),
}).show(); }).show();
} }
const url = new URL(`${UiContext.ws_prefix}home/messages-conn`, window.location.href); const url = new URL(`${UiContext.ws_prefix}websocket`, window.location.href);
// TODO handle a better way for cookie
if (url.host !== window.location.host) url.searchParams.append('sid', document.cookie.split('sid=')[1].split(';')[0]);
const endpoint = url.toString().replace('http', 'ws'); const endpoint = url.toString().replace('http', 'ws');
if (window.SharedWorker) { if (window.SharedWorker) {
try { try {
initWorkerMode(endpoint); initWorkerMode();
return; return;
} catch (e) { } catch (e) {
console.error('SharedWorker init fail: ', e.message); console.error('SharedWorker init fail: ', e.message);
@@ -129,6 +127,14 @@ const messagePage = new AutoloadPage('messagePage', (pagename) => {
localStorage.setItem('page.master', selfId); localStorage.setItem('page.master', selfId);
const masterChannel = new BroadcastChannel('hydro-messages'); const masterChannel = new BroadcastChannel('hydro-messages');
const sock = new Sock(endpoint); const sock = new Sock(endpoint);
sock.onopen = () => {
sock.send(JSON.stringify({
operation: 'subscribe',
request_id: Math.random().toString(16).substring(2),
credential: document.cookie.split('sid=')[1].split(';')[0],
channels: ['message'],
}));
};
sock.onmessage = async (message) => { sock.onmessage = async (message) => {
const payload = JSON.parse(message.data); const payload = JSON.parse(message.data);
masterChannel.postMessage({ type: 'message', payload }); masterChannel.postMessage({ type: 'message', payload });
@@ -13,7 +13,6 @@ const ports: Set<MessagePort> = new Set();
interface RequestInitSharedConnPayload { interface RequestInitSharedConnPayload {
type: 'conn'; type: 'conn';
cookie: string; cookie: string;
path: string;
} }
interface RequestAckPayload { interface RequestAckPayload {
type: 'ack'; type: 'ack';
@@ -53,19 +52,22 @@ function onMessage(payload: any) {
}, 5000); }, 5000);
} }
function initConn(path: string, port: MessagePort, cookie: any) { function initConn(port: MessagePort, cookie: any) {
ports.add(port); ports.add(port);
if (cookie !== lcookie) conn?.close(); console.log('Init connection');
else if (conn && conn.readyState === conn.OPEN) return; lcookie = cookie.split('sid=')[1].split(';')[0];
const url = new URL(path, location.origin); if (conn) return;
if (cookie) url.searchParams.set('sid', cookie.split('sid=')[1].split(';')[0]); const url = new URL('/websocket', location.origin);
lcookie = cookie; conn = new ReconnectingWebsocket(url.toString().replace('http', 'ws'));
console.log('Init connection for', path);
conn?.close();
conn = new ReconnectingWebsocket(url.toString());
conn.onopen = () => { conn.onopen = () => {
console.log('Connected to', path); console.log('Connected');
broadcastMsg({ type: 'open' }); broadcastMsg({ type: 'open' });
conn.send(JSON.stringify({
'operation': 'subscribe',
'request_id': Math.random().toString(16).substring(2),
'credential': lcookie,
'channels': ['message'],
}));
}; };
conn.onerror = () => broadcastMsg({ type: 'error' }); conn.onerror = () => broadcastMsg({ type: 'error' });
conn.onclose = (ev) => broadcastMsg({ type: 'close', error: ev.reason }); conn.onclose = (ev) => broadcastMsg({ type: 'close', error: ev.reason });
@@ -79,8 +81,8 @@ function initConn(path: string, port: MessagePort, cookie: any) {
if (['PermissionError', 'PrivilegeError'].includes(payload.error)) { if (['PermissionError', 'PrivilegeError'].includes(payload.error)) {
broadcastMsg({ type: 'close', error: payload.error }); broadcastMsg({ type: 'close', error: payload.error });
conn.close(); conn.close();
} else { } else if (payload.operation === 'event') {
onMessage(payload); onMessage(payload.payload);
} }
}; };
} }
@@ -89,7 +91,7 @@ self.onconnect = function (e) {
const port = e.ports[0]; const port = e.ports[0];
port.addEventListener('message', (msg: { data: RequestPayload }) => { port.addEventListener('message', (msg: { data: RequestPayload }) => {
if (msg.data.type === 'conn') initConn(msg.data.path, port, msg.data.cookie); if (msg.data.type === 'conn') initConn(port, msg.data.cookie);
if (msg.data.type === 'ack') ack[msg.data.id]?.(); if (msg.data.type === 'ack') ack[msg.data.id]?.();
if (msg.data.type === 'unload') ports.delete(port); if (msg.data.type === 'unload') ports.delete(port);
}); });