root
4 months ago
8 changed files with 188 additions and 458 deletions
@ -1,354 +0,0 @@ |
|||
""" |
|||
WebSocket服务模块 |
|||
提供WebSocket业务逻辑接口 |
|||
""" |
|||
from typing import Dict, List, Optional, Any, Callable |
|||
from datetime import datetime |
|||
from app.core.websocket.manager import websocket_manager |
|||
from app.core.websocket.channel import WebSocketChannel, ChannelMessage |
|||
from app.schemas.websocket import WebSocketConfig |
|||
from app.utils.log import get_logger |
|||
|
|||
logger = get_logger(__name__) |
|||
|
|||
class WebSocketService: |
|||
"""WebSocket服务 - 提供业务逻辑接口""" |
|||
|
|||
def __init__(self, config: Optional[WebSocketConfig] = None): |
|||
self._message_handlers: Dict[str, Callable] = {} |
|||
self._created_at = datetime.now() |
|||
self._config = config or WebSocketConfig() |
|||
self._total_messages_sent = 0 |
|||
self._total_messages_received = 0 |
|||
|
|||
logger.info("WebSocket服务初始化完成") |
|||
|
|||
async def initialize(self): |
|||
"""初始化服务，创建默认Channel""" |
|||
try: |
|||
# 创建默认Channel |
|||
for channel_name in self._config.default_channels: |
|||
await self.create_channel(channel_name, self._config.max_channel_size) |
|||
logger.info(f"WebSocket服务创建默认Channel: {channel_name}") |
|||
|
|||
logger.info("WebSocket服务初始化完成") |
|||
|
|||
except Exception as e: |
|||
logger.error(f"WebSocket服务初始化失败: {e}") |
|||
raise |
|||
|
|||
async def create_websocket_client(self, name: str, url: str) -> bool: |
|||
"""创建WebSocket客户端""" |
|||
try: |
|||
client = await websocket_manager.create_client(name, url) |
|||
success = await client.connect() |
|||
|
|||
if success: |
|||
logger.info(f"WebSocket服务创建客户端成功: {name} -> {url}") |
|||
# 自动为客户端订阅默认Channel |
|||
for channel_name in self._config.default_channels: |
|||
# 确保Channel存在并连接 |
|||
channel = websocket_manager.get_channel(channel_name) |
|||
if not channel: |
|||
channel = await websocket_manager.create_channel(channel_name, self._config.max_channel_size) |
|||
# 连接Channel（幂等） |
|||
await channel.connect() |
|||
# 创建并启动适配器（幂等） |
|||
await websocket_manager.create_adapter(name, channel_name) |
|||
logger.info(f"WebSocket服务已为客户端 {name} 订阅默认Channels: {self._config.default_channels}") |
|||
else: |
|||
logger.error(f"WebSocket服务创建客户端失败: {name}") |
|||
|
|||
return success |
|||
|
|||
except Exception as e: |
|||
logger.error(f"WebSocket服务创建客户端异常: {name}, 错误: {e}") |
|||
return False |
|||
|
|||
async def remove_websocket_client(self, name: str) -> bool: |
|||
"""移除WebSocket客户端""" |
|||
try: |
|||
success = await websocket_manager.remove_client(name) |
|||
|
|||
if success: |
|||
logger.info(f"WebSocket服务移除客户端成功: {name}") |
|||
else: |
|||
logger.warning(f"WebSocket服务移除客户端失败: {name}") |
|||
|
|||
return success |
|||
|
|||
except Exception as e: |
|||
logger.error(f"WebSocket服务移除客户端异常: {name}, 错误: {e}") |
|||
return False |
|||
|
|||
def get_websocket_client(self, name: str): |
|||
"""获取WebSocket客户端""" |
|||
return websocket_manager.get_client(name) |
|||
|
|||
def get_all_websocket_clients(self) -> Dict[str, Any]: |
|||
"""获取所有WebSocket客户端信息""" |
|||
clients = websocket_manager.get_all_clients() |
|||
result = {} |
|||
for name, client in clients.items(): |
|||
stats = client.get_stats() |
|||
# 添加已订阅的Channel列表 |
|||
stats['subscribed_channels'] = self._get_client_subscribed_channels(name) |
|||
result[name] = stats |
|||
return result |
|||
|
|||
def _get_client_subscribed_channels(self, client_name: str) -> List[str]: |
|||
"""获取客户端已订阅的Channel列表""" |
|||
adapters = websocket_manager.get_all_adapters() |
|||
subscribed_channels = [] |
|||
for adapter_key, adapter in adapters.items(): |
|||
if adapter_key.startswith(f"{client_name}:"): |
|||
subscribed_channels.append(adapter.channel_name) |
|||
return subscribed_channels |
|||
|
|||
async def connect_websocket_client(self, name: str) -> bool: |
|||
"""连接WebSocket客户端""" |
|||
try: |
|||
success = await websocket_manager.connect_client(name) |
|||
|
|||
if success: |
|||
logger.info(f"WebSocket服务连接客户端成功: {name}") |
|||
else: |
|||
logger.error(f"WebSocket服务连接客户端失败: {name}") |
|||
|
|||
return success |
|||
|
|||
except Exception as e: |
|||
logger.error(f"WebSocket服务连接客户端异常: {name}, 错误: {e}") |
|||
return False |
|||
|
|||
async def disconnect_websocket_client(self, name: str) -> bool: |
|||
"""断开WebSocket客户端""" |
|||
try: |
|||
success = await websocket_manager.disconnect_client(name) |
|||
|
|||
if success: |
|||
logger.info(f"WebSocket服务断开客户端成功: {name}") |
|||
else: |
|||
logger.warning(f"WebSocket服务断开客户端失败: {name}") |
|||
|
|||
return success |
|||
|
|||
except Exception as e: |
|||
logger.error(f"WebSocket服务断开客户端异常: {name}, 错误: {e}") |
|||
return False |
|||
|
|||
async def create_channel(self, name: str, max_size: int = 1000) -> Optional[WebSocketChannel]: |
|||
"""创建Channel""" |
|||
try: |
|||
channel = await websocket_manager.create_channel(name, max_size) |
|||
logger.info(f"WebSocket服务创建Channel成功: {name}") |
|||
return channel |
|||
|
|||
except Exception as e: |
|||
logger.error(f"WebSocket服务创建Channel异常: {name}, 错误: {e}") |
|||
return None |
|||
|
|||
async def remove_channel(self, name: str) -> bool: |
|||
"""移除Channel""" |
|||
try: |
|||
success = await websocket_manager.remove_channel(name) |
|||
|
|||
if success: |
|||
logger.info(f"WebSocket服务移除Channel成功: {name}") |
|||
else: |
|||
logger.warning(f"WebSocket服务移除Channel失败: {name}") |
|||
|
|||
return success |
|||
|
|||
except Exception as e: |
|||
logger.error(f"WebSocket服务移除Channel异常: {name}, 错误: {e}") |
|||
return False |
|||
|
|||
def get_channel(self, name: str) -> Optional[WebSocketChannel]: |
|||
"""获取Channel""" |
|||
return websocket_manager.get_channel(name) |
|||
|
|||
def get_all_channels(self) -> Dict[str, Any]: |
|||
"""获取所有Channel信息""" |
|||
channels = websocket_manager.get_all_channels() |
|||
result = {} |
|||
for name, channel in channels.items(): |
|||
stats = channel.get_stats() |
|||
# 添加订阅者列表 |
|||
stats['subscribers'] = self._get_channel_subscribers(name) |
|||
result[name] = stats |
|||
return result |
|||
|
|||
def _get_channel_subscribers(self, channel_name: str) -> List[str]: |
|||
"""获取Channel的订阅者列表""" |
|||
adapters = websocket_manager.get_all_adapters() |
|||
subscribers = [] |
|||
for adapter_key, adapter in adapters.items(): |
|||
if adapter_key.endswith(f":{channel_name}"): |
|||
subscribers.append(adapter.client_name) |
|||
return subscribers |
|||
|
|||
async def create_adapter(self, client_name: str, channel_name: str) -> bool: |
|||
"""创建适配器，连接客户端和Channel""" |
|||
try: |
|||
success = await websocket_manager.create_adapter(client_name, channel_name) |
|||
|
|||
if success: |
|||
logger.info(f"WebSocket服务创建适配器成功: {client_name} <-> {channel_name}") |
|||
else: |
|||
logger.error(f"WebSocket服务创建适配器失败: {client_name} <-> {channel_name}") |
|||
|
|||
return success |
|||
|
|||
except Exception as e: |
|||
logger.error(f"WebSocket服务创建适配器异常: {client_name} <-> {channel_name}, 错误: {e}") |
|||
return False |
|||
|
|||
async def remove_adapter(self, client_name: str, channel_name: str) -> bool: |
|||
"""移除适配器""" |
|||
try: |
|||
success = await websocket_manager.remove_adapter(client_name, channel_name) |
|||
|
|||
if success: |
|||
logger.info(f"WebSocket服务移除适配器成功: {client_name} <-> {channel_name}") |
|||
else: |
|||
logger.warning(f"WebSocket服务移除适配器失败: {client_name} <-> {channel_name}") |
|||
|
|||
return success |
|||
|
|||
except Exception as e: |
|||
logger.error(f"WebSocket服务移除适配器异常: {client_name} <-> {channel_name}, 错误: {e}") |
|||
return False |
|||
|
|||
def get_adapter(self, client_name: str, channel_name: str): |
|||
"""获取适配器""" |
|||
return websocket_manager.get_adapter(client_name, channel_name) |
|||
|
|||
def get_all_adapters(self) -> Dict[str, Any]: |
|||
"""获取所有适配器信息""" |
|||
adapters = websocket_manager.get_all_adapters() |
|||
return {key: adapter.get_stats() for key, adapter in adapters.items()} |
|||
|
|||
async def send_message(self, client_name: str, message_type: str, data: Any, channel_name: str = "default") -> bool: |
|||
"""发送消息到指定客户端""" |
|||
try: |
|||
success = await websocket_manager.send_message_to_client(client_name, message_type, data, channel_name) |
|||
|
|||
if success: |
|||
self._total_messages_sent += 1 |
|||
logger.info(f"WebSocket服务发送消息成功: {client_name} -> {message_type}") |
|||
else: |
|||
logger.error(f"WebSocket服务发送消息失败: {client_name} -> {message_type}") |
|||
|
|||
return success |
|||
|
|||
except Exception as e: |
|||
logger.error(f"WebSocket服务发送消息异常: {client_name} -> {message_type}, 错误: {e}") |
|||
return False |
|||
|
|||
async def broadcast_message(self, message_type: str, data: Any, channel_name: str = "default") -> Dict[str, bool]: |
|||
"""广播消息到所有客户端""" |
|||
try: |
|||
results = await websocket_manager.broadcast_message(message_type, data, channel_name) |
|||
|
|||
success_count = sum(1 for success in results.values() if success) |
|||
self._total_messages_sent += success_count |
|||
|
|||
logger.info(f"WebSocket服务广播消息完成: {message_type} -> {success_count}/{len(results)} 成功") |
|||
return results |
|||
|
|||
except Exception as e: |
|||
logger.error(f"WebSocket服务广播消息异常: {message_type}, 错误: {e}") |
|||
return {} |
|||
|
|||
async def subscribe_to_channel(self, client_name: str, channel_name: str) -> bool: |
|||
"""客户端订阅Channel""" |
|||
try: |
|||
# 检查客户端和Channel是否存在 |
|||
client = self.get_websocket_client(client_name) |
|||
channel = self.get_channel(channel_name) |
|||
|
|||
if not client: |
|||
logger.error(f"WebSocket服务订阅失败: 客户端 {client_name} 不存在") |
|||
return False |
|||
|
|||
if not channel: |
|||
logger.error(f"WebSocket服务订阅失败: Channel {channel_name} 不存在") |
|||
return False |
|||
|
|||
# 创建适配器（如果不存在） |
|||
success = await self.create_adapter(client_name, channel_name) |
|||
|
|||
if success: |
|||
logger.info(f"WebSocket服务订阅成功: {client_name} -> {channel_name}") |
|||
else: |
|||
logger.error(f"WebSocket服务订阅失败: {client_name} -> {channel_name}") |
|||
|
|||
return success |
|||
|
|||
except Exception as e: |
|||
logger.error(f"WebSocket服务订阅异常: {client_name} -> {channel_name}, 错误: {e}") |
|||
return False |
|||
|
|||
async def unsubscribe_from_channel(self, client_name: str, channel_name: str) -> bool: |
|||
"""客户端取消订阅Channel""" |
|||
try: |
|||
success = await self.remove_adapter(client_name, channel_name) |
|||
|
|||
if success: |
|||
logger.info(f"WebSocket服务取消订阅成功: {client_name} -> {channel_name}") |
|||
else: |
|||
logger.warning(f"WebSocket服务取消订阅失败: {client_name} -> {channel_name}") |
|||
|
|||
return success |
|||
|
|||
except Exception as e: |
|||
logger.error(f"WebSocket服务取消订阅异常: {client_name} -> {channel_name}, 错误: {e}") |
|||
return False |
|||
|
|||
def register_message_handler(self, message_type: str, handler: Callable): |
|||
"""注册消息处理器""" |
|||
try: |
|||
self._message_handlers[message_type] = handler |
|||
logger.info(f"WebSocket服务注册消息处理器: {message_type}") |
|||
|
|||
except Exception as e: |
|||
logger.error(f"WebSocket服务注册消息处理器异常: {message_type}, 错误: {e}") |
|||
|
|||
def unregister_message_handler(self, message_type: str): |
|||
"""取消注册消息处理器""" |
|||
try: |
|||
if message_type in self._message_handlers: |
|||
del self._message_handlers[message_type] |
|||
logger.info(f"WebSocket服务取消注册消息处理器: {message_type}") |
|||
|
|||
except Exception as e: |
|||
logger.error(f"WebSocket服务取消注册消息处理器异常: {message_type}, 错误: {e}") |
|||
|
|||
def get_service_stats(self) -> Dict[str, Any]: |
|||
"""获取服务统计信息""" |
|||
try: |
|||
stats = websocket_manager.get_stats() |
|||
stats.update({ |
|||
"total_messages_sent": self._total_messages_sent, |
|||
"total_messages_received": self._total_messages_received, |
|||
"service_created_at": self._created_at.isoformat(), |
|||
"config": self._config.dict() |
|||
}) |
|||
return stats |
|||
|
|||
except Exception as e: |
|||
logger.error(f"WebSocket服务获取统计信息异常: {e}") |
|||
return {} |
|||
|
|||
async def cleanup(self): |
|||
"""清理服务资源""" |
|||
try: |
|||
await websocket_manager.cleanup() |
|||
logger.info("WebSocket服务清理完成") |
|||
|
|||
except Exception as e: |
|||
logger.error(f"WebSocket服务清理异常: {e}") |
|||
|
|||
# 全局WebSocket服务实例 |
|||
websocket_service = WebSocketService() |
|||
Loading…
Reference in new issue