blob: 77fff722a41b475b493a9fd259754aa5d24e97f3 (
plain) (
blame)
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
|
import asyncio
from types import NoneType
from typing import Coroutine
class EventEmitter():
def __init__(self):
self.listeners = {}
def add_listener(self, event_name: str, func: Coroutine):
if not self.listeners.get(event_name, None):
self.listeners[event_name] = {func}
else:
self.listeners[event_name].add(func)
def remove_listener(self, event_name: str, func: Coroutine):
self.listeners[event_name].remove(func)
if len(self.listeners[event_name]) == 0:
del self.listeners[event_name]
def emit(self, event_name: str, args_required=False, *args, **kwargs):
listeners = self.listeners.get(event_name, [])
for func in listeners:
if args_required:
if len(args) == 0:
raise TypeError('event registered must have arguments')
else:
asyncio.create_task(func(*args, **kwargs))
else:
asyncio.create_task(func(*args, **kwargs))
|