sse(stream): fix stream payload and frequency (#52)

* fix: invalid json for sse payload

* fix: only stream a message if environment changes

Author: gagantrivedi

src/sse.py

@@ -1,4 +1,5 @@
 import asyncio
+import json
 from datetime import datetime
 from functools import lru_cache
 from typing import Optional
@@ -65,9 +66,15 @@ async def stream_environment_changes(
     request: Request, environment_key: str, settings: Settings = Depends(get_settings)
 ):
     started_at = datetime.now()
+    last_updated_at = None
 
     async def get_updated_at() -> Optional[int]:
+        nonlocal last_updated_at
         updated_at = redis_connection.get(environment_key)
+        if last_updated_at == updated_at:
+            return None
+
+        last_updated_at = updated_at
         return updated_at
 
     async def event_generator():
@@ -84,7 +91,7 @@ async def event_generator():
             if updated_at := await get_updated_at():
                 yield {
                     "event": "environment_updated",
-                    "data": {"updated_at": float(updated_at)},
+                    "data": json.dumps({"updated_at": float(updated_at)}),
                     "retry": settings.retry_timeout,
                 }
             await asyncio.sleep(settings.stream_delay)

tests/test_sse.py

@@ -13,7 +13,7 @@
 
 def get_settings_override():
     return Settings(
-        max_stream_age=3, stream_delay=1, sse_authentication_token=auth_token
+        max_stream_age=3, stream_delay=0.2, sse_authentication_token=auth_token
     )
 
 
@@ -88,6 +88,7 @@ async def test_stream_changes(client):
         await asyncio.sleep(1)
 
         second_last_updated_at = datetime.now()
+
         # Next, let's update the environment once again
         await ac.post(
             f"/sse/environments/{environment_key}/queue-change",
@@ -97,10 +98,10 @@ async def test_stream_changes(client):
         # Finally, let's wait for the stream to finish
         response = await stream_response_task
 
-        # Then
+        # Then - we only got two messages
         expected_response = (
-            "event: environment_updated\r\ndata: {'updated_at': %0.6f}\r\nretry: 15000\r\n\r\nevent:"
-            " environment_updated\r\ndata: {'updated_at': %0.6f}\r\nretry: 15000\r\n\r\n"
+            'event: environment_updated\r\ndata: {"updated_at": %0.6f}\r\nretry: 15000\r\n\r\nevent:'
+            ' environment_updated\r\ndata: {"updated_at": %0.6f}\r\nretry: 15000\r\n\r\n'
             % (first_last_updated_at.timestamp(), second_last_updated_at.timestamp())
         )
     assert response.status_code == 200