Spaces:
Sleeping
Sleeping
File size: 2,914 Bytes
0aee47a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 |
"""
bilibili_api.utils.AsyncEvent
发布-订阅模式异步事件类支持。
"""
import asyncio
from typing import Callable, Coroutine
class AsyncEvent:
"""
发布-订阅模式异步事件类支持。
特殊事件：__ALL__ 所有事件均触发
"""
def __init__(self):
self.__handlers = {}
self.__ignore_events = []
def add_event_listener(self, name: str, handler: Coroutine) -> None:
"""
注册事件监听器。
Args:
name (str): 事件名。
handler (Coroutine): 回调异步函数。
"""
name = name.upper()
if name not in self.__handlers:
self.__handlers[name] = []
self.__handlers[name].append(handler)
def on(self, event_name: str) -> Callable:
"""
装饰器注册事件监听器。
Args:
event_name (str): 事件名。
"""
def decorator(func: Coroutine):
self.add_event_listener(event_name, func)
return func
return decorator
def remove_all_event_listener(self) -> None:
"""
移除所有事件监听函数
"""
self.__handlers = {}
def remove_event_listener(self, name: str, handler: Coroutine) -> bool:
"""
移除事件监听函数。
Args:
name (str): 事件名。
handler (Coroutine): 要移除的函数。
Returns:
bool, 是否移除成功。
"""
name = name.upper()
if name in self.__handlers:
if handler in self.__handlers[name]:
self.__handlers[name].remove(handler)
return True
return False
def ignore_event(self, name: str) -> None:
"""
忽略指定事件
Args:
name (str): 事件名。
"""
name = name.upper()
self.__ignore_events.append(name)
def remove_ignore_events(self) -> None:
"""
移除所有忽略事件
"""
self.__ignore_events = []
def dispatch(self, name: str, *args, **kwargs) -> None:
"""
异步发布事件。
Args:
name (str): 事件名。
*args, **kwargs: 要传递给函数的参数。
"""
if len(args) == 0 and len(kwargs.keys()) == 0:
args = [{}]
if name.upper() in self.__ignore_events:
return
name = name.upper()
if name in self.__handlers:
for coroutine in self.__handlers[name]:
asyncio.create_task(coroutine(*args, **kwargs))
if name != "__ALL__":
kwargs.update({"name": name, "data": args})
self.dispatch("__ALL__", kwargs)
|