Added unregister_event and test for smart events

This commit is contained in:
Mattia Giambirtone 2023-06-02 11:01:24 +02:00
parent e7cb6a72f5
commit faccd4146d
Signed by: nocturn9x
GPG Key ID: 8270F9F467971E59
2 changed files with 34 additions and 1 deletions

View File

@ -384,12 +384,28 @@ async def emit(evt: str, *args, **kwargs):
def register_event(evt: str, func: Callable[[Any, Any], Coroutine[Any, Any, Any]]):
"""
Register the given async function for the given event name
Register the given async function for the given event name.
Note that if the given async function is already registered
for the chosen event, it will be called once for each time
this function is called once the associated event is fired
"""
_events[evt].append(func)
def unregister_event(evt: str, func: Callable[[Any, Any], Coroutine[Any, Any, Any]]):
"""
Unregisters the given async function from the given event.
Nothing happens if the given event or async functions are
not registered yet
"""
try:
_events[evt].remove(func)
except IndexError:
pass
def on_event(evt: str):
"""
Convenience decorator to

17
tests/smart_events.py Normal file
View File

@ -0,0 +1,17 @@
import structio
@structio.on_event("on_message")
async def test(evt, *args, **kwargs):
print(f"[test] New event {evt!r} with arguments: {args}, {kwargs}")
# Simulate some work
await structio.sleep(1)
async def main():
print(f"[main] Firing two events, exiting in two seconds")
await structio.emit("on_message", 1, 2, 3)
await structio.emit("on_message", 1, 2, 4)
structio.run(main)