sse iphone fix 1
This commit is contained in:
@@ -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]:
|
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:
|
Args:
|
||||||
request: FastAPI request object for disconnect detection
|
request: FastAPI request object for disconnect detection
|
||||||
@@ -295,17 +301,28 @@ async def event_generator(request: Request) -> AsyncGenerator[str, None]:
|
|||||||
Yields:
|
Yields:
|
||||||
str: SSE formatted event strings
|
str: SSE formatted event strings
|
||||||
"""
|
"""
|
||||||
redis_url, redis_channel = get_redis_settings()
|
redis_client = None
|
||||||
redis_client = await aioredis.from_url(redis_url, decode_responses=True)
|
pubsub = None
|
||||||
pubsub = redis_client.pubsub()
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
|
# Send retry hint immediately for EventSource reconnect behavior
|
||||||
|
yield "retry: 2500\n\n"
|
||||||
|
|
||||||
|
# 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)
|
await pubsub.subscribe(redis_channel)
|
||||||
logger.info(f"SSE client connected, subscribed to {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
|
||||||
|
|
||||||
# Create heartbeat tracking
|
# Heartbeat tracking
|
||||||
last_heartbeat = asyncio.get_event_loop().time()
|
last_heartbeat = asyncio.get_event_loop().time()
|
||||||
heartbeat_interval = 25
|
heartbeat_interval = 15 # Safari-friendly: shorter interval
|
||||||
|
|
||||||
while True:
|
while True:
|
||||||
# Check if client disconnected
|
# Check if client disconnected
|
||||||
@@ -313,29 +330,58 @@ async def event_generator(request: Request) -> AsyncGenerator[str, None]:
|
|||||||
logger.info("SSE client disconnected")
|
logger.info("SSE client disconnected")
|
||||||
break
|
break
|
||||||
|
|
||||||
# Try to get message (non-blocking)
|
# Try to get message from Redis (if available)
|
||||||
message = await pubsub.get_message(ignore_subscribe_messages=True, timeout=0.1)
|
if pubsub:
|
||||||
|
try:
|
||||||
|
message = await asyncio.wait_for(
|
||||||
|
pubsub.get_message(ignore_subscribe_messages=True),
|
||||||
|
timeout=0.1
|
||||||
|
)
|
||||||
|
|
||||||
# Handle actual data messages
|
|
||||||
if message and message["type"] == "message":
|
if message and message["type"] == "message":
|
||||||
data = message["data"]
|
data = message["data"]
|
||||||
logger.debug(f"Sending SSE message: {data[:100]}...")
|
logger.debug(f"Sending SSE message: {data[:100]}...")
|
||||||
yield f"event: message\ndata: {data}\n\n"
|
yield f"event: message\ndata: {data}\n\n"
|
||||||
last_heartbeat = asyncio.get_event_loop().time()
|
last_heartbeat = asyncio.get_event_loop().time()
|
||||||
else:
|
continue # Skip sleep, check for more messages immediately
|
||||||
# No message, sleep a bit to avoid busy loop
|
|
||||||
|
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
|
||||||
|
|
||||||
|
# Sleep briefly to avoid busy loop
|
||||||
await asyncio.sleep(0.1)
|
await asyncio.sleep(0.1)
|
||||||
|
|
||||||
# Send heartbeat every 25 seconds
|
# Send heartbeat if interval elapsed
|
||||||
current_time = asyncio.get_event_loop().time()
|
current_time = asyncio.get_event_loop().time()
|
||||||
if current_time - last_heartbeat >= heartbeat_interval:
|
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
|
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:
|
finally:
|
||||||
|
# Cleanup Redis connection
|
||||||
|
if pubsub:
|
||||||
|
try:
|
||||||
await pubsub.unsubscribe(redis_channel)
|
await pubsub.unsubscribe(redis_channel)
|
||||||
await pubsub.aclose()
|
await pubsub.aclose()
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"Error closing pubsub: {e}")
|
||||||
|
|
||||||
|
if redis_client:
|
||||||
|
try:
|
||||||
await redis_client.aclose()
|
await redis_client.aclose()
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"Error closing redis: {e}")
|
||||||
|
|
||||||
logger.info("SSE connection closed")
|
logger.info("SSE connection closed")
|
||||||
|
|
||||||
|
|
||||||
@@ -343,24 +389,29 @@ async def event_generator(request: Request) -> AsyncGenerator[str, None]:
|
|||||||
async def realtime_events(request: Request) -> StreamingResponse:
|
async def realtime_events(request: Request) -> StreamingResponse:
|
||||||
"""Server-Sent Events endpoint for real-time updates.
|
"""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:
|
Args:
|
||||||
request: FastAPI request object
|
request: FastAPI request object
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
StreamingResponse: SSE stream of Redis messages
|
StreamingResponse: SSE stream with Redis messages and heartbeats
|
||||||
"""
|
"""
|
||||||
return StreamingResponse(
|
return StreamingResponse(
|
||||||
event_generator(request),
|
event_generator(request),
|
||||||
media_type="text/event-stream",
|
media_type="text/event-stream",
|
||||||
headers={
|
headers={
|
||||||
"Cache-Control": "no-cache",
|
"Cache-Control": "no-cache, no-transform",
|
||||||
"Connection": "keep-alive",
|
"Connection": "keep-alive",
|
||||||
"X-Accel-Buffering": "no", # Disable nginx buffering
|
"X-Accel-Buffering": "no", # Disable nginx buffering
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
|
|
||||||
return {"message": f"Command sent to {device_id}"}
|
|
||||||
|
|
||||||
|
|
||||||
def main() -> None:
|
def main() -> None:
|
||||||
"""Run the API application with uvicorn."""
|
"""Run the API application with uvicorn."""
|
||||||
|
|||||||
Reference in New Issue
Block a user