fixing event manager so it can clear the queue (vocodedev#441)

* fixing event manager by moving async handle outside to accomodate future annotations

* handling events

* unit tests for events manager

* typo

Author: skirdey

tests/streaming/utils/test_events_manager.py

@@ -0,0 +1,65 @@
+import pytest
+import asyncio
+
+from vocode.streaming.models.events import PhoneCallEndedEvent, EventType
+from vocode.streaming.utils.events_manager import EventsManager
+
+CONVERSATION_ID = "1"
+
+
+@pytest.mark.asyncio
+async def test_initialization():
+    manager = EventsManager()
+    assert manager.subscriptions == set()
+    assert isinstance(manager.queue, asyncio.Queue)
+    assert not manager.active
+
+
+@pytest.mark.asyncio
+async def test_publish_event():
+    event = PhoneCallEndedEvent(
+        conversation_id=CONVERSATION_ID, type=EventType.PHONE_CALL_ENDED
+    )  # Replace with actual Event creation
+    manager = EventsManager([EventType.PHONE_CALL_ENDED])
+    manager.publish_event(event)
+    assert not manager.queue.empty()
+
+
+@pytest.mark.asyncio
+async def test_handle_event_default_implementation():
+    event = PhoneCallEndedEvent(
+        conversation_id=CONVERSATION_ID, type=EventType.PHONE_CALL_ENDED
+    )  # Replace with actual Event creation
+    manager = EventsManager([EventType.PHONE_CALL_ENDED])
+    await manager.handle_event(event)
+
+
+@pytest.mark.asyncio
+async def test_start_and_active_loop():
+    event = PhoneCallEndedEvent(
+        conversation_id=CONVERSATION_ID, type=EventType.PHONE_CALL_ENDED
+    )  # Replace with actual Event creation
+    manager = EventsManager([EventType.PHONE_CALL_ENDED])
+    asyncio.create_task(manager.start())
+    manager.publish_event(event)
+    await asyncio.sleep(0.1)
+    manager.active = False
+
+
+@pytest.mark.asyncio
+async def test_flush_method():
+    event = PhoneCallEndedEvent(
+        conversation_id=CONVERSATION_ID, type=EventType.PHONE_CALL_ENDED
+    )
+    manager = EventsManager([EventType.PHONE_CALL_ENDED])
+    for _ in range(5):
+        manager.publish_event(event)
+    await manager.flush(timeout=2)
+    assert manager.queue.empty()
+
+
+@pytest.mark.asyncio
+async def test_queue_empty_and_timeout():
+    manager = EventsManager([EventType.TRANSCRIPT])
+    await manager.flush(timeout=0)
+    assert manager.queue.empty()

vocode/streaming/utils/events_manager.py

@@ -1,13 +1,19 @@
 from __future__ import annotations
-
 import asyncio
-from typing import List
 
-from vocode.streaming.models.events import Event, EventType
+
+from vocode.streaming.models.events import Event
+
+
+async def flush_event(event):
+    if event:
+        del event
 
 
 class EventsManager:
-    def __init__(self, subscriptions: List[EventType] = []):
+    def __init__(self, subscriptions=None):
+        if subscriptions is None:
+            subscriptions = []
         self.queue: asyncio.Queue[Event] = asyncio.Queue()
         self.subscriptions = set(subscriptions)
         self.active = False
@@ -21,18 +27,19 @@ async def start(self):
         while self.active:
             try:
                 event = await self.queue.get()
+                await self.handle_event(event)
             except asyncio.QueueEmpty:
                 await asyncio.sleep(1)
-            await self.handle_event(event)
 
     async def handle_event(self, event: Event):
-        pass
+        pass  # Default implementation, can be overridden
 
-    async def flush(self):
-        self.active = False
+    async def flush(self, timeout=30):
         while True:
             try:
-                event = self.queue.get_nowait()
-                await self.handle_event(event)
+                event = await asyncio.wait_for(self.queue.get(), timeout)
+                await flush_event(event)
+            except asyncio.TimeoutError:
+                break
             except asyncio.QueueEmpty:
                 break