Merge branch 'main' into EPG-Region-Update

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

apps/proxy/ts_proxy/client_manager.py

@@ -6,6 +6,9 @@ import time
 import json
 from typing import Set, Optional
 from apps.proxy.config import TSConfig as Config
+from .constants import EventType
+from .config_helper import ConfigHelper
+from .redis_keys import RedisKeys
 
 logger = logging.getLogger("ts_proxy")
 
@@ -21,9 +24,9 @@ class ClientManager:
         self.worker_id = worker_id  # Store worker ID as instance variable
 
         # STANDARDIZED KEYS: Move client set under channel namespace
-        self.client_set_key = f"ts_proxy:channel:{channel_id}:clients"
-        self.client_ttl = getattr(Config, 'CLIENT_RECORD_TTL', 60)
-        self.heartbeat_interval = getattr(Config, 'CLIENT_HEARTBEAT_INTERVAL', 10)
+        self.client_set_key = RedisKeys.clients(channel_id)
+        self.client_ttl = ConfigHelper.get('CLIENT_RECORD_TTL', 60)
+        self.heartbeat_interval = ConfigHelper.get('CLIENT_HEARTBEAT_INTERVAL', 10)
         self.last_heartbeat_time = {}
 
         # Start heartbeat thread for local clients
@@ -143,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
@@ -172,13 +175,14 @@ 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()
 
                     # Publish client connected event with user agent
                     event_data = {
-                        "event": "client_connected",
+                        "event": EventType.CLIENT_CONNECTED,  # Use constant instead of string
                         "channel_id": self.channel_id,
                         "client_id": client_id,
                         "worker_id": self.worker_id or "unknown",
@@ -192,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)
                     )
 
