fyp

app/services/redis_pubsub.py

@@ -71,7 +71,10 @@ class RedisPubSubService:
         logger.info(f"✅ Redis Pub/Sub listener started on channel: {CHAT_CHANNEL}")
 
     async def _listener_loop(self):
-        """Background loop to listen for Redis messages with auto-reconnect"""
+        """
+        Background loop to listen for Redis messages with auto-reconnect.
+        Uses get_message with timeout to avoid Redis Cloud idle timeout (30s).
+        """
         retry_count = 0
         max_retries = 10
         base_delay = 1  # seconds
@@ -84,24 +87,33 @@ class RedisPubSubService:
                 logger.info(f"📡 Redis Pub/Sub connected to {CHAT_CHANNEL}")
                 retry_count = 0  # Reset on successful connection
                 
-                async for message in pubsub.listen():
-                    if not self.is_listening:
-                        break
+                # Use get_message with timeout instead of listen()
+                # This prevents Redis Cloud idle timeout (30s) by not blocking forever
+                while self.is_listening:
+                    try:
+                        # Wait up to 20 seconds for a message (before Redis Cloud's 30s timeout)
+                        message = await pubsub.get_message(ignore_subscribe_messages=True, timeout=20.0)
                         
-                    if message["type"] == "message":
-                        try:
-                            payload = json.loads(message["data"])
-                            if self.message_handler:
-                                await self.message_handler(payload)
-                        except json.JSONDecodeError:
-                            logger.warning("Received invalid JSON in Redis message")
-                        except Exception as e:
-                            logger.error(f"Error processing Redis message: {e}")
+                        if message is not None and message["type"] == "message":
+                            try:
+                                payload = json.loads(message["data"])
+                                if self.message_handler:
+                                    await self.message_handler(payload)
+                            except json.JSONDecodeError:
+                                logger.warning("Received invalid JSON in Redis message")
+                            except Exception as e:
+                                logger.error(f"Error processing Redis message: {e}")
+                        
+                        # If no message received, the timeout just expired - this is normal
+                        # The connection is still alive, we just loop again
+                        
+                    except asyncio.CancelledError:
+                        break
                             
             except Exception as e:
                 retry_count += 1
                 delay = min(base_delay * (2 ** retry_count), 60)  # Max 60s delay
-                logger.error(f"Redis listener crashed: {e}. Reconnecting in {delay}s (attempt {retry_count}/{max_retries})")
+                logger.error(f"Redis listener error: {e}. Reconnecting in {delay}s (attempt {retry_count}/{max_retries})")
                 await asyncio.sleep(delay)
                 
             finally: