Macbook
316 lines
8.5 KiB
TypeScript
316 lines
8.5 KiB
TypeScript
/**
|
|
* 업비트 WebSocket 단일 연결 브로커 (/upbit-ws).
|
|
* 훅·차트마다 WebSocket 을 열면 모바일에서 429·재연결 폭주가 발생한다.
|
|
*/
|
|
import type { UpbitMessage } from './upbitApi';
|
|
import { isDesktop } from './platform';
|
|
import { getSecureApiOrigin } from './backendApi';
|
|
|
|
export type UpbitWsChannel = 'ticker' | 'orderbook' | 'trade';
|
|
|
|
type RawHandler = (msg: Record<string, unknown>) => void;
|
|
type ConnectionListener = (connected: boolean) => void;
|
|
|
|
interface Listener {
|
|
channel: UpbitWsChannel;
|
|
codes: Set<string>;
|
|
handler: RawHandler;
|
|
}
|
|
|
|
const INITIAL_DELAY = 350;
|
|
const RECONNECT_BASE = 3000;
|
|
const RECONNECT_MAX = 45_000;
|
|
const PING_INTERVAL = 20_000;
|
|
const DISCONNECT_IDLE_MS = 200;
|
|
|
|
const refCounts = new Map<UpbitWsChannel, Map<string, number>>();
|
|
const listeners = new Set<Listener>();
|
|
const connectionListeners = new Set<ConnectionListener>();
|
|
|
|
let ws: WebSocket | null = null;
|
|
let connected = false;
|
|
let connectTimer: ReturnType<typeof setTimeout> | null = null;
|
|
let disconnectTimer: ReturnType<typeof setTimeout> | null = null;
|
|
let pingTimer: ReturnType<typeof setInterval> | null = null;
|
|
let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
|
|
let retryCount = 0;
|
|
let closingIntentionally = false;
|
|
let wsFailureLogged = false;
|
|
|
|
function getWsUrl(): string {
|
|
if (isDesktop()) {
|
|
const origin = getSecureApiOrigin().replace(/^http:\/\//i, 'https://');
|
|
const host = origin.replace(/^https:\/\//i, '');
|
|
return `wss://${host}/upbit-ws`;
|
|
}
|
|
const proto = window.location.protocol === 'https:' ? 'wss' : 'ws';
|
|
return `${proto}://${window.location.host}/upbit-ws`;
|
|
}
|
|
|
|
function notifyConnection(next: boolean): void {
|
|
connected = next;
|
|
connectionListeners.forEach(l => l(next));
|
|
}
|
|
|
|
function bumpRef(channel: UpbitWsChannel, code: string, delta: number): void {
|
|
const key = code.toUpperCase();
|
|
let m = refCounts.get(channel);
|
|
if (!m) {
|
|
m = new Map();
|
|
refCounts.set(channel, m);
|
|
}
|
|
const next = (m.get(key) ?? 0) + delta;
|
|
if (next <= 0) m.delete(key);
|
|
else m.set(key, next);
|
|
}
|
|
|
|
function codesForChannel(channel: UpbitWsChannel): string[] {
|
|
return Array.from(refCounts.get(channel)?.keys() ?? []);
|
|
}
|
|
|
|
function listenerCount(): number {
|
|
return listeners.size;
|
|
}
|
|
|
|
function mayOpenSocket(): boolean {
|
|
if (listenerCount() === 0) return false;
|
|
if (typeof document !== 'undefined' && document.visibilityState === 'hidden') return false;
|
|
return true;
|
|
}
|
|
|
|
function cancelPendingDisconnect(): void {
|
|
if (disconnectTimer) {
|
|
clearTimeout(disconnectTimer);
|
|
disconnectTimer = null;
|
|
}
|
|
}
|
|
|
|
function cancelPendingConnect(): void {
|
|
if (connectTimer) {
|
|
clearTimeout(connectTimer);
|
|
connectTimer = null;
|
|
}
|
|
}
|
|
|
|
function clearPing(): void {
|
|
if (pingTimer) {
|
|
clearInterval(pingTimer);
|
|
pingTimer = null;
|
|
}
|
|
}
|
|
|
|
function sendSubscribe(): void {
|
|
if (!ws || ws.readyState !== WebSocket.OPEN) return;
|
|
const packet: unknown[] = [{ ticket: `gc-${Date.now()}` }];
|
|
for (const ch of ['ticker', 'orderbook', 'trade'] as UpbitWsChannel[]) {
|
|
const codes = codesForChannel(ch);
|
|
if (codes.length > 0) packet.push({ type: ch, codes });
|
|
}
|
|
if (packet.length === 1) return;
|
|
packet.push({ format: 'DEFAULT' });
|
|
ws.send(JSON.stringify(packet));
|
|
}
|
|
|
|
function parseBinaryOrText(data: string | ArrayBuffer | Blob): Promise<string | null> {
|
|
if (typeof data === 'string') return Promise.resolve(data);
|
|
if (data instanceof ArrayBuffer) {
|
|
return Promise.resolve(new TextDecoder('utf-8').decode(data));
|
|
}
|
|
if (data instanceof Blob) return data.text();
|
|
return Promise.resolve(null);
|
|
}
|
|
|
|
function dispatch(text: string): void {
|
|
if (text === 'PING') {
|
|
ws?.send('PONG');
|
|
return;
|
|
}
|
|
let msg: Record<string, unknown>;
|
|
try {
|
|
msg = JSON.parse(text) as Record<string, unknown>;
|
|
} catch {
|
|
return;
|
|
}
|
|
const type = msg.type as string | undefined;
|
|
if (type !== 'ticker' && type !== 'orderbook' && type !== 'trade') return;
|
|
const code = String(msg.code ?? msg.market ?? '').toUpperCase();
|
|
if (!code) return;
|
|
|
|
for (const entry of listeners) {
|
|
if (entry.channel !== type) continue;
|
|
if (entry.codes.size > 0 && !entry.codes.has(code)) continue;
|
|
try {
|
|
entry.handler(msg);
|
|
} catch (err) {
|
|
console.error('[upbitWsBroker] handler error', err);
|
|
}
|
|
}
|
|
}
|
|
|
|
function scheduleReconnect(): void {
|
|
if (!mayOpenSocket()) return;
|
|
if (reconnectTimer) return;
|
|
const delay = Math.min(RECONNECT_BASE * 2 ** retryCount, RECONNECT_MAX);
|
|
retryCount += 1;
|
|
reconnectTimer = setTimeout(() => {
|
|
reconnectTimer = null;
|
|
openSocket();
|
|
}, delay);
|
|
}
|
|
|
|
function teardownSocket(): void {
|
|
clearPing();
|
|
if (reconnectTimer) {
|
|
clearTimeout(reconnectTimer);
|
|
reconnectTimer = null;
|
|
}
|
|
closingIntentionally = true;
|
|
if (ws) {
|
|
ws.onopen = null;
|
|
ws.onmessage = null;
|
|
ws.onerror = null;
|
|
ws.onclose = null;
|
|
if (ws.readyState === WebSocket.OPEN || ws.readyState === WebSocket.CONNECTING) {
|
|
ws.close(1000, 'broker idle');
|
|
}
|
|
ws = null;
|
|
}
|
|
closingIntentionally = false;
|
|
if (connected) notifyConnection(false);
|
|
}
|
|
|
|
function openSocket(): void {
|
|
if (listenerCount() === 0) return;
|
|
teardownSocket();
|
|
|
|
if (!mayOpenSocket()) return;
|
|
|
|
try {
|
|
const socket = new WebSocket(getWsUrl());
|
|
ws = socket;
|
|
socket.binaryType = 'arraybuffer';
|
|
|
|
socket.onopen = () => {
|
|
retryCount = 0;
|
|
wsFailureLogged = false;
|
|
notifyConnection(true);
|
|
sendSubscribe();
|
|
clearPing();
|
|
pingTimer = setInterval(() => {
|
|
if (socket.readyState === WebSocket.OPEN) socket.send('PING');
|
|
}, PING_INTERVAL);
|
|
};
|
|
|
|
socket.onmessage = (event: MessageEvent) => {
|
|
void parseBinaryOrText(event.data as string | ArrayBuffer | Blob).then(text => {
|
|
if (text != null) dispatch(text);
|
|
});
|
|
};
|
|
|
|
socket.onerror = () => {
|
|
if (!wsFailureLogged) {
|
|
wsFailureLogged = true;
|
|
console.warn(
|
|
'[upbitWsBroker] WebSocket 연결 실패 — 시세는 REST 폴링으로 대체됩니다. '
|
|
+ '서버 nginx에 /upbit-ws 프록시가 있는지 확인하세요.',
|
|
);
|
|
}
|
|
};
|
|
|
|
socket.onclose = () => {
|
|
clearPing();
|
|
notifyConnection(false);
|
|
ws = null;
|
|
if (!closingIntentionally && listenerCount() > 0) scheduleReconnect();
|
|
};
|
|
} catch (err) {
|
|
if (!wsFailureLogged) {
|
|
wsFailureLogged = true;
|
|
console.warn('[upbitWsBroker] connect failed', err);
|
|
}
|
|
scheduleReconnect();
|
|
}
|
|
}
|
|
|
|
function scheduleConnect(): void {
|
|
cancelPendingDisconnect();
|
|
if (ws?.readyState === WebSocket.OPEN) {
|
|
sendSubscribe();
|
|
return;
|
|
}
|
|
if (connectTimer) return;
|
|
connectTimer = setTimeout(() => {
|
|
connectTimer = null;
|
|
openSocket();
|
|
}, INITIAL_DELAY);
|
|
}
|
|
|
|
function scheduleDisconnect(): void {
|
|
cancelPendingConnect();
|
|
cancelPendingDisconnect();
|
|
disconnectTimer = setTimeout(() => {
|
|
disconnectTimer = null;
|
|
if (listenerCount() > 0) return;
|
|
teardownSocket();
|
|
}, DISCONNECT_IDLE_MS);
|
|
}
|
|
|
|
/** 연결 상태 구독 */
|
|
export function subscribeUpbitWsConnection(listener: ConnectionListener): () => void {
|
|
connectionListeners.add(listener);
|
|
listener(connected && ws?.readyState === WebSocket.OPEN);
|
|
return () => connectionListeners.delete(listener);
|
|
}
|
|
|
|
if (typeof document !== 'undefined') {
|
|
document.addEventListener('visibilitychange', () => {
|
|
if (document.visibilityState === 'hidden') {
|
|
cancelPendingConnect();
|
|
if (listenerCount() === 0) teardownSocket();
|
|
return;
|
|
}
|
|
if (mayOpenSocket()) scheduleConnect();
|
|
});
|
|
}
|
|
|
|
/**
|
|
* 업비트 WS 채널 구독 (참조 카운트). codes 가 비어 있으면 구독하지 않음.
|
|
*/
|
|
export function subscribeUpbitWs(
|
|
channel: UpbitWsChannel,
|
|
codes: string[],
|
|
handler: RawHandler,
|
|
): () => void {
|
|
const normalized = [...new Set(codes.map(c => c.toUpperCase()).filter(Boolean))];
|
|
if (normalized.length === 0) {
|
|
return () => {};
|
|
}
|
|
|
|
normalized.forEach(code => bumpRef(channel, code, 1));
|
|
const entry: Listener = { channel, codes: new Set(normalized), handler };
|
|
listeners.add(entry);
|
|
scheduleConnect();
|
|
|
|
return () => {
|
|
listeners.delete(entry);
|
|
normalized.forEach(code => bumpRef(channel, code, -1));
|
|
if (listenerCount() === 0) {
|
|
cancelPendingConnect();
|
|
scheduleDisconnect();
|
|
} else if (ws?.readyState === WebSocket.OPEN) {
|
|
sendSubscribe();
|
|
}
|
|
};
|
|
}
|
|
|
|
/** 차트용 trade 구독 — UpbitMessage 타입 핸들러 */
|
|
export function subscribeUpbitTrade(
|
|
market: string,
|
|
handler: (msg: UpbitMessage) => void,
|
|
): () => void {
|
|
return subscribeUpbitWs('trade', [market], raw => {
|
|
if (raw.type !== 'trade') return;
|
|
handler(raw as unknown as UpbitMessage);
|
|
});
|
|
}
|