From b3c82fe0d1e134fbd62c12647fee674a4a378960 Mon Sep 17 00:00:00 2001
From: Fabio Manganiello <fabio@manganiello.tech>
Date: Mon, 4 Sep 2023 02:47:18 +0200
Subject: [PATCH] More resilient termination logic for `CommandStream`.

---
 platypush/commands/_stream.py | 18 ++++++++++++++----
 1 file changed, 14 insertions(+), 4 deletions(-)

diff --git a/platypush/commands/_stream.py b/platypush/commands/_stream.py
index 3ae185d3e..b4b120142 100644
--- a/platypush/commands/_stream.py
+++ b/platypush/commands/_stream.py
@@ -1,4 +1,4 @@
-from multiprocessing import Queue
+from multiprocessing import RLock, Queue
 import os
 from queue import Empty
 import socket
@@ -35,6 +35,7 @@ class CommandStream(ControllableProcess):
         self.path = os.path.abspath(os.path.expanduser(path or self._default_sock_path))
         self._sock: Optional[socket.socket] = None
         self._cmd_queue: Queue["Command"] = Queue()
+        self._close_lock = RLock()
 
     def reset(self):
         if self._sock is not None:
@@ -68,9 +69,18 @@ class CommandStream(ControllableProcess):
         return self
 
     def __exit__(self, *_, **__):
-        self.terminate()
-        self.join()
-        self.close()
+        with self._close_lock:
+            self.terminate()
+
+            try:
+                self.close()
+            except Exception as e:
+                self.logger.warning(str(e))
+
+            try:
+                self.kill()
+            except Exception as e:
+                self.logger.warning(str(e))
 
     def _serve(self, sock: socket.socket):
         """