diff --git a/Html/apps/extension_ase/ase_client/ACK_SUPPORT_GUIDE.md b/Html/apps/extension_ase/ase_client/ACK_SUPPORT_GUIDE.md new file mode 100644 index 0000000..518f9c9 --- /dev/null +++ b/Html/apps/extension_ase/ase_client/ACK_SUPPORT_GUIDE.md @@ -0,0 +1,124 @@ +# ACK机制支持指南 + +## 概述 + +客户端已实现基于emit机制的ACK确认功能,允许客户端在发送消息后等待服务器确认,从而支持同步和异步两种通信模式。 + +## 客户端改动 + +### 1. 新增参数 + +客户端的三个主要发送方法 `send_text`、`send_voice` 和 `send_image` 都新增了以下参数: + +- `wait_for_ack: bool = False` - 是否等待服务器ACK确认 +- `timeout: int = 5` - 等待ACK的超时时间(秒) + +### 2. 消息格式变化 + +客户端发送的消息中新增了 `ack_id` 字段,格式为 `{id}_{timestamp}`,其中: +- `id` 是客户端ID +- `timestamp` 是当前时间的格式化字符串(精确到微秒) + +### 3. ACK事件 + +客户端监听 `ack` 事件,用于接收服务器的确认消息。 + +## 后端WebSocket服务器需要的支持 + +### 1. 接收消息处理 + +当服务器收到客户端发送的消息时,需要: + +1. 检查消息中是否包含 `ack_id` 字段 +2. 如果包含 `ack_id`,立即发送ACK确认消息,不需要等待后续业务逻辑执行 +3. 继续执行后续的业务逻辑 + +### 2. ACK消息格式 + +服务器需要向客户端发送 `ack` 事件,消息格式如下: + +```json +{ + "ack_id": "客户端发送的ack_id值" +} +``` + +### 3. 发送ACK的时机 + +**非常重要**:服务器必须在收到客户端消息后**立即**发送ACK确认,而不是在业务逻辑执行完成后发送。这样可以确保客户端能够快速收到确认,继续执行后续操作。 + +### 4. 实现示例(Python Socket.IO) + +```python +# 服务器端示例代码 +@sio.on('text_message') # 根据实际事件名调整 +def handle_text_message(sid, data): + # 1. 立即发送ACK确认 + if 'ack_id' in data: + sio.emit('ack', {'ack_id': data['ack_id']}, room=sid) + + # 2. 执行后续业务逻辑(可能耗时较长) + # ... 业务逻辑代码 ... + +@sio.on('voice_message') # 根据实际事件名调整 +def handle_voice_message(sid, data): + # 1. 立即发送ACK确认 + if 'ack_id' in data: + sio.emit('ack', {'ack_id': data['ack_id']}, room=sid) + + # 2. 执行后续业务逻辑(可能耗时较长) + # ... 业务逻辑代码 ... + +@sio.on('image_message') # 根据实际事件名调整 +def handle_image_message(sid, data): + # 1. 立即发送ACK确认 + if 'ack_id' in data: + sio.emit('ack', {'ack_id': data['ack_id']}, room=sid) + + # 2. 执行后续业务逻辑(可能耗时较长) + # ... 业务逻辑代码 ... +``` + +## 使用示例 + +### 客户端同步调用(等待ACK) + +```python +client = ASEngineClient(server_url, namespace, id) +client.connect() + +# 同步调用:等待ACK确认后再继续执行 +result = client.send_text('text_message', 'Hello', wait_for_ack=True, timeout=3) +if result: + print("消息已被服务器确认收到") + # 执行后续操作 +else: + print("等待ACK超时") +``` + +### 客户端异步调用(不等待ACK) + +```python +client = ASEngineClient(server_url, namespace, id) +client.connect() + +# 异步调用:发送后立即返回,不等待ACK +result = client.send_text('text_message', 'Hello', wait_for_ack=False) +if result: + print("消息已发送") + # 执行后续操作,不需要等待ACK +``` + +## 注意事项 + +1. **向后兼容**:客户端的改动保持了向后兼容性,现有代码无需修改即可继续使用。 +2. **超时处理**:客户端会在指定的超时时间后停止等待ACK,并返回False。 +3. **资源清理**:客户端会在收到ACK或超时时清理相关的ACK状态,避免内存泄漏。 +4. **唯一标识**:`ack_id` 确保了每个消息的唯一性,避免ACK混淆。 +5. **性能考虑**:等待ACK会阻塞当前线程,建议仅在必要时使用同步模式。 + +## 总结 + +通过实现基于emit机制的ACK确认功能,客户端和服务器之间的通信可以灵活支持同步和异步两种模式。服务器需要做的改动非常简单,只需要在收到消息后立即发送ACK确认即可。 + +这种设计既保证了通信的可靠性,又不会影响业务逻辑的执行效率,是一种高效且可靠的通信方式。 \ No newline at end of file diff --git a/Html/apps/extension_ase/ase_client/client.py b/Html/apps/extension_ase/ase_client/client.py index 0b88e35..ebbd306 100644 --- a/Html/apps/extension_ase/ase_client/client.py +++ b/Html/apps/extension_ase/ase_client/client.py @@ -5,6 +5,7 @@ import os from datetime import datetime from typing import Dict, Callable, Any, Optional from functools import partial +from threading import Event class ASEngineClient: def __init__(self, server_url: str, namespace: str, id:str): @@ -29,8 +30,9 @@ class ASEngineClient: self._on_connect_callback_entry=None self._on_connect_callback_initData=None - - + # ACK相关 + self.ack_events: Dict[str, Event] = {} # 存储ack_id和对应的Event对象 + # 注册基础事件处理函数 self._register_base_events() @@ -62,6 +64,15 @@ class ASEngineClient: @self.sio.on('response', namespace=self.namespace) def on_response(data): print(f"Received response: {data}") + + # ACK事件处理 + @self.sio.on('ack', namespace=self.namespace) + def on_ack(data): + ack_id = data.get('ack_id') + if ack_id and ack_id in self.ack_events: + print(f"Received ACK for message: {ack_id}") + event = self.ack_events[ack_id] + event.set() # 触发事件,通知等待的线程 def register_on_connect_entry(self, callback: Callable, entry: Any, init_data: Any)->bool: if not callable(callback): @@ -169,12 +180,15 @@ class ASEngineClient: self.sid = None print("Disconnected from server") - def send_text(self, route: str, text: str, id:str="") -> bool: + def send_text(self, route: str, text: str, id:str="", wait_for_ack: bool = False, timeout: int = 5) -> bool: """ 发送文本消息 :param route: 目标路由 :param text: 要发送的文本 - :return: 是否发送成功 + :param id: 发送者ID + :param wait_for_ack: 是否等待服务器ACK确认 + :param timeout: 等待ACK的超时时间(秒) + :return: 是否发送成功(若wait_for_ack为True,则返回是否收到ACK) """ if not self.connected: print("Not connected to server") @@ -183,19 +197,53 @@ class ASEngineClient: id = self.id try: - self.sio.emit(route, {'type': 'text', 'data': text, 'id': id}, namespace=self.namespace) + # 生成唯一ack_id + ack_id = f"{id}_{datetime.now().strftime('%Y%m%d%H%M%S%f')}" + + # 准备发送数据 + data = { + 'type': 'text', + 'data': text, + 'id': id, + 'ack_id': ack_id + } + + # 发送数据 + self.sio.emit(route, data, namespace=self.namespace) print(f"Sent text to route '{route}': {text}") - return True + + # 如果不需要等待ACK,直接返回True + if not wait_for_ack: + return True + + # 初始化ACK事件 + event = Event() + self.ack_events[ack_id] = event + + # 等待ACK或超时 + received_ack = event.wait(timeout=timeout) + + # 清理ACK事件 + del self.ack_events[ack_id] + + if received_ack: + return True + else: + print(f"ACK timeout for message: {ack_id}") + return False except Exception as e: print(f"Failed to send text to route '{route}': {e}") return False - def send_voice(self, route: str, voice_data: bytes, id:str=None) -> bool: + def send_voice(self, route: str, voice_data: bytes, id:str=None, wait_for_ack: bool = False, timeout: int = 5) -> bool: """ 发送语音数据 :param route: 目标路由 :param voice_data: 语音数据字节流 - :return: 是否发送成功 + :param id: 发送者ID + :param wait_for_ack: 是否等待服务器ACK确认 + :param timeout: 等待ACK的超时时间(秒) + :return: 是否发送成功(若wait_for_ack为True,则返回是否收到ACK) """ if id is None: id = self.id @@ -204,8 +252,9 @@ class ASEngineClient: return False try: - # 编码语音数据为base64 - # voice_data = 'data:image/jpeg;base64,' +base64.b64encode(voice_data).decode('utf-8') + # 生成唯一ack_id + ack_id = f"{id}_{datetime.now().strftime('%Y%m%d%H%M%S%f')}" + # 发送数据 self.sio.emit( route, @@ -214,24 +263,46 @@ class ASEngineClient: 'data': voice_data, 'timestamp': int(time.time() * 1000), 'sampleRate': 16000, - 'id': id + 'id': id, + 'ack_id': ack_id }, namespace=self.namespace ) - print(f"Sent voice file to route '{route}'") - return True + + # 如果不需要等待ACK,直接返回True + if not wait_for_ack: + return True + + # 初始化ACK事件 + event = Event() + self.ack_events[ack_id] = event + + # 等待ACK或超时 + received_ack = event.wait(timeout=timeout) + + # 清理ACK事件 + del self.ack_events[ack_id] + + if received_ack: + return True + else: + print(f"ACK timeout for message: {ack_id}") + return False except Exception as e: print(f"Failed to send voice to route '{route}': {e}") return False - def send_image(self, route: str, file_path: str, id:str) -> bool: + def send_image(self, route: str, file_path: str, id:str, wait_for_ack: bool = False, timeout: int = 5) -> bool: """ 发送图片文件 :param route: 目标路由 :param file_path: 图片文件路径 - :return: 是否发送成功 + :param id: 发送者ID + :param wait_for_ack: 是否等待服务器ACK确认 + :param timeout: 等待ACK的超时时间(秒) + :return: 是否发送成功(若wait_for_ack为True,则返回是否收到ACK) """ if not self.connected: print("Not connected to server") @@ -251,18 +322,41 @@ class ASEngineClient: # 获取文件名 filename = os.path.basename(file_path) + # 生成唯一ack_id + ack_id = f"{id}_{datetime.now().strftime('%Y%m%d%H%M%S%f')}" + # 发送数据 self.sio.emit( route, { 'type': 'image', 'data': image_data, - 'id': id + 'id': id, + 'ack_id': ack_id }, namespace=self.namespace ) print(f"Sent image file to route '{route}': {filename}") - return True + + # 如果不需要等待ACK,直接返回True + if not wait_for_ack: + return True + + # 初始化ACK事件 + event = Event() + self.ack_events[ack_id] = event + + # 等待ACK或超时 + received_ack = event.wait(timeout=timeout) + + # 清理ACK事件 + del self.ack_events[ack_id] + + if received_ack: + return True + else: + print(f"ACK timeout for message: {ack_id}") + return False except Exception as e: print(f"Failed to send image to route '{route}': {e}") return False