refactor(messaging): replace WebSocket with SSE for real-time message handling
Some checks failed
Frontend CI / build-android-apk (push) Has been cancelled
Frontend CI / build-and-push-web (push) Has been cancelled
Frontend CI / ota-android (push) Has been cancelled
Frontend CI / ota-android (pull_request) Has been skipped
Frontend CI / build-and-push-web (pull_request) Has been cancelled
Frontend CI / build-android-apk (pull_request) Has been cancelled
Some checks failed
Frontend CI / build-android-apk (push) Has been cancelled
Frontend CI / build-and-push-web (push) Has been cancelled
Frontend CI / ota-android (push) Has been cancelled
Frontend CI / ota-android (pull_request) Has been skipped
Frontend CI / build-and-push-web (pull_request) Has been cancelled
Frontend CI / build-android-apk (pull_request) Has been cancelled
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.
This commit is contained in:
@@ -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,
|
||||
};
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
/**
|
||||
* WebSocketClient - WebSocket连接管理
|
||||
* 只负责WebSocket连接管理,提供事件驱动架构
|
||||
* SSEClient - SSE连接管理
|
||||
* 只负责SSE连接管理,提供事件驱动架构
|
||||
*/
|
||||
|
||||
import {
|
||||
@@ -19,8 +19,8 @@ import {
|
||||
// 事件处理器类型
|
||||
type EventHandler<T> = (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<WebSocketEventType, Set<EventHandler<any>>> = new Map();
|
||||
class SSEClient {
|
||||
private handlers: Map<SSEEventType, Set<EventHandler<any>>> = 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<T extends WebSocketEventType>(
|
||||
on<T extends SSEEventType>(
|
||||
event: T,
|
||||
handler: EventHandler<WebSocketEvents[T]>
|
||||
handler: EventHandler<SSEEvents[T]>
|
||||
): () => void {
|
||||
if (!this.handlers.has(event)) {
|
||||
this.handlers.set(event, new Set());
|
||||
@@ -142,9 +142,9 @@ class WebSocketClient {
|
||||
/**
|
||||
* 触发事件
|
||||
*/
|
||||
private emit<T extends WebSocketEventType>(
|
||||
private emit<T extends SSEEventType>(
|
||||
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<boolean> {
|
||||
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;
|
||||
@@ -2,7 +2,7 @@
|
||||
* 同步状态管理类型定义
|
||||
*
|
||||
* 提供连接状态相关的类型、接口和枚举
|
||||
* 用于统一管理 WebSocket/SSE 连接状态
|
||||
* 用于统一管理 SSE 连接状态
|
||||
*/
|
||||
|
||||
/**
|
||||
|
||||
@@ -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<void> {
|
||||
});
|
||||
}
|
||||
|
||||
// 注册 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,
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
@@ -31,7 +31,7 @@ export interface MessageState {
|
||||
conversationList: Conversation[];
|
||||
messagesMap: Map<string, Message[]>;
|
||||
unreadCount: UnreadCount;
|
||||
isWebSocketConnected: boolean;
|
||||
isSSEConnected: boolean;
|
||||
currentConversationId: string | null;
|
||||
isLoading: boolean;
|
||||
typingUsersMap: Map<string, string[]>;
|
||||
@@ -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(),
|
||||
|
||||
148
src/stores/message/SSEMessageHandler.ts
Normal file
148
src/stores/message/SSEMessageHandler.ts
Normal file
@@ -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<SSEEventHandler> = 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();
|
||||
@@ -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<WebSocketEventHandler> = 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();
|
||||
@@ -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';
|
||||
@@ -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<void> {
|
||||
@@ -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(),
|
||||
});
|
||||
|
||||
|
||||
@@ -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<string, any>;
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user