#!/usr/bin/env python3 # -*- coding: utf-8 -*- import logging import asyncio from typing import Dict, List, Callable, Any from enum import Enum import json logger = logging.getLogger(__name__) class PluginMessageType(Enum): """插件消息类型枚举""" PLUGIN_EVENT = "plugin_event" PLUGIN_DATA = "plugin_data" PLUGIN_COMMAND = "plugin_command" PLUGIN_REQUEST = "plugin_request" PLUGIN_RESPONSE = "plugin_response" class PluginBridge: """插件桥接服务 - 用于框架与插件、插件间通信""" def __init__(self, core_bridge): self.core_bridge = core_bridge self.plugin_subscribers: Dict[str, Dict[str, List[Callable]]] = {} self.plugin_message_queue: asyncio.Queue = asyncio.Queue() self.is_running = False self.processing_task = None logger.debug("PluginBridge初始化开始") async def start(self): """启动插件桥接服务""" try: logger.info("启动插件桥接服务") self.is_running = True self.processing_task = asyncio.create_task(self._process_plugin_messages()) # 订阅核心桥接的相关主题 self.core_bridge.subscribe("plugin.*", self._handle_core_plugin_message) logger.debug("插件桥接服务启动完成") except Exception as e: logger.error(f"启动插件桥接服务时出错: {str(e)}", exc_info=True) raise def subscribe_plugin(self, plugin_name: str, topic: str, callback: Callable): """插件订阅主题""" try: if plugin_name not in self.plugin_subscribers: self.plugin_subscribers[plugin_name] = {} if topic not in self.plugin_subscribers[plugin_name]: self.plugin_subscribers[plugin_name][topic] = [] self.plugin_subscribers[plugin_name][topic].append(callback) logger.debug(f"插件 {plugin_name} 订阅主题: {topic}, 订阅者数: {len(self.plugin_subscribers[plugin_name][topic])}") except Exception as e: logger.error(f"插件订阅主题时出错: {str(e)}", exc_info=True) raise def unsubscribe_plugin(self, plugin_name: str, topic: str, callback: Callable): """插件取消订阅""" try: if (plugin_name in self.plugin_subscribers and topic in self.plugin_subscribers[plugin_name] and callback in self.plugin_subscribers[plugin_name][topic]): self.plugin_subscribers[plugin_name][topic].remove(callback) logger.debug(f"插件 {plugin_name} 取消订阅主题: {topic}") except Exception as e: logger.error(f"插件取消订阅时出错: {str(e)}", exc_info=True) async def publish_to_plugin(self, target_plugin: str, topic: str, message: Dict, msg_type: PluginMessageType = PluginMessageType.PLUGIN_DATA): """发布消息到指定插件""" try: message_data = { "target_plugin": target_plugin, "topic": topic, "type": msg_type.value, "data": message, "timestamp": asyncio.get_event_loop().time() } await self.plugin_message_queue.put(message_data) logger.debug(f"发布消息到插件 {target_plugin}, 主题: {topic}") except Exception as e: logger.error(f"发布消息到插件时出错: {str(e)}", exc_info=True) raise async def broadcast_to_plugins(self, topic: str, message: Dict, exclude_plugins: List[str] = None, msg_type: PluginMessageType = PluginMessageType.PLUGIN_DATA): """广播消息到所有插件""" try: exclude_plugins = exclude_plugins or [] for plugin_name in self.plugin_subscribers.keys(): if plugin_name not in exclude_plugins: await self.publish_to_plugin(plugin_name, topic, message, msg_type) logger.debug(f"广播消息到插件, 主题: {topic}, 排除: {exclude_plugins}") except Exception as e: logger.error(f"广播消息到插件时出错: {str(e)}", exc_info=True) raise async def _process_plugin_messages(self): """处理插件消息队列""" try: logger.debug("开始处理插件消息队列") while self.is_running: try: # 等待消息，带超时 message = await asyncio.wait_for(self.plugin_message_queue.get(), timeout=1.0) # 分发消息给目标插件 await self._dispatch_plugin_message(message) # 标记任务完成 self.plugin_message_queue.task_done() except asyncio.TimeoutError: continue except Exception as e: logger.error(f"处理插件消息时出错: {str(e)}", exc_info=True) continue logger.debug("插件消息队列处理结束") except Exception as e: logger.error(f"插件消息队列处理循环出错: {str(e)}", exc_info=True) async def _dispatch_plugin_message(self, message: Dict): """分发消息给插件订阅者""" try: target_plugin = message["target_plugin"] topic = message["topic"] if (target_plugin not in self.plugin_subscribers or topic not in self.plugin_subscribers[target_plugin]): logger.debug(f"插件 {target_plugin} 没有订阅主题 {topic}") return subscribers = self.plugin_subscribers[target_plugin][topic][:] # 并行调用所有订阅者 tasks = [] for callback in subscribers: task = asyncio.create_task(self._call_plugin_subscriber(callback, message)) tasks.append(task) if tasks: await asyncio.gather(*tasks, return_exceptions=True) logger.debug(f"插件消息分发完成，目标: {target_plugin}, 主题: {topic}, 订阅者数: {len(subscribers)}") except Exception as e: logger.error(f"分发插件消息时出错: {str(e)}", exc_info=True) async def _call_plugin_subscriber(self, callback: Callable, message: Dict): """调用插件订阅者回调""" try: if asyncio.iscoroutinefunction(callback): await callback(message) else: callback(message) except Exception as e: logger.error(f"调用插件订阅者回调时出错: {str(e)}", exc_info=True) async def _handle_core_plugin_message(self, message: Dict): """处理来自核心桥接的插件相关消息""" try: topic = message["topic"] data = message["data"] # 根据主题类型处理 if topic.startswith("plugin.event."): # 广播插件事件 event_type = topic.replace("plugin.event.", "") await self.broadcast_to_plugins( f"event.{event_type}", data, msg_type=PluginMessageType.PLUGIN_EVENT ) elif topic.startswith("plugin.broadcast."): # 广播消息 broadcast_topic = topic.replace("plugin.broadcast.", "") await self.broadcast_to_plugins( broadcast_topic, data, msg_type=PluginMessageType.PLUGIN_DATA ) logger.debug(f"处理核心插件消息: {topic}") except Exception as e: logger.error(f"处理核心插件消息时出错: {str(e)}", exc_info=True) def get_plugin_subscriber_count(self, plugin_name: str = None) -> int: """获取插件订阅者数量""" try: if plugin_name: if plugin_name not in self.plugin_subscribers: return 0 total = sum(len(subs) for subs in self.plugin_subscribers[plugin_name].values()) logger.debug(f"插件 {plugin_name} 的订阅者数量: {total}") return total else: total = 0 for plugin_subs in self.plugin_subscribers.values(): total += sum(len(subs) for subs in plugin_subs.values()) logger.debug(f"总插件订阅者数量: {total}") return total except Exception as e: logger.error(f"获取插件订阅者数量时出错: {str(e)}", exc_info=True) return 0 def cleanup_plugin_subscriptions(self, plugin_name: str): """清理插件的所有订阅""" try: if plugin_name in self.plugin_subscribers: del self.plugin_subscribers[plugin_name] logger.debug(f"清理插件订阅: {plugin_name}") except Exception as e: logger.error(f"清理插件订阅时出错: {str(e)}", exc_info=True) async def shutdown(self): """关闭插件桥接服务""" try: logger.info("关闭插件桥接服务") self.is_running = False # 等待处理任务结束 if self.processing_task: await asyncio.wait_for(self.processing_task, timeout=5.0) # 清理所有订阅 self.plugin_subscribers.clear() # 清空队列 while not self.plugin_message_queue.empty(): try: self.plugin_message_queue.get_nowait() self.plugin_message_queue.task_done() except asyncio.QueueEmpty: break logger.debug("插件桥接服务关闭完成") except Exception as e: logger.error(f"关闭插件桥接服务时出错: {str(e)}", exc_info=True)