/** * 차트·실시간 전략용 STOMP 단일 연결. * 웹: SockJS 1회 / Desktop: 순수 WebSocket (cross-origin SockJS 세션 불가) */ import { Client as StompClient, IMessage, StompSubscription } from '@stomp/stompjs'; import SockJS from 'sockjs-client'; import { getStompNativeWebSocketUrl, getStompSockJsUrl } from './backendApi'; import { isDesktop } from './platform'; type MessageHandler = (msg: IMessage) => void; type ConnectionListener = (connected: boolean) => void; interface TopicEntry { handlers: Set; stompSub?: StompSubscription; } let client: StompClient | null = null; let connected = false; const topics = new Map(); const connectionListeners = new Set(); let disconnectTimer: ReturnType | null = null; function notifyConnection(next: boolean): void { connected = next; connectionListeners.forEach(l => l(next)); } function handlerCount(): number { let n = 0; for (const entry of topics.values()) n += entry.handlers.size; return n; } function cancelPendingDisconnect(): void { if (disconnectTimer) { clearTimeout(disconnectTimer); disconnectTimer = null; } } function resubscribeAll(): void { if (!client?.connected) return; for (const [topic, entry] of topics) { if (entry.handlers.size === 0) continue; entry.stompSub?.unsubscribe(); entry.stompSub = client.subscribe(topic, msg => { entry.handlers.forEach(h => h(msg)); }); } } function stompConnectLabel(): string { return isDesktop() ? getStompNativeWebSocketUrl() : getStompSockJsUrl(); } function createStompClient(): StompClient { const shared = { reconnectDelay: 5000, heartbeatIncoming: 10_000, heartbeatOutgoing: 10_000, onConnect: () => { cancelPendingDisconnect(); notifyConnection(true); resubscribeAll(); }, onDisconnect: () => { notifyConnection(false); for (const entry of topics.values()) { entry.stompSub = undefined; } }, onStompError: (frame: { headers?: { message?: string } }) => { console.warn('[stompChartBroker] STOMP error', frame.headers?.message ?? frame); }, onWebSocketError: (ev: Event) => { console.warn('[stompChartBroker] WebSocket error', stompConnectLabel(), ev); }, onWebSocketClose: () => { notifyConnection(false); for (const entry of topics.values()) { entry.stompSub = undefined; } }, }; if (isDesktop()) { const brokerURL = getStompNativeWebSocketUrl(); return new StompClient({ ...shared, brokerURL }); } const sockJsUrl = getStompSockJsUrl(); return new StompClient({ ...shared, webSocketFactory: () => new SockJS(sockJsUrl), }); } function getOrCreateClient(): StompClient { if (client) return client; client = createStompClient(); return client; } function ensureActive(): void { cancelPendingDisconnect(); const c = getOrCreateClient(); if (!c.active) c.activate(); } function scheduleDisconnect(): void { cancelPendingDisconnect(); disconnectTimer = setTimeout(() => { disconnectTimer = null; if (handlerCount() > 0) return; if (!client?.active) return; if (!connected) { scheduleDisconnect(); return; } client.deactivate(); notifyConnection(false); }, 5000); } /** STOMP 연결 상태 변경 구독 */ export function subscribeStompConnection(listener: ConnectionListener): () => void { connectionListeners.add(listener); listener(connected && Boolean(client?.connected)); return () => { connectionListeners.delete(listener); }; } /** * STOMP 토픽 구독 (참조 카운트). 반환 함수로 해제. */ export function subscribeStompTopic(topic: string, handler: MessageHandler): () => void { let entry = topics.get(topic); if (!entry) { entry = { handlers: new Set() }; topics.set(topic, entry); } entry.handlers.add(handler); if (connected && client && entry.handlers.size === 1) { entry.stompSub = client.subscribe(topic, msg => { entry!.handlers.forEach(h => h(msg)); }); } ensureActive(); return () => { const e = topics.get(topic); if (!e) return; e.handlers.delete(handler); if (e.handlers.size === 0) { e.stompSub?.unsubscribe(); topics.delete(topic); scheduleDisconnect(); } }; } export function isStompChartConnected(): boolean { return connected && Boolean(client?.connected); }