From 86e6c942d79cf8c3e0ed05adfb2fd5fdbf21dc57 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Wed, 19 Mar 2025 20:22:13 -0500 Subject: [PATCH] Final refactoring and minor bug fix for transcode stream switching. --- apps/proxy/ts_proxy/channel_status.py | 35 ++++++++----------- apps/proxy/ts_proxy/client_manager.py | 11 +++--- apps/proxy/ts_proxy/server.py | 4 +-- .../ts_proxy/services/channel_service.py | 7 ++-- apps/proxy/ts_proxy/stream_manager.py | 16 +++++---- 5 files changed, 36 insertions(+), 37 deletions(-) diff --git a/apps/proxy/ts_proxy/channel_status.py b/apps/proxy/ts_proxy/channel_status.py index c9c896e0..cc8a1e5b 100644 --- a/apps/proxy/ts_proxy/channel_status.py +++ b/apps/proxy/ts_proxy/channel_status.py @@ -2,6 +2,8 @@ import logging import time import re from . import proxy_server +from .redis_keys import RedisKeys +from .constants import TS_PACKET_SIZE logger = logging.getLogger("ts_proxy") @@ -9,22 +11,15 @@ class ChannelStatus: def get_detailed_channel_info(channel_id): # Get channel metadata - metadata_key = f"ts_proxy:channel:{channel_id}:metadata" - metadata = proxy_server.redis_client.hgetall(metadata_key) - - if not metadata: - return None - - # Get detailed info - existing implementation - # Get channel metadata - metadata_key = f"ts_proxy:channel:{channel_id}:metadata" + metadata_key = RedisKeys.channel_metadata(channel_id) metadata = proxy_server.redis_client.hgetall(metadata_key) if not metadata: return None # Basic channel info - buffer_index_value = proxy_server.redis_client.get(f"ts_proxy:channel:{channel_id}:buffer:index") + buffer_index_key = RedisKeys.buffer_index(channel_id) + buffer_index_value = proxy_server.redis_client.get(buffer_index_key) info = { 'channel_id': channel_id, @@ -33,8 +28,6 @@ class ChannelStatus: 'profile': metadata.get(b'profile', b'unknown').decode('utf-8'), 'started_at': metadata.get(b'init_time', b'0').decode('utf-8'), 'owner': metadata.get(b'owner', b'unknown').decode('utf-8'), - - # Properly decode the buffer index value 'buffer_index': int(buffer_index_value.decode('utf-8')) if buffer_index_value else 0, } @@ -50,7 +43,7 @@ class ChannelStatus: info['uptime'] = time.time() - created_at # Get client information - client_set_key = f"ts_proxy:channel:{channel_id}:clients" + client_set_key = RedisKeys.clients(channel_id) client_ids = proxy_server.redis_client.smembers(client_set_key) clients = [] @@ -97,7 +90,7 @@ class ChannelStatus: # Check if the keys exist before getting for i in range(info['buffer_index']-sample_chunks+1, info['buffer_index']+1): - chunk_key = f"ts_proxy:channel:{channel_id}:buffer:chunk:{i}" + chunk_key = RedisKeys.buffer_chunk(channel_id, i) # Check if key exists first if proxy_server.redis_client.exists(chunk_key): @@ -135,9 +128,9 @@ class ChannelStatus: buffer_stats['total_sample_bytes'] = total_data # Add TS packet analysis - total_ts_packets = total_data // 188 + total_ts_packets = total_data // TS_PACKET_SIZE buffer_stats['estimated_ts_packets'] = total_ts_packets - buffer_stats['is_ts_aligned'] = all(size % 188 == 0 for size in chunk_sizes) + buffer_stats['is_ts_aligned'] = all(size % TS_PACKET_SIZE == 0 for size in chunk_sizes) else: # If no chunks found, scan for keys to help debug all_buffer_keys = [] @@ -161,7 +154,7 @@ class ChannelStatus: buffer_stats['diagnostics']['exception'] = str(e) # Add TTL information to see if chunks are expiring - chunk_ttl_key = f"ts_proxy:channel:{channel_id}:buffer:chunk:{info['buffer_index']}" + chunk_ttl_key = RedisKeys.buffer_chunk(channel_id, info['buffer_index']) chunk_ttl = proxy_server.redis_client.ttl(chunk_ttl_key) buffer_stats['latest_chunk_ttl'] = chunk_ttl @@ -182,17 +175,18 @@ class ChannelStatus: # Function for basic channel info (used for all channels summary) def get_basic_channel_info(channel_id): # Get channel metadata - metadata_key = f"ts_proxy:channel:{channel_id}:metadata" + metadata_key = RedisKeys.channel_metadata(channel_id) metadata = proxy_server.redis_client.hgetall(metadata_key) if not metadata: return None # Basic channel info only - omit diagnostics and details - buffer_index_value = proxy_server.redis_client.get(f"ts_proxy:channel:{channel_id}:buffer:index") + buffer_index_key = RedisKeys.buffer_index(channel_id) + buffer_index_value = proxy_server.redis_client.get(buffer_index_key) # Count clients (using efficient count method) - client_set_key = f"ts_proxy:channel:{channel_id}:clients" + client_set_key = RedisKeys.clients(channel_id) client_count = proxy_server.redis_client.scard(client_set_key) or 0 # Calculate uptime @@ -218,7 +212,6 @@ class ChannelStatus: # Get concise client information clients = [] - client_set_key = f"ts_proxy:channel:{channel_id}:clients" client_ids = proxy_server.redis_client.smembers(client_set_key) # Process only if we have clients and keep it limited diff --git a/apps/proxy/ts_proxy/client_manager.py b/apps/proxy/ts_proxy/client_manager.py index d756890d..ed5868a9 100644 --- a/apps/proxy/ts_proxy/client_manager.py +++ b/apps/proxy/ts_proxy/client_manager.py @@ -146,7 +146,7 @@ class ClientManager: self._registered_clients.add(client_id) - # FIX: Consistent key naming - note the 's' in 'clients' + # Use a function to get the client key client_key = f"ts_proxy:channel:{self.channel_id}:clients:{client_id}" # Prepare client data @@ -175,7 +175,8 @@ class ClientManager: self.redis_client.expire(self.client_set_key, self.client_ttl) # Clear any initialization timer - self.redis_client.delete(f"ts_proxy:channel:{self.channel_id}:init_time") + init_key = f"ts_proxy:channel:{self.channel_id}:init_time" + self.redis_client.delete(init_key) self._notify_owner_of_activity() @@ -195,7 +196,7 @@ class ClientManager: logger.debug(f"No user agent provided for client {client_id}") self.redis_client.publish( - f"ts_proxy:events:{self.channel_id}", + RedisKeys.events_channel(self.channel_id), # Use RedisKeys instead of string json.dumps(event_data) ) @@ -236,7 +237,7 @@ class ClientManager: logger.warning(f"Last client removed: {client_id} - channel may shut down soon") # Trigger disconnect time tracking even if we're not the owner - disconnect_key = f"ts_proxy:channel:{self.channel_id}:last_client_disconnect_time" + disconnect_key = RedisKeys.last_client_disconnect(self.channel_id) self.redis_client.setex(disconnect_key, 60, str(time.time())) self._notify_owner_of_activity() @@ -250,7 +251,7 @@ class ClientManager: "timestamp": time.time(), "remaining_clients": remaining }) - self.redis_client.publish(f"ts_proxy:events:{self.channel_id}", event_data) + self.redis_client.publish(RedisKeys.events_channel(self.channel_id), event_data) total_clients = self.get_total_client_count() logger.info(f"Client disconnected: {client_id} (local: {len(self.clients)}, total: {total_clients})") diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index 92449be4..427b2089 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -596,8 +596,8 @@ class ProxyServer: def _cleanup_channel(self, channel_id: str) -> None: """Remove channel resources""" - for collection in [self.stream_managers, self.stream_buffers, - self.client_managers, self.fetch_threads]: + # Removed reference to non-existent fetch_threads collection + for collection in [self.stream_managers, self.stream_buffers, self.client_managers]: collection.pop(channel_id, None) def shutdown(self) -> None: diff --git a/apps/proxy/ts_proxy/services/channel_service.py b/apps/proxy/ts_proxy/services/channel_service.py index 5b7a603b..0c190c32 100644 --- a/apps/proxy/ts_proxy/services/channel_service.py +++ b/apps/proxy/ts_proxy/services/channel_service.py @@ -11,6 +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 logger = logging.getLogger("ts_proxy") @@ -317,7 +318,7 @@ class ChannelService: return False switch_request = { - "event": "stream_switch", + "event": EventType.STREAM_SWITCH, # Use constant instead of string "channel_id": channel_id, "url": new_url, "user_agent": user_agent, @@ -338,7 +339,7 @@ class ChannelService: return False stop_request = { - "event": "channel_stop", + "event": EventType.CHANNEL_STOP, # Use constant instead of string "channel_id": channel_id, "requester_worker_id": proxy_server.worker_id, "timestamp": time.time() @@ -359,7 +360,7 @@ class ChannelService: return False stop_request = { - "event": "client_stop", + "event": EventType.CLIENT_STOP, # Use constant instead of string "channel_id": channel_id, "client_id": client_id, "requester_worker_id": proxy_server.worker_id, diff --git a/apps/proxy/ts_proxy/stream_manager.py b/apps/proxy/ts_proxy/stream_manager.py index 047f5dd7..d16173e5 100644 --- a/apps/proxy/ts_proxy/stream_manager.py +++ b/apps/proxy/ts_proxy/stream_manager.py @@ -3,6 +3,7 @@ import threading import logging import time +import socket import requests import subprocess from typing import Optional, List @@ -414,16 +415,19 @@ class StreamManager: logger.debug(f"Error closing session: {e}") self.current_session = None - # Keep backward compatibility - let's create an alias to the new method def _close_socket(self): - """Backward compatibility wrapper for _close_connection""" - if self.current_response: - return self._close_connection() + """Close socket and transcode resources as needed""" + # First try to use _close_connection for HTTP resources + if self.current_response or self.current_session: + self._close_connection() + return + + # Otherwise handle socket and transcode resources if self.socket: try: self.socket.close() except Exception as e: - logging.debug(f"Error closing socket: {e}") + logger.debug(f"Error closing socket: {e}") pass self.socket = None @@ -434,7 +438,7 @@ class StreamManager: self.transcode_process.terminate() self.transcode_process.wait() except Exception as e: - logging.debug(f"Error terminating transcode process: {e}") + logger.debug(f"Error terminating transcode process: {e}") pass self.transcode_process = None