made BSCache.run() a generator; moved dealing with the queue completely into ProdThread; allowed setting the maxsize of the queue from ProdThread()

bstrd/prodthread.py

@@ -1,7 +1,7 @@
 import signal
 
 from threading import Thread, Event
-from queue import Queue
+from queue import Queue, Full
 
 
 def prepend_signal(sig, func):
@@ -17,16 +17,30 @@ def prepend_signal(sig, func):
 
 class ProdThread:
     """
-    provided func has to accept two arguments:
-    - queue = queue.Queue()
-    - running = threading.Event()
+    Upon call of start(),
+    the provided func will be executed with the argument
+    running = threading.Event()
+    in a separate thread.
+
+    The result is expected to be iterable
+    and the yielded values will be filled into a queue.
+
+    If the queue is full, the values will be dropped.
+
+    The oldest entry can be retrieved/removed via get().
+
+    The iterator should obey the state of the running Event.
+
+    Calling stop() clears the running Event and joins the thread.
     """
 
-    def __init__(self, func):
+    def __init__(self, func, maxsize=0):
         self.func = func
 
         self.thread = None
-        self.queue = Queue()
+        self.queue = Queue(maxsize=maxsize)
+        self.get = self.queue.get
+
         self.running = Event()
 
         prepend_signal(signal.SIGINT, self.stop)
@@ -34,9 +48,15 @@ class ProdThread:
 
     def target(self):
         self.running.set()
-        self.func(self.queue, self.running)
+        gen = self.func(self.running)
+        for data in gen:
+            try:
+                self.queue.put_nowait(data)
+            except Full:
+                pass
         self.running.clear()
 
+
     def start(self):
         if not self.thread:
             self.thread = thread = Thread(target=self.target)