event_generator fix

apps/api/main.py

@@ -127,27 +127,19 @@ async def redis_state_listener():
         
         logger.info("Redis state listener connected")
         
-        while True:
-            try:
-                message = await asyncio.wait_for(
-                    pubsub.get_message(ignore_subscribe_messages=True),
-                    timeout=1.0
-                )
-                
-                if message and message["type"] == "message":
-                    data = message["data"]
-                    try:
-                        state_data = json.loads(data)
-                        if state_data.get("type") == "state" and state_data.get("device_id"):
-                            device_id = state_data["device_id"]
-                            payload = state_data.get("payload", {})
-                            device_states[device_id] = payload
-                            logger.debug(f"Updated state cache for {device_id}: {payload}")
-                    except Exception as e:
-                        logger.warning(f"Failed to parse state data: {e}")
-                        
-            except asyncio.TimeoutError:
-                pass  # No message, continue
+        # listen() blocks async and waits for messages - prevents busy loop
+        async for message in pubsub.listen():
+            if message["type"] == "message":
+                data = message["data"]
+                try:
+                    state_data = json.loads(data)
+                    if state_data.get("type") == "state" and state_data.get("device_id"):
+                        device_id = state_data["device_id"]
+                        payload = state_data.get("payload", {})
+                        device_states[device_id] = payload
+                        logger.debug(f"Updated state cache for {device_id}: {payload}")
+                except Exception as e:
+                    logger.warning(f"Failed to parse state data: {e}")
                 
     except asyncio.CancelledError:
         logger.info("Redis state listener cancelled")
@@ -567,25 +559,31 @@ async def event_generator(request: Request) -> AsyncGenerator[str, None]:
             redis_client = None
             pubsub = None
         
-        # Heartbeat tracking
-        last_heartbeat = asyncio.get_event_loop().time()
         heartbeat_interval = 15  # Safari-friendly: shorter interval
         
+        # Use listen() iterator for blocking reads with heartbeat timeout
+        if pubsub:
+            listener = pubsub.listen()
+        else:
+            listener = None
+        
         while True:
             # Check if client disconnected
             if await request.is_disconnected():
                 logger.info("SSE client disconnected")
                 break
             
-            # Try to get message from Redis (if available)
-            if pubsub:
+            # Try to get message from Redis with timeout for heartbeat
+            if listener:
                 try:
+                    # Wait for message with heartbeat timeout
+                    # If no message arrives within timeout, send heartbeat
                     message = await asyncio.wait_for(
-                        pubsub.get_message(ignore_subscribe_messages=True),
-                        timeout=0.1
+                        anext(listener),
+                        timeout=heartbeat_interval
                     )
                     
-                    if message and message["type"] == "message":
+                    if message["type"] == "message":
                         data = message["data"]
                         logger.debug(f"Sending SSE message: {data[:100]}...")
                         
@@ -598,24 +596,21 @@ async def event_generator(request: Request) -> AsyncGenerator[str, None]:
                             logger.warning(f"Failed to parse state data for cache: {e}")
                         
                         yield f"event: message\ndata: {data}\n\n"
-                        last_heartbeat = asyncio.get_event_loop().time()
-                        continue  # Skip sleep, check for more messages immediately
                         
                 except asyncio.TimeoutError:
-                    pass  # No message, continue to heartbeat check
+                    # No message within heartbeat interval - send heartbeat
+                    yield ": ping\n\n"
+                except StopAsyncIteration:
+                    logger.warning("Redis listener stopped")
+                    break
                 except Exception as e:
                     logger.error(f"Redis error: {e}")
-                    # Continue with heartbeats even if Redis fails
-            
-            # Sleep briefly to avoid busy loop
-            await asyncio.sleep(0.1)
-            
-            # Send heartbeat if interval elapsed
-            current_time = asyncio.get_event_loop().time()
-            if current_time - last_heartbeat >= heartbeat_interval:
-                # Comment-style ping (Safari-compatible, no event type)
+                    # Continue with heartbeat-only mode
+                    listener = None
+            else:
+                # Heartbeat-only mode (no Redis)
+                await asyncio.sleep(heartbeat_interval)
                 yield ": ping\n\n"
-                last_heartbeat = current_time
             
     except asyncio.CancelledError:
         logger.info("SSE connection cancelled by client")