EventBus的Python实现
- 调用规范
由于设计时考虑到可能会存在多个实例的情况,故使用ContextVar来包装EventBus实例方便使用者调用。
from eventbus import bus, Subscribe class Example: pass class A: def __init__(self): bus.get().register(a) @Subscribe(priority=100, sticky=True) def on_example(self, event: Example): bus.get().cancel_delivery(event) bus.get().unregister(a) a = A() bus.get().post(a)