mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-09-05 23:47:41 +08:00
fix(discord): close bot event loop on shutdown (#6421)
This commit is contained in:
@@ -82,6 +82,7 @@ class Discord:
|
||||
self._tree: Optional[app_commands.CommandTree] = app_commands.CommandTree(self._client)
|
||||
self._loop: asyncio.AbstractEventLoop = asyncio.new_event_loop()
|
||||
self._thread: Optional[threading.Thread] = None
|
||||
self._stop_requested = threading.Event()
|
||||
self._ready_event = threading.Event()
|
||||
self._user_dm_cache: Dict[str, discord.DMChannel] = {}
|
||||
self._user_chat_mapping: Dict[
|
||||
@@ -211,17 +212,35 @@ class Discord:
|
||||
return
|
||||
|
||||
def runner():
|
||||
asyncio.set_event_loop(self._loop)
|
||||
loop = self._loop
|
||||
client = self._client
|
||||
asyncio.set_event_loop(loop)
|
||||
start_task: Optional[asyncio.Task] = None
|
||||
try:
|
||||
self._loop.create_task(self._client.start(self._token))
|
||||
self._loop.run_forever()
|
||||
if not self._stop_requested.is_set():
|
||||
start_task = loop.create_task(client.start(self._token))
|
||||
loop.run_until_complete(start_task)
|
||||
except asyncio.CancelledError:
|
||||
if not self._stop_requested.is_set():
|
||||
logger.error("Discord Bot 启动任务被意外取消")
|
||||
except Exception as err:
|
||||
logger.error(f"Discord Bot 启动失败:{err}")
|
||||
if not self._stop_requested.is_set():
|
||||
logger.error(f"Discord Bot 启动失败:{err}")
|
||||
finally:
|
||||
self._ready_event.clear()
|
||||
if start_task and not start_task.done():
|
||||
start_task.cancel()
|
||||
try:
|
||||
loop.run_until_complete(start_task)
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
try:
|
||||
self._loop.run_until_complete(self._client.close())
|
||||
loop.run_until_complete(client.close())
|
||||
except Exception as err:
|
||||
logger.debug(f"Discord Bot 关闭失败:{err}")
|
||||
finally:
|
||||
asyncio.set_event_loop(None)
|
||||
loop.close()
|
||||
|
||||
self._thread = threading.Thread(target=runner, daemon=True)
|
||||
self._thread.start()
|
||||
@@ -229,21 +248,37 @@ class Discord:
|
||||
def stop(self):
|
||||
if not self._client or not self._loop or not self._thread:
|
||||
return
|
||||
try:
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self._stop_all_typing_tasks(), self._loop
|
||||
).result(timeout=5)
|
||||
asyncio.run_coroutine_threadsafe(self._client.close(), self._loop).result(
|
||||
timeout=10
|
||||
)
|
||||
except Exception as err:
|
||||
logger.error(f"关闭 Discord Bot 失败:{err}")
|
||||
finally:
|
||||
self._stop_requested.set()
|
||||
loop = self._loop
|
||||
thread = self._thread
|
||||
if loop.is_running():
|
||||
try:
|
||||
self._loop.call_soon_threadsafe(self._loop.stop)
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self._stop_all_typing_tasks(), loop
|
||||
).result(timeout=5)
|
||||
except Exception as err:
|
||||
logger.error(f"停止 Discord typing 状态失败:{err}")
|
||||
try:
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self._client.close(), loop
|
||||
).result(timeout=10)
|
||||
except Exception as err:
|
||||
logger.error(f"关闭 Discord Bot 失败:{err}")
|
||||
elif not loop.is_closed():
|
||||
try:
|
||||
loop.call_soon_threadsafe(loop.stop)
|
||||
except RuntimeError:
|
||||
pass
|
||||
self._ready_event.clear()
|
||||
thread.join(timeout=5)
|
||||
if thread.is_alive():
|
||||
try:
|
||||
loop.call_soon_threadsafe(loop.stop)
|
||||
except Exception as err:
|
||||
logger.error(f"停止 Discord 事件循环失败:{err}")
|
||||
self._ready_event.clear()
|
||||
thread.join(timeout=5)
|
||||
if thread.is_alive():
|
||||
logger.error("Discord Bot 线程未在超时内停止")
|
||||
|
||||
def get_state(self) -> bool:
|
||||
return self._ready_event.is_set() and self._client is not None
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
import asyncio
|
||||
import threading
|
||||
from typing import Optional
|
||||
|
||||
from app.modules.discord.discord import Discord
|
||||
|
||||
|
||||
class _DiscordClientStub:
|
||||
"""模拟 Discord 长连接,允许测试控制启动失败与正常关闭。"""
|
||||
|
||||
def __init__(self, *, start_error: Optional[Exception] = None) -> None:
|
||||
self.started = threading.Event()
|
||||
self._release: Optional[asyncio.Event] = None
|
||||
self._start_error = start_error
|
||||
self.close_calls = 0
|
||||
|
||||
async def start(self, _token: str) -> None:
|
||||
self._release = asyncio.Event()
|
||||
self.started.set()
|
||||
if self._start_error:
|
||||
raise self._start_error
|
||||
await self._release.wait()
|
||||
|
||||
async def close(self) -> None:
|
||||
self.close_calls += 1
|
||||
if self._release:
|
||||
self._release.set()
|
||||
|
||||
|
||||
def _discord(client: _DiscordClientStub) -> Discord:
|
||||
"""构造只包含线程与事件循环生命周期状态的 Discord 实例。"""
|
||||
instance = Discord.__new__(Discord)
|
||||
instance._token = "test-token"
|
||||
instance._client = client
|
||||
instance._loop = asyncio.new_event_loop()
|
||||
instance._thread = None
|
||||
instance._stop_requested = threading.Event()
|
||||
instance._ready_event = threading.Event()
|
||||
instance._typing_tasks = {}
|
||||
instance._typing_stop_events = {}
|
||||
return instance
|
||||
|
||||
|
||||
def _cleanup(instance: Discord) -> None:
|
||||
"""即使用例断言失败也回收其线程和事件循环。"""
|
||||
thread = instance._thread
|
||||
loop = instance._loop
|
||||
if thread and thread.is_alive():
|
||||
loop.call_soon_threadsafe(loop.stop)
|
||||
thread.join(timeout=1)
|
||||
if not loop.is_closed():
|
||||
loop.close()
|
||||
|
||||
|
||||
def test_stop_waits_for_discord_thread_and_closes_loop() -> None:
|
||||
"""正常停止返回时,Discord 线程与其事件循环必须已经结束。"""
|
||||
instance = _discord(_DiscordClientStub())
|
||||
instance._start()
|
||||
assert instance._client.started.wait(timeout=1)
|
||||
|
||||
try:
|
||||
instance.stop()
|
||||
|
||||
assert not instance._thread or not instance._thread.is_alive()
|
||||
assert instance._loop.is_closed()
|
||||
instance.stop()
|
||||
finally:
|
||||
_cleanup(instance)
|
||||
|
||||
|
||||
def test_start_failure_closes_discord_thread_loop() -> None:
|
||||
"""Discord 启动协程失败后不得留下空跑线程和未关闭循环。"""
|
||||
instance = _discord(_DiscordClientStub(start_error=RuntimeError("invalid token")))
|
||||
instance._start()
|
||||
assert instance._client.started.wait(timeout=1)
|
||||
|
||||
try:
|
||||
assert instance._thread
|
||||
instance._thread.join(timeout=1)
|
||||
assert not instance._thread.is_alive()
|
||||
assert instance._loop.is_closed()
|
||||
finally:
|
||||
_cleanup(instance)
|
||||
Reference in New Issue
Block a user