167 lines
4.4 KiB
TypeScript
167 lines
4.4 KiB
TypeScript
/**
|
|
* 차트·실시간 전략용 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<MessageHandler>;
|
|
stompSub?: StompSubscription;
|
|
}
|
|
|
|
let client: StompClient | null = null;
|
|
let connected = false;
|
|
const topics = new Map<string, TopicEntry>();
|
|
const connectionListeners = new Set<ConnectionListener>();
|
|
let disconnectTimer: ReturnType<typeof setTimeout> | 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);
|
|
}
|