From dc43fb41b56d0ec0a87438b038c978c0b6829a0a Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 20 Mar 2025 08:45:23 -0500 Subject: [PATCH 1/4] More refactoring. --- .../ts_proxy/services/channel_service.py | 17 ++- apps/proxy/ts_proxy/url_utils.py | 129 ++++++++++++++++++ apps/proxy/ts_proxy/views.py | 42 +----- 3 files changed, 149 insertions(+), 39 deletions(-) create mode 100644 apps/proxy/ts_proxy/url_utils.py diff --git a/apps/proxy/ts_proxy/services/channel_service.py b/apps/proxy/ts_proxy/services/channel_service.py index 0c190c32..b1f3bbb3 100644 --- a/apps/proxy/ts_proxy/services/channel_service.py +++ b/apps/proxy/ts_proxy/services/channel_service.py @@ -12,6 +12,7 @@ from apps.proxy.config import TSConfig as Config from .. import proxy_server from ..redis_keys import RedisKeys from ..constants import EventType +from ..url_utils import get_stream_info_for_switch logger = logging.getLogger("ts_proxy") @@ -43,18 +44,30 @@ class ChannelService: return success @staticmethod - def change_stream_url(channel_id, new_url, user_agent=None): + def change_stream_url(channel_id, new_url=None, user_agent=None, target_stream_id=None): """ Change the URL of an existing stream. Args: channel_id: UUID of the channel - new_url: New stream URL + new_url: New stream URL (optional if target_stream_id is provided) user_agent: Optional user agent to update + target_stream_id: Optional target stream ID to switch to Returns: dict: Result information including success status and diagnostics """ + # If no direct URL is provided but a target stream is, get URL from target stream + if not new_url and target_stream_id: + stream_info = get_stream_info_for_switch(channel_id, target_stream_id) + if 'error' in stream_info: + return { + 'status': 'error', + 'message': stream_info['error'] + } + new_url = stream_info['url'] + user_agent = stream_info['user_agent'] + # Check if channel exists in_local_managers = channel_id in proxy_server.stream_managers in_local_buffers = channel_id in proxy_server.stream_buffers diff --git a/apps/proxy/ts_proxy/url_utils.py b/apps/proxy/ts_proxy/url_utils.py new file mode 100644 index 00000000..cbce957c --- /dev/null +++ b/apps/proxy/ts_proxy/url_utils.py @@ -0,0 +1,129 @@ +""" +Utilities for handling stream URLs and transformations. +""" + +import logging +import re +from typing import Optional, Tuple +from django.shortcuts import get_object_or_404 +from apps.channels.models import Channel, Stream +from apps.m3u.models import M3UAccount, M3UAccountProfile +from core.models import UserAgent, CoreSettings + +logger = logging.getLogger("ts_proxy") + +def generate_stream_url(channel_id: str) -> Tuple[str, str, bool]: + """ + Generate the appropriate stream URL for a channel based on its profile settings. + + Args: + channel_id: The UUID of the channel + + Returns: + Tuple[str, str, bool]: (stream_url, user_agent, transcode_flag) + """ + # Get channel and related objects + channel = get_object_or_404(Channel, uuid=channel_id) + stream_id, profile_id = channel.get_stream() + + if stream_id is None or profile_id is None: + logger.error(f"No stream assigned to channel {channel_id}") + return None, None, False + + # Get the M3U account profile for URL pattern + stream = get_object_or_404(Stream, pk=stream_id) + profile = get_object_or_404(M3UAccountProfile, pk=profile_id) + + # Get the appropriate user agent + m3u_account = M3UAccount.objects.get(id=profile.m3u_account.id) + stream_user_agent = UserAgent.objects.get(id=m3u_account.user_agent.id).user_agent + + if stream_user_agent is None: + stream_user_agent = UserAgent.objects.get(id=CoreSettings.get_default_user_agent_id()) + logger.debug(f"No user agent found for account, using default: {stream_user_agent}") + + # Generate stream URL based on the selected profile + input_url = stream.url + stream_url = transform_url(input_url, profile.search_pattern, profile.replace_pattern) + + # Check if transcoding is needed + stream_profile = channel.get_stream_profile() + if stream_profile.is_proxy() or stream_profile is None: + transcode = False + else: + transcode = True + + # Get profile name as string + profile_value = str(stream_profile) + + return stream_url, stream_user_agent, transcode, profile_value + +def transform_url(input_url: str, search_pattern: str, replace_pattern: str) -> str: + """ + Transform a URL using regex pattern replacement. + + Args: + input_url: The base URL to transform + search_pattern: The regex search pattern + replace_pattern: The replacement pattern + + Returns: + str: The transformed URL + """ + try: + logger.debug("Executing URL pattern replacement:") + logger.debug(f" base URL: {input_url}") + logger.debug(f" search: {search_pattern}") + + # Handle backreferences in the replacement pattern + safe_replace_pattern = re.sub(r'\$(\d+)', r'\\\1', replace_pattern) + logger.debug(f" replace: {replace_pattern}") + logger.debug(f" safe replace: {safe_replace_pattern}") + + # Apply the transformation + stream_url = re.sub(search_pattern, safe_replace_pattern, input_url) + logger.debug(f"Generated stream url: {stream_url}") + + return stream_url + except Exception as e: + logger.error(f"Error transforming URL: {e}") + return input_url # Return original URL on error + +def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int] = None) -> dict: + """ + Get stream information for a channel switch, optionally to a specific stream ID. + + Args: + channel_id: The UUID of the channel + target_stream_id: Optional specific stream ID to switch to + + Returns: + dict: Stream information including URL, user agent and transcode flag + """ + channel = get_object_or_404(Channel, uuid=channel_id) + + # Use the target stream if specified, otherwise use current stream + if target_stream_id: + stream_id = target_stream_id + # Find a compatible profile for this stream + profiles = M3UAccountProfile.objects.filter(stream=stream_id) + if not profiles.exists(): + logger.error(f"No profile found for stream {stream_id}") + return {'error': 'No profile found for stream'} + profile_id = profiles.first().id + else: + stream_id, profile_id = channel.get_stream() + if stream_id is None or profile_id is None: + return {'error': 'No stream assigned to channel'} + + # Generate the URL using our utility + stream_url, user_agent, transcode, profile_value = generate_stream_url(channel_id) + + return { + 'url': stream_url, + 'user_agent': user_agent, + 'transcode': transcode, + 'profile': profile_value, + 'stream_id': stream_id, + 'profile_id': profile_id + } diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index a97a35f6..f84963bb 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -21,6 +21,7 @@ from rest_framework.permissions import IsAuthenticated from .constants import ChannelState, EventType, StreamType from .config_helper import ConfigHelper from .services.channel_service import ChannelService +from .url_utils import generate_stream_url, transform_url, get_stream_info_for_switch # Configure logging properly logger = logging.getLogger("ts_proxy") @@ -52,50 +53,17 @@ def stream_ts(request, channel_id): # Initialize the channel (but don't wait for completion) logger.info(f"[{client_id}] Starting channel {channel_id} initialization") - # Get stream details from channel model - stream_id, profile_id = channel.get_stream() - if stream_id is None or profile_id is None: + # Use the utility function to get stream URL and settings + stream_url, stream_user_agent, transcode, profile_value = generate_stream_url(channel_id) + if stream_url is None: return JsonResponse({'error': 'Channel not available'}, status=404) - # Load in necessary objects for the stream - logger.info(f"Fetching stream ID {stream_id}") - stream = get_object_or_404(Stream, pk=stream_id) - logger.info(f"Fetching profile ID {profile_id}") - profile = get_object_or_404(M3UAccountProfile, pk=profile_id) - - # Load in the user-agent for the STREAM connection (not client) - m3u_account = M3UAccount.objects.get(id=profile.m3u_account.id) - stream_user_agent = UserAgent.objects.get(id=m3u_account.user_agent.id).user_agent - if stream_user_agent is None: - stream_user_agent = UserAgent.objects.get(id=CoreSettings.get_default_user_agent_id()) - logger.debug(f"No user agent found for account, using default: {stream_user_agent}") - else: - logger.debug(f"User agent found for account: {stream_user_agent}") - - # Generate stream URL based on the selected profile - input_url = stream.url - logger.debug("Executing the following pattern replacement:") - logger.debug(f" search: {profile.search_pattern}") - safe_replace_pattern = re.sub(r'\$(\d+)', r'\\\1', profile.replace_pattern) - logger.debug(f" replace: {profile.replace_pattern}") - logger.debug(f" safe replace: {safe_replace_pattern}") - stream_url = re.sub(profile.search_pattern, safe_replace_pattern, input_url) - logger.debug(f"Generated stream url: {stream_url}") - # Generate transcode command if needed stream_profile = channel.get_stream_profile() if stream_profile.is_redirect(): return HttpResponseRedirect(stream_url) - # Need to check if profile is transcoded - logger.debug(f"Using profile {stream_profile} for stream {stream_id}") - if stream_profile.is_proxy() or stream_profile is None: - transcode = False - else: - transcode = True - - # Initialize channel using the service - profile_value = str(stream_profile) + # Initialize channel with the stream's user agent (not the client's) success = ChannelService.initialize_channel( channel_id, stream_url, stream_user_agent, transcode, profile_value ) From d6a452548c69cf3e92952df7ead4ff1218c817d0 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 20 Mar 2025 15:28:12 -0500 Subject: [PATCH 2/4] Automatic stream switching is working. --- apps/proxy/config.py | 15 +- apps/proxy/ts_proxy/config_helper.py | 5 + apps/proxy/ts_proxy/server.py | 40 +- .../ts_proxy/services/channel_service.py | 42 +- apps/proxy/ts_proxy/stream_manager.py | 536 ++++++++++++------ apps/proxy/ts_proxy/url_utils.py | 167 +++++- apps/proxy/ts_proxy/views.py | 6 +- 7 files changed, 593 insertions(+), 218 deletions(-) diff --git a/apps/proxy/config.py b/apps/proxy/config.py index ffa0b559..28d3b872 100644 --- a/apps/proxy/config.py +++ b/apps/proxy/config.py @@ -23,34 +23,37 @@ class HLSConfig(BaseConfig): class TSConfig(BaseConfig): """Configuration settings for TS proxy""" - + # Connection settings CONNECTION_TIMEOUT = 10 # seconds to wait for initial connection MAX_RETRIES = 3 # maximum connection retry attempts - + # Buffer settings INITIAL_BEHIND_CHUNKS = 4 # How many chunks behind to start a client (4 chunks = ~1MB) CHUNK_BATCH_SIZE = 5 # How many chunks to fetch in one batch KEEPALIVE_INTERVAL = 0.5 # Seconds between keepalive packets when at buffer head - + # Streaming settings TARGET_BITRATE = 8000000 # Target bitrate (8 Mbps) STREAM_TIMEOUT = 10 # Disconnect after this many seconds of no data HEALTH_CHECK_INTERVAL = 5 # Check stream health every N seconds - + # Resource management CLEANUP_INTERVAL = 60 # Check for inactive channels every 60 seconds CHANNEL_SHUTDOWN_DELAY = 0 # How long to wait after last client before shutdown (seconds) - + # Client tracking settings CLIENT_RECORD_TTL = 5 # How long client records persist in Redis (seconds). Client will be considered MIA after this time. CLEANUP_CHECK_INTERVAL = 1 # How often to check for disconnected clients (seconds) CHANNEL_INIT_GRACE_PERIOD = 5 # How long to wait for first client after initialization (seconds) CLIENT_HEARTBEAT_INTERVAL = 1 # How often to send client heartbeats (seconds) GHOST_CLIENT_MULTIPLIER = 5.0 # How many heartbeat intervals before client considered ghost (5 would mean 5 secondsif heartbeat interval is 1) - + # TS packets are 188 bytes # Make chunk size a multiple of TS packet size for perfect alignment # ~1MB is ideal for streaming (matches typical media buffer sizes) BUFFER_CHUNK_SIZE = 188 * 1361 # ~256KB + # Maximum number of stream switch attempts before giving up + MAX_STREAM_SWITCHES = 10 + diff --git a/apps/proxy/ts_proxy/config_helper.py b/apps/proxy/ts_proxy/config_helper.py index fbdac746..5a16581a 100644 --- a/apps/proxy/ts_proxy/config_helper.py +++ b/apps/proxy/ts_proxy/config_helper.py @@ -68,3 +68,8 @@ class ConfigHelper: def max_retries(): """Get maximum retry attempts""" return ConfigHelper.get('MAX_RETRIES', 3) + + @staticmethod + def max_stream_switches(): + """Get maximum number of stream switch attempts""" + return ConfigHelper.get('MAX_STREAM_SWITCHES', 10) diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index 427b2089..4c7897c1 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -329,7 +329,7 @@ class ProxyServer: logger.error(f"Error extending ownership: {e}") return False - def initialize_channel(self, url, channel_id, user_agent=None, transcode=False): + def initialize_channel(self, url, channel_id, user_agent=None, transcode=False, stream_id=None): """Initialize a channel without redundant active key""" try: # Create buffer and client manager instances @@ -347,6 +347,7 @@ class ProxyServer: # Get channel URL from Redis if available channel_url = url channel_user_agent = user_agent + channel_stream_id = stream_id # Store the stream ID # First check if channel metadata already exists existing_metadata = None @@ -365,6 +366,14 @@ class ProxyServer: if ua_bytes: channel_user_agent = ua_bytes.decode('utf-8') + # Get stream ID from metadata if not provided + if not channel_stream_id and b'stream_id' in existing_metadata: + try: + channel_stream_id = int(existing_metadata[b'stream_id'].decode('utf-8')) + logger.debug(f"Found stream_id {channel_stream_id} in metadata for channel {channel_id}") + except (ValueError, TypeError) as e: + logger.debug(f"Could not parse stream_id from metadata: {e}") + # Check if channel is already owned current_owner = self.get_channel_owner(channel_id) @@ -419,9 +428,23 @@ class ProxyServer: if channel_user_agent: metadata["user_agent"] = channel_user_agent - # Set channel metadata + # CRITICAL FIX: Make sure stream_id is always set in metadata and properly logged + if channel_stream_id: + metadata["stream_id"] = str(channel_stream_id) + logger.info(f"Storing stream_id {channel_stream_id} in metadata for channel {channel_id}") + else: + logger.warning(f"No stream_id provided for channel {channel_id} during initialization") + + # Set channel metadata BEFORE creating the StreamManager self.redis_client.hset(metadata_key, mapping=metadata) - self.redis_client.expire(metadata_key, 30) # 1 hour TTL + self.redis_client.expire(metadata_key, 3600) # Increased TTL from 30 seconds to 1 hour + + # Verify the stream_id was set correctly in Redis + stream_id_value = self.redis_client.hget(metadata_key, "stream_id") + if stream_id_value: + logger.info(f"Verified stream_id {stream_id_value.decode('utf-8')} is set in Redis for channel {channel_id}") + else: + logger.warning(f"Failed to set stream_id in Redis for channel {channel_id}") # Create stream buffer buffer = StreamBuffer(channel_id=channel_id, redis_client=self.redis_client) @@ -429,8 +452,15 @@ class ProxyServer: self.stream_buffers[channel_id] = buffer # Only the owner worker creates the actual stream manager - stream_manager = StreamManager(channel_id, channel_url, buffer, user_agent=channel_user_agent, transcode=transcode) - logger.debug(f"Created StreamManager for channel {channel_id}") + stream_manager = StreamManager( + channel_id, + channel_url, + buffer, + user_agent=channel_user_agent, + transcode=transcode, + stream_id=channel_stream_id # Pass stream ID to the manager + ) + logger.info(f"Created StreamManager for channel {channel_id} with stream ID {channel_stream_id}") self.stream_managers[channel_id] = stream_manager # Create client manager with channel_id, redis_client AND worker_id diff --git a/apps/proxy/ts_proxy/services/channel_service.py b/apps/proxy/ts_proxy/services/channel_service.py index b1f3bbb3..3fe6aedf 100644 --- a/apps/proxy/ts_proxy/services/channel_service.py +++ b/apps/proxy/ts_proxy/services/channel_service.py @@ -20,7 +20,7 @@ class ChannelService: """Service class for channel operations""" @staticmethod - def initialize_channel(channel_id, stream_url, user_agent, transcode=False, profile_value=None): + def initialize_channel(channel_id, stream_url, user_agent, transcode=False, profile_value=None, stream_id=None): """ Initialize a channel with the given parameters. @@ -30,16 +30,50 @@ class ChannelService: user_agent: User agent for the stream connection transcode: Whether to transcode the stream profile_value: Stream profile value to store in metadata + stream_id: ID of the stream being used Returns: bool: Success status """ - success = proxy_server.initialize_channel(stream_url, channel_id, user_agent, transcode) + # FIXED: First, ensure that Redis metadata including stream_id is set BEFORE channel initialization + # This ensures the stream ID is available when the StreamManager looks it up + if stream_id and proxy_server.redis_client: + metadata_key = RedisKeys.channel_metadata(channel_id) + # Check if metadata already exists + if proxy_server.redis_client.exists(metadata_key): + # Just update the existing metadata with stream_id + proxy_server.redis_client.hset(metadata_key, "stream_id", str(stream_id)) + logger.info(f"Pre-set stream ID {stream_id} in Redis for channel {channel_id}") + else: + # Create initial metadata with essential values + initial_metadata = { + "stream_id": str(stream_id), + "temp_init": str(time.time()) + } + proxy_server.redis_client.hset(metadata_key, mapping=initial_metadata) + logger.info(f"Created initial metadata with stream_id {stream_id} for channel {channel_id}") + + # Verify the stream_id was set + stream_id_value = proxy_server.redis_client.hget(metadata_key, "stream_id") + if stream_id_value: + logger.info(f"Verified stream_id {stream_id_value.decode('utf-8')} is now set in Redis") + else: + logger.error(f"Failed to set stream_id {stream_id} in Redis before initialization") + + # Now proceed with channel initialization + success = proxy_server.initialize_channel(stream_url, channel_id, user_agent, transcode, stream_id) # Store additional metadata if initialization was successful - if success and proxy_server.redis_client and profile_value: + if success and proxy_server.redis_client: metadata_key = RedisKeys.channel_metadata(channel_id) - proxy_server.redis_client.hset(metadata_key, "profile", profile_value) + update_data = {} + if profile_value: + update_data["profile"] = profile_value + if stream_id: + update_data["stream_id"] = str(stream_id) + + if update_data: + proxy_server.redis_client.hset(metadata_key, mapping=update_data) return success diff --git a/apps/proxy/ts_proxy/stream_manager.py b/apps/proxy/ts_proxy/stream_manager.py index d16173e5..281ba842 100644 --- a/apps/proxy/ts_proxy/stream_manager.py +++ b/apps/proxy/ts_proxy/stream_manager.py @@ -17,13 +17,14 @@ from .utils import detect_stream_type from .redis_keys import RedisKeys from .constants import ChannelState, EventType, StreamType, TS_PACKET_SIZE from .config_helper import ConfigHelper +from .url_utils import get_alternate_streams, get_stream_info_for_switch logger = logging.getLogger("ts_proxy") class StreamManager: """Manages a connection to a TS stream without using raw sockets""" - def __init__(self, channel_id, url, buffer, user_agent=None, transcode=False): + def __init__(self, channel_id, url, buffer, user_agent=None, transcode=False, stream_id=None): # Basic properties self.channel_id = channel_id self.url = url @@ -54,6 +55,38 @@ class StreamManager: self._buffer_check_timers = [] self.stopping = False + # Add tracking for tried streams and current stream + self.current_stream_id = stream_id + self.tried_stream_ids = set() + + # IMPROVED LOGGING: Better handle and track stream ID + if stream_id: + self.tried_stream_ids.add(stream_id) + logger.info(f"Initialized stream manager for channel {buffer.channel_id} with stream ID {stream_id}") + else: + # Try to get stream ID from Redis metadata if available + if hasattr(buffer, 'redis_client') and buffer.redis_client: + try: + metadata_key = RedisKeys.channel_metadata(channel_id) + + # Log all metadata for debugging purposes + metadata = buffer.redis_client.hgetall(metadata_key) + if metadata: + logger.debug(f"Redis metadata for channel {channel_id}: {metadata}") + + # Try to get stream_id specifically + stream_id_bytes = buffer.redis_client.hget(metadata_key, "stream_id") + if stream_id_bytes: + self.current_stream_id = int(stream_id_bytes.decode('utf-8')) + self.tried_stream_ids.add(self.current_stream_id) + logger.info(f"Loaded stream ID {self.current_stream_id} from Redis for channel {buffer.channel_id}") + else: + logger.warning(f"No stream_id found in Redis for channel {channel_id}") + except Exception as e: + logger.warning(f"Error loading stream ID from Redis: {e}") + else: + logger.warning(f"Unable to get stream ID for channel {channel_id} - stream switching may not work correctly") + logger.info(f"Initialized stream manager for channel {buffer.channel_id}") def _create_session(self): @@ -81,9 +114,13 @@ class StreamManager: return session def run(self): - """Main execution loop using HTTP streaming with improved connection handling""" + """Main execution loop using HTTP streaming with improved connection handling and stream switching""" # Add a stop flag to the class properties self.stop_requested = False + # Add tracking for stream switching attempts + stream_switch_attempts = 0 + # Get max stream switches from config using the helper method + max_stream_switches = ConfigHelper.max_stream_switches() # Prevent infinite switching loops try: # Check stream type before connecting @@ -102,200 +139,256 @@ class StreamManager: logger.info(f"Starting stream for URL: {self.url}") - while self.running: - if self.transcode: - if self.url_switching: - logger.debug("Skipping connection attempt during URL switch") - time.sleep(.1) - continue - # Generate transcode command - logger.debug(f"Building transcode command for channel {self.channel_id}") - channel = get_object_or_404(Channel, uuid=self.channel_id) + # Main stream switching loop - we'll try different streams if needed + while self.running and stream_switch_attempts <= max_stream_switches: + # Reset connection retry count for this specific URL + self.retry_count = 0 + url_failed = False + if self.url_switching: + logger.debug("Skipping connection attempt during URL switch") + time.sleep(0.1) + continue + # Connection retry loop for current URL + while self.running and self.retry_count < self.max_retries and not url_failed: - # Use FFmpeg specifically for HLS streams - if hasattr(self, 'force_ffmpeg') and self.force_ffmpeg: - from core.models import StreamProfile - try: - stream_profile = StreamProfile.objects.get(name='ffmpeg', locked=True) - logger.info("Using FFmpeg stream profile for HLS content") - except StreamProfile.DoesNotExist: - # Fall back to channel's profile if FFmpeg not found - stream_profile = channel.get_stream_profile() - logger.warning("FFmpeg profile not found, using channel default profile") - else: - stream_profile = channel.get_stream_profile() + logger.info(f"Connection attempt {self.retry_count + 1}/{self.max_retries} for URL: {self.url}") - self.transcode_cmd = stream_profile.build_command(self.url, self.user_agent) - # Start command process for transcoding - logger.debug(f"Starting transcode process: {self.transcode_cmd}") - self.transcode_process = subprocess.Popen( - self.transcode_cmd, - stdout=subprocess.PIPE, - stderr=subprocess.DEVNULL, # Suppress FFmpeg logs - bufsize=188 * 64 # Buffer optimized for TS packets - ) - self.socket = self.transcode_process.stdout # Read from FFmpeg output - self.connected = True - - if self.socket is not None: - # Set channel state to waiting for clients - self._set_waiting_for_clients() - - # Main fetch loop - while self.running and self.connected: - if self.fetch_chunk(): - self.last_data_time = time.time() - else: - if not self.running: - break - time.sleep(0.1) - else: + # Handle connection based on whether we transcode or not + connection_result = False try: - # Using direct HTTP streaming - if self.url_switching: - logger.debug("Skipping connection attempt during URL switch") - time.sleep(.1) - continue - logger.debug(f"Using TS Proxy to connect to stream: {self.url}") - # Create new session for each connection attempt - session = self._create_session() - self.current_session = session - - # Stream the URL with proper timeout handling - response = session.get( - self.url, - stream=True, - timeout=(10, 60) # 10s connect timeout, 60s read timeout - ) - self.current_response = response - - if response.status_code == 200: - self.connected = True - self.healthy = True - logger.info(f"Successfully connected to stream source") - - # Set channel state to waiting for clients - self._set_waiting_for_clients() - - # Process the stream in chunks with improved error handling - try: - chunk_count = 0 - for chunk in response.iter_content(chunk_size=self.chunk_size): - # Check if we've been asked to stop - if self.stop_requested: - logger.info(f"Stream loop for channel {self.channel_id} stopping due to request") - break - - if chunk: - # Add chunk to buffer with TS packet alignment - success = self.buffer.add_chunk(chunk) - - if success: - self.last_data_time = time.time() - chunk_count += 1 - - # Update last data timestamp in Redis - if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: - last_data_key = RedisKeys.last_data(self.buffer.channel_id) - self.buffer.redis_client.set(last_data_key, str(time.time()), ex=60) - except (AttributeError, ConnectionError) as e: - if self.stop_requested: - logger.debug(f"Expected connection error during shutdown: {e}") - elif hasattr(self, 'url_switching') and self.url_switching: - # This is expected during URL switching, just log at debug level - logger.debug(f"Expected connection error during URL switch: {e}") - else: - # Unexpected error during normal operation - logger.error(f"Unexpected stream error: {e}") - except Exception as e: - # Handle the specific 'NoneType' object has no attribute 'read' error - if "'NoneType' object has no attribute 'read'" in str(e): - logger.warning(f"Connection closed by server (read {chunk_count} chunks before disconnect)") - else: - # Re-raise unexpected AttributeErrors - logger.error(f"Unexpected AttributeError: {e}") - raise + if self.transcode: + connection_result = self._establish_transcode_connection() else: - logger.error(f"Failed to connect to stream: HTTP {response.status_code}") - time.sleep(2) + connection_result = self._establish_http_connection() - except requests.exceptions.ReadTimeout: - logger.warning("Read timeout - server stopped sending data") + if connection_result: + # Store connection start time to measure success duration + connection_start_time = time.time() + + # Successfully connected - read stream data until disconnect/error + self._process_stream_data() + # If we get here, the connection was closed/failed + + # Reset stream switch attempts if the connection lasted longer than threshold + # This indicates we had a stable connection for a while before failing + connection_duration = time.time() - connection_start_time + stable_connection_threshold = 30 # 30 seconds threshold + if connection_duration > stable_connection_threshold: + logger.info(f"Stream was stable for {connection_duration:.1f} seconds, resetting switch attempts counter") + stream_switch_attempts = 0 + + # Connection failed or ended - decide what to do next + if self.stop_requested or not self.running: + # Normal shutdown requested + return + + # Connection failed, increment retry count + self.retry_count += 1 self.connected = False - time.sleep(1) - except requests.RequestException as e: - logger.error(f"HTTP request error: {e}") + # If we've reached max retries, mark this URL as failed + if self.retry_count >= self.max_retries: + url_failed = True + logger.warning(f"Maximum retry attempts ({self.max_retries}) reached for URL: {self.url}") + else: + # Wait with exponential backoff before retrying + timeout = min(.25 ** self.retry_count, 3) # Cap at 3 seconds + logger.info(f"Reconnecting in {timeout} seconds... (attempt {self.retry_count}/{self.max_retries})") + time.sleep(timeout) + + except Exception as e: + logger.error(f"Connection error: {e}", exc_info=True) + self.retry_count += 1 self.connected = False - time.sleep(5) - finally: - # Clean up response and session - if self.current_response: - try: - self.current_response.close() - except Exception as e: - logger.debug(f"Error closing response: {e}") - self.current_response = None + if self.retry_count >= self.max_retries: + url_failed = True + else: + # Wait with exponential backoff before retrying + timeout = min(2 ** self.retry_count, 10) + logger.info(f"Reconnecting in {timeout} seconds after error... (attempt {self.retry_count}/{self.max_retries})") + time.sleep(timeout) - if self.current_session: - try: - self.current_session.close() - except Exception as e: - logger.debug(f"Error closing session: {e}") - self.current_session = None + # If URL failed and we're still running, try switching to another stream + if url_failed and self.running: + logger.info(f"URL {self.url} failed after {self.retry_count} attempts, trying next stream") - # Connection retry logic - if self.running and not self.connected: - self.retry_count += 1 - if self.retry_count > self.max_retries: - logger.error(f"Maximum retry attempts ({self.max_retries}) exceeded") + # Try to switch to next stream + switch_result = self._try_next_stream() + if switch_result: + # Successfully switched to a new stream, continue with the new URL + stream_switch_attempts += 1 + logger.info(f"Successfully switched to new URL: {self.url} (switch attempt {stream_switch_attempts}/{max_stream_switches})") + # Reset retry count for the new stream - important for the loop to work correctly + self.retry_count = 0 + # Continue outer loop with new URL - DON'T add a break statement here + else: + # No more streams to try + logger.error(f"Failed to find alternative streams after {stream_switch_attempts} attempts") break - - timeout = min(2 ** self.retry_count, 30) - - # When a connection fails and reconnect is needed: - self.reconnecting = True - - # Cancel all existing buffer timers during reconnect - for timer in list(self._buffer_check_timers): - try: - if timer and timer.is_alive(): - timer.cancel() - except Exception as e: - logger.error(f"Error canceling buffer timer: {e}") - self._buffer_check_timers = [] - - logger.info(f"Reconnecting in {timeout} seconds... (attempt {self.retry_count})") - time.sleep(timeout) - - self.reconnecting = False # Reset flag after sleep + elif not self.running: + # Normal shutdown was requested + break except Exception as e: logger.error(f"Stream error: {e}", exc_info=True) finally: self.connected = False - - if self.socket: - try: - self._close_socket() - except: - pass - - if self.current_response: - try: - self.current_response.close() - except: - pass - - if self.current_session: - try: - self.current_session.close() - except: - pass - + self._close_all_connections() logger.info(f"Stream manager stopped") + def _establish_transcode_connection(self): + """Establish a connection using transcoding""" + try: + logger.debug(f"Building transcode command for channel {self.channel_id}") + channel = get_object_or_404(Channel, uuid=self.channel_id) + + # Use FFmpeg specifically for HLS streams + if hasattr(self, 'force_ffmpeg') and self.force_ffmpeg: + from core.models import StreamProfile + try: + stream_profile = StreamProfile.objects.get(name='ffmpeg', locked=True) + logger.info("Using FFmpeg stream profile for HLS content") + except StreamProfile.DoesNotExist: + # Fall back to channel's profile if FFmpeg not found + stream_profile = channel.get_stream_profile() + logger.warning("FFmpeg profile not found, using channel default profile") + else: + stream_profile = channel.get_stream_profile() + + # Build and start transcode command + self.transcode_cmd = stream_profile.build_command(self.url, self.user_agent) + logger.debug(f"Starting transcode process: {self.transcode_cmd}") + + self.transcode_process = subprocess.Popen( + self.transcode_cmd, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, # Suppress FFmpeg logs + bufsize=188 * 64 # Buffer optimized for TS packets + ) + + self.socket = self.transcode_process.stdout # Read from FFmpeg output + self.connected = True + + # Set channel state to waiting for clients + self._set_waiting_for_clients() + + return True + except Exception as e: + logger.error(f"Error establishing transcode connection: {e}", exc_info=True) + self._close_socket() + return False + + def _establish_http_connection(self): + """Establish a direct HTTP connection to the stream""" + try: + logger.debug(f"Using TS Proxy to connect to stream: {self.url}") + + # Create new session for each connection attempt + session = self._create_session() + self.current_session = session + + # Stream the URL with proper timeout handling + response = session.get( + self.url, + stream=True, + timeout=(10, 60) # 10s connect timeout, 60s read timeout + ) + self.current_response = response + + if response.status_code == 200: + self.connected = True + self.healthy = True + logger.info(f"Successfully connected to stream source") + + # Set channel state to waiting for clients + self._set_waiting_for_clients() + + return True + else: + logger.error(f"Failed to connect to stream: HTTP {response.status_code}") + self._close_connection() + return False + except requests.exceptions.RequestException as e: + logger.error(f"HTTP request error: {e}") + self._close_connection() + return False + except Exception as e: + logger.error(f"Error establishing HTTP connection: {e}", exc_info=True) + self._close_connection() + return False + + def _process_stream_data(self): + """Process stream data until disconnect or error""" + try: + if self.transcode: + # Handle transcoded stream data + while self.running and self.connected: + if self.fetch_chunk(): + self.last_data_time = time.time() + else: + if not self.running: + break + time.sleep(0.1) + else: + # Handle direct HTTP connection + chunk_count = 0 + try: + for chunk in self.current_response.iter_content(chunk_size=self.chunk_size): + # Check if we've been asked to stop + if self.stop_requested or self.url_switching: + break + + if chunk: + # Add chunk to buffer with TS packet alignment + success = self.buffer.add_chunk(chunk) + + if success: + self.last_data_time = time.time() + chunk_count += 1 + + # Update last data timestamp in Redis + if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: + last_data_key = RedisKeys.last_data(self.buffer.channel_id) + self.buffer.redis_client.set(last_data_key, str(time.time()), ex=60) + except (AttributeError, ConnectionError) as e: + if self.stop_requested or self.url_switching: + logger.debug(f"Expected connection error during shutdown/URL switch: {e}") + else: + logger.error(f"Unexpected stream error: {e}") + raise + except Exception as e: + logger.error(f"Error processing stream data: {e}", exc_info=True) + + # If we exit the loop, connection is closed or failed + self.connected = False + + def _close_all_connections(self): + """Close all connection resources""" + if self.socket: + try: + self._close_socket() + except Exception as e: + logger.debug(f"Error closing socket: {e}") + + if self.current_response: + try: + self.current_response.close() + except Exception as e: + logger.debug(f"Error closing response: {e}") + + if self.current_session: + try: + self.current_session.close() + except Exception as e: + logger.debug(f"Error closing session: {e}") + + # Clear references + self.socket = None + self.current_response = None + self.current_session = None + self.transcode_process = None + def stop(self): """Stop the stream manager and cancel all timers""" # Add at the beginning of your stop method @@ -359,6 +452,9 @@ class StreamManager: # Reset retry counter to allow immediate reconnect self.retry_count = 0 + # Reset tried streams when manually switching URL + self.tried_stream_ids = set() + # Also reset buffer position to prevent stale data after URL change if hasattr(self.buffer, 'reset_buffer_position'): try: @@ -582,3 +678,89 @@ class StreamManager: except Exception as e: logger.error(f"Error in buffer check: {e}") + def _try_next_stream(self): + """ + Try to switch to the next available stream for this channel. + + Returns: + bool: True if successfully switched to a new stream, False otherwise + """ + try: + logger.info(f"Trying to find alternative stream for channel {self.channel_id}, current stream ID: {self.current_stream_id}") + + # Get alternate streams excluding the current one + alternate_streams = get_alternate_streams(self.channel_id, self.current_stream_id) + logger.info(f"Found {len(alternate_streams)} potential alternate streams for channel {self.channel_id}") + + # Filter out streams we've already tried + untried_streams = [s for s in alternate_streams if s['stream_id'] not in self.tried_stream_ids] + if untried_streams: + ids_to_try = ', '.join([str(s['stream_id']) for s in untried_streams]) + logger.info(f"Found {len(untried_streams)} untried streams for channel {self.channel_id}: [{ids_to_try}]") + else: + logger.warning(f"No untried streams available for channel {self.channel_id}, tried: {self.tried_stream_ids}") + + if not untried_streams: + # Check if we have streams but they've all been tried + if alternate_streams and len(self.tried_stream_ids) > 0: + logger.warning(f"All {len(alternate_streams)} alternate streams have been tried for channel {self.channel_id}") + return False + + # Get the next stream to try + next_stream = untried_streams[0] + stream_id = next_stream['stream_id'] + + # Add to tried streams + self.tried_stream_ids.add(stream_id) + + # Get stream info including URL + logger.info(f"Trying next stream ID {stream_id} for channel {self.channel_id}") + stream_info = get_stream_info_for_switch(self.channel_id, stream_id) + + if 'error' in stream_info or not stream_info.get('url'): + logger.error(f"Error getting info for stream {stream_id}: {stream_info.get('error', 'No URL')}") + return False + + # Update URL and user agent + new_url = stream_info['url'] + new_user_agent = stream_info['user_agent'] + new_transcode = stream_info['transcode'] + + logger.info(f"Switching from URL {self.url} to {new_url} for channel {self.channel_id}") + + # Update stream ID tracking + self.current_stream_id = stream_id + + # Store the new user agent and transcode settings + self.user_agent = new_user_agent + self.transcode = new_transcode + + # Update stream metadata in Redis + 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, mapping={ + "url": new_url, + "user_agent": new_user_agent, + "profile": stream_info['profile'], + "stream_id": str(stream_id), + "stream_switch_time": str(time.time()), + "stream_switch_reason": "max_retries_exceeded" + }) + + # Log the switch + logger.info(f"Stream metadata updated for channel {self.channel_id} to stream ID {stream_id}") + + # IMPORTANT: Just update the URL, don't stop the channel or release resources + switch_result = self.update_url(new_url) + if not switch_result: + logger.error(f"Failed to update URL for stream ID {stream_id}") + return False + + logger.info(f"Successfully switched to stream ID {stream_id} with URL {new_url}") + return True + + except Exception as e: + logger.error(f"Error trying next stream for channel {self.channel_id}: {e}", exc_info=True) + return False + + diff --git a/apps/proxy/ts_proxy/url_utils.py b/apps/proxy/ts_proxy/url_utils.py index cbce957c..e478c1c0 100644 --- a/apps/proxy/ts_proxy/url_utils.py +++ b/apps/proxy/ts_proxy/url_utils.py @@ -4,7 +4,7 @@ Utilities for handling stream URLs and transformations. import logging import re -from typing import Optional, Tuple +from typing import Optional, Tuple, List from django.shortcuts import get_object_or_404 from apps.channels.models import Channel, Stream from apps.m3u.models import M3UAccount, M3UAccountProfile @@ -100,30 +100,147 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int] Returns: dict: Stream information including URL, user agent and transcode flag """ - channel = get_object_or_404(Channel, uuid=channel_id) + try: + channel = get_object_or_404(Channel, uuid=channel_id) - # Use the target stream if specified, otherwise use current stream - if target_stream_id: - stream_id = target_stream_id - # Find a compatible profile for this stream - profiles = M3UAccountProfile.objects.filter(stream=stream_id) - if not profiles.exists(): - logger.error(f"No profile found for stream {stream_id}") - return {'error': 'No profile found for stream'} - profile_id = profiles.first().id - else: - stream_id, profile_id = channel.get_stream() - if stream_id is None or profile_id is None: - return {'error': 'No stream assigned to channel'} + # Use the target stream if specified, otherwise use current stream + if target_stream_id: + stream_id = target_stream_id - # Generate the URL using our utility - stream_url, user_agent, transcode, profile_value = generate_stream_url(channel_id) + # Get the stream object + stream = get_object_or_404(Stream, pk=stream_id) - return { - 'url': stream_url, - 'user_agent': user_agent, - 'transcode': transcode, - 'profile': profile_value, - 'stream_id': stream_id, - 'profile_id': profile_id - } + # Find compatible profile for this stream + profiles = M3UAccountProfile.objects.filter(m3u_account=stream.m3u_account) + + if not profiles.exists(): + # Try to get default profile + default_profile = M3UAccountProfile.objects.filter( + m3u_account=stream.m3u_account, + is_default=True + ).first() + + if default_profile: + profile_id = default_profile.id + else: + logger.error(f"No profile found for stream {stream_id}") + return {'error': 'No profile found for stream'} + else: + # Use first available profile + profile_id = profiles.first().id + else: + stream_id, profile_id = channel.get_stream() + if stream_id is None or profile_id is None: + return {'error': 'No stream assigned to channel'} + + # Get the stream and profile objects directly + stream = get_object_or_404(Stream, pk=stream_id) + profile = get_object_or_404(M3UAccountProfile, pk=profile_id) + + # Get the user agent from the M3U account + m3u_account = M3UAccount.objects.get(id=profile.m3u_account.id) + user_agent = UserAgent.objects.get(id=m3u_account.user_agent.id).user_agent + if not user_agent: + user_agent = UserAgent.objects.get(id=CoreSettings.get_default_user_agent_id()).user_agent + + # Generate URL using the transform function directly + stream_url = transform_url(stream.url, profile.search_pattern, profile.replace_pattern) + + # Get transcode info from the channel's stream profile + stream_profile = channel.get_stream_profile() + transcode = not (stream_profile.is_proxy() or stream_profile is None) + profile_value = str(stream_profile) + + return { + 'url': stream_url, + 'user_agent': user_agent, + 'transcode': transcode, + 'profile': profile_value, + 'stream_id': stream_id, + 'profile_id': profile_id + } + except Exception as e: + logger.error(f"Error getting stream info for switch: {e}", exc_info=True) + return {'error': f'Error: {str(e)}'} + +def get_alternate_streams(channel_id: str, current_stream_id: Optional[int] = None) -> List[dict]: + """ + Get alternative streams for a channel when the current stream fails. + + Args: + channel_id: The UUID of the channel + current_stream_id: The currently failing stream ID to exclude + + Returns: + List[dict]: List of stream information dictionaries with stream_id and profile_id + """ + try: + # Get channel object + channel = get_object_or_404(Channel, uuid=channel_id) + logger.debug(f"Looking for alternate streams for channel {channel_id}, current stream ID: {current_stream_id}") + + # Get all assigned streams for this channel + streams = channel.streams.all() + logger.debug(f"Channel {channel_id} has {streams.count()} total assigned streams") + + if not streams.exists(): + logger.warning(f"No streams assigned to channel {channel_id}") + return [] + + alternate_streams = [] + + # Process each stream + for stream in streams: + # Log each stream we're checking + logger.debug(f"Checking stream ID {stream.id} ({stream.name}) for channel {channel_id}") + + # Skip the current failing stream + if current_stream_id and stream.id == current_stream_id: + logger.debug(f"Skipping current stream ID {current_stream_id}") + continue + + # Find compatible profiles for this stream + # FIX: Looking at the error message, M3UAccountProfile doesn't have a 'stream' field + # We need to find which field relates M3UAccountProfile to Stream + try: + # Check if we can find profiles via m3u_account + profiles = M3UAccountProfile.objects.filter(m3u_account=stream.m3u_account) + if not profiles.exists(): + logger.debug(f"No profiles found via m3u_account for stream {stream.id}") + # Fallback to the default profile of the account + default_profile = M3UAccountProfile.objects.filter( + m3u_account=stream.m3u_account, + is_default=True + ).first() + if default_profile: + profiles = [default_profile] + else: + logger.warning(f"No default profile found for m3u_account {stream.m3u_account.id}") + continue + + # Get first compatible profile + profile = profiles.first() + if profile: + logger.debug(f"Found compatible profile ID {profile.id} for stream ID {stream.id}") + + alternate_streams.append({ + 'stream_id': stream.id, + 'profile_id': profile.id, + 'name': stream.name + }) + else: + logger.debug(f"No compatible profile found for stream ID {stream.id}") + except Exception as inner_e: + logger.error(f"Error finding profiles for stream {stream.id}: {inner_e}") + continue + + if alternate_streams: + stream_ids = ', '.join([str(s['stream_id']) for s in alternate_streams]) + logger.info(f"Found {len(alternate_streams)} alternate streams for channel {channel_id}: [{stream_ids}]") + else: + logger.warning(f"No alternate streams found for channel {channel_id}") + + return alternate_streams + except Exception as e: + logger.error(f"Error getting alternate streams for channel {channel_id}: {e}", exc_info=True) + return [] diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index f84963bb..3b4a966c 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -58,6 +58,10 @@ def stream_ts(request, channel_id): if stream_url is None: return JsonResponse({'error': 'Channel not available'}, status=404) + # Get the stream ID from the channel + stream_id, profile_id = channel.get_stream() + logger.info(f"Channel {channel_id} using stream ID {stream_id}, profile ID {profile_id}") + # Generate transcode command if needed stream_profile = channel.get_stream_profile() if stream_profile.is_redirect(): @@ -65,7 +69,7 @@ def stream_ts(request, channel_id): # Initialize channel with the stream's user agent (not the client's) success = ChannelService.initialize_channel( - channel_id, stream_url, stream_user_agent, transcode, profile_value + channel_id, stream_url, stream_user_agent, transcode, profile_value, stream_id ) if not success: From 6d84821942d723ed238ca66dca91823f3094e2b8 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 20 Mar 2025 15:56:34 -0500 Subject: [PATCH 3/4] Adds better checks for a channel that is partially initialized and clears it. --- apps/proxy/ts_proxy/redis_keys.py | 5 ++ apps/proxy/ts_proxy/server.py | 86 ++++++++++++++++++- .../ts_proxy/services/channel_service.py | 65 +++++++++++++- apps/proxy/ts_proxy/views.py | 31 ++++++- 4 files changed, 181 insertions(+), 6 deletions(-) diff --git a/apps/proxy/ts_proxy/redis_keys.py b/apps/proxy/ts_proxy/redis_keys.py index 3b5a3ae8..ebbcbc24 100644 --- a/apps/proxy/ts_proxy/redis_keys.py +++ b/apps/proxy/ts_proxy/redis_keys.py @@ -73,3 +73,8 @@ class RedisKeys: def switch_status(channel_id): """Key for stream switch status""" return f"ts_proxy:channel:{channel_id}:switch_status" + + @staticmethod + def worker_heartbeat(worker_id): + """Key for worker heartbeat""" + return f"ts_proxy:worker:{worker_id}:heartbeat" diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index 4c7897c1..aa2e7ffc 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -498,7 +498,10 @@ class ProxyServer: return False def check_if_channel_exists(self, channel_id): - """Check if a channel exists using standardized key structure""" + """ + Check if a channel exists and is in a valid state. + Enhanced to detect zombie channels after server restarts. + """ # Check local memory first if channel_id in self.stream_managers or channel_id in self.stream_buffers: return True @@ -508,9 +511,48 @@ class ProxyServer: # Primary check - look for channel metadata metadata_key = RedisKeys.channel_metadata(channel_id) - # If metadata exists, return true + # If metadata exists, validate it's in a healthy state if self.redis_client.exists(metadata_key): - return True + metadata = self.redis_client.hgetall(metadata_key) + + # Get channel state and owner + state = metadata.get(b'state', b'unknown').decode('utf-8') + owner = metadata.get(b'owner', b'').decode('utf-8') + + # States that indicate the channel is running properly + valid_states = [ChannelState.ACTIVE, ChannelState.WAITING_FOR_CLIENTS, ChannelState.CONNECTING] + + # If the channel is in a valid state, check if the owner is still active + if state in valid_states: + # Check if owner still exists by checking heartbeat + owner_heartbeat_key = f"ts_proxy:worker:{owner}:heartbeat" + owner_alive = self.redis_client.exists(owner_heartbeat_key) + + if owner_alive: + return True + else: + # This is a zombie channel - owner is gone but metadata still exists + logger.warning(f"Detected zombie channel {channel_id} - owner {owner} is no longer active") + self._clean_zombie_channel(channel_id, metadata) + return False + elif state in [ChannelState.STOPPING, ChannelState.STOPPED, ChannelState.ERROR]: + # These states indicate the channel should be reinitialized + logger.info(f"Channel {channel_id} exists but in terminal state: {state}") + return False + else: + # Unknown or initializing state, check how long it's been in this state + if b'state_changed_at' in metadata: + state_changed_at = float(metadata[b'state_changed_at'].decode('utf-8')) + state_age = time.time() - state_changed_at + + # If in initializing state for too long, consider it stale + if state_age > 60: # 60 seconds threshold + logger.warning(f"Channel {channel_id} stuck in {state} state for {state_age:.1f}s - treating as zombie") + self._clean_zombie_channel(channel_id, metadata) + return False + + # Otherwise assume it's still in progress + return True # Additional checks if metadata doesn't exist additional_keys = [ @@ -521,10 +563,41 @@ class ProxyServer: for key in additional_keys: if self.redis_client.exists(key): - return True + # Found orphaned keys without metadata - clean them up + logger.warning(f"Found orphaned keys for channel {channel_id} without metadata - cleaning up") + self._clean_redis_keys(channel_id) + return False return False + def _clean_zombie_channel(self, channel_id, metadata=None): + """Clean up a zombie channel (channel with Redis keys but no active owner)""" + try: + logger.info(f"Cleaning up zombie channel {channel_id}") + + # If we have metadata, log details for debugging + if metadata: + state = metadata.get(b'state', b'unknown').decode('utf-8') + owner = metadata.get(b'owner', b'unknown').decode('utf-8') + logger.info(f"Zombie channel details - state: {state}, owner: {owner}") + + # Clean up Redis keys + self._clean_redis_keys(channel_id) + + # Force release resources in the Channel model + try: + from apps.channels.models import Channel + channel = Channel.objects.get(uuid=channel_id) + channel.release_stream() + logger.info(f"Released stream allocation for zombie channel {channel_id}") + except Exception as e: + logger.error(f"Error releasing stream for zombie channel {channel_id}: {e}") + + return True + except Exception as e: + logger.error(f"Error cleaning zombie channel {channel_id}: {e}", exc_info=True) + return False + def stop_channel(self, channel_id): """Stop a channel with proper ownership handling""" try: @@ -640,6 +713,11 @@ class ProxyServer: def cleanup_task(): while True: try: + # Send worker heartbeat first + if self.redis_client: + worker_heartbeat_key = f"ts_proxy:worker:{self.worker_id}:heartbeat" + self.redis_client.setex(worker_heartbeat_key, 30, str(time.time())) + # Refresh channel registry self.refresh_channel_registry() diff --git a/apps/proxy/ts_proxy/services/channel_service.py b/apps/proxy/ts_proxy/services/channel_service.py index 3fe6aedf..e00e680d 100644 --- a/apps/proxy/ts_proxy/services/channel_service.py +++ b/apps/proxy/ts_proxy/services/channel_service.py @@ -11,7 +11,7 @@ from apps.channels.models import Channel from apps.proxy.config import TSConfig as Config from .. import proxy_server from ..redis_keys import RedisKeys -from ..constants import EventType +from ..constants import EventType, ChannelState from ..url_utils import get_stream_info_for_switch logger = logging.getLogger("ts_proxy") @@ -318,6 +318,69 @@ class ChannelService: 'event_published': event_published } + @staticmethod + def validate_channel_state(channel_id): + """ + Validate if a channel is in a healthy state and has an active owner. + + Args: + channel_id: UUID of the channel + + Returns: + tuple: (valid, state, owner, details) - validity status, current state, owner, and diagnostic info + """ + if not proxy_server.redis_client: + return False, None, None, {"error": "Redis not available"} + + try: + metadata_key = RedisKeys.channel_metadata(channel_id) + if not proxy_server.redis_client.exists(metadata_key): + return False, None, None, {"error": "No channel metadata"} + + metadata = proxy_server.redis_client.hgetall(metadata_key) + + # Extract state and owner + state = metadata.get(b'state', b'unknown').decode('utf-8') + owner = metadata.get(b'owner', b'unknown').decode('utf-8') + + # Valid states indicate channel is running properly + valid_states = [ChannelState.ACTIVE, ChannelState.WAITING_FOR_CLIENTS, ChannelState.CONNECTING] + + if state not in valid_states: + return False, state, owner, {"error": f"Invalid state: {state}"} + + # Check if owner is still active + owner_heartbeat_key = f"ts_proxy:worker:{owner}:heartbeat" + owner_alive = proxy_server.redis_client.exists(owner_heartbeat_key) + + if not owner_alive: + return False, state, owner, {"error": "Owner not active"} + + # Check for recent activity + last_data_key = RedisKeys.last_data(channel_id) + last_data = proxy_server.redis_client.get(last_data_key) + + details = { + "state": state, + "owner": owner, + "owner_alive": owner_alive + } + + if last_data: + last_data_time = float(last_data.decode('utf-8')) + data_age = time.time() - last_data_time + details["last_data_age"] = data_age + + # If no data for too long, consider invalid + if data_age > 30: # 30 seconds threshold + return False, state, owner, {"error": f"No data for {data_age:.1f}s", **details} + + return True, state, owner, details + + except Exception as e: + logger.error(f"Error validating channel state: {e}", exc_info=True) + return False, None, None, {"error": f"Exception: {str(e)}"} + # Helper methods for Redis operations @staticmethod diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index 3b4a966c..75ef4fc5 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -47,9 +47,38 @@ def stream_ts(request, channel_id): logger.debug(f"[{client_id}] Client connected with user agent: {client_user_agent}") break + # Check if we need to reinitialize the channel + needs_initialization = True + channel_state = None + + # Get current channel state from Redis if available + if proxy_server.redis_client: + metadata_key = RedisKeys.channel_metadata(channel_id) + if proxy_server.redis_client.exists(metadata_key): + metadata = proxy_server.redis_client.hgetall(metadata_key) + if b'state' in metadata: + channel_state = metadata[b'state'].decode('utf-8') + + # Only skip initialization if channel is in a healthy state + valid_states = [ChannelState.ACTIVE, ChannelState.WAITING_FOR_CLIENTS] + if channel_state in valid_states: + # Verify the owner is still active + if b'owner' in metadata: + owner = metadata[b'owner'].decode('utf-8') + owner_heartbeat_key = f"ts_proxy:worker:{owner}:heartbeat" + if proxy_server.redis_client.exists(owner_heartbeat_key): + # Owner is active and channel is in good state + needs_initialization = False + logger.info(f"[{client_id}] Channel {channel_id} in state {channel_state} with active owner {owner}") + # Start initialization if needed channel_initializing = False - if not proxy_server.check_if_channel_exists(channel_id): + if needs_initialization or not proxy_server.check_if_channel_exists(channel_id): + # Force cleanup of any previous instance + if channel_state in [ChannelState.ERROR, ChannelState.STOPPING, ChannelState.STOPPED]: + logger.warning(f"[{client_id}] Channel {channel_id} in state {channel_state}, forcing cleanup") + proxy_server.stop_channel(channel_id) + # Initialize the channel (but don't wait for completion) logger.info(f"[{client_id}] Starting channel {channel_id} initialization") From ed00f2f8767ca33e286da611db7c532eb85a4244 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 20 Mar 2025 16:12:08 -0500 Subject: [PATCH 4/4] Fixed streams not switching to FFmpeg if HLS is detected. --- apps/proxy/ts_proxy/stream_manager.py | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/apps/proxy/ts_proxy/stream_manager.py b/apps/proxy/ts_proxy/stream_manager.py index 281ba842..93ea4960 100644 --- a/apps/proxy/ts_proxy/stream_manager.py +++ b/apps/proxy/ts_proxy/stream_manager.py @@ -123,15 +123,7 @@ class StreamManager: max_stream_switches = ConfigHelper.max_stream_switches() # Prevent infinite switching loops try: - # Check stream type before connecting - stream_type = detect_stream_type(self.url) - if self.transcode == False and stream_type == StreamType.HLS: - logger.info(f"Detected HLS stream: {self.url}") - logger.info(f"HLS streams will be handled with FFmpeg for now - future version will support HLS natively") - # Enable transcoding for HLS streams - self.transcode = True - # We'll override the stream profile selection with ffmpeg in the transcoding section - self.force_ffmpeg = True + # Start health monitor thread health_thread = threading.Thread(target=self._monitor_health, daemon=True) @@ -141,6 +133,15 @@ class StreamManager: # Main stream switching loop - we'll try different streams if needed while self.running and stream_switch_attempts <= max_stream_switches: + # Check stream type before connecting + stream_type = detect_stream_type(self.url) + if self.transcode == False and stream_type == StreamType.HLS: + logger.info(f"Detected HLS stream: {self.url}") + logger.info(f"HLS streams will be handled with FFmpeg for now - future version will support HLS natively") + # Enable transcoding for HLS streams + self.transcode = True + # We'll override the stream profile selection with ffmpeg in the transcoding section + self.force_ffmpeg = True # Reset connection retry count for this specific URL self.retry_count = 0 url_failed = False