@@ -233,21 +237,21 @@ 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()
 
                 # Publish client disconnected event
                 event_data = json.dumps({
-                    "event": "client_disconnected",
+                    "event": EventType.CLIENT_DISCONNECTED,  # Use constant instead of string
                     "channel_id": self.channel_id,
                     "client_id": client_id,
                     "worker_id": self.worker_id or "unknown",
                     "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})")

apps/proxy/ts_proxy/config_helper.py

@@ -0,0 +1,70 @@
+"""
+Helper module to access configuration values with proper defaults.
+"""
+
+import logging
+from apps.proxy.config import TSConfig as Config
+
+logger = logging.getLogger("ts_proxy")
+
+class ConfigHelper:
+    """
+    Helper class for accessing configuration values with sensible defaults.
+    This simplifies code and ensures consistent defaults across the application.
+    """
+
+    @staticmethod
+    def get(name, default=None):
+        """Get a configuration value with a default fallback"""
+        return getattr(Config, name, default)
+
+    # Commonly used configuration values
+    @staticmethod
+    def connection_timeout():
+        """Get connection timeout in seconds"""
+        return ConfigHelper.get('CONNECTION_TIMEOUT', 10)
+
+    @staticmethod
+    def client_wait_timeout():
+        """Get client wait timeout in seconds"""
+        return ConfigHelper.get('CLIENT_WAIT_TIMEOUT', 30)
+
+    @staticmethod
+    def stream_timeout():
+        """Get stream timeout in seconds"""
+        return ConfigHelper.get('STREAM_TIMEOUT', 60)
+
+    @staticmethod
+    def channel_shutdown_delay():
+        """Get channel shutdown delay in seconds"""
+        return ConfigHelper.get('CHANNEL_SHUTDOWN_DELAY', 5)
+
+    @staticmethod
+    def initial_behind_chunks():
+        """Get number of chunks to start behind"""
+        return ConfigHelper.get('INITIAL_BEHIND_CHUNKS', 10)
+
+    @staticmethod
+    def keepalive_interval():
+        """Get keepalive interval in seconds"""
+        return ConfigHelper.get('KEEPALIVE_INTERVAL', 0.5)
+
+    @staticmethod
+    def cleanup_check_interval():
+        """Get cleanup check interval in seconds"""
+        return ConfigHelper.get('CLEANUP_CHECK_INTERVAL', 3)
+
+    @staticmethod
+    def redis_chunk_ttl():
+        """Get Redis chunk TTL in seconds"""
+        return ConfigHelper.get('REDIS_CHUNK_TTL', 60)
+
+    @staticmethod
+    def chunk_size():
+        """Get chunk size in bytes"""
+        return ConfigHelper.get('CHUNK_SIZE', 8192)
+
+    @staticmethod
+    def max_retries():
+        """Get maximum retry attempts"""
+        return ConfigHelper.get('MAX_RETRIES', 3)

apps/proxy/ts_proxy/constants.py

@@ -0,0 +1,42 @@
+"""
+Constants used throughout the TS Proxy application.
+Centralizing constants makes it easier to maintain and modify them.
+"""
+
+# Redis related constants
+REDIS_KEY_PREFIX = "ts_proxy"
+REDIS_TTL_DEFAULT = 3600  # 1 hour
+REDIS_TTL_SHORT = 60      # 1 minute
+REDIS_TTL_MEDIUM = 300    # 5 minutes
+
+# Channel states
+class ChannelState:
+    INITIALIZING = "initializing"
+    CONNECTING = "connecting"
+    WAITING_FOR_CLIENTS = "waiting_for_clients"
+    ACTIVE = "active"
+    ERROR = "error"
+    STOPPING = "stopping"
+    STOPPED = "stopped"
+
+# Event types
+class EventType:
+    STREAM_SWITCH = "stream_switch"
+    STREAM_SWITCHED = "stream_switched"
+    CHANNEL_STOP = "channel_stop"
+    CHANNEL_STOPPED = "channel_stopped"
+    CLIENT_CONNECTED = "client_connected"
+    CLIENT_DISCONNECTED = "client_disconnected"
+    CLIENT_STOP = "client_stop"
+
+# Stream types
+class StreamType:
+    HLS = "hls"
+    TS = "ts"
+    UNKNOWN = "unknown"
+
+# TS packet constants
+TS_PACKET_SIZE = 188
+TS_SYNC_BYTE = 0x47
+NULL_PID_HIGH = 0x1F
+NULL_PID_LOW = 0xFF

apps/proxy/ts_proxy/redis_keys.py

@@ -0,0 +1,75 @@
+"""
+Defines Redis key patterns used throughout the TS proxy service.
+Centralizing these key patterns makes it easier to maintain and change them if needed.
+"""
+
+class RedisKeys:
+    @staticmethod
+    def channel_metadata(channel_id):
+        """Key for channel metadata hash"""
+        return f"ts_proxy:channel:{channel_id}:metadata"
+
+    @staticmethod
+    def buffer_index(channel_id):
+        """Key for tracking buffer index"""
+        return f"ts_proxy:channel:{channel_id}:buffer:index"
+
+    @staticmethod
+    def buffer_chunk(channel_id, chunk_index):
+        """Key for specific buffer chunk"""
+        return f"ts_proxy:channel:{channel_id}:buffer:chunk:{chunk_index}"
+
+    @staticmethod
+    def buffer_chunk_prefix(channel_id):
+        """Prefix for buffer chunks"""
+        return f"ts_proxy:channel:{channel_id}:buffer:chunk:"
+
+    @staticmethod
+    def channel_stopping(channel_id):
+        """Key indicating channel is stopping"""
+        return f"ts_proxy:channel:{channel_id}:stopping"
+
+    @staticmethod
+    def client_stop(channel_id, client_id):
+        """Key requesting client stop"""
+        return f"ts_proxy:channel:{channel_id}:client:{client_id}:stop"
+
+    @staticmethod
+    def events_channel(channel_id):
+        """PubSub channel for events"""
+        return f"ts_proxy:events:{channel_id}"
+
+    @staticmethod
+    def switch_request(channel_id):
+        """Key for stream switch request"""
+        return f"ts_proxy:channel:{channel_id}:switch_request"
+
+    @staticmethod
+    def channel_owner(channel_id):
+        """Key for storing channel owner worker ID"""
+        return f"ts_proxy:channel:{channel_id}:owner"
+
+    @staticmethod
+    def clients(channel_id):
+        """Key for set of client IDs"""
+        return f"ts_proxy:channel:{channel_id}:clients"
+
+    @staticmethod
+    def last_client_disconnect(channel_id):
+        """Key for last client disconnect timestamp"""
+        return f"ts_proxy:channel:{channel_id}:last_client_disconnect_time"
+
+    @staticmethod
+    def connection_attempt(channel_id):
+        """Key for connection attempt timestamp"""
+        return f"ts_proxy:channel:{channel_id}:connection_attempt_time"
+
+    @staticmethod
+    def last_data(channel_id):
+        """Key for last data timestamp"""
+        return f"ts_proxy:channel:{channel_id}:last_data"
+
+    @staticmethod
+    def switch_status(channel_id):
+        """Key for stream switch status"""
+        return f"ts_proxy:channel:{channel_id}:switch_status"

apps/proxy/ts_proxy/server.py

@@ -21,6 +21,9 @@ from apps.channels.models import Channel
 from .stream_manager import StreamManager
 from .stream_buffer import StreamBuffer
 from .client_manager import ClientManager
+from .redis_keys import RedisKeys
+from .constants import ChannelState, EventType, StreamType
+from .config_helper import ConfigHelper
 
 logger = logging.getLogger("ts_proxy")
 
@@ -87,25 +90,24 @@ class ProxyServer:
                         if channel_id and event_type:
                             # For owner, update client status immediately
                             if self.am_i_owner(channel_id):
-                                if event_type == "client_connected":
-                                    logger.debug(f"Owner received client_connected event for channel {channel_id}")
+                                if event_type == EventType.CLIENT_CONNECTED:
+                                    logger.debug(f"Owner received {EventType.CLIENT_CONNECTED} event for channel {channel_id}")
                                     # Reset any disconnect timer
-                                    # RENAMED: no_clients_since → last_client_disconnect_time
-                                    disconnect_key = f"ts_proxy:channel:{channel_id}:last_client_disconnect_time"
+                                    disconnect_key = RedisKeys.last_client_disconnect(channel_id)
                                     self.redis_client.delete(disconnect_key)
 
-                                elif event_type == "client_disconnected":
-                                    logger.debug(f"Owner received client_disconnected event for channel {channel_id}")
+                                elif event_type == EventType.CLIENT_DISCONNECTED:
+                                    logger.debug(f"Owner received {EventType.CLIENT_DISCONNECTED} event for channel {channel_id}")
                                     # Check if any clients remain
                                     if channel_id in self.client_managers:
                                         # VERIFY REDIS CLIENT COUNT DIRECTLY
-                                        client_set_key = f"ts_proxy:channel:{channel_id}:clients"
+                                        client_set_key = RedisKeys.clients(channel_id)
                                         total = self.redis_client.scard(client_set_key) or 0
 
                                         if total == 0:
                                             logger.debug(f"No clients left after disconnect event - stopping channel {channel_id}")
                                             # Set the disconnect timer for other workers to see
-                                            disconnect_key = f"ts_proxy:channel:{channel_id}:last_client_disconnect_time"
+                                            disconnect_key = RedisKeys.last_client_disconnect(channel_id)
                                             self.redis_client.setex(disconnect_key, 60, str(time.time()))
 
                                             # Get configured shutdown delay or default
@@ -126,8 +128,8 @@ class ProxyServer:
                                             self.stop_channel(channel_id)
 
 
-                                elif event_type == "stream_switch":
-                                    logger.info(f"Owner received stream switch request for channel {channel_id}")
+                                elif event_type == EventType.STREAM_SWITCH:
+                                    logger.info(f"Owner received {EventType.STREAM_SWITCH} request for channel {channel_id}")
                                     # Handle stream switch request
                                     new_url = data.get("url")
                                     user_agent = data.get("user_agent")
@@ -135,13 +137,13 @@ class ProxyServer:
                                     if new_url and channel_id in self.stream_managers:
                                         # Update metadata in Redis
                                         if self.redis_client:
-                                            metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
+                                            metadata_key = RedisKeys.channel_metadata(channel_id)
                                             self.redis_client.hset(metadata_key, "url", new_url)
                                             if user_agent:
                                                 self.redis_client.hset(metadata_key, "user_agent", user_agent)
 
                                             # Set switch status
-                                            status_key = f"ts_proxy:channel:{channel_id}:switch_status"
+                                            status_key = RedisKeys.switch_status(channel_id)
                                             self.redis_client.set(status_key, "switching")
 
                                         # Perform the stream switch
@@ -153,7 +155,7 @@ class ProxyServer:
 
                                             # Publish confirmation
                                             switch_result = {
-                                                "event": "stream_switched",
+                                                "event": EventType.STREAM_SWITCHED,  # Use constant instead of string
                                                 "channel_id": channel_id,
                                                 "success": True,
                                                 "url": new_url,
@@ -172,7 +174,7 @@ class ProxyServer:
 
                                             # Publish failure
                                             switch_result = {
-                                                "event": "stream_switched",
+                                                "event": EventType.STREAM_SWITCHED,
                                                 "channel_id": channel_id,
                                                 "success": False,
                                                 "url": new_url,
@@ -182,15 +184,15 @@ class ProxyServer:
                                                 f"ts_proxy:events:{channel_id}",
                                                 json.dumps(switch_result)
                                             )
-                                elif event_type == "channel_stop":
-                                    logger.info(f"Received channel stop event for channel {channel_id}")
+                                elif event_type == EventType.CHANNEL_STOP:
+                                    logger.info(f"Received {EventType.CHANNEL_STOP} event for channel {channel_id}")
                                     # First mark channel as stopping in Redis
                                     if self.redis_client:
                                         # Set stopping state in metadata
-                                        metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
+                                        metadata_key = RedisKeys.channel_metadata(channel_id)
                                         if self.redis_client.exists(metadata_key):
                                             self.redis_client.hset(metadata_key, mapping={
-                                                "state": "stopping",
+                                                "state": ChannelState.STOPPING,
                                                 "state_changed_at": str(time.time())
                                             })
 
@@ -202,7 +204,7 @@ class ProxyServer:
 
                                     # Acknowledge stop by publishing a response
                                     stop_response = {
-                                        "event": "channel_stopped",
+                                        "event": EventType.CHANNEL_STOPPED,
                                         "channel_id": channel_id,
                                         "worker_id": self.worker_id,
                                         "timestamp": time.time()
@@ -211,7 +213,7 @@ class ProxyServer:
                                         f"ts_proxy:events:{channel_id}",
                                         json.dumps(stop_response)
                                     )
-                                elif event_type == "client_stop":
+                                elif event_type == EventType.CLIENT_STOP:
                                     client_id = data.get("client_id")
                                     if client_id and channel_id:
                                         logger.info(f"Received request to stop client {client_id} on channel {channel_id}")
@@ -225,7 +227,7 @@ class ProxyServer:
 
                                         # Set a Redis key for the generator to detect
                                         if self.redis_client:
-                                            stop_key = f"ts_proxy:channel:{channel_id}:client:{client_id}:stop"
+                                            stop_key = RedisKeys.client_stop(channel_id, client_id)
                                             self.redis_client.setex(stop_key, 30, "true")  # 30 second TTL
                                             logger.info(f"Set stop key for client {client_id}")
                     except Exception as e:
@@ -246,7 +248,7 @@ class ProxyServer:
             return None
 
         try:
-            lock_key = f"ts_proxy:channel:{channel_id}:owner"
+            lock_key = RedisKeys.channel_owner(channel_id)
             owner = self.redis_client.get(lock_key)
             if owner:
                 return owner.decode('utf-8')
@@ -267,7 +269,7 @@ class ProxyServer:
 
         try:
             # Create a lock key with proper namespace
-            lock_key = f"ts_proxy:channel:{channel_id}:owner"
+            lock_key = RedisKeys.channel_owner(channel_id)
 
             # Use Redis SETNX for atomic locking - only succeeds if the key doesn't exist
             acquired = self.redis_client.setnx(lock_key, self.worker_id)
@@ -299,7 +301,7 @@ class ProxyServer:
             return
 
         try:
-            lock_key = f"ts_proxy:channel:{channel_id}:owner"
+            lock_key = RedisKeys.channel_owner(channel_id)
 
             # Only delete if we're the current owner to prevent race conditions
             current = self.redis_client.get(lock_key)
@@ -315,7 +317,7 @@ class ProxyServer:
             return False
 
         try:
-            lock_key = f"ts_proxy:channel:{channel_id}:owner"
+            lock_key = RedisKeys.channel_owner(channel_id)
             current = self.redis_client.get(lock_key)
 
             # Only extend if we're still the owner
@@ -348,7 +350,7 @@ class ProxyServer:
 
             # First check if channel metadata already exists
             existing_metadata = None
-            metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
+            metadata_key = RedisKeys.channel_metadata(channel_id)
 
             if self.redis_client:
                 existing_metadata = self.redis_client.hgetall(metadata_key)
@@ -412,7 +414,7 @@ class ProxyServer:
                     "init_time": str(time.time()),
                     "last_active": str(time.time()),
                     "owner": self.worker_id,
-                    "state": "initializing"  # Only the owner sets this initial state
+                    "state": ChannelState.INITIALIZING  # Use constant instead of string literal
                 }
                 if channel_user_agent:
                     metadata["user_agent"] = channel_user_agent
@@ -447,16 +449,16 @@ class ProxyServer:
 
             # If we're the owner, we need to set the channel state rather than starting a grace period immediately
             if self.am_i_owner(channel_id):
-                self.update_channel_state(channel_id, "connecting", {
+                self.update_channel_state(channel_id, ChannelState.CONNECTING, {
                     "init_time": str(time.time()),
                     "owner": self.worker_id
                 })
 
                 # Set connection attempt start time
-                attempt_key = f"ts_proxy:channel:{channel_id}:connection_attempt_time"
+                attempt_key = RedisKeys.connection_attempt(channel_id)
                 self.redis_client.setex(attempt_key, 60, str(time.time()))
 
-                logger.info(f"Channel {channel_id} in connecting state - will start grace period after connection")
+                logger.info(f"Channel {channel_id} in {ChannelState.CONNECTING} state - will start grace period after connection")
             return True
 
         except Exception as e:
@@ -474,7 +476,7 @@ class ProxyServer:
         # Check Redis using the standard key pattern
         if self.redis_client:
             # Primary check - look for channel metadata
-            metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
+            metadata_key = RedisKeys.channel_metadata(channel_id)
 
             # If metadata exists, return true
             if self.redis_client.exists(metadata_key):
@@ -482,9 +484,9 @@ class ProxyServer:
 
             # Additional checks if metadata doesn't exist
             additional_keys = [
-                f"ts_proxy:channel:{channel_id}:clients",
-                f"ts_proxy:channel:{channel_id}:buffer:index",
-                f"ts_proxy:channel:{channel_id}:owner"
+                RedisKeys.clients(channel_id),
+                RedisKeys.buffer_index(channel_id),
+                RedisKeys.channel_owner(channel_id)
             ]
 
             for key in additional_keys:
@@ -497,12 +499,12 @@ class ProxyServer:
         """Stop a channel with proper ownership handling"""
         try:
             logger.info(f"Stopping channel {channel_id}")
-            
+
             # First set a stopping key that clients will check
             if self.redis_client:
-                stop_key = f"ts_proxy:channel:{channel_id}:stopping"
+                stop_key = RedisKeys.channel_stopping(channel_id)
                 self.redis_client.setex(stop_key, 10, "true")
-                
+
             # Only stop the actual stream manager if we're the owner
             if self.am_i_owner(channel_id):
                 logger.info(f"This worker ({self.worker_id}) is the owner - closing provider connection")
@@ -544,7 +546,7 @@ class ProxyServer:
             if channel_id in self.stream_managers:
                 del self.stream_managers[channel_id]
                 logger.info(f"Removed stream manager for channel {channel_id}")
-            
+
             # Stop buffer and ensure all its timers are cancelled - SAFE CHECK HERE
             if channel_id in self.stream_buffers:
                 buffer = self.stream_buffers[channel_id]
@@ -555,7 +557,7 @@ class ProxyServer:
                         logger.debug(f"Buffer for channel {channel_id} properly stopped")
                     except Exception as e:
                         logger.error(f"Error stopping buffer: {e}")
-                
+
                 # Save reference and check again before deleting
                 try:
                     if channel_id in self.stream_buffers:  # Check again to prevent race conditions
@@ -563,7 +565,7 @@ class ProxyServer:
                         logger.info(f"Removed stream buffer for channel {channel_id}")
                 except KeyError:
                     logger.debug(f"Buffer for channel {channel_id} already removed")
-            
+
             # Clean up client manager - SAFE CHECK HERE TOO
             if channel_id in self.client_managers:
                 try:
@@ -571,10 +573,10 @@ class ProxyServer:
                     logger.info(f"Removed client manager for channel {channel_id}")
                 except KeyError:
                     logger.debug(f"Client manager for channel {channel_id} already removed")
-                
+
             # Clean up Redis keys
             self._clean_redis_keys(channel_id)
-            
+
             return True
         except Exception as e:
             logger.error(f"Error stopping channel {channel_id}: {e}", exc_info=True)
@@ -594,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:
@@ -624,7 +626,7 @@ class ProxyServer:
                             # Get channel state from metadata hash
                             channel_state = "unknown"
                             if self.redis_client:
-                                metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
+                                metadata_key = RedisKeys.channel_metadata(channel_id)
                                 metadata = self.redis_client.hgetall(metadata_key)
                                 if metadata and b'state' in metadata:
                                     channel_state = metadata[b'state'].decode('utf-8')
@@ -640,7 +642,7 @@ class ProxyServer:
                                 logger.info(f"Channel {channel_id} has {total_clients} clients, state: {channel_state}")
 
                             # If in connecting or waiting_for_clients state, check grace period
-                            if channel_state in ["connecting", "waiting_for_clients"]:
+                            if channel_state in [ChannelState.CONNECTING, ChannelState.WAITING_FOR_CLIENTS]:
                                 # Get connection ready time from metadata
                                 connection_ready_time = None
                                 if metadata and b'connection_ready_time' in metadata:
@@ -650,13 +652,13 @@ class ProxyServer:
                                         pass
 
                                 # If still connecting, give it more time
-                                if channel_state == "connecting":
+                                if channel_state == ChannelState.CONNECTING:
                                     logger.debug(f"Channel {channel_id} still connecting - not checking for clients yet")
                                     continue
 
                                 # If waiting for clients, check grace period
                                 if connection_ready_time:
-                                    grace_period = getattr(Config, 'CHANNEL_INIT_GRACE_PERIOD', 20)
+                                    grace_period = ConfigHelper.get('CHANNEL_INIT_GRACE_PERIOD', 20)
                                     time_since_ready = time.time() - connection_ready_time
 
                                     # Add this debug log
@@ -678,15 +680,15 @@ class ProxyServer:
                                         old_state = "unknown"
                                         if metadata and b'state' in metadata:
                                             old_state = metadata[b'state'].decode('utf-8')
-                                        if self.update_channel_state(channel_id, "active", {
+                                        if self.update_channel_state(channel_id, ChannelState.ACTIVE, {
                                             "grace_period_ended_at": str(time.time()),
                                             "clients_at_activation": str(total_clients)
                                         }):
                                             logger.info(f"Channel {channel_id} activated with {total_clients} clients after grace period")
                             # If active and no clients, start normal shutdown procedure
-                            elif channel_state not in ["connecting", "waiting_for_clients"] and total_clients == 0:
+                            elif channel_state not in [ChannelState.CONNECTING, ChannelState.WAITING_FOR_CLIENTS] and total_clients == 0:
                                 # Check if there's a pending no-clients timeout
-                                disconnect_key = f"ts_proxy:channel:{channel_id}:last_client_disconnect_time"
+                                disconnect_key = RedisKeys.last_client_disconnect(channel_id)
                                 disconnect_time = None
 
                                 if self.redis_client:
@@ -704,7 +706,7 @@ class ProxyServer:
                                     if self.redis_client:
                                         self.redis_client.setex(disconnect_key, 60, str(current_time))
                                     logger.warning(f"No clients detected for channel {channel_id}, starting shutdown timer")
-                                elif current_time - disconnect_time > getattr(Config, 'CHANNEL_SHUTDOWN_DELAY', 5):
+                                elif current_time - disconnect_time > ConfigHelper.channel_shutdown_delay():
                                     # We've had no clients for the shutdown delay period
                                     logger.warning(f"No clients for {current_time - disconnect_time:.1f}s, stopping channel {channel_id}")
                                     self.stop_channel(channel_id)
@@ -712,7 +714,7 @@ class ProxyServer:
                                     # Still in shutdown delay period
                                     logger.debug(f"Channel {channel_id} shutdown timer: "
                                                 f"{current_time - disconnect_time:.1f}s of "
-                                                f"{getattr(Config, 'CHANNEL_SHUTDOWN_DELAY', 5)}s elapsed")
+                                                f"{ConfigHelper.channel_shutdown_delay()}s elapsed")
                             else:
                                 # There are clients or we're still connecting - clear any disconnect timestamp
                                 if self.redis_client:
@@ -723,21 +725,21 @@ class ProxyServer:
                             # For channels we don't own, check if they've been stopped/cleaned up in Redis
                             if self.redis_client:
                                 # Method 1: Check for stopping key
-                                stop_key = f"ts_proxy:channel:{channel_id}:stopping"
+                                stop_key = RedisKeys.channel_stopping(channel_id)
                                 if self.redis_client.exists(stop_key):
                                     logger.debug(f"Non-owner cleanup: Channel {channel_id} has stopping flag in Redis, cleaning up local resources")
                                     self._cleanup_local_resources(channel_id)
                                     continue
 
                                 # Method 2: Check if owner still exists
-                                owner_key = f"ts_proxy:channel:{channel_id}:owner"
+                                owner_key = RedisKeys.channel_owner(channel_id)
                                 if not self.redis_client.exists(owner_key):
                                     logger.debug(f"Non-owner cleanup: Channel {channel_id} has no owner in Redis, cleaning up local resources")
                                     self._cleanup_local_resources(channel_id)
                                     continue
 
                                 # Method 3: Check if metadata still exists
-                                metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
+                                metadata_key = RedisKeys.channel_metadata(channel_id)
                                 if not self.redis_client.exists(metadata_key):
                                     logger.debug(f"Non-owner cleanup: Channel {channel_id} has no metadata in Redis, cleaning up local resources")
                                     self._cleanup_local_resources(channel_id)
@@ -752,12 +754,12 @@ class ProxyServer:
                 except Exception as e:
                     logger.error(f"Error in cleanup thread: {e}", exc_info=True)
 
-                time.sleep(getattr(Config, 'CLEANUP_CHECK_INTERVAL', 1))
+                time.sleep(ConfigHelper.cleanup_check_interval())
 
         thread = threading.Thread(target=cleanup_task, daemon=True)
         thread.name = "ts-proxy-cleanup"
         thread.start()
-        logger.info(f"Started TS proxy cleanup thread (interval: {getattr(Config, 'CLEANUP_CHECK_INTERVAL', 3)}s)")
+        logger.info(f"Started TS proxy cleanup thread (interval: {ConfigHelper.cleanup_check_interval()}s)")
 
     def _check_orphaned_channels(self):
         """Check for orphaned channels in Redis (owner worker crashed)"""
@@ -782,7 +784,7 @@ class ProxyServer:
 
                     if not owner:
                         # Check if there are any clients
-                        client_set_key = f"ts_proxy:channel:{channel_id}:clients"
+                        client_set_key = RedisKeys.clients(channel_id)
                         client_count = self.redis_client.scard(client_set_key) or 0
 
                         if client_count > 0:
@@ -811,7 +813,7 @@ class ProxyServer:
             # Define key patterns to scan for
             patterns = [
                 f"ts_proxy:channel:{channel_id}:*",  # All channel keys
-                f"ts_proxy:events:{channel_id}"      # Event channel
+                RedisKeys.events_channel(channel_id)  # Event channel
             ]
 
             total_deleted = 0
@@ -843,7 +845,7 @@ class ProxyServer:
         # Refresh registry entries for channels we own
         for channel_id in list(self.stream_buffers.keys()):
             # Use standard key pattern
-            metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
+            metadata_key = RedisKeys.channel_metadata(channel_id)
 
             # Update activity timestamp in metadata only
             self.redis_client.hset(metadata_key, "last_active", str(time.time()))
@@ -856,7 +858,7 @@ class ProxyServer:
             return False
 
         try:
-            metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
+            metadata_key = RedisKeys.channel_metadata(channel_id)
 
             # Get current state for logging
             current_state = None

apps/proxy/ts_proxy/services/channel_service.py

@@ -0,0 +1,374 @@
+"""
+Channel service layer for handling business logic related to channel operations.
+This separates business logic from HTTP handling in views.
+"""
+
+import logging
+import time
+import json
+from django.shortcuts import get_object_or_404
+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")
+
+class ChannelService:
+    """Service class for channel operations"""
+
+    @staticmethod
+    def initialize_channel(channel_id, stream_url, user_agent, transcode=False, profile_value=None):
+        """
+        Initialize a channel with the given parameters.
+
+        Args:
+            channel_id: UUID of the channel
+            stream_url: URL of the stream
+            user_agent: User agent for the stream connection
+            transcode: Whether to transcode the stream
+            profile_value: Stream profile value to store in metadata
+
+        Returns:
+            bool: Success status
+        """
+        success = proxy_server.initialize_channel(stream_url, channel_id, user_agent, transcode)
+
+        # Store additional metadata if initialization was successful
+        if success and proxy_server.redis_client and profile_value:
+            metadata_key = RedisKeys.channel_metadata(channel_id)
+            proxy_server.redis_client.hset(metadata_key, "profile", profile_value)
+
+        return success
+
+    @staticmethod
+    def change_stream_url(channel_id, new_url, user_agent=None):
+        """
+        Change the URL of an existing stream.
+
+        Args:
+            channel_id: UUID of the channel
+            new_url: New stream URL
+            user_agent: Optional user agent to update
+
+        Returns:
+            dict: Result information including success status and diagnostics
+        """
+        # Check if channel exists
+        in_local_managers = channel_id in proxy_server.stream_managers
+        in_local_buffers = channel_id in proxy_server.stream_buffers
+
+        # Check Redis for keys
+        redis_keys = None
+        if proxy_server.redis_client:
+            try:
+                # This is inefficient but used for diagnostics - in production would use more targeted checks
+                redis_keys = proxy_server.redis_client.keys(f"ts_proxy:*:{channel_id}*")
+                redis_keys = [k.decode('utf-8') for k in redis_keys] if redis_keys else []
+            except Exception as e:
+                logger.error(f"Error checking Redis keys: {e}")
+
+        # Check if channel exists using standard method
+        channel_exists = proxy_server.check_if_channel_exists(channel_id)
+
+        # Log detailed diagnostics
+        logger.info(f"Channel {channel_id} diagnostics: "
+                   f"in_local_managers={in_local_managers}, "
+                   f"in_local_buffers={in_local_buffers}, "
+                   f"redis_keys_count={len(redis_keys) if redis_keys else 0}, "
+                   f"channel_exists={channel_exists}")
+
+        if not channel_exists:
+            # Try to recover if Redis keys exist but channel check failed
+            if redis_keys:
+                logger.warning(f"Channel {channel_id} not detected but Redis keys exist. Forcing initialization.")
+                proxy_server.initialize_channel(new_url, channel_id, user_agent)
+                result = {
+                    'status': 'recovered',
+                    'message': 'Channel was recovered and initialized'
+                }
+            else:
+                logger.error(f"Channel {channel_id} not found in any worker or Redis")
+                return {
+                    'status': 'error',
+                    'message': 'Channel not found',
+                    'diagnostics': {
+                        'in_local_managers': in_local_managers,
+                        'in_local_buffers': in_local_buffers,
+                        'redis_keys': redis_keys,
+                    }
+                }
+        else:
+            result = {'status': 'success'}
+
+        # Update metadata in Redis regardless of ownership
+        if proxy_server.redis_client:
+            try:
+                ChannelService._update_channel_metadata(channel_id, new_url, user_agent)
+                result['metadata_updated'] = True
+            except Exception as e:
+                logger.error(f"Error updating Redis metadata: {e}", exc_info=True)
+                result['metadata_updated'] = False
+
+        # If we're the owner, update directly
+        if proxy_server.am_i_owner(channel_id) and channel_id in proxy_server.stream_managers:
+            logger.info(f"This worker is the owner, changing stream URL for channel {channel_id}")
+            manager = proxy_server.stream_managers[channel_id]
+            old_url = manager.url
+
+            # Update the stream
+            success = manager.update_url(new_url)
+            logger.info(f"Stream URL changed from {old_url} to {new_url}, result: {success}")
+
+            result.update({
+                'direct_update': True,
+                'success': success,
+                'worker_id': proxy_server.worker_id
+            })
+        else:
+            # If we're not the owner, publish an event for the owner to pick up
+            logger.info(f"Not the owner, requesting URL change via Redis PubSub")
+            if proxy_server.redis_client:
+                ChannelService._publish_stream_switch_event(channel_id, new_url, user_agent)
+                result.update({
+                    'direct_update': False,
+                    'event_published': True,
+                    'worker_id': proxy_server.worker_id
+                })
+            else:
+                result.update({
+                    'direct_update': False,
+                    'event_published': False,
+                    'error': 'Redis not available for pubsub'
+                })
+
+        return result
+
+    @staticmethod
+    def stop_channel(channel_id):
+        """
+        Stop a channel and release all resources.
+
+        Args:
+            channel_id: UUID of the channel
+
+        Returns:
+            dict: Result information including previous state if available
+        """
+        # Check if channel exists
+        channel_exists = proxy_server.check_if_channel_exists(channel_id)
+        if not channel_exists:
+            logger.warning(f"Channel {channel_id} not found in any worker or Redis")
+            return {'status': 'error', 'message': 'Channel not found'}
+
+        # Get channel state information for result
+        channel_info = None
+        if proxy_server.redis_client:
+            metadata_key = RedisKeys.channel_metadata(channel_id)
+            try:
+                metadata = proxy_server.redis_client.hgetall(metadata_key)
+                if metadata and b'state' in metadata:
+                    state = metadata[b'state'].decode('utf-8')
+                    channel_info = {"state": state}
+            except Exception as e:
+                logger.error(f"Error fetching channel state: {e}")
+
+        # Broadcast stop event to all workers via PubSub
+        if proxy_server.redis_client:
+            ChannelService._publish_channel_stop_event(channel_id)
+
+            # Also stop locally to ensure this worker cleans up right away
+            local_result = proxy_server.stop_channel(channel_id)
+        else:
+            # No Redis, just stop locally
+            local_result = proxy_server.stop_channel(channel_id)
+
+        # Release the channel in the channel model if applicable
+        try:
+            channel = Channel.objects.get(uuid=channel_id)
+            channel.release_stream()
+            logger.info(f"Released channel {channel_id} stream allocation")
+            model_released = True
+        except Channel.DoesNotExist:
+            logger.warning(f"Could not find Channel model for UUID {channel_id}")
+            model_released = False
+        except Exception as e:
+            logger.error(f"Error releasing channel stream: {e}")
+            model_released = False
+
+        return {
+            'status': 'success',
+            'message': 'Channel stop request sent',
+            'channel_id': channel_id,
+            'previous_state': channel_info,
+            'model_released': model_released,
+            'local_stop_result': local_result
+        }
+
+    @staticmethod
+    def stop_client(channel_id, client_id):
+        """
+        Stop a specific client connection.
+
+        Args:
+            channel_id: UUID of the channel
+            client_id: ID of the client to stop
+
+        Returns:
+            dict: Result information
+        """
+        logger.info(f"Request to stop client {client_id} on channel {channel_id}")
+
+        # Set a Redis key for immediate detection
+        key_set = False
+        if proxy_server.redis_client:
+            stop_key = RedisKeys.client_stop(channel_id, client_id)
+            try:
+                proxy_server.redis_client.setex(stop_key, 30, "true")  # 30 second TTL
+                logger.info(f"Set stop key for client {client_id}")
+                key_set = True
+            except Exception as e:
+                logger.error(f"Error setting client stop key: {e}")
+
+        # Check if channel exists
+        channel_exists = proxy_server.check_if_channel_exists(channel_id)
+        if not channel_exists:
+            logger.warning(f"Channel {channel_id} not found")
+            return {
+                'status': 'error',
+                'message': 'Channel not found',
+                'stop_key_set': key_set
+            }
+
+        # Try to stop locally if client is on this worker
+        local_client_stopped = False
+        if channel_id in proxy_server.client_managers:
+            client_manager = proxy_server.client_managers[channel_id]
+            with client_manager.lock:
+                if client_id in client_manager.clients:
+                    client_manager.remove_client(client_id)
+                    local_client_stopped = True
+                    logger.info(f"Client {client_id} stopped locally on channel {channel_id}")
+
+        # If client wasn't found locally, broadcast stop event for other workers
+        event_published = False
+        if not local_client_stopped and proxy_server.redis_client:
+            try:
+                ChannelService._publish_client_stop_event(channel_id, client_id)
+                event_published = True
+                logger.info(f"Published stop request for client {client_id} on channel {channel_id}")
+            except Exception as e:
+                logger.error(f"Error publishing client stop event: {e}")
+
+        return {
+            'status': 'success',
+            'message': 'Client stop request processed',
+            'channel_id': channel_id,
+            'client_id': client_id,
+            'locally_processed': local_client_stopped,
+            'stop_key_set': key_set,
+            'event_published': event_published
+        }
+
+    # Helper methods for Redis operations
+
+    @staticmethod
+    def _update_channel_metadata(channel_id, url, user_agent=None):
+        """Update channel metadata in Redis"""
+        if not proxy_server.redis_client:
+            return False
+
+        metadata_key = RedisKeys.channel_metadata(channel_id)
+
+        # First check if the key exists and what type it is
+        key_type = proxy_server.redis_client.type(metadata_key).decode('utf-8')
+        logger.debug(f"Redis key {metadata_key} is of type: {key_type}")
+
+        # Use the appropriate method based on the key type
+        if key_type == 'hash':
+            proxy_server.redis_client.hset(metadata_key, "url", url)
+            if user_agent:
+                proxy_server.redis_client.hset(metadata_key, "user_agent", user_agent)
+        elif key_type == 'none':  # Key doesn't exist yet
+            # Create new hash with all required fields
+            metadata = {"url": url}
+            if user_agent:
+                metadata["user_agent"] = user_agent
+            proxy_server.redis_client.hset(metadata_key, mapping=metadata)
+        else:
+            # If key exists with wrong type, delete it and recreate
+            proxy_server.redis_client.delete(metadata_key)
+            metadata = {"url": url}
+            if user_agent:
+                metadata["user_agent"] = user_agent
+            proxy_server.redis_client.hset(metadata_key, mapping=metadata)
+
+        # Set switch request flag to ensure all workers see it
+        switch_key = RedisKeys.switch_request(channel_id)
+        proxy_server.redis_client.setex(switch_key, 30, url)  # 30 second TTL
+
+        logger.info(f"Updated metadata for channel {channel_id} in Redis")
+        return True
+
+    @staticmethod
+    def _publish_stream_switch_event(channel_id, new_url, user_agent=None):
+        """Publish a stream switch event to Redis pubsub"""
+        if not proxy_server.redis_client:
+            return False
+
+        switch_request = {
+            "event": EventType.STREAM_SWITCH,  # Use constant instead of string
+            "channel_id": channel_id,
+            "url": new_url,
+            "user_agent": user_agent,
+            "requester": proxy_server.worker_id,
+            "timestamp": time.time()
+        }
+
+        proxy_server.redis_client.publish(
+            RedisKeys.events_channel(channel_id),
+            json.dumps(switch_request)
+        )
+        return True
+
+    @staticmethod
+    def _publish_channel_stop_event(channel_id):
+        """Publish a channel stop event to Redis pubsub"""
+        if not proxy_server.redis_client:
+            return False
+
+        stop_request = {
+            "event": EventType.CHANNEL_STOP,  # Use constant instead of string
+            "channel_id": channel_id,
+            "requester_worker_id": proxy_server.worker_id,
+            "timestamp": time.time()
+        }
+
+        proxy_server.redis_client.publish(
+            RedisKeys.events_channel(channel_id),
+            json.dumps(stop_request)
+        )
+
+        logger.info(f"Published channel stop event for {channel_id}")
+        return True
+
+    @staticmethod
+    def _publish_client_stop_event(channel_id, client_id):
+        """Publish a client stop event to Redis pubsub"""
+        if not proxy_server.redis_client:
+            return False
+
+        stop_request = {
+            "event": EventType.CLIENT_STOP,  # Use constant instead of string
+            "channel_id": channel_id,
+            "client_id": client_id,
+            "requester_worker_id": proxy_server.worker_id,
+            "timestamp": time.time()
+        }
+
+        proxy_server.redis_client.publish(
+            RedisKeys.events_channel(channel_id),
+            json.dumps(stop_request)
+        )
+        return True

apps/proxy/ts_proxy/stream_buffer.py

@@ -7,6 +7,9 @@ from collections import deque
 from typing import Optional, Deque
 import random
 from apps.proxy.config import TSConfig as Config
+from .redis_keys import RedisKeys
+from .config_helper import ConfigHelper
+from .constants import TS_PACKET_SIZE
 
 logger = logging.getLogger("ts_proxy")
 
@@ -18,13 +21,13 @@ class StreamBuffer:
         self.redis_client = redis_client
         self.lock = threading.Lock()
         self.index = 0
-        self.TS_PACKET_SIZE = 188
+        self.TS_PACKET_SIZE = TS_PACKET_SIZE
 
-        # STANDARDIZED KEYS: Move buffer keys under channel namespace
-        self.buffer_index_key = f"ts_proxy:channel:{channel_id}:buffer:index"
-        self.buffer_prefix = f"ts_proxy:channel:{channel_id}:buffer:chunk:"
+        # STANDARDIZED KEYS: Use RedisKeys class instead of hardcoded patterns
+        self.buffer_index_key = RedisKeys.buffer_index(channel_id) if channel_id else ""
+        self.buffer_prefix = RedisKeys.buffer_chunk_prefix(channel_id) if channel_id else ""
 
-        self.chunk_ttl = getattr(Config, 'REDIS_CHUNK_TTL', 60)
+        self.chunk_ttl = ConfigHelper.redis_chunk_ttl()
 
         # Initialize from Redis if available
         if self.redis_client and channel_id:
@@ -37,7 +40,7 @@ class StreamBuffer:
                 logger.error(f"Error initializing buffer from Redis: {e}")
 
         self._write_buffer = bytearray()
-        self.target_chunk_size = getattr(Config, 'BUFFER_CHUNK_SIZE', 188 * 5644)  # ~1MB default
+        self.target_chunk_size = ConfigHelper.get('BUFFER_CHUNK_SIZE', TS_PACKET_SIZE * 5644)  # ~1MB default
 
         # Track timers for proper cleanup
         self.stopping = False
@@ -57,7 +60,7 @@ class StreamBuffer:
             combined_data = bytearray(self._partial_packet) + bytearray(chunk)
 
             # Calculate complete packets
-            complete_packets_size = (len(combined_data) // 188) * 188
+            complete_packets_size = (len(combined_data) // self.TS_PACKET_SIZE) * self.TS_PACKET_SIZE
 
             if complete_packets_size == 0:
                 # Not enough data for a complete packet
@@ -82,7 +85,7 @@ class StreamBuffer:
                     # Write optimized chunk to Redis
                     if self.redis_client:
                         chunk_index = self.redis_client.incr(self.buffer_index_key)
-                        chunk_key = f"{self.buffer_prefix}{chunk_index}"
+                        chunk_key = RedisKeys.buffer_chunk(self.channel_id, chunk_index)
                         self.redis_client.setex(chunk_key, self.chunk_ttl, bytes(chunk_data))
 
                         # Update local tracking
@@ -146,7 +149,7 @@ class StreamBuffer:
             # Directly fetch from Redis using pipeline for efficiency
             pipe = self.redis_client.pipeline()
             for idx in range(start_id, end_id):
-                chunk_key = f"{self.buffer_prefix}{idx}"
+                chunk_key = RedisKeys.buffer_chunk(self.channel_id, idx)
                 pipe.get(chunk_key)
 
             results = pipe.execute()
@@ -200,7 +203,7 @@ class StreamBuffer:
             # Directly fetch from Redis using pipeline
             pipe = self.redis_client.pipeline()
             for idx in range(start_id, end_id):
-                chunk_key = f"{self.buffer_prefix}{idx}"
+                chunk_key = RedisKeys.buffer_chunk(self.channel_id, idx)
                 pipe.get(chunk_key)
 
             results = pipe.execute()
@@ -222,7 +225,7 @@ class StreamBuffer:
         """Stop the buffer and cancel all timers"""
         # Set stopping flag first to prevent new timer creation
         self.stopping = True
-        
+
         # Cancel all pending timers
         timers_cancelled = 0
         for timer in list(self.fill_timers):
@@ -232,10 +235,10 @@ class StreamBuffer:
                     timers_cancelled += 1
             except Exception as e:
                 logger.error(f"Error canceling timer: {e}")
-        
+
         if timers_cancelled:
             logger.info(f"Cancelled {timers_cancelled} buffer timers for channel {self.channel_id}")
-            
+
         # Clear timer list
         self.fill_timers.clear()
 
@@ -315,7 +318,7 @@ class StreamBuffer:
         """Schedule a timer and track it for proper cleanup"""
         if self.stopping:
             return None
-            
+
         timer = threading.Timer(delay, callback, args=args, kwargs=kwargs)
         timer.daemon = True
         timer.start()

apps/proxy/ts_proxy/stream_generator.py

@@ -0,0 +1,344 @@
+"""
+Stream generation and client-side handling for TS streams.
+This module handles generating and delivering video streams to clients.
+"""
+
+import time
+import logging
+import threading
+from apps.proxy.config import TSConfig as Config
+from . import proxy_server
+from .utils import create_ts_packet
+from .redis_keys import RedisKeys
+
+logger = logging.getLogger("ts_proxy")
+
+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):
+        """
+        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
+        """
+        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
+
+        # 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
+
+    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 = 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
+
+            # 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 _wait_for_initialization(self):
+        """Wait for channel initialization to complete, sending keepalive packets."""
+        initialization_start = time.time()
+        max_init_wait = getattr(Config, 'CLIENT_WAIT_TIMEOUT', 30)
+        keepalive_interval = 0.5
+        last_keepalive = 0
+
+        # While init is happening, send keepalive packets
+        while time.time() - initialization_start < max_init_wait:
+            # Check if initialization has completed
+            if proxy_server.redis_client:
+                metadata_key = RedisKeys.channel_metadata(self.channel_id)
+                metadata = proxy_server.redis_client.hgetall(metadata_key)
+
+                if metadata and b'state' in metadata:
+                    state = metadata[b'state'].decode('utf-8')
+                    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']:
+                        error_message = metadata.get(b'error_message', b'Unknown error').decode('utf-8')
+                        logger.error(f"[{self.client_id}] Channel {self.channel_id} in error state: {state}, message: {error_message}")
+                        # Send error packet before giving up
+                        yield create_ts_packet('error', f"Error: {error_message}")
+                        return False
+                    else:
+                        # Still initializing - send keepalive if needed
+                        if time.time() - last_keepalive >= keepalive_interval:
+                            status_msg = f"Initializing: {state}"
+                            keepalive_packet = create_ts_packet('keepalive', status_msg)
+                            logger.debug(f"[{self.client_id}] Sending keepalive packet during initialization, state={state}")
+                            yield keepalive_packet
+                            self.bytes_sent += len(keepalive_packet)
+                            last_keepalive = time.time()
+
+            # Wait a bit before checking again
+            time.sleep(0.1)
+
+        # Timed out waiting
+        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."""
+        # Get buffer - stream manager may not exist in this worker
+        buffer = proxy_server.stream_buffers.get(self.channel_id)
+        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 - use config for initial position
+        initial_behind = getattr(Config, 'INITIAL_BEHIND_CHUNKS', 10)
+        current_buffer_index = buffer.index
+        self.local_index = max(0, current_buffer_index - initial_behind)
+
+        # Store important objects as instance variables
+        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."""
+        bytes_sent = 0
+        chunks_sent = 0
+        stream_start_time = time.time()
+        local_index = self.local_index
+
+        # 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(local_index)
+
+            if chunks:
+                yield from self._process_chunks(chunks, next_index, bytes_sent, chunks_sent, stream_start_time)
+                local_index = next_index
+                self.local_index = local_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
+
+                if self._should_send_keepalive(local_index):
+                    keepalive_packet = create_ts_packet('keepalive')
+                    logger.debug(f"[{self.client_id}] Sending keepalive packet while waiting at buffer head")
+                    yield keepalive_packet
+                    bytes_sent += len(keepalive_packet)
+                    self.last_yield_time = time.time()
+                    self.consecutive_empty = 0  # Reset consecutive counter but keep total empty_reads
+                    time.sleep(Config.KEEPALIVE_INTERVAL)
+                else:
+                    # Standard wait with backoff
+                    sleep_time = min(0.1 * self.consecutive_empty, 1.0)
+                    time.sleep(sleep_time)
+
+                # 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 {local_index} (buffer at {self.buffer.index}, stream: {stream_status})")
+
+                # Check for ghost clients
+                if self._is_ghost_client(local_index):
+                    logger.warning(f"[{self.client_id}] Possible ghost client: buffer has advanced {self.buffer.index - local_index} chunks ahead but client stuck at {local_index}")
+                    break
+
+                # Check for timeouts
+                if self._is_timeout():
+                    break
+
+    def _check_resources(self):
+        """Check if required resources still exist."""
+        # Enhanced resource checks
+        if self.channel_id not in proxy_server.stream_buffers:
+            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
+
+        # Check if this specific client has been stopped (Redis keys, etc.)
+        if proxy_server.redis_client:
+            # 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
+
+            # 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
+
+            # Also check if client has been removed from client_manager
+            if self.channel_id in proxy_server.client_managers:
+                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
+
+        return True
+
+    def _process_chunks(self, chunks, next_index, bytes_sent, chunks_sent, stream_start_time):
+        """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}")
+
+        # Send the chunks to the client
+        for chunk in chunks:
+            try:
+                yield chunk
+                bytes_sent += len(chunk)
+                chunks_sent += 1
+
+                # Log every 100 chunks for visibility
+                if chunks_sent % 100 == 0:
+                    elapsed = time.time() - stream_start_time
+                    rate = bytes_sent / elapsed / 1024 if elapsed > 0 else 0
+                    logger.info(f"[{self.client_id}] Stats: {chunks_sent} chunks, {bytes_sent/1024:.1f}KB, {rate:.1f}KB/s")
+            except Exception as e:
+                logger.error(f"[{self.client_id}] Error sending chunk to client: {e}")
+                raise  # Re-raise to exit the generator
+
+        return bytes_sent, chunks_sent
+
+    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 we're at buffer head and no data is coming, send keepalive
+        stream_healthy = self.stream_manager.healthy if self.stream_manager else True
+        return at_buffer_head and not stream_healthy and self.consecutive_empty >= 5
+
+    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."""
+        # Disconnect after long inactivity
+        if time.time() - self.last_yield_time > Config.STREAM_TIMEOUT:
+            if self.stream_manager and not self.stream_manager.healthy:
+                logger.warning(f"[{self.client_id}] No data for {Config.STREAM_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 {Config.STREAM_TIMEOUT}s, disconnecting")
+                return True
+        return False
+
+    def _cleanup(self):
+        """Clean up resources and report final statistics."""
+        # Client cleanup
+        elapsed = time.time() - self.stream_start_time
+        local_clients = 0
+        total_clients = 0
+
+        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)
+            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})")
+
+            # Schedule channel shutdown if no clients left
+            self._schedule_channel_shutdown_if_needed(local_clients)
+
+    def _schedule_channel_shutdown_if_needed(self, local_clients):
+        """
+        Schedule channel shutdown if there are no clients left and we're the owner.
+        """
+        # If no clients left and we're the owner, schedule shutdown using the config value
+        if local_clients == 0 and proxy_server.am_i_owner(self.channel_id):
+            logger.info(f"No local clients left for channel {self.channel_id}, scheduling shutdown")
+
+            def delayed_shutdown():
+                # Use the config setting instead of hardcoded value
+                shutdown_delay = getattr(Config, 'CHANNEL_SHUTDOWN_DELAY', 5)
+                logger.info(f"Waiting {shutdown_delay}s before checking if channel should be stopped")
+                time.sleep(shutdown_delay)
+
+                # After delay, check global client count
+                if self.channel_id in proxy_server.client_managers:
+                    total = proxy_server.client_managers[self.channel_id].get_total_client_count()
+                    if total == 0:
+                        logger.info(f"Shutting down channel {self.channel_id} as no clients connected")
+                        proxy_server.stop_channel(self.channel_id)
+                    else:
+                        logger.info(f"Not shutting down channel {self.channel_id}, {total} clients still connected")
+
+            shutdown_thread = threading.Thread(target=delayed_shutdown)
+            shutdown_thread.daemon = True
+            shutdown_thread.start()
+
+def create_stream_generator(channel_id, client_id, client_ip, client_user_agent, channel_initializing=False):
+    """
+    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)
+    return generator.generate

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
@@ -13,6 +14,9 @@ from apps.m3u.models import M3UAccount, M3UAccountProfile
 from core.models import UserAgent, CoreSettings
 from .stream_buffer import StreamBuffer
 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
 
 logger = logging.getLogger("ts_proxy")
 
@@ -27,7 +31,7 @@ class StreamManager:
         self.running = True
         self.connected = False
         self.retry_count = 0
-        self.max_retries = Config.MAX_RETRIES
+        self.max_retries = ConfigHelper.max_retries()
         self.current_response = None
         self.current_session = None
         self.url_switching = False
@@ -43,8 +47,8 @@ class StreamManager:
         # Stream health monitoring
         self.last_data_time = time.time()
         self.healthy = True
-        self.health_check_interval = Config.HEALTH_CHECK_INTERVAL
-        self.chunk_size = getattr(Config, 'CHUNK_SIZE', 8192)
+        self.health_check_interval = ConfigHelper.get('HEALTH_CHECK_INTERVAL', 5)
+        self.chunk_size = ConfigHelper.chunk_size()
 
         # Add to your __init__ method
         self._buffer_check_timers = []
@@ -84,7 +88,7 @@ class StreamManager:
         try:
             # Check stream type before connecting
             stream_type = detect_stream_type(self.url)
-            if self.transcode == False and stream_type == 'hls':
+            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
@@ -192,7 +196,7 @@ class StreamManager:
 
                                             # Update last data timestamp in Redis
                                             if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
-                                                last_data_key = f"ts_proxy:channel:{self.buffer.channel_id}:last_data"
+                                                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:
@@ -411,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
@@ -431,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
@@ -466,7 +473,7 @@ class StreamManager:
 
             # Update last data timestamp in Redis if successful
             if success and hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
-                last_data_key = f"ts_proxy:channel:{self.buffer.channel_id}:last_data"
+                last_data_key = RedisKeys.last_data(self.buffer.channel_id)
                 self.buffer.redis_client.set(last_data_key, str(time.time()), ex=60)
 
             return True
@@ -491,7 +498,7 @@ class StreamManager:
 
                 if channel_id and redis_client:
                     current_time = str(time.time())
-                    metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
+                    metadata_key = RedisKeys.channel_metadata(channel_id)
 
                     # Check current state first
                     current_state = None
@@ -503,16 +510,16 @@ class StreamManager:
                         logger.error(f"Error checking current state: {e}")
 
                     # Only update if not already past connecting
-                    if not current_state or current_state in ["initializing", "connecting"]:
+                    if not current_state or current_state in [ChannelState.INITIALIZING, ChannelState.CONNECTING]:
                         # NEW CODE: Check if buffer has enough chunks
                         current_buffer_index = getattr(self.buffer, 'index', 0)
-                        initial_chunks_needed = getattr(Config, 'INITIAL_BEHIND_CHUNKS', 10)
+                        initial_chunks_needed = ConfigHelper.initial_behind_chunks()
 
                         if current_buffer_index < initial_chunks_needed:
                             # Not enough buffer yet - set to connecting state if not already
-                            if current_state != "connecting":
+                            if current_state != ChannelState.CONNECTING:
                                 update_data = {
-                                    "state": "connecting",
+                                    "state": ChannelState.CONNECTING,
                                     "state_changed_at": current_time
                                 }
                                 redis_client.hset(metadata_key, mapping=update_data)
@@ -526,7 +533,7 @@ class StreamManager:
 
                         # We have enough buffer, proceed with state change
                         update_data = {
-                            "state": "waiting_for_clients",
+                            "state": ChannelState.WAITING_FOR_CLIENTS,
                             "connection_ready_time": current_time,
                             "state_changed_at": current_time,
                             "buffer_chunks": str(current_buffer_index)
@@ -534,8 +541,8 @@ class StreamManager:
                         redis_client.hset(metadata_key, mapping=update_data)
 
                         # Get configured grace period or default
-                        grace_period = getattr(Config, 'CHANNEL_INIT_GRACE_PERIOD', 20)
-                        logger.info(f"STREAM MANAGER: Updated channel {channel_id} state: {current_state or 'None'} → waiting_for_clients with {current_buffer_index} buffer chunks")
+                        grace_period = ConfigHelper.get('CHANNEL_INIT_GRACE_PERIOD', 20)
+                        logger.info(f"STREAM MANAGER: Updated channel {channel_id} state: {current_state or 'None'} → {ChannelState.WAITING_FOR_CLIENTS} with {current_buffer_index} buffer chunks")
                         logger.info(f"Started initial connection grace period ({grace_period}s) for channel {channel_id}")
                     else:
                         logger.debug(f"Not changing state: channel {channel_id} already in {current_state} state")

apps/proxy/ts_proxy/utils.py

@@ -34,4 +34,47 @@ def detect_stream_type(url):
         return 'hls'
 
     # Default to TS
-    return 'ts'
+    return 'ts'
+
+def get_client_ip(request):
+    """
+    Extract client IP address from request.
+    Handles cases where request is behind a proxy by checking X-Forwarded-For.
+    """
+    x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
+    if x_forwarded_for:
+        ip = x_forwarded_for.split(',')[0]
+    else:
+        ip = request.META.get('REMOTE_ADDR')
+    return ip
+
+def create_ts_packet(packet_type='null', message=None):
+    """
+    Create a Transport Stream (TS) packet for various purposes.
+
+    Args:
+        packet_type (str): Type of packet - 'null', 'error', 'keepalive', etc.
+        message (str): Optional message to include in packet payload
+
+    Returns:
+        bytes: A properly formatted 188-byte TS packet
+    """
+    packet = bytearray(188)
+
+    # TS packet header
+    packet[0] = 0x47  # Sync byte
+
+    # PID - Use different PIDs based on packet type
+    if packet_type == 'error':
+        packet[1] = 0x1F  # PID high bits
+        packet[2] = 0xFF  # PID low bits
+    else:  # null/keepalive packets
+        packet[1] = 0x1F  # PID high bits (null packet)
+        packet[2] = 0xFF  # PID low bits (null packet)
+
+    # Add message to payload if provided
+    if message:
+        msg_bytes = message.encode('utf-8')
+        packet[4:4+min(len(msg_bytes), 180)] = msg_bytes[:180]
+
+    return bytes(packet)

apps/proxy/ts_proxy/views.py

@@ -5,30 +5,27 @@ import random
 import re
 from django.http import StreamingHttpResponse, JsonResponse, HttpResponseRedirect
 from django.views.decorators.csrf import csrf_exempt
-from django.views.decorators.http import require_http_methods, require_GET
 from django.shortcuts import get_object_or_404
 from apps.proxy.config import TSConfig as Config
 from . import proxy_server
 from .channel_status import ChannelStatus
+from .stream_generator import create_stream_generator
+from .utils import get_client_ip
+from .redis_keys import RedisKeys
 import logging
 from apps.channels.models import Channel, Stream
 from apps.m3u.models import M3UAccount, M3UAccountProfile
 from core.models import UserAgent, CoreSettings, PROXY_PROFILE_NAME
 from rest_framework.decorators import api_view, permission_classes
 from rest_framework.permissions import IsAuthenticated
+from .constants import ChannelState, EventType, StreamType
+from .config_helper import ConfigHelper
+from .services.channel_service import ChannelService
 
 # Configure logging properly
 logger = logging.getLogger("ts_proxy")
 
 
-def get_client_ip(request):
-    x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
-    if x_forwarded_for:
-        ip = x_forwarded_for.split(',')[0]
-    else:
-        ip = request.META.get('REMOTE_ADDR')
-    return ip
-
 @api_view(['GET'])
 def stream_ts(request, channel_id):
     """Stream TS data to client with immediate response and keep-alive packets during initialization"""
@@ -97,12 +94,12 @@ def stream_ts(request, channel_id):
             else:
                 transcode = True
 
-            # Initialize channel with the stream's user agent (not the client's)
-            success = proxy_server.initialize_channel(stream_url, channel_id, stream_user_agent, transcode)
-            if proxy_server.redis_client:
-                metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
-                profile_value = str(stream_profile)
-                proxy_server.redis_client.hset(metadata_key, "profile", profile_value)
+            # Initialize channel using the service
+            profile_value = str(stream_profile)
+            success = ChannelService.initialize_channel(
+                channel_id, stream_url, stream_user_agent, transcode, profile_value
+            )
+
             if not success:
                 return JsonResponse({'error': 'Failed to initialize channel'}, status=500)
 
@@ -111,8 +108,9 @@ def stream_ts(request, channel_id):
                 manager = proxy_server.stream_managers.get(channel_id)
                 if manager:
                     wait_start = time.time()
+                    timeout = ConfigHelper.connection_timeout()
                     while not manager.connected:
-                        if time.time() - wait_start > Config.CONNECTION_TIMEOUT:
+                        if time.time() - wait_start > timeout:
                             proxy_server.stop_channel(channel_id)
                             return JsonResponse({'error': 'Connection timeout'}, status=504)
                         if not manager.should_retry():
@@ -134,7 +132,7 @@ def stream_ts(request, channel_id):
             stream_user_agent = None  # Initialize the variable
 
             if proxy_server.redis_client:
-                metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
+                metadata_key = RedisKeys.channel_metadata(channel_id)
                 url_bytes = proxy_server.redis_client.hget(metadata_key, "url")
                 ua_bytes = proxy_server.redis_client.hget(metadata_key, "user_agent")
                 profile_bytes = proxy_server.redis_client.hget(metadata_key, "profile")
@@ -168,291 +166,18 @@ def stream_ts(request, channel_id):
         client_manager.add_client(client_id, client_ip, client_user_agent)
         logger.info(f"[{client_id}] Client registered with channel {channel_id}")
 
-        # Define a single generate function
-        def generate():
-            stream_start_time = time.time()
-            bytes_sent = 0
-            chunks_sent = 0
+        # Create a stream generator for this client
+        generate = create_stream_generator(
+            channel_id, client_id, client_ip, client_user_agent, channel_initializing
+        )
 
-            # Keep track of initialization state
-            initialization_start = time.time()
-            max_init_wait = getattr(Config, 'CLIENT_WAIT_TIMEOUT', 30)
-            channel_ready = not channel_initializing
-            keepalive_interval = 0.5
-            last_keepalive = 0
-
-            try:
-                logger.info(f"[{client_id}] Stream generator started, channel_ready={channel_ready}")
-
-                # Wait for initialization to complete if needed
-                if not channel_ready:
-                    # While init is happening, send keepalive packets
-                    while time.time() - initialization_start < max_init_wait:
-                        # Check if initialization has completed
-                        if proxy_server.redis_client:
-                            metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
-                            metadata = proxy_server.redis_client.hgetall(metadata_key)
-
-                            if metadata and b'state' in metadata:
-                                state = metadata[b'state'].decode('utf-8')
-                                if state in ['waiting_for_clients', 'active']:
-                                    logger.info(f"[{client_id}] Channel {channel_id} now ready (state={state})")
-                                    channel_ready = True
-                                    break
-                                elif state in ['error', 'stopped']:
-                                    error_message = metadata.get(b'error_message', b'Unknown error').decode('utf-8')
-                                    logger.error(f"[{client_id}] Channel {channel_id} in error state: {state}, message: {error_message}")
-                                    # Send error in a comment TS packet before giving up
-                                    error_packet = bytearray(188)
-                                    error_packet[0] = 0x47  # Sync byte
-                                    error_packet[1] = 0x1F  # PID high bits
-                                    error_packet[2] = 0xFF  # PID low bits
-                                    error_msg = f"Error: {error_message}".encode('utf-8')
-                                    error_packet[4:4+min(len(error_msg), 180)] = error_msg[:180]
-                                    yield bytes(error_packet)
-                                    return
-                                else:
-                                    # Still initializing - send keepalive if needed
-                                    if time.time() - last_keepalive >= keepalive_interval:
-                                        keepalive_packet = bytearray(188)
-                                        keepalive_packet[0] = 0x47  # Sync byte
-                                        keepalive_packet[1] = 0x1F  # PID high bits (null packet)
-                                        keepalive_packet[2] = 0xFF  # PID low bits (null packet)
-
-                                        # Add status info in packet payload (will be ignored by players)
-                                        status_msg = f"Initializing: {state}".encode('utf-8')
-                                        keepalive_packet[4:4+min(len(status_msg), 180)] = status_msg[:180]
-
-                                        logger.debug(f"[{client_id}] Sending keepalive packet during initialization, state={state}")
-                                        yield bytes(keepalive_packet)
-                                        bytes_sent += len(keepalive_packet)
-                                        last_keepalive = time.time()
-
-                        # Wait a bit before checking again (don't send too many keepalives)
-                        time.sleep(0.1)
-
-                    # Check if we timed out waiting
-                    if not channel_ready:
-                        logger.warning(f"[{client_id}] Timed out waiting for initialization")
-                        error_packet = bytearray(188)
-                        error_packet[0] = 0x47  # Sync byte
-                        error_packet[1] = 0x1F  # PID high bits
-                        error_packet[2] = 0xFF  # PID low bits
-                        error_msg = f"Error: Initialization timeout".encode('utf-8')
-                        error_packet[4:4+min(len(error_msg), 180)] = error_msg[:180]
-                        yield bytes(error_packet)
-                        return
-
-                # Channel is now ready - original streaming code goes here
-                logger.info(f"[{client_id}] Channel {channel_id} ready, starting normal streaming")
-
-                # Reset start time for real streaming
-                stream_start_time = time.time()
-
-                # Get buffer - stream manager may not exist in this worker
-                buffer = proxy_server.stream_buffers.get(channel_id)
-                stream_manager = proxy_server.stream_managers.get(channel_id)
-
-                if not buffer:
-                    logger.error(f"[{client_id}] No buffer found for channel {channel_id}")
-                    return
-
-                # Client state tracking - use config for initial position
-                initial_behind = getattr(Config, 'INITIAL_BEHIND_CHUNKS', 10)
-                current_buffer_index = buffer.index
-                local_index = max(0, current_buffer_index - initial_behind)
-                logger.debug(f"[{client_id}] Buffer at {current_buffer_index}, starting {initial_behind} chunks behind at index {local_index}")
-
-                initial_position = local_index
-                last_yield_time = time.time()
-                empty_reads = 0
-                bytes_sent = 0
-                chunks_sent = 0
-                stream_start_time = time.time()
-                consecutive_empty = 0  # Track consecutive empty reads
-
-                # Timing parameters from config
-                ts_packet_size = 188
-                target_bitrate = Config.TARGET_BITRATE
-                packets_per_second = target_bitrate / (8 * ts_packet_size)
-
-                logger.info(f"[{client_id}] Starting stream at index {local_index} (buffer at {buffer.index})")
-
-                # Check if we're the owner worker
-                is_owner_worker = proxy_server.am_i_owner(channel_id) if hasattr(proxy_server, 'am_i_owner') else True
-
-                # Main streaming loop
-                while True:
-                    # Enhanced resource checks
-                    if channel_id not in proxy_server.stream_buffers:
-                        logger.info(f"[{client_id}] Channel buffer no longer exists, terminating stream")
-                        break
-
-                    if channel_id not in proxy_server.client_managers:
-                        logger.info(f"[{client_id}] Client manager no longer exists, terminating stream")
-                        break
-
-                    # Check if this specific client has been stopped
-                    if proxy_server.redis_client:
-                        # Channel stop check
-                        stop_key = f"ts_proxy:channel:{channel_id}:stopping"
-                        if proxy_server.redis_client.exists(stop_key):
-                            logger.info(f"[{client_id}] Detected channel stop signal, terminating stream")
-                            break
-
-                        # Client stop check - NEW
-                        client_stop_key = f"ts_proxy:channel:{channel_id}:client:{client_id}:stop"
-                        if proxy_server.redis_client.exists(client_stop_key):
-                            logger.info(f"[{client_id}] Detected client stop signal, terminating stream")
-                            break
-
-                        # Also check if client has been removed from client_manager
-                        if channel_id in proxy_server.client_managers:
-                            client_manager = proxy_server.client_managers[channel_id]
-                            if client_id not in client_manager.clients:
-                                logger.info(f"[{client_id}] Client no longer in client manager, terminating stream")
-                                break
-
-                    # Get chunks at client's position using improved strategy
-                    chunks, next_index = buffer.get_optimized_client_data(local_index)
-
-                    if chunks:
-                        empty_reads = 0
-                        consecutive_empty = 0
-
-                        # Process and send chunks
-                        total_size = sum(len(c) for c in chunks)
-                        logger.debug(f"[{client_id}] Retrieved {len(chunks)} chunks ({total_size} bytes) from index {local_index+1} to {next_index}")
-
-                        # CRITICAL FIX: Actually send the chunks to the client
-                        for chunk in chunks:
-                            try:
-                                # This is the crucial line that was likely missing
-                                yield chunk
-                                bytes_sent += len(chunk)
-                                chunks_sent += 1
-
-                                # Log every 100 chunks for visibility
-                                if chunks_sent % 100 == 0:
-                                    elapsed = time.time() - stream_start_time
-                                    rate = bytes_sent / elapsed / 1024 if elapsed > 0 else 0
-                                    logger.info(f"[{client_id}] Stats: {chunks_sent} chunks, {bytes_sent/1024:.1f}KB, {rate:.1f}KB/s")
-                            except Exception as e:
-                                logger.error(f"[{client_id}] Error sending chunk to client: {e}")
-                                raise  # Re-raise to exit the generator
-
-                        # Update index after successfully sending all chunks
-                        local_index = next_index
-                        last_yield_time = time.time()
-                    else:
-                        # No chunks available
-                        empty_reads += 1
-                        consecutive_empty += 1
-
-                        # Check if we're caught up to buffer head
-                        at_buffer_head = local_index >= buffer.index
-
-                        # If we're at buffer head and no data is coming, send keepalive
-                        # Only check stream manager health if it exists
-                        stream_healthy = stream_manager.healthy if stream_manager else True
-
-                        if at_buffer_head and not stream_healthy and consecutive_empty >= 5:
-                            # Create a null TS packet as keepalive (188 bytes filled with padding)
-                            # This prevents VLC from hitting EOF
-                            keepalive_packet = bytearray(188)
-                            keepalive_packet[0] = 0x47  # Sync byte
-                            keepalive_packet[1] = 0x1F  # PID high bits (null packet)
-                            keepalive_packet[2] = 0xFF  # PID low bits (null packet)
-
-                            logger.debug(f"[{client_id}] Sending keepalive packet while waiting at buffer head")
-                            yield bytes(keepalive_packet)
-                            bytes_sent += len(keepalive_packet)
-                            last_yield_time = time.time()
-                            consecutive_empty = 0  # Reset consecutive counter but keep total empty_reads
-                            time.sleep(Config.KEEPALIVE_INTERVAL)
-                        else:
-                            # Standard wait
-                            sleep_time = min(0.1 * consecutive_empty, 1.0)  # Progressive backoff up to 1s
-                            time.sleep(sleep_time)
-
-                        # Log empty reads periodically
-                        if empty_reads % 50 == 0:
-                            stream_status = "healthy" if (stream_manager and stream_manager.healthy) else "unknown"
-                            logger.debug(f"[{client_id}] Waiting for chunks beyond {local_index} (buffer at {buffer.index}, stream: {stream_status})")
-
-                        # CRITICAL FIX: Check for client disconnect during wait periods
-                        # Django/WSGI might not immediately detect disconnections, but we can check periodically
-                        if consecutive_empty > 10:  # After some number of empty reads
-                            if hasattr(request, 'META') and request.META.get('wsgi.input'):
-                                try:
-                                    # Try to check if the connection is still alive
-                                    available = request.META['wsgi.input'].read(0)
-                                    if available is None:  # Connection closed
-                                        logger.info(f"[{client_id}] Detected client disconnect during wait")
-                                        break
-                                except Exception:
-                                    # Error reading from connection, likely closed
-                                    logger.info(f"[{client_id}] Connection error, client likely disconnected")
-                                    break
-
-                        # Disconnect after long inactivity
-                        # For non-owner workers, we're more lenient with timeout
-                        if time.time() - last_yield_time > Config.STREAM_TIMEOUT:
-                            if stream_manager and not stream_manager.healthy:
-                                logger.warning(f"[{client_id}] No data for {Config.STREAM_TIMEOUT}s and stream unhealthy, disconnecting")
-                                break
-                            elif not is_owner_worker and consecutive_empty > 100:
-                                # Non-owner worker without data for too long
-                                logger.warning(f"[{client_id}] Non-owner worker with no data for {Config.STREAM_TIMEOUT}s, disconnecting")
-                                break
-
-                        # ADD THIS: Check if worker has more recent chunks but still stuck
-                        # This can indicate the client is disconnected but we're not detecting it
-                        if consecutive_empty > 100 and buffer.index > local_index + 50:
-                            logger.warning(f"[{client_id}] Possible ghost client: buffer has advanced {buffer.index - local_index} chunks ahead but client stuck at {local_index}")
-                            break
-
-            except Exception as e:
-                logger.error(f"[{client_id}] Stream error: {e}", exc_info=True)
-            finally:
-                # Client cleanup
-                elapsed = time.time() - stream_start_time
-                local_clients = 0
-
-                if channel_id in proxy_server.client_managers:
-                    local_clients = proxy_server.client_managers[channel_id].remove_client(client_id)
-                    total_clients = proxy_server.client_managers[channel_id].get_total_client_count()
-                    logger.info(f"[{client_id}] Disconnected after {elapsed:.2f}s, {bytes_sent/1024:.1f}KB in {chunks_sent} chunks (local: {local_clients}, total: {total_clients})")
-
-                    # If no clients left and we're the owner, schedule shutdown using the config value
-                    if local_clients == 0 and proxy_server.am_i_owner(channel_id):
-                        logger.info(f"No local clients left for channel {channel_id}, scheduling shutdown")
-                        def delayed_shutdown():
-                            # Use the config setting instead of hardcoded value
-                            shutdown_delay = getattr(Config, 'CHANNEL_SHUTDOWN_DELAY', 5)
-                            logger.info(f"Waiting {shutdown_delay}s before checking if channel should be stopped")
-                            time.sleep(shutdown_delay)
-
-                            # After delay, check global client count
-                            if channel_id in proxy_server.client_managers:
-                                total = proxy_server.client_managers[channel_id].get_total_client_count()
-                                if total == 0:
-                                    logger.info(f"Shutting down channel {channel_id} as no clients connected")
-                                    proxy_server.stop_channel(channel_id)
-                                else:
-                                    logger.info(f"Not shutting down channel {channel_id}, {total} clients still connected")
-
-                        shutdown_thread = threading.Thread(target=delayed_shutdown)
-                        shutdown_thread.daemon = True
-                        shutdown_thread.start()
-
-        # IMPORTANT: Return the StreamingHttpResponse from the main function
+        # Return the StreamingHttpResponse from the main function
         response = StreamingHttpResponse(
             streaming_content=generate(),
             content_type='video/mp2t'
         )
         response['Cache-Control'] = 'no-cache'
-        return response  # This now properly returns from stream_ts
+        return response
 
     except Exception as e:
         logger.error(f"Error in stream_ts: {e}", exc_info=True)
@@ -473,91 +198,17 @@ def change_stream(request, channel_id):
 
         logger.info(f"Attempting to change stream URL for channel {channel_id} to {new_url}")
 
-        # Enhanced channel detection
-        in_local_managers = channel_id in proxy_server.stream_managers
-        in_local_buffers = channel_id in proxy_server.stream_buffers
+        # Use the service layer instead of direct implementation
+        result = ChannelService.change_stream_url(channel_id, new_url, user_agent)
 
-        # First check Redis directly before using our wrapper method
-        redis_keys = None
-        if proxy_server.redis_client:
-            try:
-                redis_keys = proxy_server.redis_client.keys(f"ts_proxy:*:{channel_id}*")
-                redis_keys = [k.decode('utf-8') for k in redis_keys] if redis_keys else []
-            except Exception as e:
-                logger.error(f"Error checking Redis keys: {e}")
+        if result.get('status') == 'error':
+            return JsonResponse({
+                'error': result.get('message', 'Unknown error'),
+                'diagnostics': result.get('diagnostics', {})
+            }, status=404)
 
-        # Now use our standard check
-        channel_exists = proxy_server.check_if_channel_exists(channel_id)
-
-        # Log detailed diagnostics
-        logger.info(f"Channel {channel_id} diagnostics: "
-                   f"in_local_managers={in_local_managers}, "
-                   f"in_local_buffers={in_local_buffers}, "
-                   f"redis_keys_count={len(redis_keys) if redis_keys else 0}, "
-                   f"channel_exists={channel_exists}")
-
-        if not channel_exists:
-            # If channel doesn't exist but we found Redis keys, force initialize it
-            if redis_keys:
-                logger.warning(f"Channel {channel_id} not detected by check_if_channel_exists but Redis keys exist. Forcing initialization.")
-                proxy_server.initialize_channel(new_url, channel_id, user_agent)
-            else:
-                logger.error(f"Channel {channel_id} not found in any worker or Redis")
-                return JsonResponse({
-                    'error': 'Channel not found',
-                    'diagnostics': {
-                        'in_local_managers': in_local_managers,
-                        'in_local_buffers': in_local_buffers,
-                        'redis_keys': redis_keys,
-                    }
-                }, status=404)
-
-        # Update metadata in Redis regardless of ownership - this ensures URL is updated
-        # even if the owner worker is handling another request
-        if proxy_server.redis_client:
-            try:
-                metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
-
-                # First check if the key exists and what type it is
-                key_type = proxy_server.redis_client.type(metadata_key).decode('utf-8')
-                logger.debug(f"Redis key {metadata_key} is of type: {key_type}")
-
-                # Use the appropriate method based on the key type
-                if key_type == 'hash':
-                    proxy_server.redis_client.hset(metadata_key, "url", new_url)
-                    if user_agent:
-                        proxy_server.redis_client.hset(metadata_key, "user_agent", user_agent)
-                elif key_type == 'none':  # Key doesn't exist yet
-                    # Create new hash with all required fields
-                    metadata = {"url": new_url}
-                    if user_agent:
-                        metadata["user_agent"] = user_agent
-                    proxy_server.redis_client.hset(metadata_key, mapping=metadata)
-                else:
-                    # If key exists with wrong type, delete it and recreate
-                    proxy_server.redis_client.delete(metadata_key)
-                    metadata = {"url": new_url}
-                    if user_agent:
-                        metadata["user_agent"] = user_agent
-                    proxy_server.redis_client.hset(metadata_key, mapping=metadata)
-
-                # Set switch request flag to ensure all workers see it
-                switch_key = f"ts_proxy:channel:{channel_id}:switch_request"
-                proxy_server.redis_client.setex(switch_key, 30, new_url)  # 30 second TTL
-
-                logger.info(f"Updated metadata for channel {channel_id} in Redis")
-            except Exception as e:
-                logger.error(f"Error updating Redis metadata: {e}", exc_info=True)
-
-        # If we're the owner, update directly
-        if proxy_server.am_i_owner(channel_id) and channel_id in proxy_server.stream_managers:
-            logger.info(f"This worker is the owner, changing stream URL for channel {channel_id}")
-            manager = proxy_server.stream_managers[channel_id]
-            old_url = manager.url
-
-            # Update the stream
-            result = manager.update_url(new_url)
-            logger.info(f"Stream URL changed from {old_url} to {new_url}, result: {result}")
+        # Format response based on whether it was a direct update or event-based
+        if result.get('direct_update'):
             return JsonResponse({
                 'message': 'Stream URL updated',
                 'channel': channel_id,
@@ -565,25 +216,7 @@ def change_stream(request, channel_id):
                 'owner': True,
                 'worker_id': proxy_server.worker_id
             })
-
-        # If we're not the owner, publish an event for the owner to pick up
         else:
-            logger.info(f"This worker is not the owner, requesting URL change via Redis PubSub")
-            # Publish switch request event
-            switch_request = {
-                "event": "stream_switch",
-                "channel_id": channel_id,
-                "url": new_url,
-                "user_agent": user_agent,
-                "requester": proxy_server.worker_id,
-                "timestamp": time.time()
-            }
-
-            proxy_server.redis_client.publish(
-                f"ts_proxy:events:{channel_id}",
-                json.dumps(switch_request)
-            )
-
             return JsonResponse({
                 'message': 'Stream URL change requested',
                 'channel': channel_id,
@@ -653,61 +286,16 @@ def stop_channel(request, channel_id):
     try:
         logger.info(f"Request to stop channel {channel_id} received")
 
-        # Check if channel exists
-        channel_exists = proxy_server.check_if_channel_exists(channel_id)
-        if not channel_exists:
-            logger.warning(f"Channel {channel_id} not found in any worker or Redis")
-            return JsonResponse({'error': 'Channel not found'}, status=404)
+        # Use the service layer instead of direct implementation
+        result = ChannelService.stop_channel(channel_id)
 
-        # Get channel state information for response
-        channel_info = None
-        if proxy_server.redis_client:
-            metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
-            try:
-                metadata = proxy_server.redis_client.hgetall(metadata_key)
-                if metadata and b'state' in metadata:
-                    state = metadata[b'state'].decode('utf-8')
-                    channel_info = {"state": state}
-            except Exception as e:
-                logger.error(f"Error fetching channel state: {e}")
-
-        # Broadcast stop event to all workers via PubSub
-        if proxy_server.redis_client:
-            stop_request = {
-                "event": "channel_stop",
-                "channel_id": channel_id,
-                "requester_worker_id": proxy_server.worker_id,
-                "timestamp": time.time()
-            }
-
-            # Publish the stop event
-            proxy_server.redis_client.publish(
-                f"ts_proxy:events:{channel_id}",
-                json.dumps(stop_request)
-            )
-
-            logger.info(f"Published channel stop event for {channel_id}")
-
-            # Also stop locally to ensure this worker cleans up right away
-            result = proxy_server.stop_channel(channel_id)
-        else:
-            # No Redis, just stop locally
-            result = proxy_server.stop_channel(channel_id)
-
-        # Release the channel in the channel model if applicable
-        try:
-            channel = Channel.objects.get(uuid=channel_id)
-            channel.release_stream()
-            logger.info(f"Released channel {channel_id} stream allocation")
-        except Channel.DoesNotExist:
-            logger.warning(f"Could not find Channel model for UUID {channel_id}")
-        except Exception as e:
-            logger.error(f"Error releasing channel stream: {e}")
+        if result.get('status') == 'error':
+            return JsonResponse({'error': result.get('message', 'Unknown error')}, status=404)
 
         return JsonResponse({
             'message': 'Channel stop request sent',
             'channel_id': channel_id,
-            'previous_state': channel_info
+            'previous_state': result.get('previous_state')
         })
 
     except Exception as e:
@@ -727,55 +315,17 @@ def stop_client(request, channel_id):
         if not client_id:
             return JsonResponse({'error': 'No client_id provided'}, status=400)
 
-        logger.info(f"Request to stop client {client_id} on channel {channel_id}")
+        # Use the service layer instead of direct implementation
+        result = ChannelService.stop_client(channel_id, client_id)
 
-        # Set a Redis key for the generator to detect regardless of whether client is local
-        if proxy_server.redis_client:
-            stop_key = f"ts_proxy:channel:{channel_id}:client:{client_id}:stop"
-            proxy_server.redis_client.setex(stop_key, 30, "true")  # 30 second TTL
-            logger.info(f"Set stop key for client {client_id}")
-
-        # Check if channel exists
-        channel_exists = proxy_server.check_if_channel_exists(channel_id)
-        if not channel_exists:
-            logger.warning(f"Channel {channel_id} not found")
-            return JsonResponse({'error': 'Channel not found'}, status=404)
-
-        # Two-part approach:
-        # 1. Handle locally if client is on this worker
-        # 2. Use events to inform other workers if needed
-
-        local_client_stopped = False
-        if channel_id in proxy_server.client_managers:
-            client_manager = proxy_server.client_managers[channel_id]
-            # Use the existing remove_client method directly
-            with client_manager.lock:
-                if client_id in client_manager.clients:
-                    client_manager.remove_client(client_id)
-                    local_client_stopped = True
-                    logger.info(f"Client {client_id} stopped locally on channel {channel_id}")
-
-        # If client wasn't found locally, broadcast stop event for other workers
-        if not local_client_stopped and proxy_server.redis_client:
-            stop_request = {
-                "event": "client_stop",
-                "channel_id": channel_id,
-                "client_id": client_id,
-                "requester_worker_id": proxy_server.worker_id,
-                "timestamp": time.time()
-            }
-
-            proxy_server.redis_client.publish(
-                f"ts_proxy:events:{channel_id}",
-                json.dumps(stop_request)
-            )
-            logger.info(f"Published stop request for client {client_id} on channel {channel_id}")
+        if result.get('status') == 'error':
+            return JsonResponse({'error': result.get('message')}, status=404)
 
         return JsonResponse({
             'message': 'Client stop request processed',
             'channel_id': channel_id,
             'client_id': client_id,
-            'locally_processed': local_client_stopped
+            'locally_processed': result.get('locally_processed', False)
         })
 
     except json.JSONDecodeError: