xudan/web-map
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 优化 WebSocket 连接逻辑，使用增强的 WebSocket 服务，支持心跳检测和重连机制，提升稳定性和性能

Browse Source
This commit is contained in:
xudan 2025-09-29 15:48:41 +08:00
parent c921da0c24
commit f4e6424104
2 changed files with 65 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

95
src/hooks/usePlaybackWebSocket.ts
View File

@ -1,9 +1,10 @@
import { message } from 'ant-design-vue';
import { type Ref, ref, type ShallowRef,shallowRef } from 'vue';
import { type Ref, ref, type ShallowRef, shallowRef } from 'vue';
import type { RobotRealtimeInfo } from '../apis/robot';
import type { EditorService } from '../services/editor.service';
import ws from '../services/ws';
// Define the structure of WebSocket messages for playback
type PlaybackMessage = {
@ -48,54 +49,59 @@ export function usePlaybackWebSocket(editorService: ShallowRef<EditorService | u
const latestRobotData = new Map<string, RobotRealtimeInfo>();
let animationFrameId: number;
const connect = (historySceneId: string) => {
const connect = async (historySceneId: string) => {
if (client.value) {
disconnect();
}
// TODO: Replace with the actual WebSocket host from environment variables or config.
const wsHost = import.meta.env.VITE_WEBSOCKET_URL || `ws://${window.location.host}`;
const wsUrl = `${wsHost}/history/scene/logplayback/${historySceneId}`;
const wsPath = `/scene/monitor/logplayback/${historySceneId}`;
const ws = new WebSocket(wsUrl);
client.value = ws;
try {
const wsInstance = await ws.create(wsPath, {
heartbeatInterval: 3600000, // 1 hour
});
client.value = wsInstance;
ws.onopen = () => {
isConnected.value = true;
message.success('回放连接已建立');
startRenderLoop();
};
wsInstance.onopen = () => {
isConnected.value = true;
message.success('回放连接已建立');
startRenderLoop();
};
ws.onmessage = (event) => {
const msg = JSON.parse(event.data) as PlaybackMessage;
wsInstance.onmessage = (event) => {
const msg = JSON.parse(event.data) as PlaybackMessage;
if (msg.timestamp) {
currentTime.value = msg.timestamp;
}
if (msg.timestamp) {
currentTime.value = msg.timestamp;
}
if (msg.type === 'SCENE') {
// As per user feedback, data is the full scene JSON string
sceneJson.value = msg.data;
latestRobotData.clear(); // Clear robot data when scene changes
} else if (msg.type === 'AMR') {
(msg.data as RobotRealtimeInfo[]).forEach(robot => {
latestRobotData.set(robot.id, robot);
});
}
};
if (msg.type === 'SCENE') {
sceneJson.value = msg.data;
latestRobotData.clear();
} else if (msg.type === 'AMR') {
(msg.data as RobotRealtimeInfo[]).forEach(robot => {
latestRobotData.set(robot.id, robot);
});
}
};
ws.onclose = () => {
isConnected.value = false;
isPlaying.value = false;
stopRenderLoop();
message.info('回放连接已断开');
};
wsInstance.onclose = () => {
isConnected.value = false;
isPlaying.value = false;
stopRenderLoop();
message.info('回放连接已断开');
client.value = null; // Clean up the client ref
};
ws.onerror = (error) => {
console.error('回放 WebSocket 发生错误:', error);
message.error('回放连接发生错误');
disconnect();
};
wsInstance.onerror = (error) => {
console.error('回放 WebSocket 发生错误:', error);
message.error('回放连接发生错误');
disconnect();
};
} catch (error) {
console.error('创建 WebSocket 连接失败:', error);
message.error('创建回放连接失败');
}
};
const disconnect = () => {
@ -119,19 +125,15 @@ export function usePlaybackWebSocket(editorService: ShallowRef<EditorService | u
};
const pause = () => {
// The API doc says PAUSE:{timestamp}, using current time.
sendCommand(`PAUSE:${currentTime.value}`);
isPlaying.value = false;
};
const seek = (timestamp: number) => {
sendCommand(`SEEK:${timestamp}`);
// If not playing, seeking should not start playing.
};
const changeSpeed = (speed: number) => {
// The provided API doc does not include speed control.
// This is a placeholder for future implementation if the backend supports it.
console.warn(`播放速度控制 (${speed}x) 尚未实现`);
};
@ -139,7 +141,6 @@ export function usePlaybackWebSocket(editorService: ShallowRef<EditorService | u
const editor = editorService.value;
if (!editor || updates.length === 0) return;
// This logic is adapted from movement-supervision.vue for consistency.
updates.forEach(({ id, data }) => {
if (editor.checkRobotById(id)) {
const { x, y, angle, ...rest } = data;
@ -157,9 +158,6 @@ export function usePlaybackWebSocket(editorService: ShallowRef<EditorService | u
const renderLoop = () => {
const updates: Array<{ id: string; data: RobotRealtimeInfo }> = [];
// Naive implementation: process all buffered data in one frame.
// Can be optimized with time slicing like in movement-supervision.vue if needed.
for (const [id, data] of latestRobotData.entries()) {
updates.push({ id, data });
}
@ -169,20 +167,17 @@ export function usePlaybackWebSocket(editorService: ShallowRef<EditorService | u
batchUpdateRobots(updates);
}
// Check for playback end
if (isPlaying.value && currentTime.value >= totalDuration.value && totalDuration.value > 0) {
pause();
// Optionally, set current time to total duration to prevent overflow on slider
currentTime.value = totalDuration.value;
}
editorService.value?.render();
animationFrameId = requestAnimationFrame(renderLoop);
};
const startRenderLoop = () => {
stopRenderLoop(); // Ensure no multiple loops are running
stopRenderLoop();
renderLoop();
};

36
src/services/ws.ts
View File

@ -9,6 +9,8 @@ const WS_CONFIG = {
heartbeatResponseType: 'pong', // 心跳响应类型
};
type WSConfig = typeof WS_CONFIG;
// WebSocket关闭码说明
const WS_CLOSE_CODES: Record<number, string> = {
1000: '正常关闭',
@ -50,15 +52,17 @@ class EnhancedWebSocket {
private userOnMessage: ((event: MessageEvent) => void) | null = null;
private userOnClose: ((event: CloseEvent) => void) | null = null;
private userOnError: ((event: Event) => void) | null = null;
private config: WSConfig;
private userOnOpen: ((event: Event) => void) | null = null;
private connectionStartTime: number = 0;
private lastHeartbeatTime: number = 0;
private heartbeatSentCount: number = 0;
private heartbeatReceivedCount: number = 0;
constructor(path: string, baseUrl: string) {
constructor(path: string, baseUrl: string, config: Partial<WSConfig> = {}) {
this.path = path;
this.baseUrl = baseUrl;
this.config = { ...WS_CONFIG, ...config };
this.connectionStartTime = Date.now();
console.log(`🔗 开始创建WebSocket连接: ${this.path}, 基础URL: ${baseUrl}`);
this.ws = new WebSocket(baseUrl + path);
@ -78,7 +82,7 @@ class EnhancedWebSocket {
// 🔧 优化：连接建立后立即发送一次心跳，然后开始定期心跳
if (this.ws.readyState === WebSocket.OPEN) {
// console.log(`💓 连接建立后发送初始心跳: ${this.path}`);
this.ws.send(WS_CONFIG.heartbeatMessage);
this.ws.send(this.config.heartbeatMessage);
this.heartbeatSentCount++;
this.lastHeartbeatTime = Date.now();
this.startHeartbeatTimeout();
@ -100,7 +104,7 @@ class EnhancedWebSocket {
let isHeartbeatResponse = false;
// 1. 检查是否为简单字符串心跳响应
if (typeof messageData === 'string' && messageData === WS_CONFIG.heartbeatResponseType) {
if (typeof messageData === 'string' && messageData === this.config.heartbeatResponseType) {
isHeartbeatResponse = true;
}
@ -108,7 +112,7 @@ class EnhancedWebSocket {
if (!isHeartbeatResponse && typeof messageData === 'string') {
try {
const data = JSON.parse(messageData);
if (data.type === WS_CONFIG.heartbeatResponseType) {
if (data.type === this.config.heartbeatResponseType) {
isHeartbeatResponse = true;
}
} catch {
@ -240,13 +244,13 @@ class EnhancedWebSocket {
// 开始心跳检测
private startHeartbeat(): void {
this.stopHeartbeat();
console.log(`💓 开始心跳检测: ${this.path}, 间隔: ${WS_CONFIG.heartbeatInterval}ms`);
console.log(`💓 开始心跳检测: ${this.path}, 间隔: ${this.config.heartbeatInterval}ms`);
this.heartbeatTimer = setInterval(() => {
if (this.ws.readyState === WebSocket.OPEN) {
this.heartbeatSentCount++;
this.lastHeartbeatTime = Date.now();
// console.log(`💓 发送心跳: ${this.path} (#${this.heartbeatSentCount})`);
this.ws.send(WS_CONFIG.heartbeatMessage);
this.ws.send(this.config.heartbeatMessage);
// 只有在没有进行超时检测时才设置新的超时检测
if (!this.heartbeatTimeoutTimer) {
@ -255,7 +259,7 @@ class EnhancedWebSocket {
} else {
console.log(`⚠️ 心跳检测时发现连接状态异常: ${this.path}, 状态: ${this.getReadyStateText()}`);
}
}, WS_CONFIG.heartbeatInterval);
}, this.config.heartbeatInterval);
}
// 停止心跳检测
@ -272,12 +276,12 @@ class EnhancedWebSocket {
private startHeartbeatTimeout(): void {
// 不再自动清除，只在收到响应时清除
this.heartbeatTimeoutTimer = setTimeout(() => {
console.log(`💔 心跳响应超时: ${this.path}, ${WS_CONFIG.heartbeatTimeout}ms内未收到响应，主动断开连接`);
console.log(`💔 心跳响应超时: ${this.path}, ${this.config.heartbeatTimeout}ms内未收到响应，主动断开连接`);
console.log(` └─ 心跳统计: 发送${this.heartbeatSentCount}次, 接收${this.heartbeatReceivedCount}次`);
// 设置心跳超时标志，触发重连
this.isHeartbeatTimeout = true;
this.ws.close(1000, 'Heartbeat timeout'); // 使用正常关闭状态码，通过标志来判断是否重连
}, WS_CONFIG.heartbeatTimeout);
}, this.config.heartbeatTimeout);
}
// 清除心跳响应超时检测
@ -290,7 +294,7 @@ class EnhancedWebSocket {
// 安排重连
private scheduleReconnect(): void {
if (this.isManualClose || this.reconnectAttempts >= WS_CONFIG.maxReconnectAttempts) {
if (this.isManualClose || this.reconnectAttempts >= this.config.maxReconnectAttempts) {
console.log(`🚫 停止重连: ${this.path}, 手动关闭: ${this.isManualClose}, 重连次数: ${this.reconnectAttempts}`);
return;
}
@ -299,12 +303,12 @@ class EnhancedWebSocket {
// 指数退避重连策略
const delay = Math.min(
WS_CONFIG.reconnectBaseDelay * Math.pow(2, this.reconnectAttempts - 1),
WS_CONFIG.maxReconnectDelay,
this.config.reconnectBaseDelay * Math.pow(2, this.reconnectAttempts - 1),
this.config.maxReconnectDelay,
);
console.log(
`🔄 WebSocket将在${delay}ms后重连: ${this.path} (${this.reconnectAttempts}/${WS_CONFIG.maxReconnectAttempts})`,
`🔄 WebSocket将在${delay}ms后重连: ${this.path} (${this.reconnectAttempts}/${this.config.maxReconnectAttempts})`,
);
this.reconnectTimer = setTimeout(() => {
@ -316,7 +320,7 @@ class EnhancedWebSocket {
private reconnect(): void {
if (this.isManualClose) return;
console.log(`🔄 WebSocket重连尝试: ${this.path} (${this.reconnectAttempts}/${WS_CONFIG.maxReconnectAttempts})`);
console.log(`🔄 WebSocket重连尝试: ${this.path} (${this.reconnectAttempts}/${this.config.maxReconnectAttempts})`);
this.connectionStartTime = Date.now();
// 创建新的WebSocket连接
@ -440,7 +444,7 @@ class EnhancedWebSocket {
readonly CLOSED = WebSocket.CLOSED;
}
function create(path: string): Promise<WebSocket> {
function create(path: string, config?: Partial<WSConfig>): Promise<WebSocket> {
let baseUrl = '';
if (path.includes(import.meta.env.ENV_STORAGE_WEBSOCKET_BASE)) {
baseUrl = '';
@ -448,7 +452,7 @@ function create(path: string): Promise<WebSocket> {
baseUrl = import.meta.env.ENV_WEBSOCKET_BASE ?? '';
}
const ws = new EnhancedWebSocket(path, baseUrl) as WebSocket;
const ws = new EnhancedWebSocket(path, baseUrl, config) as WebSocket;
return new Promise((resolve, reject) => {
const timeout = setTimeout(() => {