From 10523dbbd7df12c1c64465b560f2f96b0570e25a Mon Sep 17 00:00:00 2001
From: Jakob Ketterl <jakob.ketterl@gmx.de>
Date: Fri, 28 Feb 2020 16:13:53 +0100
Subject: [PATCH] use threading to uncouble the queues

---
 csdr/csdr.py | 25 +++++++++++++++++++++++--
 1 file changed, 23 insertions(+), 2 deletions(-)

diff --git a/csdr/csdr.py b/csdr/csdr.py
index 95f6a0d..4ef95f6 100644
--- a/csdr/csdr.py
+++ b/csdr/csdr.py
@@ -110,15 +110,36 @@ class Pipe(object):
 
 class WritingPipe(Pipe):
     def __init__(self, path, encoding=None):
+        self.queue = []
+        self.queueLock = threading.Lock()
         super().__init__(path, "w", encoding=encoding)
+        self.open()
+
+    def open_and_dequeue(self):
+        super().open()
+        with self.queueLock:
+            for i in self.queue:
+                self.file.write(i)
+            self.file.flush()
+            self.queue = None
+
+    def open(self):
+        threading.Thread(target=self.open_and_dequeue).start()
 
     def write(self, data):
         if self.file is None:
-            self.open()
-        r =self.file.write(data)
+            with self.queueLock:
+                self.queue.append(data)
+            return
+        r = self.file.write(data)
         self.file.flush()
         return r
 
+    def close(self):
+        if self.file is None:
+            logger.warning("queue %s never successfully opened - thread leak!", self.path)
+        super().close()
+
 
 class ReadingPipe(Pipe):
     def __init__(self, path, encoding=None):