streaming logic fix

Author: Mehak Bindra

packages/apps/src/microsoft/teams/apps/http_stream.py

@@ -104,9 +104,6 @@ def emit(self, activity: Union[MessageActivityInput, TypingActivityInput, str])
         Args:
             activity: The activity to emit.
         """
-        if self._timeout is not None:
-            self._timeout.cancel()
-            self._timeout = None
 
         if isinstance(activity, str):
             activity = MessageActivityInput(text=activity, type="message")
@@ -115,7 +112,9 @@ def emit(self, activity: Union[MessageActivityInput, TypingActivityInput, str])
         # Clear the queue empty event since we just added an item
         self._queue_empty_event.clear()
 
-        self._timeout = Timeout(0.5, self._flush)
+        if not self._timeout:
+            loop = asyncio.get_running_loop()
+            loop.create_task(self._flush())
 
     def update(self, text: str) -> None:
         """
@@ -171,10 +170,15 @@ async def _flush(self) -> None:
         Flush the current activity queue.
         """
         # If there are no items in the queue, nothing to flush
-        async with self._lock:
-            if not self._queue:
-                return
+        if self._lock.locked():
+            return
+
+        await self._lock.acquire()
 
+        if not self._queue:
+            return
+
+        try:
             if self._timeout is not None:
                 self._timeout.cancel()
                 self._timeout = None
@@ -224,6 +228,10 @@ async def _flush(self) -> None:
             if self._queue and not self._timeout:
                 self._timeout = Timeout(0.5, self._flush)
 
+        finally:
+            # Reset flushing flag so future emits can trigger another flush
+            self._lock.release()
+
     async def _send_activity(self, to_send: TypingActivityInput):
         """
         Send an activity to the Teams conversation with the ID.

packages/apps/src/microsoft/teams/apps/utils/timer.py

@@ -23,7 +23,7 @@ def __init__(self, delay: float, callback: TimerCallback) -> None:
         self._handle: Optional[asyncio.TimerHandle] = None
         self._cancelled: bool = False
 
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         self._handle = loop.call_later(delay, self._run)
 
     def _run(self) -> None:

packages/apps/tests/test_http_stream.py

@@ -5,7 +5,6 @@
 # pyright: basic
 
 import asyncio
-from datetime import datetime, timedelta
 from unittest.mock import MagicMock
 
 import pytest
@@ -35,13 +34,12 @@ def mock_api_client(self):
         client.conversations = mock_conversations
 
         client.send_call_count = 0
-        client.send_times = []
         client.sent_activities = []
 
         async def mock_send(activity):
             client.send_call_count += 1
-            client.send_times.append(datetime.now())
             client.sent_activities.append(activity)
+            await asyncio.sleep(0.05)  # Simulate network delay
             return SentActivity(id=f"test-id-{client.send_call_count}", activity_params=activity)
 
         client.conversations.activities().create = mock_send
@@ -63,37 +61,45 @@ def http_stream(self, mock_api_client, conversation_reference, mock_logger):
         return HttpStream(mock_api_client, conversation_reference, mock_logger)
 
     @pytest.mark.asyncio
-    async def test_stream_emit_message_flushes_after_500ms(self, mock_api_client, conversation_reference, mock_logger):
-        """Test that messages are flushed after 500ms timeout."""
+    async def test_stream_emit_message_flushes_immediately(self, mock_api_client, conversation_reference, mock_logger):
+        """Test that messages are flushed immediately."""
 
         stream = HttpStream(mock_api_client, conversation_reference, mock_logger)
-        start_time = datetime.now()
-
         stream.emit("Test message")
-        await asyncio.sleep(0.6)  # Wait longer than 500ms timeout
-
-        assert mock_api_client.send_call_count > 0, "Should have sent at least one message"
-        assert any(t >= start_time + timedelta(milliseconds=450) for t in mock_api_client.send_times), (
-            "Should have waited approximately 500ms before sending"
-        )
+        await asyncio.sleep(0.07)  # Wait for the flush task to complete
+        assert mock_api_client.send_call_count == 1
 
     @pytest.mark.asyncio
-    async def test_stream_multiple_emits_restarts_timer(self, mock_api_client, conversation_reference, mock_logger):
+    async def test_stream_multiple_emits_timer_check(self, mock_api_client, conversation_reference, mock_logger):
         """Test that multiple emits reset the timer."""
 
         stream = HttpStream(mock_api_client, conversation_reference, mock_logger)
 
         stream.emit("First message")
