[New feature]

Switch streams when buffering is detected.

apps/proxy/config.py

@@ -11,6 +11,8 @@ class BaseConfig:
     BUFFER_CHUNK_SIZE = 188 * 1361  # ~256KB
     # Redis settings
     REDIS_CHUNK_TTL = 60  # Number in seconds - Chunks expire after 1 minute
+    BUFFERING_TIMEOUT = 15  # Seconds to wait for buffering before switching streams
+    BUFFER_SPEED = 1 # What speed to condsider the stream buffering, 1x is normal speed, 2x is double speed, etc.
 
 class HLSConfig(BaseConfig):
     MIN_SEGMENTS = 12

apps/proxy/ts_proxy/config_helper.py

@@ -85,3 +85,11 @@ class ConfigHelper:
     def failover_grace_period():
         """Get extra time (in seconds) to allow for stream switching before disconnecting clients"""
         return ConfigHelper.get('FAILOVER_GRACE_PERIOD', 20)  # Default to 20 seconds
+    @staticmethod
+    def buffering_timeout():
+        """Get buffering timeout in seconds"""
+        return ConfigHelper.get('BUFFERING_TIMEOUT', 15)  # Default to 15 seconds
+    @staticmethod
+    def buffering_speed():
+        """Get buffering speed in bytes per second"""
+        return ConfigHelper.get('BUFFERING_SPEED',1) # Default to 1x

apps/proxy/ts_proxy/constants.py

@@ -18,6 +18,7 @@ class ChannelState:
     ERROR = "error"
     STOPPING = "stopping"
     STOPPED = "stopped"
+    BUFFERING = "buffering"
 
 # Event types
 class EventType:

apps/proxy/ts_proxy/stream_manager.py

@@ -40,6 +40,10 @@ class StreamManager:
         self.url_switching = False
         self.url_switch_start_time = 0
         self.url_switch_timeout = ConfigHelper.url_switch_timeout()
+        self.buffering = False
+        self.buffering_timeout = ConfigHelper.buffering_timeout()
+        self.buffering_speed = ConfigHelper.buffering_speed()
+        self.buffering_start_time = None
         # Store worker_id for ownership checks
         self.worker_id = worker_id
 
@@ -545,7 +549,6 @@ class StreamManager:
             actual_fps = None
             if ffmpeg_fps is not None and ffmpeg_speed is not None and ffmpeg_speed > 0:
                 actual_fps = ffmpeg_fps / ffmpeg_speed
-
             # Store in Redis if we have valid data
             if any(x is not None for x in [ffmpeg_speed, ffmpeg_fps, actual_fps, ffmpeg_bitrate]):
                 self._update_ffmpeg_stats_in_redis(ffmpeg_speed, ffmpeg_fps, actual_fps, ffmpeg_bitrate)
@@ -553,10 +556,51 @@ class StreamManager:
             # Fix the f-string formatting
             actual_fps_str = f"{actual_fps:.1f}" if actual_fps is not None else "N/A"
             ffmpeg_bitrate_str = f"{ffmpeg_bitrate:.1f}" if ffmpeg_bitrate is not None else "N/A"
-
+            # Log the stats
             logger.debug(f"FFmpeg stats - Speed: {ffmpeg_speed}x, FFmpeg FPS: {ffmpeg_fps}, "
                         f"Actual FPS: {actual_fps_str}, "
                         f"Bitrate: {ffmpeg_bitrate_str} kbps")
+            # If we have a valid speed, check for buffering
+            if ffmpeg_speed is not None and ffmpeg_speed < self.buffering_speed:
+                if self.buffering:
+                    # Buffering is still ongoing, check for how long
+                    if self.buffering_start_time is None:
+                        self.buffering_start_time = time.time()
+                    else:
+                        buffering_duration = time.time() - self.buffering_start_time
+                        if buffering_duration > self.buffering_timeout:
+                            # Buffering timeout reached, log error and try next stream
+                            logger.error(f"Buffering timeout reached for channel {self.channel_id} after {buffering_duration:.1f} seconds")
+                            # Send next stream request
+                            if self._try_next_stream():
+                                logger.info(f"Switched to next stream for channel {self.channel_id} after buffering timeout")
+                                # Reset buffering state
+                                self.buffering = False
+                                self.buffering_start_time = None
+                            else:
+                                logger.error(f"Failed to switch to next stream for channel {self.channel_id} after buffering timeout")
+                else:
+                    # Buffering just started, set the flag and start timer
+                    self.buffering = True
+                    self.buffering_start_time = time.time()
+                    logger.warning(f"Buffering started for channel {self.channel_id} - speed: {ffmpeg_speed}x")
+                # Log buffering warning
+                logger.debug(f"FFmpeg speed on channel {self.channel_id} is below {self.buffering_speed} ({ffmpeg_speed}x) - buffering detected")
+                # Set channel state to buffering
+                if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
+                    metadata_key = RedisKeys.channel_metadata(self.channel_id)
+                    self.buffer.redis_client.hset(metadata_key, ChannelMetadataField.STATE, ChannelState.BUFFERING)
+            elif ffmpeg_speed is not None and ffmpeg_speed >= self.buffering_speed:
+                # Speed is good, check if we were buffering
+                if self.buffering:
+                    # Reset buffering state
+                    logger.info(f"Buffering ended for channel {self.channel_id} - speed: {ffmpeg_speed}x")
+                    self.buffering = False
+                    self.buffering_start_time = None
+                    # Set channel state to active if speed is good
+                    if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
+                        metadata_key = RedisKeys.channel_metadata(self.channel_id)
+                        self.buffer.redis_client.hset(metadata_key, ChannelMetadataField.STATE, ChannelState.ACTIVE)
 
         except Exception as e:
             logger.debug(f"Error parsing FFmpeg stats: {e}")