client: 增加ACK支持
This commit is contained in:
124
Html/apps/extension_ase/ase_client/ACK_SUPPORT_GUIDE.md
Normal file
124
Html/apps/extension_ase/ase_client/ACK_SUPPORT_GUIDE.md
Normal file
@@ -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确认即可。
|
||||||
|
|
||||||
|
这种设计既保证了通信的可靠性,又不会影响业务逻辑的执行效率,是一种高效且可靠的通信方式。
|
||||||
@@ -5,6 +5,7 @@ import os
|
|||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from typing import Dict, Callable, Any, Optional
|
from typing import Dict, Callable, Any, Optional
|
||||||
from functools import partial
|
from functools import partial
|
||||||
|
from threading import Event
|
||||||
|
|
||||||
class ASEngineClient:
|
class ASEngineClient:
|
||||||
def __init__(self, server_url: str, namespace: str, id:str):
|
def __init__(self, server_url: str, namespace: str, id:str):
|
||||||
@@ -29,7 +30,8 @@ class ASEngineClient:
|
|||||||
self._on_connect_callback_entry=None
|
self._on_connect_callback_entry=None
|
||||||
self._on_connect_callback_initData=None
|
self._on_connect_callback_initData=None
|
||||||
|
|
||||||
|
# ACK相关
|
||||||
|
self.ack_events: Dict[str, Event] = {} # 存储ack_id和对应的Event对象
|
||||||
|
|
||||||
# 注册基础事件处理函数
|
# 注册基础事件处理函数
|
||||||
self._register_base_events()
|
self._register_base_events()
|
||||||
@@ -63,6 +65,15 @@ class ASEngineClient:
|
|||||||
def on_response(data):
|
def on_response(data):
|
||||||
print(f"Received 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:
|
def register_on_connect_entry(self, callback: Callable, entry: Any, init_data: Any)->bool:
|
||||||
if not callable(callback):
|
if not callable(callback):
|
||||||
return False
|
return False
|
||||||
@@ -169,12 +180,15 @@ class ASEngineClient:
|
|||||||
self.sid = None
|
self.sid = None
|
||||||
print("Disconnected from server")
|
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 route: 目标路由
|
||||||
:param text: 要发送的文本
|
:param text: 要发送的文本
|
||||||
:return: 是否发送成功
|
:param id: 发送者ID
|
||||||
|
:param wait_for_ack: 是否等待服务器ACK确认
|
||||||
|
:param timeout: 等待ACK的超时时间(秒)
|
||||||
|
:return: 是否发送成功(若wait_for_ack为True,则返回是否收到ACK)
|
||||||
"""
|
"""
|
||||||
if not self.connected:
|
if not self.connected:
|
||||||
print("Not connected to server")
|
print("Not connected to server")
|
||||||
@@ -183,19 +197,53 @@ class ASEngineClient:
|
|||||||
id = self.id
|
id = self.id
|
||||||
|
|
||||||
try:
|
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}")
|
print(f"Sent text to route '{route}': {text}")
|
||||||
|
|
||||||
|
# 如果不需要等待ACK,直接返回True
|
||||||
|
if not wait_for_ack:
|
||||||
return True
|
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:
|
except Exception as e:
|
||||||
print(f"Failed to send text to route '{route}': {e}")
|
print(f"Failed to send text to route '{route}': {e}")
|
||||||
return False
|
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 route: 目标路由
|
||||||
:param voice_data: 语音数据字节流
|
: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:
|
if id is None:
|
||||||
id = self.id
|
id = self.id
|
||||||
@@ -204,8 +252,9 @@ class ASEngineClient:
|
|||||||
return False
|
return False
|
||||||
|
|
||||||
try:
|
try:
|
||||||
# 编码语音数据为base64
|
# 生成唯一ack_id
|
||||||
# voice_data = 'data:image/jpeg;base64,' +base64.b64encode(voice_data).decode('utf-8')
|
ack_id = f"{id}_{datetime.now().strftime('%Y%m%d%H%M%S%f')}"
|
||||||
|
|
||||||
# 发送数据
|
# 发送数据
|
||||||
self.sio.emit(
|
self.sio.emit(
|
||||||
route,
|
route,
|
||||||
@@ -214,24 +263,46 @@ class ASEngineClient:
|
|||||||
'data': voice_data,
|
'data': voice_data,
|
||||||
'timestamp': int(time.time() * 1000),
|
'timestamp': int(time.time() * 1000),
|
||||||
'sampleRate': 16000,
|
'sampleRate': 16000,
|
||||||
'id': id
|
'id': id,
|
||||||
|
'ack_id': ack_id
|
||||||
},
|
},
|
||||||
namespace=self.namespace
|
namespace=self.namespace
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
print(f"Sent voice file to route '{route}'")
|
print(f"Sent voice file to route '{route}'")
|
||||||
|
|
||||||
|
# 如果不需要等待ACK,直接返回True
|
||||||
|
if not wait_for_ack:
|
||||||
return True
|
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:
|
except Exception as e:
|
||||||
print(f"Failed to send voice to route '{route}': {e}")
|
print(f"Failed to send voice to route '{route}': {e}")
|
||||||
return False
|
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 route: 目标路由
|
||||||
:param file_path: 图片文件路径
|
: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:
|
if not self.connected:
|
||||||
print("Not connected to server")
|
print("Not connected to server")
|
||||||
@@ -251,18 +322,41 @@ class ASEngineClient:
|
|||||||
# 获取文件名
|
# 获取文件名
|
||||||
filename = os.path.basename(file_path)
|
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(
|
self.sio.emit(
|
||||||
route,
|
route,
|
||||||
{
|
{
|
||||||
'type': 'image',
|
'type': 'image',
|
||||||
'data': image_data,
|
'data': image_data,
|
||||||
'id': id
|
'id': id,
|
||||||
|
'ack_id': ack_id
|
||||||
},
|
},
|
||||||
namespace=self.namespace
|
namespace=self.namespace
|
||||||
)
|
)
|
||||||
print(f"Sent image file to route '{route}': {filename}")
|
print(f"Sent image file to route '{route}': {filename}")
|
||||||
|
|
||||||
|
# 如果不需要等待ACK,直接返回True
|
||||||
|
if not wait_for_ack:
|
||||||
return True
|
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:
|
except Exception as e:
|
||||||
print(f"Failed to send image to route '{route}': {e}")
|
print(f"Failed to send image to route '{route}': {e}")
|
||||||
return False
|
return False
|
||||||
|
|||||||
Reference in New Issue
Block a user