From 01b60671dbaef8af046e792697525619dafae7cf Mon Sep 17 00:00:00 2001 From: Wolfgang Hottgenroth Date: Sun, 9 Nov 2025 20:05:35 +0100 Subject: [PATCH] sse iphone fix 1 --- apps/api/main.py | 107 ++++++++++++++++++++++++++++++++++------------- 1 file changed, 79 insertions(+), 28 deletions(-) diff --git a/apps/api/main.py b/apps/api/main.py index 6acbf58..b3c0664 100644 --- a/apps/api/main.py +++ b/apps/api/main.py @@ -287,7 +287,13 @@ async def set_device(device_id: str, request: SetDeviceRequest) -> dict[str, str async def event_generator(request: Request) -> AsyncGenerator[str, None]: - """Generate SSE events from Redis Pub/Sub. + """Generate SSE events from Redis Pub/Sub with Safari compatibility. + + Safari-compatible features: + - Immediate retry hint on connection + - Regular heartbeats every 15s (comment-only, no data) + - Proper flushing after each yield + - Graceful disconnect handling Args: request: FastAPI request object for disconnect detection @@ -295,17 +301,28 @@ async def event_generator(request: Request) -> AsyncGenerator[str, None]: Yields: str: SSE formatted event strings """ - redis_url, redis_channel = get_redis_settings() - redis_client = await aioredis.from_url(redis_url, decode_responses=True) - pubsub = redis_client.pubsub() + redis_client = None + pubsub = None try: - await pubsub.subscribe(redis_channel) - logger.info(f"SSE client connected, subscribed to {redis_channel}") + # Send retry hint immediately for EventSource reconnect behavior + yield "retry: 2500\n\n" - # Create heartbeat tracking + # Try to connect to Redis + redis_url, redis_channel = get_redis_settings() + try: + redis_client = await aioredis.from_url(redis_url, decode_responses=True) + pubsub = redis_client.pubsub() + await pubsub.subscribe(redis_channel) + logger.info(f"SSE client connected, subscribed to {redis_channel}") + except Exception as e: + logger.warning(f"Redis unavailable, running in heartbeat-only mode: {e}") + redis_client = None + pubsub = None + + # Heartbeat tracking last_heartbeat = asyncio.get_event_loop().time() - heartbeat_interval = 25 + heartbeat_interval = 15 # Safari-friendly: shorter interval while True: # Check if client disconnected @@ -313,29 +330,58 @@ async def event_generator(request: Request) -> AsyncGenerator[str, None]: logger.info("SSE client disconnected") break - # Try to get message (non-blocking) - message = await pubsub.get_message(ignore_subscribe_messages=True, timeout=0.1) + # Try to get message from Redis (if available) + if pubsub: + try: + message = await asyncio.wait_for( + pubsub.get_message(ignore_subscribe_messages=True), + timeout=0.1 + ) + + if message and message["type"] == "message": + data = message["data"] + logger.debug(f"Sending SSE message: {data[:100]}...") + 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 + except Exception as e: + logger.error(f"Redis error: {e}") + # Continue with heartbeats even if Redis fails - # Handle actual data messages - if message and message["type"] == "message": - data = message["data"] - logger.debug(f"Sending SSE message: {data[:100]}...") - yield f"event: message\ndata: {data}\n\n" - last_heartbeat = asyncio.get_event_loop().time() - else: - # No message, sleep a bit to avoid busy loop - await asyncio.sleep(0.1) + # Sleep briefly to avoid busy loop + await asyncio.sleep(0.1) - # Send heartbeat every 25 seconds + # Send heartbeat if interval elapsed current_time = asyncio.get_event_loop().time() if current_time - last_heartbeat >= heartbeat_interval: - yield "event: ping\ndata: heartbeat\n\n" + # Comment-style ping (Safari-compatible, no event type) + yield ": ping\n\n" last_heartbeat = current_time + except asyncio.CancelledError: + logger.info("SSE connection cancelled by client") + raise + except Exception as e: + logger.error(f"SSE error: {e}") + raise finally: - await pubsub.unsubscribe(redis_channel) - await pubsub.aclose() - await redis_client.aclose() + # Cleanup Redis connection + if pubsub: + try: + await pubsub.unsubscribe(redis_channel) + await pubsub.aclose() + except Exception as e: + logger.error(f"Error closing pubsub: {e}") + + if redis_client: + try: + await redis_client.aclose() + except Exception as e: + logger.error(f"Error closing redis: {e}") + logger.info("SSE connection closed") @@ -343,23 +389,28 @@ async def event_generator(request: Request) -> AsyncGenerator[str, None]: async def realtime_events(request: Request) -> StreamingResponse: """Server-Sent Events endpoint for real-time updates. + Safari-compatible SSE implementation: + - Immediate retry hint (2.5s reconnect delay) + - Heartbeat every 15s using comment syntax ": ping" + - Proper Cache-Control headers + - No buffering (nginx compatibility) + - Graceful Redis fallback (heartbeat-only mode) + Args: request: FastAPI request object Returns: - StreamingResponse: SSE stream of Redis messages + StreamingResponse: SSE stream with Redis messages and heartbeats """ return StreamingResponse( event_generator(request), media_type="text/event-stream", headers={ - "Cache-Control": "no-cache", + "Cache-Control": "no-cache, no-transform", "Connection": "keep-alive", "X-Accel-Buffering": "no", # Disable nginx buffering } ) - - return {"message": f"Command sent to {device_id}"} def main() -> None: