Merge pull request #24 from IthacaDream/fix/pops-empty-deque

fix: pop an empty deque issue

Author: thirdgerb

src/ghoshell_moss/core/helpers/stream.py

@@ -68,6 +68,8 @@ def __iter__(self):
     def __next__(self) -> ItemT:
         if len(self._queue) > 0:
             item = self._queue.popleft()
+            if len(self._queue) == 0:
+                self._added.clear()
             if isinstance(item, Exception):
                 raise item
             elif item is None:
@@ -104,6 +106,8 @@ def __aiter__(self):
     async def __anext__(self) -> ItemT:
         if len(self._queue) > 0:
             item = self._queue.popleft()
+            if len(self._queue) == 0:
+                self._added.clear()
             if isinstance(item, Exception):
                 raise item
             elif item is None:

tests/helpers/test_stream.py

@@ -1,5 +1,6 @@
 import asyncio
 import threading
+import time
 
 from ghoshell_moss.core.helpers.stream import (
     create_thread_safe_stream,
@@ -62,3 +63,31 @@ def sync_receiving():
     t1.join()
     t2.join()
     assert content == done[0]
+
+
+def test_receiver_waits_after_queue_empty_until_new_item_sync():
+    sender, receiver = create_thread_safe_stream(timeout=1.0)
+    consumed: list[str] = []
+
+    def producer():
+        with sender:
+            sender.append("A")  # queue has one item; not completed yet
+            time.sleep(0.1)  # ensure consumer attempts the next() on empty queue
+            sender.append("B")
+            sender.commit()
+
+    def consumer():
+        with receiver:
+            a = next(receiver)
+            consumed.append(a)
+            b = next(receiver)
+            consumed.append(b)
+
+    t1 = threading.Thread(target=producer)
+    t2 = threading.Thread(target=consumer)
+    t1.start()
+    t2.start()
+    t1.join()
+    t2.join()
+
+    assert consumed == ["A", "B"]