Wait for All Callbacks in _emit to Finish, Add More Thread Tracing

deepgram/clients/live/v1/async_client.py

@@ -6,6 +6,7 @@
 import logging
 from typing import Dict, Union, Optional, cast, Any
 from datetime import datetime
+import threading
 
 import websockets
 from websockets.client import WebSocketClientProtocol
@@ -157,6 +158,11 @@ async def start(
             )
             self._exit_event.clear()
 
+            # debug the threads
+            for thread in threading.enumerate():
+                self._logger.debug("after running thread: %s", thread.name)
+            self._logger.debug("number of active threads: %s", threading.active_count())
+
             # listen thread
             self._listen_thread = asyncio.create_task(self._listening())
 
@@ -174,6 +180,11 @@ async def start(
             else:
                 self._logger.notice("autoflush is disabled")
 
+            # debug the threads
+            for thread in threading.enumerate():
+                self._logger.debug("after running thread: %s", thread.name)
+            self._logger.debug("number of active threads: %s", threading.active_count())
+
             # push open event
             await self._emit(
                 LiveTranscriptionEvents(LiveTranscriptionEvents.Open),
@@ -217,12 +228,28 @@ async def _emit(self, event: LiveTranscriptionEvents, *args, **kwargs) -> None:
         """
         Emits events to the registered event handlers.
         """
+        self._logger.debug("AsyncLiveClient._emit ENTER")
         self._logger.debug("callback handlers for: %s", event)
+
+        # debug the threads
+        for thread in threading.enumerate():
+            self._logger.debug("after running thread: %s", thread.name)
+        self._logger.debug("number of active threads: %s", threading.active_count())
+
+        tasks = []
         for handler in self._event_handlers[event]:
-            if asyncio.iscoroutinefunction(handler):
-                await handler(self, *args, **kwargs)
-            else:
-                asyncio.create_task(handler(self, *args, **kwargs))
+            tasks.append(asyncio.create_task(handler(self, *args, **kwargs)))
+
+        if len(tasks) > 0:
+            self._logger.debug("waiting for tasks to finish...")
+            await asyncio.gather(*filter(None, tasks), return_exceptions=True)
+
+        # debug the threads
+        for thread in threading.enumerate():
+            self._logger.debug("after running thread: %s", thread.name)
+        self._logger.debug("number of active threads: %s", threading.active_count())
+
+        self._logger.debug("AsyncLiveClient._emit LEAVE")
 
     # pylint: disable=too-many-return-statements,too-many-statements,too-many-locals,too-many-branches
     async def _listening(self) -> None:
@@ -796,6 +823,11 @@ async def finish(self) -> bool:
         self._logger.verbose("cancelling tasks...")
         try:
             # Before cancelling, check if the tasks were created
+            # debug the threads
+            for thread in threading.enumerate():
+                self._logger.debug("before running thread: %s", thread.name)
+            self._logger.debug("number of active threads: %s", threading.active_count())
+
             tasks = []
             if self._keep_alive_thread is not None:
                 self._keep_alive_thread.cancel()
@@ -816,6 +848,11 @@ async def finish(self) -> bool:
             await asyncio.gather(*filter(None, tasks), return_exceptions=True)
             self._logger.notice("threads joined")
 
+            # debug the threads
+            for thread in threading.enumerate():
+                self._logger.debug("after running thread: %s", thread.name)
+            self._logger.debug("number of active threads: %s", threading.active_count())
+
             self._logger.notice("finish succeeded")
             self._logger.spam("AsyncLiveClient.finish LEAVE")
             return True

deepgram/clients/live/v1/client.py

@@ -2,11 +2,11 @@
 # Use of this source code is governed by a MIT license that can be found in the LICENSE file.
 # SPDX-License-Identifier: MIT
 import json
-import threading
 import time
 import logging
 from typing import Dict, Union, Optional, cast, Any
 from datetime import datetime
+import threading
 
 from websockets.sync.client import connect, ClientConnection
 import websockets
@@ -160,6 +160,11 @@ def start(
             self._socket = connect(url_with_params, additional_headers=combined_headers)
             self._exit_event.clear()
 
+            # debug the threads
+            for thread in threading.enumerate():
+                self._logger.debug("after running thread: %s", thread.name)
+            self._logger.debug("number of active threads: %s", threading.active_count())
+
             # listening thread
             self._listen_thread = threading.Thread(target=self._listening)
             self._listen_thread.start()
@@ -180,6 +185,11 @@ def start(
             else:
                 self._logger.notice("autoflush is disabled")
 
+            # debug the threads
+            for thread in threading.enumerate():
+                self._logger.debug("after running thread: %s", thread.name)
+            self._logger.debug("number of active threads: %s", threading.active_count())
+
             # push open event
             self._emit(
                 LiveTranscriptionEvents(LiveTranscriptionEvents.Open),
@@ -780,6 +790,11 @@ def finish(self) -> bool:
         """
         self._logger.spam("LiveClient.finish ENTER")
 
+        # debug the threads
+        for thread in threading.enumerate():
+            self._logger.debug("before running thread: %s", thread.name)
+        self._logger.debug("number of active threads: %s", threading.active_count())
+
         # signal exit
         self._signal_exit()
 
@@ -800,6 +815,11 @@ def finish(self) -> bool:
             self._listen_thread = None
         self._logger.notice("listening thread joined")
 
+        # debug the threads
+        for thread in threading.enumerate():
+            self._logger.debug("before running thread: %s", thread.name)
+        self._logger.debug("number of active threads: %s", threading.active_count())
+
         self._logger.notice("finish succeeded")
         self._logger.spam("LiveClient.finish LEAVE")
         return True