Improved shutdown dispatch operations

Author: ritchiecarroll

src/sttp/transport/datasubscriber.py

@@ -227,11 +227,11 @@ def dispose(self):
         """
 
         self._disposing = True
-        self._connector.cancel()
+        self._connector.dispose()
         self._disconnect(True, False)
 
-        # Allow a moment for connection terminated event to complete
-        Event().wait(0.01)  # 10ms
+        # Wait for connection terminated event to complete
+        self._threadpool.shutdown(wait=True)
 
     @property
     def connected(self) -> bool:
@@ -574,15 +574,16 @@ def _run_disconnectthread(self, autoreconnecting: bool):  # sourcery skip: extra
     # by the peer. Additionally, this allows the user to automatically reconnect in their
     # callback function without having to spawn their own separate thread.
     def _dispatch_connectionterminated(self):
-        self._connection_terminationthread_mutex.acquire()
-
-        if self._connection_terminationthread is not None:
-            return
+        try:
+            self._connection_terminationthread_mutex.acquire()
 
-        self._connection_terminationthread = Thread(target=self._handle_connectionterminated, name="ConnectionTerminationThread")
-        self._connection_terminationthread.start()
+            if self._connection_terminationthread is not None:
+                return
 
-        self._connection_terminationthread_mutex.release()
+            self._connection_terminationthread = Thread(target=self._handle_connectionterminated, name="ConnectionTerminationThread")
+            self._connection_terminationthread.start()
+        finally:
+            self._connection_terminationthread_mutex.release()
 
     def _handle_connectionterminated(self):
         self._disconnect(False, True)

src/sttp/transport/subscriberconnector.py

@@ -110,6 +110,14 @@ def __init__(self,
 
         self._threadpool = ThreadPoolExecutor(thread_name_prefix="SC-PoolThread")
 
+    def dispose(self):
+        """
+        Cleanly shuts down a `DataConnector` that is no longer being used, e.g., during a normal application exit.
+        """
+
+        self.cancel()
+        self._threadpool.shutdown(wait=False)
+
     def _autoreconnect(self, ds: DataSubscriber):
         if self._cancel or ds.disposing:
             return