use get_event_loop() and move event handlers to BaseTransport

src/pipecat/transports/base_transport.py

@@ -4,7 +4,12 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
+import asyncio
+import inspect
+import types
+
 from abc import ABC, abstractmethod
+from functools import partial
 
 from pydantic import ConfigDict
 from pydantic.main import BaseModel
@@ -12,6 +17,8 @@ from pydantic.main import BaseModel
 from pipecat.processors.frame_processor import FrameProcessor
 from pipecat.vad.vad_analyzer import VADAnalyzer
 
+from loguru import logger
+
 
 class TransportParams(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True)
@@ -36,6 +43,10 @@ class TransportParams(BaseModel):
 
 class BaseTransport(ABC):
 
+    def __init__(self, loop: asyncio.AbstractEventLoop):
+        self._loop = loop
+        self._event_handlers: dict = {}
+
     @abstractmethod
     def input(self) -> FrameProcessor:
         raise NotImplementedError
@@ -43,3 +54,42 @@ class BaseTransport(ABC):
     @abstractmethod
     def output(self) -> FrameProcessor:
         raise NotImplementedError
+
+    def event_handler(self, event_name: str):
+        def decorator(handler):
+            self._add_event_handler(event_name, handler)
+            return handler
+        return decorator
+
+    def _register_event_handler(self, event_name: str):
+        methods = inspect.getmembers(self, predicate=inspect.ismethod)
+        if event_name not in [method[0] for method in methods]:
+            raise Exception(f"Event handler {event_name} not found")
+
+        self._event_handlers[event_name] = [getattr(self, event_name)]
+
+        patch_method = types.MethodType(partial(self._patch_method, event_name), self)
+        setattr(self, event_name, patch_method)
+
+    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(types.MethodType(handler, self))
+
+    def _patch_method(self, event_name, *args, **kwargs):
+        try:
+            for handler in self._event_handlers[event_name]:
+                if inspect.iscoroutinefunction(handler):
+                    # Beware, if handler() calls another event handler it
+                    # will deadlock. You shouldn't do that anyways.
+                    future = asyncio.run_coroutine_threadsafe(
+                        handler(*args[1:], **kwargs), self._loop)
+
+                    # wait for the coroutine to finish. This will also
+                    # raise any exceptions raised by the coroutine.
+                    future.result()
+                else:
+                    handler(*args[1:], **kwargs)
+        except Exception as e:
+            logger.error(f"Exception in event handler {event_name}: {e}")
+            raise e