BaseObject: run each handler for the same event in a separate task

CHANGELOG.md

@@ -65,6 +65,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed an issue where multiple handlers for an event would not run in parallel.
+
 - Fixed `DailyTransport.sip_call_transfer()` to automatically use the session
   ID from the `on_dialin_connected` event, when not explicitly provided. Now
   supports cold transfers (from incoming dial-in calls) by automatically

src/pipecat/utils/base_object.py

@@ -154,35 +154,40 @@ class BaseObject(ABC):
 
         event_handler = self._event_handlers[event_name]
 
-        if event_handler.is_sync:
+        for handler in event_handler.handlers:
-            # Just run the handler.
+            if event_handler.is_sync:
-            await self._run_handler(event_handler, *args, **kwargs)
+                # Just run the handler.
-        else:
+                await self._run_handler(event_handler.name, handler, *args, **kwargs)
-            # Create the task.
+            else:
-            task = asyncio.create_task(self._run_handler(event_handler, *args, **kwargs))
+                # Create the task. Note that this is a task per each function
+                # handler. Users can register to an event handler multiple
+                # times.
+                task = asyncio.create_task(
+                    self._run_handler(event_handler.name, handler, *args, **kwargs)
+                )
 
-            # Add it to our list of event tasks.
+                # Add it to our list of event tasks.
-            self._event_tasks.add((event_name, task))
+                self._event_tasks.add((event_name, task))
 
-            # Remove the task from the event tasks list when the task completes.
+                # Remove the task from the event tasks list when the task completes.
-            task.add_done_callback(self._event_task_finished)
+                task.add_done_callback(self._event_task_finished)
 
-    async def _run_handler(self, event_handler: EventHandler, *args, **kwargs):
+    async def _run_handler(self, event_name: str, handler, *args, **kwargs):
         """Execute all handlers for an event.
 
         Args:
-            event_handler: The event handler to run.
+            event_name: The event name for this handler.
+            handler: The handler function to run.
             *args: Positional arguments to pass to handlers.
             **kwargs: Keyword arguments to pass to handlers.
         """
         try:
-            for handler in event_handler.handlers:
+            if inspect.iscoroutinefunction(handler):
-                if inspect.iscoroutinefunction(handler):
+                await handler(self, *args, **kwargs)
-                    await handler(self, *args, **kwargs)
+            else:
-                else:
+                handler(self, *args, **kwargs)
-                    handler(self, *args, **kwargs)
         except Exception as e:
-            logger.exception(f"Exception in event handler {event_handler.name}: {e}")
+            logger.exception(f"Exception in event handler {event_name}: {e}")
 
     def _event_task_finished(self, task: asyncio.Task):
         """Clean up completed event handler tasks.