89 lines
3.4 KiB
Python
89 lines
3.4 KiB
Python
"""
|
|
同步/异步函数测试插件
|
|
|
|
用于演示 SyncHandlerError 异常以及如何将同步函数放入线程池执行。
|
|
"""
|
|
import time
|
|
from typing import Any
|
|
from core.command_manager import matcher
|
|
from core.executor import run_in_thread_pool
|
|
from core.bot import Bot
|
|
from core.logger import logger
|
|
|
|
# 插件元数据
|
|
__plugin_meta__ = {
|
|
"name": "SyncAsyncTestPlugin",
|
|
"description": "用于测试同步/异步函数处理的插件。",
|
|
"usage": (
|
|
"/test_sync_error - 尝试注册一个同步函数作为异步处理器，会触发错误。\n"
|
|
"/test_blocking_task <duration> - 演示将同步阻塞任务放入线程池执行。"
|
|
),
|
|
}
|
|
|
|
# --- 示例 1: 触发 SyncHandlerError (此函数不会被成功注册) ---
|
|
|
|
# 这是一个同步函数，如果直接用 @matcher.message_handler 装饰，
|
|
# 并且 event_handler 检查到它是同步的，就会抛出 SyncHandlerError。
|
|
# 注意：为了演示错误，我们不会真正注册它，因为注册会失败。
|
|
def _sync_function_that_should_fail(bot: Bot, event: Any):
|
|
"""
|
|
一个同步函数，如果直接作为异步事件处理器注册，会触发 SyncHandlerError。
|
|
"""
|
|
logger.info("这个同步函数不应该被直接调用。")
|
|
return "这是一个同步函数的结果。"
|
|
|
|
# --- 示例 2: 将同步阻塞任务放入线程池运行 ---
|
|
|
|
def _blocking_task(duration: int) -> str:
|
|
"""
|
|
一个模拟耗时操作的同步函数。
|
|
Args:
|
|
duration (int): 模拟阻塞的秒数。
|
|
Returns:
|
|
str: 任务完成消息。
|
|
"""
|
|
logger.info(f"同步阻塞任务开始，持续 {duration} 秒...")
|
|
time.sleep(duration)
|
|
logger.info("同步阻塞任务结束。")
|
|
return f"阻塞任务完成，耗时 {duration} 秒。"
|
|
|
|
@matcher.message_handler.command("test_blocking_task")
|
|
async def test_blocking_task_handler(bot: Bot, event: Any, args: list):
|
|
"""
|
|
处理 /test_blocking_task 命令，将同步阻塞任务放入线程池执行。
|
|
Args:
|
|
bot (Bot): 机器人实例。
|
|
event (Any): 接收到的事件对象。
|
|
args (list): 命令参数列表。
|
|
"""
|
|
if not args:
|
|
await bot.send(event, "请提供阻塞时长，例如：/test_blocking_task 5")
|
|
return
|
|
|
|
try:
|
|
duration = int(args[0])
|
|
if duration <= 0:
|
|
raise ValueError("时长必须是正整数。")
|
|
except ValueError:
|
|
await bot.send(event, "无效的时长，请提供一个正整数。")
|
|
return
|
|
|
|
await bot.send(event, f"开始执行同步阻塞任务，预计耗时 {duration} 秒...")
|
|
|
|
# 将同步函数放入线程池执行
|
|
result = await run_in_thread_pool(_blocking_task, duration)
|
|
|
|
await bot.send(event, f"同步阻塞任务已完成：{result}")
|
|
|
|
# --- 示例 3: 尝试注册一个同步函数作为异步处理器 (会失败) ---
|
|
# 这个函数不会被成功注册，因为 event_handler 会检测到它是同步的并抛出 SyncHandlerError。
|
|
# 插件管理器会捕获这个错误并跳过加载此插件。
|
|
# 为了演示，我们故意尝试注册它。
|
|
# @matcher.message_handler.command("test_sync_error")
|
|
# def test_sync_error_handler(bot: Bot, event: Any):
|
|
# """
|
|
# 这个同步函数尝试作为异步处理器注册，会触发 SyncHandlerError。
|
|
# """
|
|
# logger.error("这个同步函数不应该被直接注册为异步处理器。")
|
|
# return "这个消息不应该被看到。"
|