""" Stream generation and client-side handling for TS streams. This module handles generating and delivering video streams to clients. """ import time import gevent from apps.proxy.config import TSConfig as Config from apps.channels.models import Channel, Stream from django.db import close_old_connections from core.utils import log_system_event from ...server import ProxyServer from ...utils import create_ts_packet, get_logger, resolve_channel_display_name from ...redis_keys import RedisKeys from ...constants import ChannelMetadataField from ...config_helper import ConfigHelper logger = get_logger() class StreamGenerator: """ Handles generating streams for clients, including initialization, data delivery, and cleanup. """ def __init__( self, channel_id, client_id, client_ip, client_user_agent, channel_initializing=False, user=None, buffer=None, channel_name=None, ): """ Initialize the stream generator with client and channel details. Args: channel_id: The UUID of the channel to stream client_id: Unique ID for this client connection client_ip: Client's IP address client_user_agent: User agent string from client channel_initializing: Whether the channel is still initializing user: Authenticated user making the request buffer: Source StreamBuffer to read from. Resolved via ProxyServer.get_buffer() before construction; passed in so the generator is buffer-agnostic. channel_name: Optional display name (avoids ORM during construction) """ self.channel_id = channel_id self.client_id = client_id self.client_ip = client_ip self.client_user_agent = client_user_agent self.channel_initializing = channel_initializing self.user = user self._source_buffer = buffer self.channel_name = resolve_channel_display_name(channel_id, channel_name=channel_name) # Performance and state tracking self.stream_start_time = time.time() self.bytes_sent = 0 self.chunks_sent = 0 self.local_index = 0 self.consecutive_empty = 0 # Add tracking for current transfer rate calculation self.last_stats_time = time.time() self.last_stats_bytes = 0 self.current_rate = 0.0 # TTL refresh tracking self.last_ttl_refresh = time.time() self.ttl_refresh_interval = 3 # Refresh TTL every 3 seconds of active streaming # Throttle per-client stats writes to Redis. # channels with many viewers. self.last_stats_write = 0.0 self.stats_write_interval = 1.0 # Cached proxy server reference self.proxy_server = None # Non-owner health check throttle: avoid Redis GET on every loop iteration self._last_health_check_time = 0.0 self._last_health_check_result = False self._health_check_interval = 2.0 # seconds # Resource check throttle: Redis stop/state checks are expensive; throttle # them while allowing cheap in-memory checks to run every iteration. self._last_resource_check_time = 0.0 self._resource_check_interval = 1.0 # seconds def generate(self): """ Generator function that produces the stream content for the client. Handles initialization state, data delivery, and client disconnection. Yields: bytes: Chunks of TS stream data """ self.stream_start_time = time.time() self.bytes_sent = 0 self.chunks_sent = 0 try: logger.info(f"[{self.client_id}] Stream generator started, channel_ready={not self.channel_initializing}") # First handle initialization if needed if self.channel_initializing: channel_ready = yield from self._wait_for_initialization() if not channel_ready: # If initialization failed or timed out, we've already sent error packets return # Channel is now ready - start normal streaming logger.info(f"[{self.client_id}] Channel {self.channel_id} ready, starting normal streaming") # Reset start time for real streaming self.stream_start_time = time.time() # Setup streaming parameters and verify resources if not self._setup_streaming(): return # Log client connect event try: log_system_event( 'client_connect', channel_id=self.channel_id, channel_name=self.channel_name, client_ip=self.client_ip, client_id=self.client_id, user_agent=self.client_user_agent[:100] if self.client_user_agent else None, username=self.user.username if self.user else None ) except Exception as e: logger.error(f"Could not log client connect event: {e}") # Main streaming loop for chunk in self._stream_data_generator(): yield chunk except Exception as e: logger.error(f"[{self.client_id}] Stream error: {e}", exc_info=True) finally: self._cleanup() def _init_wait_abort_reason(self, proxy_server, initialization_start): """Return a reason string when init wait should end early, else None.""" from ...services.channel_service import ChannelService if ChannelService.is_channel_teardown_active(self.channel_id): return "stopping" client_mgr = proxy_server.client_managers.get(self.channel_id) if client_mgr is not None: if self.client_id not in client_mgr.clients: return "client_gone" elif proxy_server.redis_client: client_key = RedisKeys.client_metadata(self.channel_id, self.client_id) if not proxy_server.redis_client.exists(client_key): total = proxy_server.redis_client.scard(RedisKeys.clients(self.channel_id)) or 0 if total == 0: return "client_gone" elapsed = time.time() - initialization_start init_grace_period = ConfigHelper.channel_init_grace_period() if elapsed < init_grace_period or not proxy_server.redis_client: return None metadata_key = RedisKeys.channel_metadata(self.channel_id) state_raw = proxy_server.redis_client.hget(metadata_key, 'state') if not state_raw: return None state = state_raw.decode() if isinstance(state_raw, bytes) else state_raw if state not in ('connecting', 'initializing', 'buffering'): return None buffer = proxy_server.stream_buffers.get(self.channel_id) if buffer is not None and buffer.index > 0: return None if proxy_server.redis_client.get(RedisKeys.last_data(self.channel_id)): return None return "stalled" def _wait_for_initialization(self): initialization_start = time.time() max_init_wait = ConfigHelper.client_wait_timeout() proxy_server = ProxyServer.get_instance() while time.time() - initialization_start < max_init_wait: abort_reason = self._init_wait_abort_reason(proxy_server, initialization_start) if abort_reason == "stopping": logger.error( f"[{self.client_id}] Channel {self.channel_id} teardown active during initialization" ) yield create_ts_packet('error', "Error: Channel is stopping") return False if abort_reason == "client_gone": logger.info( f"[{self.client_id}] Client disconnected during initialization wait, aborting" ) yield create_ts_packet('error', "Error: Client disconnected") return False if abort_reason == "stalled": logger.warning( f"[{self.client_id}] Channel {self.channel_id} stalled in connecting state " f"with no buffer data after " f"{ConfigHelper.channel_init_grace_period()}s, aborting init wait" ) yield create_ts_packet('error', "Error: Connection stalled") return False if proxy_server.redis_client: metadata_key = RedisKeys.channel_metadata(self.channel_id) metadata = proxy_server.redis_client.hgetall(metadata_key) if metadata and 'state' in metadata: state = metadata['state'] if state in ['waiting_for_clients', 'active']: logger.info(f"[{self.client_id}] Channel {self.channel_id} now ready (state={state})") return True elif state in ['error', 'stopped', 'stopping']: error_message = metadata.get('error_message', 'Unknown error') logger.error(f"[{self.client_id}] Channel {self.channel_id} in error state: {state}, message: {error_message}") yield create_ts_packet('error', f"Error: {error_message}") return False stop_key = RedisKeys.channel_stopping(self.channel_id) if proxy_server.redis_client.exists(stop_key): logger.error(f"[{self.client_id}] Channel {self.channel_id} stopping flag detected during initialization") yield create_ts_packet('error', "Error: Channel is stopping") return False gevent.sleep(0.1) logger.warning(f"[{self.client_id}] Timed out waiting for initialization") yield create_ts_packet('error', "Error: Initialization timeout") return False def _setup_streaming(self): """Setup streaming parameters and check resources.""" proxy_server = ProxyServer.get_instance() buffer = self._source_buffer stream_manager = proxy_server.stream_managers.get(self.channel_id) if not buffer: logger.error(f"[{self.client_id}] No buffer found for channel {self.channel_id}") return False # Client state tracking — determine start position # When behind_seconds > 0, use time-based positioning to start # the client that many seconds behind live. # When behind_seconds == 0, start at live (buffer head). behind_seconds = ConfigHelper.new_client_behind_seconds() current_buffer_index = buffer.index if behind_seconds > 0: time_index = buffer.find_chunk_index_by_time(behind_seconds) if time_index is not None: self.local_index = max(0, time_index) logger.info( f"[{self.client_id}] Time-based positioning: " f"{behind_seconds}s behind -> index {self.local_index} " f"(buffer head at {current_buffer_index})" ) else: # Not enough buffer for the requested time — start as far # back as possible (oldest available chunk). oldest = buffer.find_oldest_available_chunk(0) if oldest is not None: self.local_index = max(0, oldest) logger.info( f"[{self.client_id}] Buffer shorter than {behind_seconds}s, " f"starting at oldest available chunk {self.local_index} " f"(buffer head at {current_buffer_index})" ) else: # No timestamp data at all — start at live self.local_index = current_buffer_index logger.info( f"[{self.client_id}] No timestamp data, starting at live: " f"index {self.local_index} (buffer head at {current_buffer_index})" ) else: # 0 = start at live (buffer head) self.local_index = current_buffer_index logger.info( f"[{self.client_id}] Starting at live (behind_seconds=0): " f"index {self.local_index} (buffer head at {current_buffer_index})" ) # Store important objects as instance variables self.proxy_server = proxy_server self.buffer = buffer self.stream_manager = stream_manager self.last_yield_time = time.time() self.empty_reads = 0 self.consecutive_empty = 0 self.is_owner_worker = proxy_server.am_i_owner(self.channel_id) if hasattr(proxy_server, 'am_i_owner') else True logger.info(f"[{self.client_id}] Starting stream at index {self.local_index} (buffer at {buffer.index})") return True def _stream_data_generator(self): """Generate stream data chunks based on buffer contents.""" # Keepalive packets refresh last_yield_time, so _is_timeout() never fires # during sustained stream failure. This timer enforces a wall-clock cap. keepalive_start_time = None # Main streaming loop while True: # Check if resources still exist if not self._check_resources(): break # Get chunks at client's position using improved strategy chunks, next_index = self.buffer.get_optimized_client_data(self.local_index) if chunks: keepalive_start_time = None # Each recovery restarts the cap independently. yield from self._process_chunks(chunks, next_index) self.local_index = next_index self.last_yield_time = time.time() self.empty_reads = 0 self.consecutive_empty = 0 else: # Handle no data condition (with possible keepalive packets) self.empty_reads += 1 self.consecutive_empty += 1 # We got no data despite being behind the buffer head. # The read itself is the authoritative signal — no separate # existence check needed, avoiding TOCTOU races with Redis TTL. chunks_behind = self.buffer.index - self.local_index if chunks_behind > 0: # Next chunk has expired — find the oldest chunk still in Redis new_index = self.buffer.find_oldest_available_chunk(self.local_index) if new_index is not None: skipped = new_index - self.local_index logger.warning( f"[{self.client_id}] Next chunk expired (index {self.local_index + 1}), " f"jumping to oldest available: {new_index + 1} " f"(skipped {skipped} chunks, buffer head at {self.buffer.index})" ) self.local_index = new_index else: # No chunks available at all — jump to near the buffer head initial_behind = ConfigHelper.initial_behind_chunks() new_index = max(self.local_index, self.buffer.index - initial_behind) logger.warning( f"[{self.client_id}] No chunks available in buffer, " f"jumping to near buffer head: {new_index} " f"(buffer head at {self.buffer.index})" ) self.local_index = new_index self.consecutive_empty = 0 continue # Retry immediately with the new position if self._should_send_keepalive(self.local_index): if keepalive_start_time is None: keepalive_start_time = time.time() max_keepalive = getattr(Config, 'MAX_KEEPALIVE_DURATION', 300) if time.time() - keepalive_start_time > max_keepalive: logger.warning( f"[{self.client_id}] Keepalive duration exceeded {max_keepalive}s " f"with no stream recovery, disconnecting" ) break keepalive_packet = create_ts_packet('keepalive') logger.debug(f"[{self.client_id}] Sending keepalive packet while waiting at buffer head") yield keepalive_packet self.bytes_sent += len(keepalive_packet) self.last_yield_time = time.time() self.consecutive_empty = 0 # Reset consecutive counter but keep total empty_reads # Update last_active so clients waiting during failover aren't flagged as ghosts proxy_server = ProxyServer.get_instance() if proxy_server and proxy_server.redis_client: client_key = RedisKeys.client_metadata(self.channel_id, self.client_id) proxy_server.redis_client.hset(client_key, "last_active", str(time.time())) gevent.sleep(Config.KEEPALIVE_INTERVAL) # Replace time.sleep else: # Standard wait with backoff sleep_time = min(0.1 * self.consecutive_empty, 1.0) gevent.sleep(sleep_time) # Replace time.sleep # Log empty reads periodically if self.empty_reads % 50 == 0: stream_status = "healthy" if (self.stream_manager and self.stream_manager.healthy) else "unknown" logger.debug(f"[{self.client_id}] Waiting for chunks beyond {self.local_index} for channel: {self.channel_id} (buffer at {self.buffer.index}, stream: {stream_status})") # Check for ghost clients if self._is_ghost_client(self.local_index): logger.warning(f"[{self.client_id}] Possible ghost client: buffer has advanced {self.buffer.index - self.local_index} chunks ahead but client stuck at {self.local_index}") break # Check for timeouts if self._is_timeout(): break def _check_resources(self): """Check if required resources still exist.""" proxy_server = self.proxy_server or ProxyServer.get_instance() if self.buffer is None: logger.info(f"[{self.client_id}] Channel buffer no longer exists, terminating stream") return False if self.channel_id not in proxy_server.client_managers: logger.info(f"[{self.client_id}] Client manager no longer exists, terminating stream") return False client_manager = proxy_server.client_managers[self.channel_id] if self.client_id not in client_manager.clients: logger.info(f"[{self.client_id}] Client no longer in client manager, terminating stream") return False # --- Redis checks: throttled to _resource_check_interval (default 1s) --- # 3 Redis round-trips on every iteration is expensive at stream rates; # stop/state signals change infrequently so a 1-second poll is sufficient. if not proxy_server.redis_client: return True now = time.time() if now - self._last_resource_check_time < self._resource_check_interval: return True self._last_resource_check_time = now # Channel stop check stop_key = RedisKeys.channel_stopping(self.channel_id) if proxy_server.redis_client.exists(stop_key): logger.info(f"[{self.client_id}] Detected channel stop signal, terminating stream") return False # Channel state in metadata metadata_key = RedisKeys.channel_metadata(self.channel_id) metadata = proxy_server.redis_client.hgetall(metadata_key) if metadata and 'state' in metadata: state = metadata['state'] if state in ['error', 'stopped', 'stopping']: logger.info(f"[{self.client_id}] Channel in {state} state, terminating stream") return False # Client stop check client_stop_key = RedisKeys.client_stop(self.channel_id, self.client_id) if proxy_server.redis_client.exists(client_stop_key): logger.info(f"[{self.client_id}] Detected client stop signal, terminating stream") return False return True def _process_chunks(self, chunks, next_index): """Process and yield chunks to the client.""" # Process and send chunks total_size = sum(len(c) for c in chunks) logger.debug(f"[{self.client_id}] Retrieved {len(chunks)} chunks ({total_size} bytes) from index {self.local_index+1} to {next_index}") proxy_server = self.proxy_server or ProxyServer.get_instance() # Send the chunks to the client for chunk in chunks: try: yield chunk self.bytes_sent += len(chunk) self.chunks_sent += 1 logger.debug(f"[{self.client_id}] Sent chunk {self.chunks_sent} ({len(chunk)} bytes) for channel {self.channel_id} to client") current_time = time.time() # Calculate average rate (since stream start) elapsed_total = current_time - self.stream_start_time avg_rate = self.bytes_sent / elapsed_total / 1024 if elapsed_total > 0 else 0 # Calculate current rate (since last measurement) elapsed_current = current_time - self.last_stats_time bytes_since_last = self.bytes_sent - self.last_stats_bytes if elapsed_current > 0: self.current_rate = bytes_since_last / elapsed_current / 1024 # Update last stats values self.last_stats_time = current_time self.last_stats_bytes = self.bytes_sent # Log every 10 chunks if self.chunks_sent % 10 == 0: logger.debug(f"[{self.client_id}] Stats: {self.chunks_sent} chunks, {self.bytes_sent/1024:.1f} KB, " f"avg: {avg_rate:.1f} KB/s, current: {self.current_rate:.1f} KB/s") # Store stats in Redis client metadata, throttled to avoid an # hset on every chunk. Frontend stats panels poll on the order # of seconds, so 1s resolution is sufficient. if proxy_server.redis_client and (current_time - self.last_stats_write) >= self.stats_write_interval: try: self.last_stats_write = current_time client_key = RedisKeys.client_metadata(self.channel_id, self.client_id) stats = { ChannelMetadataField.CHUNKS_SENT: str(self.chunks_sent), ChannelMetadataField.BYTES_SENT: str(self.bytes_sent), ChannelMetadataField.AVG_RATE_KBPS: str(round(avg_rate, 1)), ChannelMetadataField.CURRENT_RATE_KBPS: str(round(self.current_rate, 1)), ChannelMetadataField.STATS_UPDATED_AT: str(current_time), "last_active": str(current_time) } proxy_server.redis_client.hset(client_key, mapping=stats) # Refresh TTL periodically while actively streaming # This provides proof-of-life independent of heartbeat thread if current_time - self.last_ttl_refresh > self.ttl_refresh_interval: try: # Refresh TTL on client key proxy_server.redis_client.expire(client_key, Config.CLIENT_RECORD_TTL) # Also refresh the client set TTL client_set_key = f"live:channel:{self.channel_id}:clients" proxy_server.redis_client.expire(client_set_key, Config.CLIENT_RECORD_TTL) self.last_ttl_refresh = current_time logger.debug(f"[{self.client_id}] Refreshed client TTL (active streaming)") except Exception as ttl_error: logger.debug(f"[{self.client_id}] Failed to refresh TTL: {ttl_error}") except Exception as e: logger.warning(f"[{self.client_id}] Failed to store stats in Redis: {e}") except Exception as e: logger.error(f"[{self.client_id}] Error sending chunk to client: {e}") raise # Re-raise to exit the generator def _should_send_keepalive(self, local_index): """Determine if a keepalive packet should be sent.""" # Check if we're caught up to buffer head at_buffer_head = local_index >= self.buffer.index if not at_buffer_head or self.consecutive_empty < 5: return False if self.stream_manager is not None: # Owner worker: use the in-memory health flag directly. return not self.stream_manager.healthy else: # Non-owner worker: stream_manager only exists in the owner process. # Approximate health from the Redis last_data timestamp; if stale # beyond CONNECTION_TIMEOUT, send keepalives to prevent DVR timeout. # Throttled: only re-query Redis every _health_check_interval seconds # to avoid a Redis GET on every loop iteration during sustained waits. now = time.time() if now - self._last_health_check_time < self._health_check_interval: return self._last_health_check_result try: proxy_server = self.proxy_server or ProxyServer.get_instance() if proxy_server.redis_client: raw = proxy_server.redis_client.get(RedisKeys.last_data(self.channel_id)) if raw: age = now - float(raw) timeout_threshold = getattr(Config, 'CONNECTION_TIMEOUT', 10) result = age >= timeout_threshold else: # No timestamp in Redis → key missing or expired → unhealthy result = True self._last_health_check_time = now self._last_health_check_result = result return result except Exception: pass return False def _is_ghost_client(self, local_index): """Check if this appears to be a ghost client (stuck but buffer advancing).""" return self.consecutive_empty > 100 and self.buffer.index > local_index + 50 def _is_timeout(self): """Check if the stream has timed out.""" # Get a more generous timeout for stream switching stream_timeout = ConfigHelper.stream_timeout() failover_grace_period = ConfigHelper.failover_grace_period() total_timeout = stream_timeout + failover_grace_period # Disconnect after long inactivity if time.time() - self.last_yield_time > total_timeout: if self.stream_manager and not self.stream_manager.healthy: # Check if stream manager is actively switching or reconnecting if (hasattr(self.stream_manager, 'url_switching') and self.stream_manager.url_switching): logger.info(f"[{self.client_id}] Stream switching in progress, giving more time") return False logger.warning(f"[{self.client_id}] No data for {total_timeout}s and stream unhealthy, disconnecting") return True elif not self.is_owner_worker and self.consecutive_empty > 100: # Non-owner worker without data for too long logger.warning(f"[{self.client_id}] Non-owner worker with no data for {total_timeout}s, disconnecting") return True return False def _cleanup(self): """Clean up resources and report final statistics.""" try: # Client cleanup elapsed = time.time() - self.stream_start_time local_clients = 0 total_clients = 0 proxy_server = ProxyServer.get_instance() # Release M3U profile stream allocation if this is the last client stream_released = False if proxy_server.redis_client: try: if self.channel_id in proxy_server.client_managers: client_count = proxy_server.client_managers[self.channel_id].get_total_client_count() # Pool slots are global; the last client on any worker must release. # During shutdown_delay, keep the slot until coordinated stop runs. if client_count <= 1 and ConfigHelper.channel_shutdown_delay() <= 0: try: try: obj = Channel.objects.get(uuid=self.channel_id) except (Channel.DoesNotExist, Exception): obj = Stream.objects.get(stream_hash=self.channel_id) stream_released = obj.release_stream() if stream_released: logger.debug(f"[{self.client_id}] Released stream for channel {self.channel_id}") else: logger.warning(f"[{self.client_id}] release_stream found no keys for channel {self.channel_id}") except Exception as e: logger.error(f"[{self.client_id}] Error releasing stream for channel {self.channel_id}: {e}") except Exception as e: logger.error(f"[{self.client_id}] Error checking stream data for release: {e}") if self.channel_id in proxy_server.client_managers: client_manager = proxy_server.client_managers[self.channel_id] if self.client_id in client_manager.clients: local_clients = client_manager.remove_client(self.client_id) else: local_clients = client_manager.get_client_count() total_clients = client_manager.get_total_client_count() logger.info(f"[{self.client_id}] Disconnected after {elapsed:.2f}s (local: {local_clients}, total: {total_clients})") # Log client disconnect event try: log_system_event( 'client_disconnect', channel_id=self.channel_id, channel_name=self.channel_name, client_ip=self.client_ip, client_id=self.client_id, user_agent=self.client_user_agent[:100] if self.client_user_agent else None, duration=round(elapsed, 2), bytes_sent=self.bytes_sent, username=self.user.username if self.user else None ) except Exception as e: logger.error(f"Could not log client disconnect event: {e}") finally: close_old_connections() def create_stream_generator( channel_id, client_id, client_ip, client_user_agent, channel_initializing=False, user=None, buffer=None, channel_name=None, ): """ Factory function to create a new stream generator. Returns a function that can be passed to StreamingHttpResponse. """ generator = StreamGenerator( channel_id, client_id, client_ip, client_user_agent, channel_initializing, user=user, buffer=buffer, channel_name=channel_name, ) return generator.generate