diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py index f566c4ab..8f1c09d9 100644 --- a/apps/proxy/ts_proxy/stream_generator.py +++ b/apps/proxy/ts_proxy/stream_generator.py @@ -354,6 +354,33 @@ class StreamGenerator: 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: + metadata_key = RedisKeys.channel_metadata(self.channel_id) + metadata = proxy_server.redis_client.hgetall(metadata_key) + if metadata: + stream_id_bytes = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.STREAM_ID) + if stream_id_bytes: + stream_id = int(stream_id_bytes.decode('utf-8')) + + # Check if we're the last client + if self.channel_id in proxy_server.client_managers: + client_count = proxy_server.client_managers[self.channel_id].get_total_client_count() + # Only the last client or owner should release the stream + if client_count <= 1 and proxy_server.am_i_owner(self.channel_id): + from apps.channels.models import Stream + try: + stream = Stream.objects.get(pk=stream_id) + stream.release_stream() + stream_released = True + logger.debug(f"[{self.client_id}] Released stream {stream_id} for channel {self.channel_id}") + except Exception as e: + logger.error(f"[{self.client_id}] Error releasing stream {stream_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] local_clients = client_manager.remove_client(self.client_id) @@ -361,7 +388,8 @@ class StreamGenerator: logger.info(f"[{self.client_id}] Disconnected after {elapsed:.2f}s (local: {local_clients}, total: {total_clients})") # Schedule channel shutdown if no clients left - self._schedule_channel_shutdown_if_needed(local_clients) + if not stream_released: # Only if we haven't already released the stream + self._schedule_channel_shutdown_if_needed(local_clients) def _schedule_channel_shutdown_if_needed(self, local_clients): """ diff --git a/apps/proxy/ts_proxy/stream_manager.py b/apps/proxy/ts_proxy/stream_manager.py index 317ec176..5d06c079 100644 --- a/apps/proxy/ts_proxy/stream_manager.py +++ b/apps/proxy/ts_proxy/stream_manager.py @@ -490,6 +490,21 @@ class StreamManager: # Add at the beginning of your stop method self.stopping = True + # Release stream resources if we're the owner + if self.current_stream_id and hasattr(self, 'worker_id') and self.worker_id: + if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: + owner_key = RedisKeys.channel_owner(self.channel_id) + current_owner = self.buffer.redis_client.get(owner_key) + + if current_owner and current_owner.decode('utf-8') == self.worker_id: + try: + from apps.channels.models import Stream + stream = Stream.objects.get(pk=self.current_stream_id) + stream.release_stream() + logger.info(f"Released stream {self.current_stream_id} for channel {self.channel_id}") + except Exception as e: + logger.error(f"Error releasing stream {self.current_stream_id}: {e}") + # Cancel all buffer check timers for timer in list(self._buffer_check_timers): try: diff --git a/apps/proxy/ts_proxy/url_utils.py b/apps/proxy/ts_proxy/url_utils.py index 641a794e..6f1b6495 100644 --- a/apps/proxy/ts_proxy/url_utils.py +++ b/apps/proxy/ts_proxy/url_utils.py @@ -24,7 +24,7 @@ def get_stream_object(id: str): logger.info(f"Fetching stream hash {id}") return get_object_or_404(Stream, stream_hash=id) -def generate_stream_url(channel_id: str) -> Tuple[str, str, bool]: +def generate_stream_url(channel_id: str) -> Tuple[str, str, bool, Optional[int]]: """ Generate the appropriate stream URL for a channel based on its profile settings. @@ -32,7 +32,7 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool]: channel_id: The UUID of the channel Returns: - Tuple[str, str, bool]: (stream_url, user_agent, transcode_flag) + Tuple[str, str, bool, Optional[int]]: (stream_url, user_agent, transcode_flag, profile_id) """ # Get channel and related objects channel = get_stream_object(channel_id) @@ -40,7 +40,7 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool]: if stream_id is None or profile_id is None: logger.error(f"No stream assigned to channel {channel_id}") - return None, None, False + return None, None, False, None # Get the M3U account profile for URL pattern stream = get_object_or_404(Stream, pk=stream_id) diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index ad3ccbc9..de140035 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -89,7 +89,16 @@ def stream_ts(request, channel_id): # 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) + # Make sure to release any stream locks that might have been acquired + if hasattr(channel, 'streams') and channel.streams.exists(): + for stream in channel.streams.all(): + try: + stream.release_stream() + logger.info(f"[{client_id}] Released stream {stream.id} for channel {channel_id}") + except Exception as e: + logger.error(f"[{client_id}] Error releasing stream: {e}") + + return JsonResponse({'error': 'No available streams for this channel'}, status=404) # Get the stream ID from the channel stream_id, m3u_profile_id = channel.get_stream()