import pako from 'pako'; import { AppState, AppStateStatus } from 'react-native'; import { api, WS_URL } from './api'; import { eventBus } from '@/core/events/EventBus'; import { MessageCategory, SystemMessageType, SystemMessageExtraData, MessageSegment } from '@/types/dto'; import { systemNotificationService } from '../notification/systemNotificationService'; import { showVerificationModal } from './verification'; export type WSMessageType = | 'chat' | 'message' | 'read' | 'typing' | 'recall' | 'notification' | 'announcement' | 'group_message' | 'group_typing' | 'group_notice' | 'group_mention' | 'group_read' | 'group_recall' | 'notice' | 'request' | 'meta' | 'private' | 'group' | 'follow' | 'like' | 'comment' | 'heartbeat' | 'pong' | 'message_sent' | 'message_recalled' | 'error' | 'call_incoming' | 'call_accepted' | 'call_rejected' | 'call_busy' | 'call_ready' | 'call_ended' | 'call_peer_muted' | 'call_invited' | 'call_answered_elsewhere' | 'call_participant_joined' | 'call_participant_left' | 'sync_required'; export interface WSCallIncomingMessage { type: 'call_incoming'; call_id: string; conversation_id: string; caller_id: string; call_type: string; media_type?: string; created_at: number; lifetime?: number; } export interface WSCallAcceptedMessage { type: 'call_accepted'; call_id: string; started_at: number; } export interface WSCallRejectedMessage { type: 'call_rejected'; call_id: string; reason?: string; } export interface WSCallBusyMessage { type: 'call_busy'; call_id: string; } export interface WSCallEndedMessage { type: 'call_ended'; call_id: string; ended_by?: string; reason?: string; duration?: number; ended_at?: number; } export interface WSCallPeerMutedMessage { type: 'call_peer_muted'; call_id: string; user_id: string; muted: boolean; } export interface WSCallInvitedMessage { type: 'call_invited'; call_id: string; conversation_id: string; callee_id: string; } export interface WSCallAnsweredElsewhereMessage { type: 'call_answered_elsewhere'; call_id: string; reason?: string; } export interface WSCallParticipantJoinedMessage { type: 'call_participant_joined'; call_id: string; user_id: string; } export interface WSCallParticipantLeftMessage { type: 'call_participant_left'; call_id: string; user_id: string; reason?: string; } export interface WSErrorMessage { type: 'error'; code: string; message: string; } export interface WSChatMessage { type: 'chat'; conversation_id: string; id: string; sender_id: string; seq: number; segments?: MessageSegment[]; created_at: string; } export interface WSReadMessage { type: 'read'; conversation_id: string; user_id: string; seq: number; } export interface WSTypingMessage { type: 'typing'; conversation_id: string; user_id: string; is_typing: boolean; } export interface WSRecallMessage { type: 'recall'; conversation_id: string; message_id: string; } export interface WSNotificationMessage { type: 'notification'; id: string; sender_id?: string; receiver_id?: string; content: string; category?: MessageCategory; system_type?: SystemMessageType; extra_data?: SystemMessageExtraData; is_read?: boolean; created_at: string; } export interface WSAnnouncementMessage { type: 'announcement'; id: string; sender_id?: string; receiver_id?: string; content: string; category?: MessageCategory; system_type?: SystemMessageType; extra_data?: SystemMessageExtraData; created_at: string; } export interface WSGroupMentionMessage { type: 'group_mention'; group_id: string; conversation_id: string; message_id: string; from_user_id: string; content: string; mention_all: boolean; created_at: string; } export interface WSGroupChatMessage { type: 'group_message'; conversation_id: string; group_id: string; id: string; sender_id: string; seq: number; segments?: MessageSegment[]; created_at: string; } export interface WSGroupTypingMessage { type: 'group_typing'; group_id: string; user_id: string; is_typing: boolean; } export type GroupNoticeType = 'member_join' | 'member_leave' | 'member_removed' | 'role_changed' | 'muted' | 'unmuted'; export interface WSGroupNoticeMessage { type: 'group_notice'; notice_type: GroupNoticeType; group_id: string; data: { user_id?: string; operator_id?: string; role?: string; [key: string]: any; }; timestamp: number; message_id?: string; seq?: number; } export interface WSGroupReadMessage { type: 'group_read'; group_id: string; conversation_id: string; user_id: string; seq: number; } export interface WSGroupRecallMessage { type: 'group_recall'; group_id: string; conversation_id: string; message_id: string; } export interface WSSyncRequiredMessage { type: 'sync_required'; } export interface WSServerMessage { event_id?: number; type: string; ts?: number; payload?: any; } export type WSMessage = | WSChatMessage | WSReadMessage | WSTypingMessage | WSRecallMessage | WSNotificationMessage | WSAnnouncementMessage | WSGroupChatMessage | WSGroupTypingMessage | WSGroupNoticeMessage | WSGroupMentionMessage | WSGroupReadMessage | WSGroupRecallMessage | WSCallIncomingMessage | WSCallAcceptedMessage | WSCallRejectedMessage | WSCallBusyMessage | WSCallEndedMessage | WSCallPeerMutedMessage | WSCallInvitedMessage | WSCallAnsweredElsewhereMessage | WSCallParticipantJoinedMessage | WSCallParticipantLeftMessage | WSErrorMessage | WSSyncRequiredMessage; type MessageHandler = (message: T) => void; type ConnectionHandler = () => void; // 发送队列中的消息 interface PendingMessage { id: string; type: string; payload: any; resolve: (value: any) => void; reject: (reason: any) => void; timestamp: number; } class WebSocketService { private ws: WebSocket | null = null; private isConnecting = false; private connected = false; private reconnectAttempts = 0; private maxReconnectAttempts = 20; private reconnectDelay = 3000; private reconnectTimer: ReturnType | null = null; private heartbeatTimer: ReturnType | null = null; private heartbeatInterval = 30000; // 30秒心跳 private messageHandlers: Map = new Map(); private connectionHandlers: ConnectionHandler[] = []; private disconnectionHandlers: ConnectionHandler[] = []; private appStateSubscription: any = null; private lastAppState: AppStateStatus = 'active'; private shouldRun = false; private lastEventId = ''; private lastActivityAt = 0; // 发送队列 private pendingMessages: PendingMessage[] = []; private messageIdCounter = 0; // 降级状态 private isFallbackMode = false; private fallbackCheckTimer: ReturnType | null = null; // 通话中阻止后台断开(对方请求权限等场景会导致 AppState 切换,不应断开 WebSocket) preventDisconnectOnBackground = false; private getWSUrl(token: string | null): string { return `${WS_URL}?token=${encodeURIComponent(token || '')}&compress=1`; } async connect(): Promise { this.shouldRun = true; if (this.isConnecting || this.isConnected()) { console.log('[WebSocket] Already connected or connecting'); return true; } this.isConnecting = true; console.log('[WebSocket] Connecting...'); try { const token = await api.getToken(); if (!token) { console.log('[WebSocket] No token available, skipping connection'); this.isConnecting = false; // 没有 token 时不要重连,等待用户登录后再连接 return false; } const url = this.getWSUrl(token); console.log('[WebSocket] Connecting to:', url.replace(token, '***')); this.ws = new WebSocket(url); this.ws.binaryType = 'arraybuffer'; this.ws.onopen = () => { console.log('[WebSocket] Connected successfully'); this.isConnecting = false; this.connected = true; this.reconnectAttempts = 0; this.isFallbackMode = false; this.markActivity(); this.startHeartbeat(); this.connectionHandlers.forEach(h => h()); // 发送队列中的消息 this.flushPendingMessages(); }; this.ws.onmessage = (event) => { this.markActivity(); this.handleIncoming(event.data); }; this.ws.onerror = (error) => { console.error('[WebSocket] Error:', error); }; this.ws.onclose = (event) => { console.log('[WebSocket] Closed:', event.code, event.reason); this.handleDisconnected(); }; return true; } catch (error) { console.error('[WebSocket] Connection error:', error); this.isConnecting = false; this.scheduleReconnect(); return false; } } private handleIncoming(data: string | ArrayBuffer): void { let textData: string; if (typeof data === 'string') { textData = data; } else { // Binary frame — gzip decompress const decompressed = pako.inflate(new Uint8Array(data as ArrayBuffer)); textData = new TextDecoder('utf-8').decode(decompressed); } // 后端 writePump 会将多条消息以 \n 分隔合并发送,需要逐条解析 const lines = textData.split('\n'); for (const line of lines) { const trimmed = line.trim(); if (!trimmed) continue; try { const msg: WSServerMessage = JSON.parse(trimmed); this.processMessage(msg); } catch (error) { console.error('Failed to parse WebSocket message:', error); } } } private processMessage(msg: WSServerMessage): void { // 处理pong响应 if (msg.type === 'pong') { return; } // 处理错误消息 - emit 到上层处理 if (msg.type === 'error') { console.error('WebSocket error:', msg.payload); const errorCode: string = msg.payload?.code || 'unknown'; const err: WSErrorMessage = { type: 'error', code: errorCode, message: msg.payload?.message || 'Unknown error', }; if (errorCode === 'VERIFICATION_REQUIRED') { this.handleVerificationRequired(); } this.emit('error', err); return; } // 处理消息发送成功响应 if (msg.type === 'message_sent') { this.handleMessageSent(msg.payload); return; } // 处理消息撤回响应 if (msg.type === 'message_recalled') { this.handleMessageRecalled(msg.payload); return; } // 保存事件ID用于断线重连 if (msg.event_id) { this.lastEventId = String(msg.event_id); } // 分发事件 this.dispatchServerMessage(msg); } private dispatchServerMessage(msg: WSServerMessage): void { const { type, payload } = msg; switch (type) { case 'chat_message': this.handleChatMessage(payload); break; case 'message_read': this.handleMessageRead(payload); break; case 'typing': this.handleTyping(payload); break; case 'message_recall': this.handleMessageRecall(payload); break; case 'group_notice': this.handleGroupNotice(payload); break; case 'system_notification': this.handleSystemNotification(payload); break; // Call signaling messages case 'call_incoming': this.handleCallIncoming(payload); break; case 'call_accepted': this.handleCallAccepted(payload); break; case 'call_rejected': this.handleCallRejected(payload); break; case 'call_busy': this.handleCallBusy(payload); break; case 'call_ended': this.handleCallEnded(payload); break; case 'call_peer_muted': this.handleCallPeerMuted(payload); break; case 'call_invited': this.handleCallInvited(payload); break; case 'call_answered_elsewhere': this.handleCallAnsweredElsewhere(payload); break; case 'sync_required': this.emit('sync_required', { type: 'sync_required' }); break; default: console.log('Unknown message type:', type); } } private handleChatMessage(payload: any): void { const detailType = payload?.detail_type || 'private'; const m = payload?.message || payload; if (detailType === 'group') { const gm: WSGroupChatMessage = { type: 'group_message', conversation_id: m.conversation_id, group_id: String(m.group_id ?? ''), id: m.id, sender_id: m.sender_id, seq: Number(m.seq || 0), segments: m.segments || [], created_at: m.created_at || new Date().toISOString(), }; this.emit('group_message', gm); } else { const cm: WSChatMessage = { type: 'chat', conversation_id: m.conversation_id, id: m.id, sender_id: m.sender_id, seq: Number(m.seq || 0), segments: m.segments || [], created_at: m.created_at || new Date().toISOString(), }; this.emit('chat', cm); } } private handleMessageRead(payload: any): void { const detailType = payload?.detail_type || 'private'; if (detailType === 'group') { const m: WSGroupReadMessage = { type: 'group_read', group_id: String(payload.group_id ?? ''), conversation_id: payload.conversation_id, user_id: payload.user_id, seq: Number(payload.seq || 0), }; this.emit('group_read', m); } else { const m: WSReadMessage = { type: 'read', conversation_id: payload.conversation_id, user_id: payload.user_id, seq: Number(payload.seq || 0), }; this.emit('read', m); } } private handleTyping(payload: any): void { const detailType = payload?.detail_type || 'private'; if (detailType === 'group') { const m: WSGroupTypingMessage = { type: 'group_typing', group_id: String(payload.group_id ?? ''), user_id: payload.user_id, is_typing: payload.is_typing !== false, }; this.emit('group_typing', m); } else { const m: WSTypingMessage = { type: 'typing', conversation_id: payload.conversation_id, user_id: payload.user_id, is_typing: payload.is_typing !== false, }; this.emit('typing', m); } } private handleMessageRecall(payload: any): void { const detailType = payload?.detail_type || 'private'; if (detailType === 'group') { const m: WSGroupRecallMessage = { type: 'group_recall', group_id: String(payload.group_id ?? ''), conversation_id: payload.conversation_id, message_id: payload.message_id, }; this.emit('group_recall', m); } else { const m: WSRecallMessage = { type: 'recall', conversation_id: payload.conversation_id, message_id: payload.message_id, }; this.emit('recall', m); } } private handleGroupNotice(payload: any): void { const m: WSGroupNoticeMessage = { type: 'group_notice', notice_type: payload.notice_type, group_id: String(payload.group_id ?? ''), data: payload.data || {}, timestamp: payload.timestamp || Date.now(), message_id: payload.message_id, seq: payload.seq, }; this.emit('group_notice', m); } private handleSystemNotification(payload: any): void { const m: WSNotificationMessage = { type: 'notification', id: String(payload.id ?? ''), content: payload.content || '', sender_id: payload.sender_id, receiver_id: payload.receiver_id, category: payload.category, system_type: payload.system_type, extra_data: payload.extra_data || payload.extra || payload.data, is_read: !!payload.is_read, created_at: payload.created_at || new Date().toISOString(), }; this.emit('notification', m); if (!m.is_read) { systemNotificationService.handleWSMessage(m as any).catch(() => {}); } } // Call signaling handlers private handleCallIncoming(payload: any): void { console.log('[WSService] call_incoming payload:', JSON.stringify({ call_type: payload.call_type, media_type: payload.media_type, })); const m: WSCallIncomingMessage = { type: 'call_incoming', call_id: payload.call_id, conversation_id: payload.conversation_id, caller_id: payload.caller_id, call_type: payload.media_type || payload.call_type, created_at: payload.created_at, lifetime: payload.lifetime, }; this.emit('call_incoming', m); } private handleCallAccepted(payload: any): void { const m: WSCallAcceptedMessage = { type: 'call_accepted', call_id: payload.call_id, started_at: payload.started_at, }; this.emit('call_accepted', m); } private handleCallRejected(payload: any): void { const m: WSCallRejectedMessage = { type: 'call_rejected', call_id: payload.call_id, reason: payload.reason, }; this.emit('call_rejected', m); } private handleCallBusy(payload: any): void { const m: WSCallBusyMessage = { type: 'call_busy', call_id: payload.call_id, }; this.emit('call_busy', m); } private handleCallEnded(payload: any): void { const m: WSCallEndedMessage = { type: 'call_ended', call_id: payload.call_id, ended_by: payload.ended_by, reason: payload.reason, duration: payload.duration, ended_at: payload.ended_at, }; this.emit('call_ended', m); } private handleCallPeerMuted(payload: any): void { const m: WSCallPeerMutedMessage = { type: 'call_peer_muted', call_id: payload.call_id, user_id: payload.user_id, muted: payload.muted, }; this.emit('call_peer_muted', m); } private handleCallInvited(payload: any): void { const m: WSCallInvitedMessage = { type: 'call_invited', call_id: payload.call_id, conversation_id: payload.conversation_id, callee_id: payload.callee_id, }; this.emit('call_invited', m); } private handleCallAnsweredElsewhere(payload: any): void { const m: WSCallAnsweredElsewhereMessage = { type: 'call_answered_elsewhere', call_id: payload.call_id, reason: payload.reason, }; this.emit('call_answered_elsewhere', m); } // Call signaling send methods sendCallInvite(conversationId: string, calleeId: string, callType: 'voice' | 'video' = 'voice'): void { this.sendFireAndForget('call_invite', { conversation_id: conversationId, callee_id: calleeId, call_type: callType, }); } /** * 查询当前活跃的通话 * 用于 WebSocket 重连后恢复通话状态 */ async getActiveCall(): Promise { try { const { api } = await import('./api'); const response = await api.get('/calls/active'); return response.data || null; } catch (error) { console.log('[WSService] No active call found'); return null; } } sendCallAnswer(callId: string): void { this.sendFireAndForget('call_answer', { call_id: callId }); } sendCallReject(callId: string): void { this.sendFireAndForget('call_reject', { call_id: callId }); } sendCallBusy(callId: string): void { this.sendFireAndForget('call_busy', { call_id: callId }); } sendCallEnd(callId: string, reason?: string): void { this.sendFireAndForget('call_end', { call_id: callId, reason }); } sendCallMute(callId: string, muted: boolean): void { this.sendFireAndForget('call_mute', { call_id: callId, muted }); } sendCallReady(callId: string): void { this.sendFireAndForget('call_ready', { call_id: callId }); } sendCallGroupInvite(groupId: string, conversationId: string, callType: string): void { this.sendFireAndForget('call_group_invite', { group_id: groupId, conversation_id: conversationId, call_type: callType, }); } private handleMessageSent(payload: any): void { // 使用后端回带的 client_msg_id 精确匹配单条消息 // 避免同会话连续发送时,第一条 ACK 误清掉第二条的 pending 记录 const clientId = payload?.client_msg_id; let pendingMsg: PendingMessage | undefined; if (clientId) { pendingMsg = this.pendingMessages.find(p => p.id === clientId); } // 兜底:旧版后端不回带 client_msg_id 时,按 conversation_id + 类型匹配最旧一条 if (!pendingMsg) { pendingMsg = this.pendingMessages.find(p => p.type === 'chat' && p.payload.conversation_id === payload.conversation_id ); } if (pendingMsg) { pendingMsg.resolve(payload); this.removePendingMessage(pendingMsg.id); } } private handleMessageRecalled(payload: any): void { const clientId = payload?.client_msg_id; let pendingMsg: PendingMessage | undefined; if (clientId) { pendingMsg = this.pendingMessages.find(p => p.id === clientId); } if (!pendingMsg) { pendingMsg = this.pendingMessages.find(p => p.type === 'recall' && p.payload.message_id === payload.message_id ); } if (pendingMsg) { pendingMsg.resolve(payload); this.removePendingMessage(pendingMsg.id); } } // 发送消息(支持降级到HTTP) async sendMessage( conversationId: string, detailType: 'private' | 'group', segments: MessageSegment[], replyToId?: string ): Promise { const payload = { conversation_id: conversationId, detail_type: detailType, segments, reply_to_id: replyToId, }; // 如果WebSocket连接正常,优先使用WebSocket发送 if (this.isConnected() && !this.isFallbackMode) { return this.sendViaWebSocket('chat', payload); } // 降级到HTTP发送 return this.sendViaHTTP(conversationId, detailType, segments, replyToId); } // 标记已读 async markRead(conversationId: string, seq: number): Promise { const payload = { conversation_id: conversationId, seq, }; if (this.isConnected() && !this.isFallbackMode) { this.sendViaWebSocket('read', payload).catch(() => {}); } // 已读不需要等待响应,失败也不影响 } // 发送输入状态 sendTyping(conversationId: string, isTyping: boolean): void { const payload = { conversation_id: conversationId, is_typing: isTyping, }; if (this.isConnected() && !this.isFallbackMode) { this.sendViaWebSocket('typing', payload).catch(() => {}); } } // 撤回消息 async recallMessage(messageId: string): Promise { const payload = { message_id: messageId }; if (this.isConnected() && !this.isFallbackMode) { return this.sendViaWebSocket('recall', payload); } // 降级到HTTP const { api } = await import('./api'); return api.post('/messages/delete_msg', { message_id: messageId }); } // 通过WebSocket发送(fire-and-forget,不等待确认) private sendFireAndForget(type: string, payload: any): void { if (!this.ws || this.ws.readyState !== WebSocket.OPEN) { return; } const message = { type, payload }; this.ws.send(JSON.stringify(message)); } // 发送 ACK 确认(fire-and-forget) sendAck(messageId: string): void { this.sendFireAndForget('ack', { message_id: messageId }); } // 通过WebSocket发送 private sendViaWebSocket(type: string, payload: any): Promise { return new Promise((resolve, reject) => { if (!this.ws || this.ws.readyState !== WebSocket.OPEN) { reject(new Error('WebSocket not connected')); return; } const messageId = `msg_${++this.messageIdCounter}_${Date.now()}`; // 携带 client_msg_id:用于服务器回 ACK / message_sent 时精确匹配单条消息 const payloadWithId = { ...payload, client_msg_id: messageId }; const message = { type, payload: payloadWithId, }; // 添加到待处理队列 const pendingMsg: PendingMessage = { id: messageId, type, payload: payloadWithId, resolve, reject, timestamp: Date.now(), }; this.pendingMessages.push(pendingMsg); // 发送消息 this.ws.send(JSON.stringify(message)); // 设置超时 setTimeout(() => { const index = this.pendingMessages.findIndex(p => p.id === messageId); if (index >= 0) { this.pendingMessages[index].reject(new Error('Message timeout')); this.removePendingMessage(messageId); } }, 10000); }); } // 通过HTTP发送(降级方案) private async sendViaHTTP( conversationId: string, detailType: 'private' | 'group', segments: MessageSegment[], replyToId?: string ): Promise { const { api } = await import('./api'); const body: any = { detail_type: detailType, segments, }; if (replyToId) { body.reply_to_id = replyToId; } return api.post(`/conversations/${conversationId}/messages`, body); } // 刷新待发送消息队列 // 过期判定: // - 单条 sendViaWebSocket 自带 10s 超时,正常情况下 pending 消息不会超过这么久 // - 重连后链路刚恢复时,距离"刚入队"的消息一律视为有效(避免把刚刚挂起的消息误清) // - 兜底:超过 STALE_PENDING_MS(60s)一定视为过期丢弃 private static readonly STALE_PENDING_MS = 60000; private flushPendingMessages(): void { if (this.pendingMessages.length === 0) return; const now = Date.now(); const expiredMessages: PendingMessage[] = []; const validMessages: PendingMessage[] = []; // 分离过期和有效的消息 for (const msg of this.pendingMessages) { const ageMs = now - msg.timestamp; const isStale = ageMs > WebSocketService.STALE_PENDING_MS; if (isStale) { expiredMessages.push(msg); } else { validMessages.push(msg); } } // 拒绝过期消息 expiredMessages.forEach(msg => { msg.reject(new Error('Message expired')); }); // 重新发送有效消息:直接复用原 client_msg_id,否则重连后服务端 ACK 无法回带到原 pending 记录 // 注意:不能调 sendViaWebSocket(会重新生成 id 并入队),必须直接 ws.send this.pendingMessages = validMessages; for (let i = 0; i < validMessages.length; i++) { const msg = validMessages[i]; if (!this.ws || this.ws.readyState !== WebSocket.OPEN) { // 链路断开,剩余消息留在队列等下次重连 break; } try { this.ws.send(JSON.stringify({ type: msg.type, payload: msg.payload })); } catch { // ws.send 抛异常时 reject 并移出队列,避免无限残留 msg.reject(new Error('WebSocket send failed on flush')); this.removePendingMessage(msg.id); continue; } // 重置 10s 超时 setTimeout(() => { const index = this.pendingMessages.findIndex(p => p.id === msg.id); if (index >= 0) { this.pendingMessages[index].reject(new Error('Message timeout')); this.removePendingMessage(msg.id); } }, 10000); } } private removePendingMessage(id: string): void { const index = this.pendingMessages.findIndex(p => p.id === id); if (index >= 0) { this.pendingMessages.splice(index, 1); } } disconnect(): void { this.shouldRun = false; this.connected = false; this.isConnecting = false; this.isFallbackMode = false; this.reconnectAttempts = 0; // 重置重连计数 this.lastEventId = ''; // 重置事件ID this.lastActivityAt = 0; // 重置活动时间 if (this.ws) { this.ws.close(); this.ws = null; } this.stopHeartbeat(); this.stopReconnect(); this.stopFallbackCheck(); // 拒绝所有待处理消息 this.pendingMessages.forEach(msg => { msg.reject(new Error('WebSocket disconnected')); }); this.pendingMessages = []; } private scheduleReconnect(): void { if (this.reconnectAttempts >= this.maxReconnectAttempts) { // 超过最大重试次数,进入降级模式 this.enterFallbackMode(); return; } this.stopReconnect(); console.log(`[WebSocket] Reconnecting in ${this.reconnectDelay}ms (attempt ${this.reconnectAttempts + 1}/${this.maxReconnectAttempts})`); this.reconnectTimer = setTimeout(() => { this.reconnectAttempts += 1; this.connect(); }, this.reconnectDelay); } private stopReconnect() { if (this.reconnectTimer) { clearTimeout(this.reconnectTimer); this.reconnectTimer = null; } } isConnected(): boolean { return this.connected && this.ws?.readyState === WebSocket.OPEN; } isInFallbackMode(): boolean { return this.isFallbackMode; } // 进入降级模式 private enterFallbackMode(): void { if (this.isFallbackMode) return; this.isFallbackMode = true; console.log('WebSocket entering fallback mode, using HTTP API'); // 启动降级模式检查定时器,定期尝试重连WebSocket this.startFallbackCheck(); } // 启动降级模式检查 private startFallbackCheck(): void { this.stopFallbackCheck(); this.fallbackCheckTimer = setInterval(() => { if (!this.isConnected()) { this.reconnectAttempts = 0; this.connect(); } }, 30000); // 每30秒尝试重连 } private stopFallbackCheck(): void { if (this.fallbackCheckTimer) { clearInterval(this.fallbackCheckTimer); this.fallbackCheckTimer = null; } } private emit(type: T, message: Extract) { const handlers = this.messageHandlers.get(type) || []; handlers.forEach(h => h(message as WSMessage)); } on(type: T, handler: MessageHandler>): () => void { const list = this.messageHandlers.get(type) || []; list.push(handler as MessageHandler); this.messageHandlers.set(type, list); return () => { const current = this.messageHandlers.get(type) || []; const idx = current.indexOf(handler as MessageHandler); if (idx >= 0) current.splice(idx, 1); }; } onConnect(handler: ConnectionHandler): () => void { this.connectionHandlers.push(handler); return () => { const i = this.connectionHandlers.indexOf(handler); if (i >= 0) this.connectionHandlers.splice(i, 1); }; } onDisconnect(handler: ConnectionHandler): () => void { this.disconnectionHandlers.push(handler); return () => { const i = this.disconnectionHandlers.indexOf(handler); if (i >= 0) this.disconnectionHandlers.splice(i, 1); }; } private setupAppStateListener(): void { if (this.appStateSubscription) return; this.lastAppState = AppState.currentState; this.appStateSubscription = AppState.addEventListener('change', (nextState: AppStateStatus) => { // 进入后台时断开 WebSocket,由 JPush 负责后台推送通知 // 但如果正在通话中则不断开,避免对方请求权限等场景导致通话中断 if (this.lastAppState === 'active' && nextState.match(/inactive|background/)) { if (this.preventDisconnectOnBackground) { console.log('[WebSocket] App entering background but call is active, skipping disconnect'); } else { console.log('[WebSocket] App entering background, disconnecting'); this.disconnect(); } } // 回到前台时重连 WebSocket if (this.lastAppState.match(/inactive|background/) && nextState === 'active' && !this.isConnected()) { console.log('[WebSocket] App returning to foreground, reconnecting'); this.reconnectAttempts = 0; this.connect(); } this.lastAppState = nextState; }); } async start(): Promise { this.setupAppStateListener(); return this.connect(); } stop(): void { if (this.appStateSubscription) { this.appStateSubscription.remove(); this.appStateSubscription = null; } this.disconnect(); } private markActivity(): void { this.lastActivityAt = Date.now(); } private handleVerificationRequired(): void { showVerificationModal(); } private startHeartbeat(): void { this.stopHeartbeat(); this.heartbeatTimer = setInterval(() => { if (this.ws?.readyState === WebSocket.OPEN) { this.ws.send(JSON.stringify({ type: 'ping' })); } }, this.heartbeatInterval); } private stopHeartbeat(): void { if (this.heartbeatTimer) { clearInterval(this.heartbeatTimer); this.heartbeatTimer = null; } } private handleDisconnected(): void { // 防止重复调用 if (!this.connected && !this.ws && !this.isConnecting) { console.log('[WebSocket] Already disconnected, ignoring'); return; } console.log('[WebSocket] Handling disconnect...'); const wasActive = this.connected || this.ws != null || this.isConnecting; this.isConnecting = false; this.connected = false; if (this.ws) { this.ws.close(); this.ws = null; } this.stopHeartbeat(); if (wasActive) { this.disconnectionHandlers.forEach(h => h()); } if (this.shouldRun) { this.scheduleReconnect(); } else { console.log('[WebSocket] shouldRun is false, not reconnecting'); } } } export const wsService = new WebSocketService();