From c83189dfcce83c7febf3f39b9850f5bb6c3293d4 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 9 Mar 2025 12:25:36 -0500 Subject: [PATCH] Initial test switching to redis for buffer. Only 1 client connecting. --- apps/proxy/ts_proxy/server.py | 514 +++++++++++++++++++++++++++++----- apps/proxy/ts_proxy/views.py | 79 ++---- dispatcharr/settings.py | 4 + 3 files changed, 469 insertions(+), 128 deletions(-) diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index 9f762808..19d78420 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -10,24 +10,37 @@ Handles live TS stream proxying with support for: import requests import threading import logging +import socket from collections import deque import time from typing import Optional, Set, Deque, Dict from apps.proxy.config import TSConfig as Config class StreamManager: - """Manages TS stream state and connection handling""" + """Manages a connection to a TS stream with continuity tracking""" - def __init__(self, initial_url: str, channel_id: str, user_agent: Optional[str] = None): - self.current_url: str = initial_url - self.channel_id: str = channel_id - self.user_agent: str = user_agent or Config.DEFAULT_USER_AGENT - self.url_changed: threading.Event = threading.Event() - self.running: bool = True - self.session: requests.Session = self._create_session() - self.connected: bool = False - self.retry_count: int = 0 - logging.info(f"Initialized stream manager for channel {channel_id}") + def __init__(self, url, buffer, user_agent=None): + self.url = url + self.buffer = buffer + self.running = True + self.connected = False + self.socket = None + self.ready_event = threading.Event() + self.retry_count = 0 + self.max_retries = 3 + + # User agent for connection + self.user_agent = user_agent or 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36' + + # TS packet handling + self.TS_PACKET_SIZE = 188 + self.recv_buffer = bytearray() + self.sync_found = False + self.continuity_counters = {} # Track continuity counters for each PID + + # Buffer management + self._last_buffer_check = time.time() + logging.info(f"Initialized stream manager for channel {buffer.channel_id}") def _create_session(self) -> requests.Session: """Create and configure requests session""" @@ -40,31 +53,332 @@ class StreamManager: def update_url(self, new_url: str) -> bool: """Update stream URL and signal connection change""" - if new_url != self.current_url: - logging.info(f"Stream switch initiated: {self.current_url} -> {new_url}") - self.current_url = new_url + if new_url != self.url: + logging.info(f"Stream switch initiated: {self.url} -> {new_url}") + self.url = new_url self.connected = False - self.url_changed.set() + self.ready_event.set() return True return False def should_retry(self) -> bool: """Check if connection retry is allowed""" - return self.retry_count < Config.MAX_RETRIES + return self.retry_count < self.max_retries def stop(self) -> None: """Clean shutdown of stream manager""" self.running = False - if self.session: - self.session.close() + if self.socket: + self.socket.close() + + def _process_complete_packets(self): + """Process only complete TS packets from buffer with continuity tracking""" + try: + # Find sync byte if needed + if not self.sync_found and len(self.recv_buffer) >= 376: + for i in range(min(188, len(self.recv_buffer) - 188)): + # Look for at least two sync bytes (0x47) at 188-byte intervals + if (self.recv_buffer[i] == 0x47 and + self.recv_buffer[i + 188] == 0x47): + + # Trim buffer to start at first sync byte + self.recv_buffer = self.recv_buffer[i:] + self.sync_found = True + logging.debug(f"TS sync found at position {i}") + break + + # If sync not found, keep last 188 bytes and return + if not self.sync_found: + if len(self.recv_buffer) > 188: + self.recv_buffer = self.recv_buffer[-188:] + return False + + # If we don't have a complete packet yet, wait for more data + if len(self.recv_buffer) < 188: + return False + + # Calculate how many complete packets we have + packet_count = len(self.recv_buffer) // 188 + + if packet_count == 0: + return False + + # Extract and process packets with continuity correction + processed_buffer = bytearray() + + for i in range(packet_count): + packet_start = i * 188 + packet = bytearray(self.recv_buffer[packet_start:packet_start + 188]) + + # Check sync byte + if packet[0] == 0x47: + # Extract PID (13 bits from bytes 1-2) + pid = ((packet[1] & 0x1F) << 8) | packet[2] + + # Only process continuity for packets that have it + # Skip adaptation-only packets and null packets + if (packet[3] & 0x10) and pid != 0x1FFF: # Has payload and not null packet + # Extract continuity counter (4 bits from byte 3) + cc = packet[3] & 0x0F + + # Correct continuity counter if we're tracking this PID + if pid in self.continuity_counters: + expected = (self.continuity_counters[pid] + 1) & 0x0F + if cc != expected: + # Rewrite continuity counter to maintain sequence + packet[3] = (packet[3] & 0xF0) | expected + + # Update counter for next time + self.continuity_counters[pid] = packet[3] & 0x0F + + # Add processed packet to buffer + processed_buffer.extend(packet) + + # Keep remaining data in buffer + self.recv_buffer = self.recv_buffer[packet_count * 188:] + + # Send processed packets to buffer + if processed_buffer: + self.buffer.add_chunk(bytes(processed_buffer)) + return True + + return False + + except Exception as e: + logging.error(f"Error processing TS packets: {e}") + self.sync_found = False # Reset sync state on error + return False + + def _process_ts_data(self, chunk): + """Process received data and add to buffer""" + if not chunk: + return False + + # Add to existing buffer + self.recv_buffer.extend(chunk) + + # Process complete packets now + return self._process_complete_packets() + + def run(self): + """Main execution loop for stream manager""" + try: + # Establish network connection + import socket + import requests + + logging.info(f"Starting stream for URL: {self.url}") + + # Parse URL + if self.url.startswith("http"): + # HTTP connection + session = self._create_session() + + try: + # Create an initial connection to get socket + response = session.get(self.url, stream=True) + if response.status_code == 200: + self.connected = True + self.socket = response.raw._fp.fp.raw + + # Main fetch loop + while self.running: + if not self.fetch_chunk(): + if not self.running: + break + time.sleep(1) + continue + + else: + logging.error(f"Failed to connect to stream: HTTP {response.status_code}") + + finally: + session.close() + else: + # Direct socket connection (UDP/TCP) + logging.error(f"Unsupported URL scheme: {self.url}") + + except Exception as e: + logging.error(f"Stream error: {e}") + self._close_socket() + finally: + self._close_socket() + logging.info("Stream manager stopped") + + def _close_socket(self): + """Close the socket connection safely""" + if self.socket: + try: + self.socket.close() + except Exception: + pass + self.socket = None + self.connected = False + + def fetch_chunk(self): + """Fetch data from socket with improved buffer management""" + if not self.connected or not self.socket: + return False + + try: + # SocketIO objects use read instead of recv and don't support settimeout + try: + # Try to read data chunk - use a multiple of TS packet size + if hasattr(self.socket, 'recv'): + chunk = self.socket.recv(188 * 64) # Standard socket + else: + chunk = self.socket.read(188 * 64) # SocketIO object + + except AttributeError: + # Fall back to read() if recv() isn't available + chunk = self.socket.read(188 * 64) + + if not chunk: + # Connection closed by server + logging.warning("Server closed connection") + self._close_socket() + self.connected = False + return False + + # Process this chunk + self._process_ts_data(chunk) + + # Memory management - clear any internal buffers periodically + current_time = time.time() + if current_time - self._last_buffer_check > 60: # Check every minute + self._last_buffer_check = current_time + if len(self.recv_buffer) > 188 * 1024: # If buffer is extremely large + logging.warning(f"Receive buffer unusually large ({len(self.recv_buffer)} bytes), trimming") + # Keep only recent data, aligned to TS packet boundary + keep_size = 188 * 128 # Keep reasonable buffer + self.recv_buffer = self.recv_buffer[-keep_size:] + + return True + + except (socket.timeout, socket.error) as e: + # Socket error + logging.error(f"Socket error: {e}") + self._close_socket() + self.connected = False + return False + + except Exception as e: + logging.error(f"Error in fetch_chunk: {e}") + return False class StreamBuffer: - """Manages stream data buffering""" + """Manages stream data buffering using Redis for persistence""" - def __init__(self): - self.buffer: Deque[bytes] = deque(maxlen=Config.BUFFER_SIZE) - self.lock: threading.Lock = threading.Lock() - self.index: int = 0 + def __init__(self, channel_id=None, redis_client=None): + self.channel_id = channel_id + self.redis_client = redis_client + self.lock = threading.Lock() + self.index = 0 + self.TS_PACKET_SIZE = 188 + + # Redis keys + self.buffer_index_key = f"ts_proxy:buffer:{channel_id}:index" + self.buffer_prefix = f"ts_proxy:buffer:{channel_id}:chunk:" + + # Expiration time for chunks + self.chunk_ttl = getattr(Config, 'REDIS_CHUNK_TTL', 60) # Default 60 seconds + + # Local tracking for performance + self.local_cache = {} + self.local_cache_size = 50 # Keep last 50 chunks in local memory + + def add_chunk(self, chunk): + """Add a chunk to the buffer""" + if not chunk: + return False + + try: + # Ensure chunk is properly aligned with TS packets + if len(chunk) % self.TS_PACKET_SIZE != 0: + logging.warning(f"Received non-aligned chunk of size {len(chunk)}") + aligned_size = (len(chunk) // self.TS_PACKET_SIZE) * self.TS_PACKET_SIZE + if (aligned_size == 0): + return False + chunk = chunk[:aligned_size] + + with self.lock: + # Increment index atomically + if self.redis_client: + # Use Redis to store and track chunks + chunk_index = self.redis_client.incr(self.buffer_index_key) + chunk_key = f"{self.buffer_prefix}{chunk_index}" + self.redis_client.setex(chunk_key, self.chunk_ttl, chunk) + + # Update local tracking + self.index = chunk_index + self.local_cache[chunk_index] = chunk + + # Trim local cache if needed + if len(self.local_cache) > self.local_cache_size: + oldest = min(self.local_cache.keys()) + del self.local_cache[oldest] + + return True + else: + # Fallback to local storage if Redis not available + logging.warning("Redis not available, using local storage only") + self.index += 1 + self.local_cache[self.index] = chunk + + if len(self.local_cache) > self.local_cache_size: + oldest = min(self.local_cache.keys()) + del self.local_cache[oldest] + + return True + + except Exception as e: + logging.error(f"Error adding chunk to buffer: {e}") + return False + + def get_chunks(self, start_index=None): + """Get chunks from the buffer starting at the given index""" + try: + if not self.redis_client: + # Local fallback + chunks = [] + with self.lock: + for idx in sorted(self.local_cache.keys()): + if start_index is None or idx > start_index: + chunks.append(self.local_cache[idx]) + return chunks + + # If no start_index provided, use most recent chunks + if start_index is None: + start_index = max(0, self.index - 50) + + # Get current index from Redis + current_index = int(self.redis_client.get(self.buffer_index_key) or 0) + + # Calculate range of chunks to retrieve (up to 50 at a time) + start_id = start_index + 1 + end_id = min(current_index + 1, start_id + 50) + + if start_id >= end_id: + return [] + + # Retrieve chunks in a pipeline + pipe = self.redis_client.pipeline() + for idx in range(start_id, end_id): + chunk_key = f"{self.buffer_prefix}{idx}" + pipe.get(chunk_key) + + results = pipe.execute() + chunks = [r for r in results if r] + + # Update local tracking + if chunks: + self.index = end_id - 1 + + return chunks + + except Exception as e: + logging.error(f"Error getting chunks from buffer: {e}") + return [] class ClientManager: """Manages active client connections""" @@ -130,7 +444,7 @@ class StreamFetcher: if not self._handle_connection(): continue - with self.manager.session.get(self.manager.current_url, stream=True) as response: + with self.manager.session.get(self.manager.url, stream=True) as response: if response.status_code == 200: self._handle_successful_connection() self._process_stream(response) @@ -142,15 +456,15 @@ class StreamFetcher: """Handle connection state and retries""" if not self.manager.connected: if not self.manager.should_retry(): - logging.error(f"Failed to connect after {Config.MAX_RETRIES} attempts") + logging.error(f"Failed to connect after {self.manager.max_retries} attempts") return False if not self.manager.running: return False self.manager.retry_count += 1 - logging.info(f"Connecting to stream: {self.manager.current_url} " - f"(attempt {self.manager.retry_count}/{Config.MAX_RETRIES})") + logging.info(f"Connecting to stream: {self.manager.url} " + f"(attempt {self.manager.retry_count}/{self.manager.max_retries})") return True def _handle_successful_connection(self) -> None: @@ -168,9 +482,9 @@ class StreamFetcher: return if chunk: - if self.manager.url_changed.is_set(): + if self.manager.ready_event.is_set(): logging.info("Stream switch in progress, closing connection") - self.manager.url_changed.clear() + self.manager.ready_event.clear() break with self.buffer.lock: @@ -201,57 +515,103 @@ def wait_for_running(manager: StreamManager, delay: float) -> bool: class ProxyServer: """Manages TS proxy server instance""" - def __init__(self, user_agent: Optional[str] = None): - self.stream_managers: Dict[str, StreamManager] = {} - self.stream_buffers: Dict[str, StreamBuffer] = {} - self.client_managers: Dict[str, ClientManager] = {} - self.fetch_threads: Dict[str, threading.Thread] = {} - self.user_agent: str = user_agent or Config.DEFAULT_USER_AGENT - - def initialize_channel(self, url: str, channel_id: str) -> None: - """Initialize a new channel stream""" - if channel_id in self.stream_managers: - self.stop_channel(channel_id) + def __init__(self): + self.stream_managers = {} # Maps channel_id to StreamManager + self.stream_buffers = {} # Maps channel_id to StreamBuffer + self.client_managers = {} # Maps channel_id to ClientManager + self.fetch_threads = {} # Add this missing attribute + + # Initialize Redis connection + try: + import redis + from django.conf import settings - self.stream_managers[channel_id] = StreamManager( - url, - channel_id, - user_agent=self.user_agent - ) - self.stream_buffers[channel_id] = StreamBuffer() - self.client_managers[channel_id] = ClientManager() - - # Start cleanup timer immediately after initialization - self.client_managers[channel_id].start_cleanup_timer(self, channel_id) - - fetcher = StreamFetcher( - self.stream_managers[channel_id], - self.stream_buffers[channel_id] - ) - - self.fetch_threads[channel_id] = threading.Thread( - target=fetcher.fetch_loop, - name=f"StreamFetcher-{channel_id}", - daemon=True - ) - self.fetch_threads[channel_id].start() - logging.info(f"Initialized channel {channel_id} with URL {url}") - - def stop_channel(self, channel_id: str) -> None: - """Stop and cleanup a channel""" - if channel_id in self.stream_managers: - logging.info(f"Stopping channel {channel_id}") - try: + redis_url = getattr(settings, 'REDIS_URL', 'redis://localhost:6379/0') + self.redis_client = redis.from_url(redis_url) + logging.info(f"Connected to Redis at {redis_url}") + except Exception as e: + self.redis_client = None + logging.error(f"Failed to connect to Redis: {e}") + + def initialize_channel(self, url, channel_id): + """Initialize a channel""" + try: + # Clean up any existing Redis entries for this channel + if self.redis_client: + # Delete index key + index_key = f"ts_proxy:buffer:{channel_id}:index" + self.redis_client.delete(index_key) + + # Delete all chunks for this channel + pattern = f"ts_proxy:buffer:{channel_id}:chunk:*" + cursor = 0 + while True: + cursor, keys = self.redis_client.scan(cursor, pattern, 100) + if keys: + self.redis_client.delete(*keys) + if cursor == 0: + break + + # Register this channel as active + self.redis_client.set(f"ts_proxy:active_channel:{channel_id}", "1", ex=60) + + # Create buffer and stream manager + buffer = StreamBuffer(channel_id=channel_id, redis_client=self.redis_client) + self.stream_buffers[channel_id] = buffer + + stream_manager = StreamManager(url, buffer) + self.stream_managers[channel_id] = stream_manager + + # FIX: Create client manager without passing channel_id + client_manager = ClientManager() # No arguments here + self.client_managers[channel_id] = client_manager + # Set up the cleanup timer afterwards + client_manager.start_cleanup_timer(self, channel_id) + + # Start stream manager + threading.Thread(target=stream_manager.run, daemon=True).start() + return True + + except Exception as e: + logging.error(f"Error initializing channel: {e}") + return False + + def stop_channel(self, channel_id): + """Stop channel and clean up resources""" + try: + # Stop stream manager + if channel_id in self.stream_managers: self.stream_managers[channel_id].stop() - if channel_id in self.fetch_threads: - self.fetch_threads[channel_id].join(timeout=5) - if self.fetch_threads[channel_id].is_alive(): - logging.warning(f"Fetch thread for channel {channel_id} did not stop cleanly") - except Exception as e: - logging.error(f"Error stopping channel {channel_id}: {e}") - finally: - self._cleanup_channel(channel_id) + del self.stream_managers[channel_id] + # Remove buffer + if channel_id in self.stream_buffers: + del self.stream_buffers[channel_id] + + # Cancel cleanup timer + if channel_id in self.client_managers: + if self.client_managers[channel_id].cleanup_timer: + self.client_managers[channel_id].cleanup_timer.cancel() + del self.client_managers[channel_id] + + # Clean up fetch thread if exists + if channel_id in self.fetch_threads: + if self.fetch_threads[channel_id].is_alive(): + # We can't forcibly terminate threads in Python, + # but we can make sure they're not referenced + pass + del self.fetch_threads[channel_id] + + # Remove Redis active channel marker + if self.redis_client: + self.redis_client.delete(f"ts_proxy:active_channel:{channel_id}") + + logging.info(f"Stopped channel {channel_id}") + return True + except Exception as e: + logging.error(f"Error stopping channel {channel_id}: {e}") + return False + def _cleanup_channel(self, channel_id: str) -> None: """Remove channel resources""" for collection in [self.stream_managers, self.stream_buffers, diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index 16148f05..36e84790 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -2,6 +2,7 @@ import json import threading import logging import time +import random from django.http import StreamingHttpResponse, JsonResponse from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_http_methods @@ -54,77 +55,53 @@ def initialize_stream(request, channel_id): @csrf_exempt @require_http_methods(["GET"]) def stream_ts(request, channel_id): - """Handle TS stream requests with improved multi-client support""" + """Stream TS data to client""" if channel_id not in proxy_server.stream_managers: return JsonResponse({'error': 'Channel not found'}, status=404) - + def generate(): - # Use a truly unique client ID (timestamp + random component) - import random client_id = int(time.time() * 1000) + random.randint(1, 1000) try: - buffer = proxy_server.stream_buffers[channel_id] + # Add client to manager client_manager = proxy_server.client_managers[channel_id] - - # Record this client client_manager.add_client(client_id) - # Each client starts at current buffer position - with buffer.lock: - last_index = buffer.index + # Get buffer + buffer = proxy_server.stream_buffers.get(channel_id) + if not buffer: + logger.error(f"No buffer found for channel {channel_id}") + return - logger.info(f"New client {client_id} connected to channel {channel_id} (starting at index {last_index})") + # Start at current position + local_index = None - # Yield some initial headers or empty data to establish connection - yield b'' - - # Stream indefinitely + # Main streaming loop while True: - new_chunks = False + chunks = buffer.get_chunks(local_index) - # Minimize lock time - only lock briefly to check and grab data - with buffer.lock: - if buffer.index > last_index: - chunks_behind = buffer.index - last_index - # Calculate start position in circular buffer - start_pos = max(0, len(buffer.buffer) - chunks_behind) - - # Get chunks to send (make a copy to avoid long lock) - chunks_to_send = [buffer.buffer[i] for i in range(start_pos, len(buffer.buffer))] - last_index = buffer.index - new_chunks = True - - if new_chunks: - # Send all collected chunks outside the lock - for chunk in chunks_to_send: + if chunks: + # Send chunks to client + for chunk in chunks: yield chunk + + # Update local index + local_index = buffer.index else: - # Shorter sleep to be more responsive - time.sleep(0.05) - + # No data available, wait briefly + time.sleep(0.1) + except Exception as e: logger.error(f"Streaming error for client {client_id}, channel {channel_id}: {e}") finally: - try: - if channel_id in proxy_server.client_managers: - remaining = proxy_server.client_managers[channel_id].remove_client(client_id) - logger.info(f"Client {client_id} disconnected from channel {channel_id} ({remaining} clients remaining)") - - # Keep channel active with at least one client - if remaining == 0: - logger.info(f"No clients remaining, stopping channel {channel_id}") - proxy_server.stop_channel(channel_id) - except Exception as e: - logger.error(f"Error during client cleanup: {e}") + # Clean up client + if channel_id in proxy_server.client_managers: + remaining = proxy_server.client_managers[channel_id].remove_client(client_id) + logger.info(f"Client {client_id} disconnected from channel {channel_id} ({remaining} clients remaining)") - # Create response with appropriate streaming settings - response = StreamingHttpResponse( - generate(), - content_type='video/MP2T' - ) + # Create streaming response + response = StreamingHttpResponse(generate(), content_type='video/MP2T') response['Cache-Control'] = 'no-cache, no-store' - response['X-Accel-Buffering'] = 'no' # Disable nginx buffering return response @csrf_exempt diff --git a/dispatcharr/settings.py b/dispatcharr/settings.py index 5de5a496..cc844b9d 100644 --- a/dispatcharr/settings.py +++ b/dispatcharr/settings.py @@ -172,6 +172,9 @@ SIMPLE_JWT = { 'BLACKLIST_AFTER_ROTATION': True, # Optional: Whether to blacklist refresh tokens } +# Redis settings for TS proxy +REDIS_URL = 'redis://localhost:6379/0' + # Proxy Settings PROXY_SETTINGS = { 'HLS': { @@ -191,5 +194,6 @@ PROXY_SETTINGS = { 'BUFFER_SIZE': 1000, 'RECONNECT_DELAY': 5, 'USER_AGENT': 'VLC/3.0.20 LibVLC/3.0.20', + 'REDIS_CHUNK_TTL': 60, # How long to keep chunks in Redis (seconds) } }