wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

transports(base): allow registering event handlers without decorators

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2024-07-25 11:23:19 -07:00
parent 4d606aefb3
commit 64f5119b08
2 changed files with 10 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/pipecat/transports/base_transport.py
View File

@@ -60,20 +60,20 @@ class BaseTransport(ABC):
def event_handler(self, event_name: str):
def decorator(handler):
self._add_event_handler(event_name, handler)
self.add_event_handler(event_name, handler)
return handler
return decorator
def add_event_handler(self, event_name: str, handler):
if event_name not in self._event_handlers:
raise Exception(f"Event handler {event_name} not registered")
self._event_handlers[event_name].append(handler)
def _register_event_handler(self, event_name: str):
if event_name in self._event_handlers:
raise Exception(f"Event handler {event_name} already registered")
self._event_handlers[event_name] = []
def _add_event_handler(self, event_name: str, handler):
if event_name not in self._event_handlers:
raise Exception(f"Event handler {event_name} not registered")
self._event_handlers[event_name].append(handler)
async def _call_event_handler(self, event_name: str, *args, **kwargs):
try:
for handler in self._event_handlers[event_name]: