From a91637466cf863732bb0e0e7a481c8660843b8fa Mon Sep 17 00:00:00 2001 From: lan Date: Thu, 19 Mar 2026 00:56:41 +0800 Subject: [PATCH] refactor(messaging): replace WebSocket with SSE for real-time message handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace WebSocket-based real-time communication with Server-Sent Events (SSE) across the messaging infrastructure. This includes: - Create new SSEClient datasource to manage SSE connections - Create new SSEMessageHandler to process SSE events - Update ProcessMessageUseCase to use SSEClient instead of WebSocketClient - Update MessageManager and MessageStateManager to work with SSE handlers - Rename connection state variables from `isWebSocketConnected` to `isSSEConnected` - Update type definitions in dto.ts (WSEventType → SSEEventType, etc.) - Delete obsolete WebSocketClient and WebSocketMessageHandler files - Update comments and documentation to reflect SSE terminology This refactoring aligns with the backend's SSE implementation for better compatibility with React Native's networking capabilities. --- src/core/usecases/ProcessMessageUseCase.ts | 32 ++-- .../{WebSocketClient.ts => SSEClient.ts} | 36 ++--- src/infrastructure/sync/types.ts | 2 +- src/services/backgroundService.ts | 12 +- src/stores/message/MessageManager.ts | 42 ++--- src/stores/message/MessageStateManager.ts | 12 +- src/stores/message/SSEMessageHandler.ts | 148 ++++++++++++++++++ src/stores/message/WebSocketMessageHandler.ts | 136 ---------------- src/stores/message/index.ts | 4 +- src/stores/messageManager.ts | 38 ++--- src/types/dto.ts | 24 +-- 11 files changed, 249 insertions(+), 237 deletions(-) rename src/data/datasources/{WebSocketClient.ts => SSEClient.ts} (81%) create mode 100644 src/stores/message/SSEMessageHandler.ts delete mode 100644 src/stores/message/WebSocketMessageHandler.ts diff --git a/src/core/usecases/ProcessMessageUseCase.ts b/src/core/usecases/ProcessMessageUseCase.ts index 2deb33c..aaab9a0 100644 --- a/src/core/usecases/ProcessMessageUseCase.ts +++ b/src/core/usecases/ProcessMessageUseCase.ts @@ -1,11 +1,11 @@ /** * ProcessMessageUseCase - 处理消息用例 - * 处理消息的业务逻辑，协调 Repository 和 WebSocketClient + * 处理消息的业务逻辑，协调 Repository 和 SSEClient * 纯业务逻辑，无UI依赖 */ import { messageRepository } from '../../data/repositories/MessageRepository'; -import { webSocketClient, WebSocketEventType } from '../../data/datasources/WebSocketClient'; +import { sseClient, SSEEventType } from '../../data/datasources/SSEClient'; import { Message, Conversation, @@ -62,7 +62,7 @@ class ProcessMessageUseCase { */ initialize(currentUserId: string | null): void { this.currentUserId = currentUserId; - this.setupWebSocketListeners(); + this.setupSSEListeners(); } /** @@ -79,51 +79,51 @@ class ProcessMessageUseCase { } /** - * 设置WebSocket监听器 + * 设置SSE监听器 */ - private setupWebSocketListeners(): void { + private setupSSEListeners(): void { // 监听私聊消息 - const unsubChat = webSocketClient.on('chat', (message) => { + const unsubChat = sseClient.on('chat', (message) => { this.handleNewMessage(message); }); // 监听群聊消息 - const unsubGroupMessage = webSocketClient.on('group_message', (message) => { + const unsubGroupMessage = sseClient.on('group_message', (message) => { this.handleNewMessage(message); }); // 监听私聊已读回执 - const unsubRead = webSocketClient.on('read', (message) => { + const unsubRead = sseClient.on('read', (message) => { this.handleReadReceipt(message); }); // 监听群聊已读回执 - const unsubGroupRead = webSocketClient.on('group_read', (message) => { + const unsubGroupRead = sseClient.on('group_read', (message) => { this.handleGroupReadReceipt(message); }); // 监听私聊消息撤回 - const unsubRecall = webSocketClient.on('recall', (message) => { + const unsubRecall = sseClient.on('recall', (message) => { this.handleRecallMessage(message); }); // 监听群聊消息撤回 - const unsubGroupRecall = webSocketClient.on('group_recall', (message) => { + const unsubGroupRecall = sseClient.on('group_recall', (message) => { this.handleGroupRecallMessage(message); }); // 监听群聊输入状态 - const unsubGroupTyping = webSocketClient.on('group_typing', (message) => { + const unsubGroupTyping = sseClient.on('group_typing', (message) => { this.handleGroupTyping(message); }); // 监听群通知 - const unsubGroupNotice = webSocketClient.on('group_notice', (message) => { + const unsubGroupNotice = sseClient.on('group_notice', (message) => { this.handleGroupNotice(message); }); // 监听连接状态 - const unsubConnected = webSocketClient.on('connected', () => { + const unsubConnected = sseClient.on('connected', () => { this.notifySubscribers({ type: 'connection_changed', payload: { connected: true }, @@ -131,7 +131,7 @@ class ProcessMessageUseCase { }); }); - const unsubDisconnected = webSocketClient.on('disconnected', () => { + const unsubDisconnected = sseClient.on('disconnected', () => { this.notifySubscribers({ type: 'connection_changed', payload: { connected: false }, @@ -397,7 +397,7 @@ class ProcessMessageUseCase { groupId: groupIdStr, data: data || {}, timestamp: timestamp || Date.now(), - messageId, + messageId: message_id, seq, }; diff --git a/src/data/datasources/WebSocketClient.ts b/src/data/datasources/SSEClient.ts similarity index 81% rename from src/data/datasources/WebSocketClient.ts rename to src/data/datasources/SSEClient.ts index 8c7398e..ae955ac 100644 --- a/src/data/datasources/WebSocketClient.ts +++ b/src/data/datasources/SSEClient.ts @@ -1,6 +1,6 @@ /** - * WebSocketClient - WebSocket连接管理 - * 只负责WebSocket连接管理，提供事件驱动架构 + * SSEClient - SSE连接管理 + * 只负责SSE连接管理，提供事件驱动架构 */ import { @@ -19,8 +19,8 @@ import { // 事件处理器类型 type EventHandler = (data: T) => void; -// WebSocket事件类型 -export interface WebSocketEvents { +// SSE事件类型 +export interface SSEEvents { 'chat': WSChatMessage; 'group_message': WSGroupChatMessage; 'read': WSReadMessage; @@ -34,15 +34,15 @@ export interface WebSocketEvents { 'error': Error; } -export type WebSocketEventType = keyof WebSocketEvents; +export type SSEEventType = keyof SSEEvents; -class WebSocketClient { - private handlers: Map>> = new Map(); +class SSEClient { + private handlers: Map>> = new Map(); private unsubscribeFns: Array<() => void> = []; private isInitialized = false; /** - * 初始化WebSocket监听 + * 初始化SSE监听 */ initialize(): void { if (this.isInitialized) return; @@ -113,7 +113,7 @@ class WebSocketClient { } /** - * 销毁WebSocket监听 + * 销毁SSE监听 */ destroy(): void { this.unsubscribeFns.forEach((fn) => fn()); @@ -125,9 +125,9 @@ class WebSocketClient { /** * 订阅事件 */ - on( + on( event: T, - handler: EventHandler + handler: EventHandler ): () => void { if (!this.handlers.has(event)) { this.handlers.set(event, new Set()); @@ -142,9 +142,9 @@ class WebSocketClient { /** * 触发事件 */ - private emit( + private emit( event: T, - data: WebSocketEvents[T] + data: SSEEvents[T] ): void { const handlers = this.handlers.get(event); if (handlers) { @@ -152,7 +152,7 @@ class WebSocketClient { try { handler(data); } catch (error) { - console.error(`[WebSocketClient] 事件处理器执行失败: ${event}`, error); + console.error(`[SSEClient] 事件处理器执行失败: ${event}`, error); } }); } @@ -166,19 +166,19 @@ class WebSocketClient { } /** - * 启动WebSocket连接 + * 启动SSE连接 */ async connect(): Promise { return sseService.start(); } /** - * 断开WebSocket连接 + * 断开SSE连接 */ disconnect(): void { sseService.stop(); } } -export const webSocketClient = new WebSocketClient(); -export default webSocketClient; +export const sseClient = new SSEClient(); +export default sseClient; diff --git a/src/infrastructure/sync/types.ts b/src/infrastructure/sync/types.ts index 62d8e8c..bffe10b 100644 --- a/src/infrastructure/sync/types.ts +++ b/src/infrastructure/sync/types.ts @@ -2,7 +2,7 @@ * 同步状态管理类型定义 * * 提供连接状态相关的类型、接口和枚举 - * 用于统一管理 WebSocket/SSE 连接状态 + * 用于统一管理 SSE 连接状态 */ /** diff --git a/src/services/backgroundService.ts b/src/services/backgroundService.ts index 3db82c5..efebf06 100644 --- a/src/services/backgroundService.ts +++ b/src/services/backgroundService.ts @@ -4,7 +4,7 @@ * * 功能： * - 定期后台任务保持 App 在内存中 - * - 配合 WebSocket 服务保持长连接 + * - 配合 SSE 服务保持长连接 * - 收到消息时触发震动提示 */ @@ -47,7 +47,7 @@ let appStateSubscription: any = null; // 定义后台任务 TaskManager.defineTask(BACKGROUND_FETCH_TASK, async () => { try { - // 检查 WebSocket 连接状态 + // 检查 SSE 连接状态 if (!sseService.isConnected()) { await sseService.connect(); } @@ -60,7 +60,7 @@ TaskManager.defineTask(BACKGROUND_FETCH_TASK, async () => { } }); -// WebSocket 保活任务 +// SSE 保活任务 TaskManager.defineTask(REALTIME_KEEPALIVE_TASK, async () => { try { if (!sseService.isConnected()) { @@ -174,9 +174,9 @@ async function registerBackgroundTasks(): Promise { }); } - // 注册 WebSocket 保活任务 - const isWsKeepaliveRegistered = await TaskManager.isTaskRegisteredAsync(REALTIME_KEEPALIVE_TASK); - if (!isWsKeepaliveRegistered) { + // 注册 SSE 保活任务 + const isSseKeepaliveRegistered = await TaskManager.isTaskRegisteredAsync(REALTIME_KEEPALIVE_TASK); + if (!isSseKeepaliveRegistered) { await BackgroundFetch.registerTaskAsync(REALTIME_KEEPALIVE_TASK, { minimumInterval: 60, // 1 分钟检查一次 stopOnTerminate: false, diff --git a/src/stores/message/MessageManager.ts b/src/stores/message/MessageManager.ts index e89ba73..62aad2c 100644 --- a/src/stores/message/MessageManager.ts +++ b/src/stores/message/MessageManager.ts @@ -3,40 +3,40 @@ * 作为协调者，整合各个专注的服务模块 */ -import { messageStateManager, MessageStateManager } from './message/MessageStateManager'; -import { webSocketMessageHandler, WebSocketMessageHandler } from './message/WebSocketMessageHandler'; -import { messageSyncService, MessageSyncService } from './message/MessageSyncService'; -import { readReceiptManager, ReadReceiptManager } from './message/ReadReceiptManager'; -import { messageRepository } from '../data/repositories/MessageRepository'; -import { useAuthStore } from './authStore'; -import type { Message, Conversation } from '../core/entities/Message'; -import type { ConversationResponse, MessageResponse } from '../types/dto'; +import { messageStateManager, MessageStateManager } from './MessageStateManager'; +import { sseMessageHandler, SSEMessageHandler } from './SSEMessageHandler'; +import { messageSyncService, MessageSyncService } from './MessageSyncService'; +import { readReceiptManager, ReadReceiptManager } from './ReadReceiptManager'; +import { messageRepository } from '../../data/repositories/MessageRepository'; +import { useAuthStore } from '../authStore'; +import type { Message, Conversation } from '../../core/entities/Message'; +import type { ConversationResponse, MessageResponse } from '../../types/dto'; export type { MessageEventType, MessageEvent, MessageSubscriber, -} from './message/MessageStateManager'; +} from './MessageStateManager'; export class MessageManager { private stateManager: MessageStateManager; - private wsHandler: WebSocketMessageHandler; + private sseHandler: SSEMessageHandler; private syncService: MessageSyncService; private readManager: ReadReceiptManager; private initialized: boolean = false; private authUnsubscribe: (() => void) | null = null; - private wsUnsubscribe: (() => void) | null = null; + private sseUnsubscribe: (() => void) | null = null; private currentUserId: string | null = null; constructor() { this.stateManager = messageStateManager; - this.wsHandler = webSocketMessageHandler; + this.sseHandler = sseMessageHandler; this.syncService = messageSyncService; this.readManager = readReceiptManager; this.readManager.setStateManager(this.stateManager); - this.setupWebSocketHandlers(); + this.setupSSEHandlers(); this.initAuthListener(); } @@ -58,8 +58,8 @@ export class MessageManager { }); } - private setupWebSocketHandlers(): void { - this.wsUnsubscribe = this.wsHandler.subscribe((event) => { + private setupSSEHandlers(): void { + this.sseUnsubscribe = this.sseHandler.subscribe((event) => { switch (event.type) { case 'chat_message': case 'group_message': @@ -89,7 +89,7 @@ export class MessageManager { try { console.log('[MessageManager] Initializing...'); - this.wsHandler.connect(); + this.sseHandler.connect(); const conversations = await this.syncService.syncConversations(); this.stateManager.setConversations(conversations); @@ -224,14 +224,14 @@ export class MessageManager { } isConnected(): boolean { - return this.wsHandler.isConnected(); + return this.sseHandler.isConnected(); } reset(): void { this.initialized = false; this.stateManager.reset(); this.readManager.reset(); - this.wsHandler.disconnect(); + this.sseHandler.disconnect(); } destroy(): void { @@ -239,9 +239,9 @@ export class MessageManager { this.authUnsubscribe(); this.authUnsubscribe = null; } - if (this.wsUnsubscribe) { - this.wsUnsubscribe(); - this.wsUnsubscribe = null; + if (this.sseUnsubscribe) { + this.sseUnsubscribe(); + this.sseUnsubscribe = null; } this.reset(); } diff --git a/src/stores/message/MessageStateManager.ts b/src/stores/message/MessageStateManager.ts index 80b2529..df7e160 100644 --- a/src/stores/message/MessageStateManager.ts +++ b/src/stores/message/MessageStateManager.ts @@ -31,7 +31,7 @@ export interface MessageState { conversationList: Conversation[]; messagesMap: Map; unreadCount: UnreadCount; - isWebSocketConnected: boolean; + isSSEConnected: boolean; currentConversationId: string | null; isLoading: boolean; typingUsersMap: Map; @@ -50,7 +50,7 @@ export class MessageStateManager { conversationList: [], messagesMap: new Map(), unreadCount: { total: 0, system: 0 }, - isWebSocketConnected: false, + isSSEConnected: false, currentConversationId: null, isLoading: false, typingUsersMap: new Map(), @@ -86,7 +86,7 @@ export class MessageStateManager { } isConnected(): boolean { - return this.state.isWebSocketConnected; + return this.state.isSSEConnected; } getTypingUsers(groupId: string): string[] { @@ -188,8 +188,8 @@ export class MessageStateManager { this.notify({ type: 'unread_count_updated', payload: count }); } - setWebSocketConnected(connected: boolean): void { - this.state.isWebSocketConnected = connected; + setSSEConnected(connected: boolean): void { + this.state.isSSEConnected = connected; this.notify({ type: 'connection_changed', payload: connected }); } @@ -287,7 +287,7 @@ export class MessageStateManager { conversationList: [], messagesMap: new Map(), unreadCount: { total: 0, system: 0 }, - isWebSocketConnected: false, + isSSEConnected: false, currentConversationId: null, isLoading: false, typingUsersMap: new Map(), diff --git a/src/stores/message/SSEMessageHandler.ts b/src/stores/message/SSEMessageHandler.ts new file mode 100644 index 0000000..afbad45 --- /dev/null +++ b/src/stores/message/SSEMessageHandler.ts @@ -0,0 +1,148 @@ +/** + * SSE 消息处理器 + * 只负责处理 SSE 消息，不管理状态 + */ + +import { sseService, WSChatMessage, WSGroupChatMessage, WSReadMessage, WSGroupReadMessage, WSRecallMessage, WSGroupRecallMessage, WSGroupTypingMessage, WSGroupNoticeMessage, GroupNoticeType, WSMessage } from '../../services/sseService'; +import type { Message, GroupNotice } from '../../core/entities/Message'; + +export type SSEEventType = + | 'chat_message' + | 'group_message' + | 'read_receipt' + | 'group_read_receipt' + | 'message_recalled' + | 'group_message_recalled' + | 'typing' + | 'group_notice'; + +export interface SSEEvent { + type: SSEEventType; + payload: any; + raw: any; +} + +export type SSEEventHandler = (event: SSEEvent) => void; + +export class SSEMessageHandler { + private unsubscribeFns: Array<() => void> = []; + private handlers: Set = new Set(); + + connect(): void { + if (this.unsubscribeFns.length > 0) return; + + // 监听私聊消息 + const unsubChat = sseService.on('chat', (message) => { + this.emit('chat_message', this.parseChatMessage(message), message); + }); + + // 监听群聊消息 + const unsubGroupMessage = sseService.on('group_message', (message) => { + this.emit('group_message', this.parseGroupMessage(message), message); + }); + + // 监听私聊已读回执 + const unsubRead = sseService.on('read', (message) => { + this.emit('read_receipt', message, message); + }); + + // 监听群聊已读回执 + const unsubGroupRead = sseService.on('group_read', (message) => { + this.emit('group_read_receipt', message, message); + }); + + // 监听私聊消息撤回 + const unsubRecall = sseService.on('recall', (message) => { + this.emit('message_recalled', message, message); + }); + + // 监听群聊消息撤回 + const unsubGroupRecall = sseService.on('group_recall', (message) => { + this.emit('group_message_recalled', message, message); + }); + + // 监听群聊输入状态 + const unsubGroupTyping = sseService.on('group_typing', (message) => { + this.emit('typing', message, message); + }); + + // 监听群通知 + const unsubGroupNotice = sseService.on('group_notice', (message) => { + this.emit('group_notice', this.parseGroupNotice(message), message); + }); + + this.unsubscribeFns = [ + unsubChat, + unsubGroupMessage, + unsubRead, + unsubGroupRead, + unsubRecall, + unsubGroupRecall, + unsubGroupTyping, + unsubGroupNotice, + ]; + } + + disconnect(): void { + this.unsubscribeFns.forEach((fn) => fn()); + this.unsubscribeFns = []; + this.handlers.clear(); + } + + subscribe(handler: SSEEventHandler): () => void { + this.handlers.add(handler); + return () => this.handlers.delete(handler); + } + + private emit(type: SSEEventType, payload: any, raw: any): void { + const event: SSEEvent = { type, payload, raw }; + this.handlers.forEach(handler => { + try { + handler(event); + } catch (error) { + console.error('[SSEMessageHandler] Handler error:', error); + } + }); + } + + private parseChatMessage(data: WSChatMessage): Message { + return { + id: data.id || '', + conversationId: data.conversation_id || '', + senderId: data.sender_id || '', + seq: data.seq || 0, + segments: data.segments || [], + createdAt: data.created_at || new Date().toISOString(), + status: 'normal', + }; + } + + private parseGroupMessage(data: WSGroupChatMessage): Message { + return { + id: data.id || '', + conversationId: data.conversation_id || '', + senderId: data.sender_id || '', + seq: data.seq || 0, + segments: data.segments || [], + createdAt: data.created_at || new Date().toISOString(), + status: 'normal', + }; + } + + private parseGroupNotice(data: WSGroupNoticeMessage): GroupNotice { + return { + type: data.notice_type || 'member_join', + groupId: String(data.group_id || ''), + data: data.data || {}, + timestamp: data.timestamp || Date.now(), + messageId: data.message_id, + seq: data.seq, + }; + } + + isConnected(): boolean { + return sseService.isConnected(); + } +} + +export const sseMessageHandler = new SSEMessageHandler(); \ No newline at end of file diff --git a/src/stores/message/WebSocketMessageHandler.ts b/src/stores/message/WebSocketMessageHandler.ts deleted file mode 100644 index d9ee386..0000000 --- a/src/stores/message/WebSocketMessageHandler.ts +++ /dev/null @@ -1,136 +0,0 @@ -/** - * WebSocket 消息处理器 - * 只负责处理 WebSocket 消息，不管理状态 - */ - -import { sseService, WSChatMessage, WSGroupChatMessage, WSReadMessage, WSGroupReadMessage, WSRecallMessage, WSGroupRecallMessage, WSGroupTypingMessage, WSGroupNoticeMessage, GroupNoticeType } from '../../services/sseService'; -import type { Message, GroupNotice } from '../../core/entities/Message'; - -export type WebSocketEventType = - | 'chat_message' - | 'group_message' - | 'read_receipt' - | 'group_read_receipt' - | 'message_recalled' - | 'group_message_recalled' - | 'typing' - | 'group_notice'; - -export interface WebSocketEvent { - type: WebSocketEventType; - payload: any; - raw: any; -} - -export type WebSocketEventHandler = (event: WebSocketEvent) => void; - -export class WebSocketMessageHandler { - private unsubscribe: (() => void) | null = null; - private handlers: Set = new Set(); - - connect(): void { - if (this.unsubscribe) return; - - this.unsubscribe = sseService.subscribe(this.handleSSEEvent.bind(this)); - } - - disconnect(): void { - if (this.unsubscribe) { - this.unsubscribe(); - this.unsubscribe = null; - } - this.handlers.clear(); - } - - subscribe(handler: WebSocketEventHandler): () => void { - this.handlers.add(handler); - return () => this.handlers.delete(handler); - } - - private handleSSEEvent(event: any): void { - const { type, data } = event; - - switch (type) { - case 'chat': - this.emit('chat_message', this.parseChatMessage(data), data); - break; - case 'group_message': - this.emit('group_message', this.parseGroupMessage(data), data); - break; - case 'read': - this.emit('read_receipt', data, data); - break; - case 'group_read': - this.emit('group_read_receipt', data, data); - break; - case 'recall': - this.emit('message_recalled', data, data); - break; - case 'group_recall': - this.emit('group_message_recalled', data, data); - break; - case 'typing': - this.emit('typing', data, data); - break; - case 'group_notice': - this.emit('group_notice', this.parseGroupNotice(data), data); - break; - default: - console.log('[WebSocketMessageHandler] Unknown event type:', type); - } - } - - private emit(type: WebSocketEventType, payload: any, raw: any): void { - const event: WebSocketEvent = { type, payload, raw }; - this.handlers.forEach(handler => { - try { - handler(event); - } catch (error) { - console.error('[WebSocketMessageHandler] Handler error:', error); - } - }); - } - - private parseChatMessage(data: WSChatMessage): Message { - return { - id: data.id || '', - conversationId: data.conversationId || data.conversation_id || '', - senderId: data.senderId || data.sender_id || '', - seq: data.seq || 0, - segments: data.segments || [], - createdAt: data.createdAt || data.created_at || new Date().toISOString(), - status: 'normal', - sender: data.sender, - }; - } - - private parseGroupMessage(data: WSGroupChatMessage): Message { - return { - id: data.id || '', - conversationId: data.groupId || data.group_id || '', - senderId: data.senderId || data.sender_id || '', - seq: data.seq || 0, - segments: data.segments || [], - createdAt: data.createdAt || data.created_at || new Date().toISOString(), - status: 'normal', - sender: data.sender, - }; - } - - private parseGroupNotice(data: WSGroupNoticeMessage): GroupNotice { - return { - type: data.noticeType || data.notice_type || 'member_join', - groupId: data.groupId || data.group_id || '', - data: data.data || {}, - timestamp: data.timestamp || Date.now(), - messageId: data.messageId || data.message_id, - seq: data.seq, - }; - } - - isConnected(): boolean { - return sseService.isConnected(); - } -} - -export const webSocketMessageHandler = new WebSocketMessageHandler(); \ No newline at end of file diff --git a/src/stores/message/index.ts b/src/stores/message/index.ts index c13fbcf..0961ba2 100644 --- a/src/stores/message/index.ts +++ b/src/stores/message/index.ts @@ -12,8 +12,8 @@ export type { MessageState, MessageEvent, MessageSubscriber, MessageEventType } // 导出同步服务 export { messageSyncService, MessageSyncService } from './MessageSyncService'; -// 导出WebSocket处理器 -export { webSocketMessageHandler, WebSocketMessageHandler } from './WebSocketMessageHandler'; +// 导出SSE处理器 +export { sseMessageHandler, SSEMessageHandler } from './SSEMessageHandler'; // 导出已读管理器 export { readReceiptManager, ReadReceiptManager } from './ReadReceiptManager'; \ No newline at end of file diff --git a/src/stores/messageManager.ts b/src/stores/messageManager.ts index d3ebe39..5db8812 100644 --- a/src/stores/messageManager.ts +++ b/src/stores/messageManager.ts @@ -8,7 +8,7 @@ * * 架构特点： * - 管理conversations和messages状态 - * - 统一处理WebSocket消息 + * - 统一处理SSE消息 * - 统一处理本地数据库读写 * - 提供订阅机制供React组件使用 */ @@ -83,7 +83,7 @@ interface MessageManagerState { systemUnreadCount: number; // 连接状态 - isWebSocketConnected: boolean; + isSSEConnected: boolean; // 当前活动会话ID（用户正在查看的会话） currentConversationId: string | null; @@ -131,7 +131,7 @@ interface ReadStateRecord { class MessageManager { private state: MessageManagerState; - private wsUnsubscribe: (() => void) | null = null; + private sseUnsubscribe: (() => void) | null = null; private currentUserId: string | null = null; private authUnsubscribe: (() => void) | null = null; private lastReconnectSyncAt: number = 0; @@ -158,7 +158,7 @@ class MessageManager { messagesMap: new Map(), totalUnreadCount: 0, systemUnreadCount: 0, - isWebSocketConnected: false, + isSSEConnected: false, currentConversationId: null, isLoadingConversations: false, loadingMessagesSet: new Set(), @@ -281,8 +281,8 @@ class MessageManager { return; } - // 2. 初始化WebSocket监听 - this.initWebSocketListeners(); + // 2. 初始化SSE监听 + this.initSSEListeners(); // 3. 加载会话列表 await this.fetchConversations(); @@ -320,9 +320,9 @@ class MessageManager { } destroy(): void { - if (this.wsUnsubscribe) { - this.wsUnsubscribe(); - this.wsUnsubscribe = null; + if (this.sseUnsubscribe) { + this.sseUnsubscribe(); + this.sseUnsubscribe = null; } this.state.subscribers.clear(); this.state.isInitialized = false; @@ -330,11 +330,11 @@ class MessageManager { this.activatingConversationTasks.clear(); } - // ==================== WebSocket 处理 ==================== + // ==================== SSE 处理 ==================== - private initWebSocketListeners(): void { - if (this.wsUnsubscribe) { - this.wsUnsubscribe(); + private initSSEListeners(): void { + if (this.sseUnsubscribe) { + this.sseUnsubscribe(); } @@ -380,7 +380,7 @@ class MessageManager { // 监听连接状态 sseService.onConnect(() => { - this.state.isWebSocketConnected = true; + this.state.isSSEConnected = true; this.notifySubscribers({ type: 'connection_changed', payload: { connected: true }, @@ -403,7 +403,7 @@ class MessageManager { }); sseService.onDisconnect(() => { - this.state.isWebSocketConnected = false; + this.state.isSSEConnected = false; this.notifySubscribers({ type: 'connection_changed', payload: { connected: false }, @@ -585,7 +585,7 @@ class MessageManager { } /** - * 处理新消息（WebSocket推送） + * 处理新消息（SSE推送） * 这是核心方法，必须立即处理并同步到所有订阅者 */ private async handleNewMessage(message: (WSChatMessage | WSGroupChatMessage) & { _isAck?: boolean }): Promise { @@ -2014,10 +2014,10 @@ class MessageManager { } /** - * 获取WebSocket连接状态 + * 获取SSE连接状态 */ isConnected(): boolean { - return this.state.isWebSocketConnected; + return this.state.isSSEConnected; } /** @@ -2122,7 +2122,7 @@ class MessageManager { subscriber({ type: 'connection_changed', - payload: { connected: this.state.isWebSocketConnected }, + payload: { connected: this.state.isSSEConnected }, timestamp: Date.now(), }); diff --git a/src/types/dto.ts b/src/types/dto.ts index 67a2cce..8394b1d 100644 --- a/src/types/dto.ts +++ b/src/types/dto.ts @@ -672,28 +672,28 @@ export interface GroupAnnouncementListResponse { total_pages: number; } -// ==================== WebSocket Event Types ==================== +// ==================== SSE Event Types ==================== -// WebSocket 事件类型 -export type WSEventType = 'message' | 'notice' | 'request' | 'meta'; +// SSE 事件类型 +export type SSEEventType = 'message' | 'notice' | 'request' | 'meta'; -// WebSocket 消息详细类型 -export type WSDetailType = 'private' | 'group' | 'follow' | 'like' | 'comment' | 'request' | 'typing' | 'heartbeat' | 'read'; +// SSE 消息详细类型 +export type SSEDetailType = 'private' | 'group' | 'follow' | 'like' | 'comment' | 'request' | 'typing' | 'heartbeat' | 'read'; -// WebSocket 事件消息段（与后端 message 字段一致） -export interface WSSegment { +// SSE 事件消息段（与后端 message 字段一致） +export interface SSESegment { type: string; data: Record; } -// WebSocket 事件（后端推送的事件格式） -export interface WSEvent { +// SSE 事件（后端推送的事件格式） +export interface SSEEvent { id: string; // 事件唯一ID time: number; // 事件时间戳（毫秒） - type: WSEventType; // 事件类型: message(消息), notice(通知), request(请求), meta(元事件) - detail_type: WSDetailType; // 详细类型: private(私聊), group(群聊), follow(关注), like(点赞)等 + type: SSEEventType; // 事件类型: message(消息), notice(通知), request(请求), meta(元事件) + detail_type: SSEDetailType; // 详细类型: private(私聊), group(群聊), follow(关注), like(点赞)等 seq: string; // 消息序号 - message?: WSSegment[]; // 消息内容数组 + message?: SSESegment[]; // 消息内容数组 conversation_id: string; // 会话ID }