-        await asyncio.sleep(0.3)  # Wait less than 500ms
-
-        stream.emit("Second message")  # This should reset the timer
-        await asyncio.sleep(0.3)  # Still less than 500ms from second emit
-        assert mock_api_client.send_call_count == 0, "Should not have sent yet"
-        await asyncio.sleep(0.3)  # Now over 500ms from second emit
-        assert mock_api_client.send_call_count > 0, "Should have sent messages after timer expired"
+        stream.emit("Second message")
+        stream.emit("Third message")
+        stream.emit("Fourth message")
+        stream.emit("Fifth message")
+        stream.emit("Sixth message")
+        stream.emit("Seventh message")
+        stream.emit("Eighth message")
+        stream.emit("Ninth message")
+        stream.emit("Tenth message")
+        stream.emit("Eleventh message")
+        stream.emit("Twelfth message")
+
+        await asyncio.sleep(0.07)  # Wait for the flush task to complete
+        assert mock_api_client.send_call_count == 1  # First message should trigger flush immediately
+
+        stream.emit("Thirteenth message")
+        await asyncio.sleep(0.3)  # Less than 500ms from first flush
+        assert mock_api_client.send_call_count == 1, "No new flush should have occurred yet"
+
+        await asyncio.sleep(0.3)  # Now exceed 500ms from last emit
+        assert mock_api_client.send_call_count == 2, "Second flush should have occurred"
 
     @pytest.mark.asyncio
-    async def test_stream_send_timeout_handled_gracefully(self, mock_api_client, conversation_reference, mock_logger):
+    async def test_stream_error_handled_gracefully(self, mock_api_client, conversation_reference, mock_logger):
         """Test that send timeouts are handled gracefully with retries."""
         call_count = 0
 
@@ -104,14 +110,15 @@ async def mock_send_with_timeout(activity):
                 raise TimeoutError("Operation timed out")
 
             # Succeed on second attempt
+            await asyncio.sleep(0.05)  # Simulate delay
             return SentActivity(id=f"success-after-timeout-{call_count}", activity_params=activity)
 
         mock_api_client.conversations.activities().create = mock_send_with_timeout
 
         stream = HttpStream(mock_api_client, conversation_reference, mock_logger)
 
         stream.emit("Test message with timeout")
-        await asyncio.sleep(0.8)  # Wait for flush and retries
+        await asyncio.sleep(0.6)  # Wait for flush and 1 retry to complete
 
         result = await stream.close()
 
@@ -148,7 +155,7 @@ async def test_stream_update_status_sends_typing_activity(
         stream = HttpStream(mock_api_client, conversation_reference, mock_logger)
 
         stream.update("Thinking...")
-        await asyncio.sleep(0.6)  # Wait for the flush task to complete
+        await asyncio.sleep(0.07)  # Wait for the flush task to complete
 
         assert stream.count > 0 or len(mock_api_client.sent_activities) > 0, "Should have processed the update"
         assert stream.sequence >= 2, "Should increment sequence after sending"
@@ -167,10 +174,9 @@ async def test_stream_sequence_of_update_and_emit(self, mock_api_client, convers
         stream = HttpStream(mock_api_client, conversation_reference, mock_logger)
 
         stream.update("Preparing response...")
-        await asyncio.sleep(0.6)
-
         stream.emit("Final response message")
-        await asyncio.sleep(0.6)
+
+        await asyncio.sleep(0.5)  # Wait for the flush task to complete
 
         assert len(mock_api_client.sent_activities) >= 2, "Should have sent typing activity and message"
 
@@ -186,3 +192,41 @@ async def test_stream_sequence_of_update_and_emit(self, mock_api_client, convers
 
         # Sequence numbers should have increased
         assert stream.sequence >= 3, "Sequence should increment for both update and emit"
+
+    @pytest.mark.asyncio
+    async def test_stream_concurrent_emits_do_not_flush_simultaneously(
+        self, mock_api_client, conversation_reference, mock_logger
+    ):
+        """
+        Test that multiple concurrent emits do not allow simultaneous flush execution.
+        """
+        concurrent_entries = 0
+        max_concurrent_entries = 0
+        lock = asyncio.Lock()
+
+        async def mock_send(activity):
+            nonlocal concurrent_entries, max_concurrent_entries
+            async with lock:
+                concurrent_entries += 1
+                max_concurrent_entries = max(max_concurrent_entries, concurrent_entries)
+            await asyncio.sleep(0.05)  # simulate delay in sending
+            async with lock:
+                concurrent_entries -= 1
+            return activity
+
+        mock_api_client.conversations.activities().create = mock_send
+
+        stream = HttpStream(mock_api_client, conversation_reference, mock_logger)
+
+        # Schedule multiple emits concurrently
+        async def emit_task():
+            stream.emit("Concurrent message")
+
+        tasks = [asyncio.create_task(emit_task()) for _ in range(10)]
+        await asyncio.gather(*tasks)
+
+        # Wait for flushes to complete
+        await asyncio.sleep(0.07)
+
+        # Only one flush should have entered the critical section at a time
+        assert max_concurrent_entries == 1, f"Flush entered concurrently {max_concurrent_entries} times, expected 1"