mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-21 01:05:30 +00:00
Switched back to http connections instead of socket.
This commit is contained in:
SergeantPanda
parent
3483f01637
commit
d00aa3444e
3 changed files with 345 additions and 187 deletions
|
|
@ -48,4 +48,9 @@ class TSConfig(BaseConfig):
|
|||
CHANNEL_INIT_GRACE_PERIOD = 5 # How long to wait for first client after initialization (seconds)
|
||||
CLIENT_HEARTBEAT_INTERVAL = 1 # How often to send client heartbeats (seconds)
|
||||
GHOST_CLIENT_MULTIPLIER = 5.0 # How many heartbeat intervals before client considered ghost (5 would mean 5 secondsif heartbeat interval is 1)
|
||||
|
||||
# TS packets are 188 bytes
|
||||
# Make chunk size a multiple of TS packet size for perfect alignment
|
||||
# ~1MB is ideal for streaming (matches typical media buffer sizes)
|
||||
BUFFER_CHUNK_SIZE = 188 * 5644 # ~1MB (exactly 1,061,072 bytes)
|
||||
|
||||
|
|
|
|||
|
|
@ -30,18 +30,18 @@ logging.basicConfig(
|
|||
print("TS PROXY SERVER MODULE LOADED", file=sys.stderr)
|
||||
|
||||
class StreamManager:
|
||||
"""Manages a connection to a TS stream with continuity tracking"""
|
||||
"""Manages a connection to a TS stream without using raw sockets"""
|
||||
|
||||
def __init__(self, url, buffer, user_agent=None):
|
||||
# Existing initialization code
|
||||
# Basic properties
|
||||
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 = Config.MAX_RETRIES
|
||||
self.current_response = None
|
||||
self.current_session = None
|
||||
|
||||
# User agent for connection
|
||||
self.user_agent = user_agent or Config.DEFAULT_USER_AGENT
|
||||
|
|
@ -50,32 +50,187 @@ class StreamManager:
|
|||
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)
|
||||
|
||||
# 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"""
|
||||
|
||||
def _create_session(self):
|
||||
"""Create and configure requests session with optimal settings"""
|
||||
session = requests.Session()
|
||||
|
||||
# Configure session headers
|
||||
session.headers.update({
|
||||
'User-Agent': self.user_agent,
|
||||
'Connection': 'keep-alive'
|
||||
})
|
||||
|
||||
# Set up connection pooling for better performance
|
||||
adapter = requests.adapters.HTTPAdapter(
|
||||
pool_connections=1, # Single connection for this stream
|
||||
pool_maxsize=1, # Max size of connection pool
|
||||
max_retries=3, # Auto-retry for failed requests
|
||||
pool_block=False # Don't block when pool is full
|
||||
)
|
||||
|
||||
# Apply adapter to both HTTP and HTTPS
|
||||
session.mount('http://', adapter)
|
||||
session.mount('https://', adapter)
|
||||
|
||||
return session
|
||||
|
||||
def run(self):
|
||||
"""Main execution loop using HTTP streaming with improved connection handling"""
|
||||
try:
|
||||
# Start health monitor thread
|
||||
health_thread = threading.Thread(target=self._monitor_health, daemon=True)
|
||||
health_thread.start()
|
||||
|
||||
logging.info(f"Starting stream for URL: {self.url}")
|
||||
|
||||
while self.running:
|
||||
try:
|
||||
# Create new session for each connection attempt
|
||||
session = self._create_session()
|
||||
self.current_session = session
|
||||
|
||||
# Stream the URL with proper timeout handling
|
||||
response = session.get(
|
||||
self.url,
|
||||
stream=True,
|
||||
timeout=(10, 60) # 10s connect timeout, 60s read timeout
|
||||
)
|
||||
self.current_response = response
|
||||
|
||||
if response.status_code == 200:
|
||||
self.connected = True
|
||||
self.healthy = True
|
||||
logging.info("Successfully connected to stream source")
|
||||
|
||||
# Set channel state to waiting for clients
|
||||
self._set_waiting_for_clients()
|
||||
|
||||
# Process the stream in chunks with improved error handling
|
||||
try:
|
||||
chunk_count = 0
|
||||
for chunk in response.iter_content(chunk_size=self.chunk_size):
|
||||
if not self.running:
|
||||
break
|
||||
|
||||
if chunk:
|
||||
# Add chunk to buffer with TS packet alignment
|
||||
success = self.buffer.add_chunk(chunk)
|
||||
|
||||
if success:
|
||||
self.last_data_time = time.time()
|
||||
chunk_count += 1
|
||||
|
||||
# Update last data timestamp in Redis
|
||||
if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
|
||||
last_data_key = f"ts_proxy:channel:{self.buffer.channel_id}:last_data"
|
||||
self.buffer.redis_client.set(last_data_key, str(time.time()), ex=60)
|
||||
except AttributeError as e:
|
||||
# Handle the specific 'NoneType' object has no attribute 'read' error
|
||||
if "'NoneType' object has no attribute 'read'" in str(e):
|
||||
logging.warning(f"Connection closed by server (read {chunk_count} chunks before disconnect)")
|
||||
else:
|
||||
# Re-raise unexpected AttributeError
|
||||
raise
|
||||
else:
|
||||
logging.error(f"Failed to connect to stream: HTTP {response.status_code}")
|
||||
time.sleep(2)
|
||||
|
||||
except requests.exceptions.ReadTimeout:
|
||||
logging.warning("Read timeout - server stopped sending data")
|
||||
self.connected = False
|
||||
time.sleep(1)
|
||||
|
||||
except requests.RequestException as e:
|
||||
logging.error(f"HTTP request error: {e}")
|
||||
self.connected = False
|
||||
time.sleep(5)
|
||||
|
||||
finally:
|
||||
# Clean up response and session
|
||||
if self.current_response:
|
||||
try:
|
||||
self.current_response.close()
|
||||
except Exception as e:
|
||||
logging.debug(f"Error closing response: {e}")
|
||||
self.current_response = None
|
||||
|
||||
if self.current_session:
|
||||
try:
|
||||
self.current_session.close()
|
||||
except Exception as e:
|
||||
logging.debug(f"Error closing session: {e}")
|
||||
self.current_session = None
|
||||
|
||||
# Connection retry logic
|
||||
if self.running and not self.connected:
|
||||
self.retry_count += 1
|
||||
if self.retry_count > self.max_retries:
|
||||
logging.error(f"Maximum retry attempts ({self.max_retries}) exceeded")
|
||||
break
|
||||
|
||||
timeout = min(2 ** self.retry_count, 30)
|
||||
logging.info(f"Reconnecting in {timeout} seconds... (attempt {self.retry_count})")
|
||||
time.sleep(timeout)
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"Stream error: {e}", exc_info=True)
|
||||
finally:
|
||||
self.connected = False
|
||||
|
||||
if self.current_response:
|
||||
try:
|
||||
self.current_response.close()
|
||||
except:
|
||||
pass
|
||||
|
||||
if self.current_session:
|
||||
try:
|
||||
self.current_session.close()
|
||||
except:
|
||||
pass
|
||||
|
||||
logging.info("Stream manager stopped")
|
||||
|
||||
def stop(self):
|
||||
"""Stop the stream manager and clean up resources"""
|
||||
self.running = False
|
||||
|
||||
if self.current_response:
|
||||
try:
|
||||
self.current_response.close()
|
||||
except:
|
||||
pass
|
||||
|
||||
if self.current_session:
|
||||
try:
|
||||
self.current_session.close()
|
||||
except:
|
||||
pass
|
||||
|
||||
logging.info("Stream manager resources released")
|
||||
|
||||
def update_url(self, new_url: str) -> bool:
|
||||
"""Update stream URL and reconnect"""
|
||||
def update_url(self, new_url):
|
||||
"""Update stream URL and reconnect with HTTP streaming approach"""
|
||||
if new_url == self.url:
|
||||
logging.info(f"URL unchanged: {new_url}")
|
||||
return False
|
||||
|
||||
logging.info(f"Switching stream URL from {self.url} to {new_url}")
|
||||
|
||||
# Close existing HTTP connection resources instead of socket
|
||||
self._close_connection() # Use our new method instead of _close_socket
|
||||
|
||||
# Update URL and reset connection state
|
||||
old_url = self.url
|
||||
self.url = new_url
|
||||
self.connected = False
|
||||
self._close_socket() # Close existing connection
|
||||
|
||||
# Signal health monitor to reconnect immediately
|
||||
self.last_data_time = 0
|
||||
# Reset retry counter to allow immediate reconnect
|
||||
self.retry_count = 0
|
||||
|
||||
return True
|
||||
|
||||
|
|
@ -83,119 +238,6 @@ class StreamManager:
|
|||
"""Check if connection retry is allowed"""
|
||||
return self.retry_count < self.max_retries
|
||||
|
||||
def stop(self) -> None:
|
||||
"""Stop the stream manager and close all resources"""
|
||||
self.running = False
|
||||
self._close_socket()
|
||||
logging.info("Stream manager resources released")
|
||||
|
||||
def run(self):
|
||||
"""Main execution loop with stream health monitoring"""
|
||||
try:
|
||||
# Check if buffer already has data - in which case we might not need to connect
|
||||
if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client:
|
||||
buffer_index = self.buffer.redis_client.get(f"ts_proxy:buffer:{self.buffer.channel_id}:index")
|
||||
if buffer_index and int(buffer_index) > 0:
|
||||
# There's already data in Redis, check if it's recent (within last 10 seconds)
|
||||
last_data_key = f"ts_proxy:channel:{self.buffer.channel_id}:last_data"
|
||||
last_data = self.buffer.redis_client.get(last_data_key)
|
||||
if last_data:
|
||||
last_time = float(last_data)
|
||||
if time.time() - last_time < 10:
|
||||
logging.info(f"Recent data found in Redis, no need to reconnect")
|
||||
self.connected = True
|
||||
self.healthy = True
|
||||
return
|
||||
|
||||
# Start health monitor thread
|
||||
health_thread = threading.Thread(target=self._monitor_health, daemon=True)
|
||||
health_thread.start()
|
||||
|
||||
current_response = None # Track the current response object
|
||||
current_session = None # Track the current session
|
||||
|
||||
# Establish network connection
|
||||
import socket
|
||||
import requests
|
||||
|
||||
logging.info(f"Starting stream for URL: {self.url}")
|
||||
|
||||
while self.running:
|
||||
try:
|
||||
# Parse URL
|
||||
if self.url.startswith("http"):
|
||||
# HTTP connection
|
||||
session = self._create_session()
|
||||
current_session = session
|
||||
|
||||
try:
|
||||
# Create an initial connection to get socket
|
||||
response = session.get(self.url, stream=True)
|
||||
current_response = response
|
||||
|
||||
if response.status_code == 200:
|
||||
self.connected = True
|
||||
self.socket = response.raw._fp.fp.raw
|
||||
self.healthy = True
|
||||
logging.info("Successfully connected to stream source")
|
||||
|
||||
# Connection successful - START GRACE PERIOD HERE
|
||||
self._set_waiting_for_clients()
|
||||
|
||||
# Main fetch loop
|
||||
while self.running and self.connected:
|
||||
if self.fetch_chunk():
|
||||
self.last_data_time = time.time()
|
||||
else:
|
||||
if not self.running:
|
||||
break
|
||||
time.sleep(0.1)
|
||||
else:
|
||||
logging.error(f"Failed to connect to stream: HTTP {response.status_code}")
|
||||
time.sleep(2)
|
||||
finally:
|
||||
# Properly close response before session
|
||||
if current_response:
|
||||
try:
|
||||
# Close the response explicitly to avoid the urllib3 error
|
||||
current_response.close()
|
||||
except Exception as e:
|
||||
logging.debug(f"Error closing response: {e}")
|
||||
current_response = None
|
||||
|
||||
if current_session:
|
||||
try:
|
||||
current_session.close()
|
||||
except Exception as e:
|
||||
logging.debug(f"Error closing session: {e}")
|
||||
current_session = None
|
||||
else:
|
||||
logging.error(f"Unsupported URL scheme: {self.url}")
|
||||
|
||||
# Connection retry logic
|
||||
if self.running and not self.connected:
|
||||
self.retry_count += 1
|
||||
if self.retry_count > self.max_retries:
|
||||
logging.error(f"Maximum retry attempts ({self.max_retries}) exceeded")
|
||||
break
|
||||
|
||||
timeout = min(2 ** self.retry_count, 30)
|
||||
logging.info(f"Reconnecting in {timeout} seconds... (attempt {self.retry_count})")
|
||||
time.sleep(timeout)
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"Connection error: {e}")
|
||||
self._close_socket()
|
||||
time.sleep(5)
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"Stream error: {e}")
|
||||
self._close_socket()
|
||||
finally:
|
||||
# Final cleanup
|
||||
self._close_socket()
|
||||
logging.info("Stream manager stopped")
|
||||
|
||||
def _monitor_health(self):
|
||||
"""Monitor stream health and attempt recovery if needed"""
|
||||
while self.running:
|
||||
|
|
@ -223,16 +265,28 @@ class StreamManager:
|
|||
|
||||
time.sleep(self.health_check_interval)
|
||||
|
||||
def _close_socket(self):
|
||||
"""Close the socket connection safely"""
|
||||
if self.socket:
|
||||
def _close_connection(self):
|
||||
"""Close HTTP connection resources"""
|
||||
# Close response if it exists
|
||||
if hasattr(self, 'current_response') and self.current_response:
|
||||
try:
|
||||
self.socket.close()
|
||||
self.current_response.close()
|
||||
except Exception as e:
|
||||
logging.debug(f"Error closing socket: {e}")
|
||||
pass
|
||||
self.socket = None
|
||||
self.connected = False
|
||||
logging.debug(f"Error closing response: {e}")
|
||||
self.current_response = None
|
||||
|
||||
# Close session if it exists
|
||||
if hasattr(self, 'current_session') and self.current_session:
|
||||
try:
|
||||
self.current_session.close()
|
||||
except Exception as e:
|
||||
logging.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"""
|
||||
return self._close_connection()
|
||||
|
||||
def fetch_chunk(self):
|
||||
"""Fetch data from socket with direct pass-through to buffer"""
|
||||
|
|
@ -355,6 +409,7 @@ class StreamManager:
|
|||
class StreamBuffer:
|
||||
"""Manages stream data buffering using Redis for persistence"""
|
||||
|
||||
# Add a memory buffer to collect data before writing to Redis
|
||||
def __init__(self, channel_id=None, redis_client=None):
|
||||
self.channel_id = channel_id
|
||||
self.redis_client = redis_client
|
||||
|
|
@ -377,39 +432,70 @@ class StreamBuffer:
|
|||
logging.info(f"Initialized buffer from Redis with index {self.index}")
|
||||
except Exception as e:
|
||||
logging.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
|
||||
|
||||
def add_chunk(self, chunk):
|
||||
"""Add a chunk to the buffer with minimal processing"""
|
||||
"""Add data with optimized Redis storage"""
|
||||
if not chunk:
|
||||
return False
|
||||
|
||||
try:
|
||||
# Accumulate partial packets between chunks
|
||||
if not hasattr(self, '_partial_packet'):
|
||||
self._partial_packet = bytearray()
|
||||
|
||||
# Combine with any previous partial packet
|
||||
combined_data = bytearray(self._partial_packet) + bytearray(chunk)
|
||||
|
||||
# Calculate complete packets
|
||||
complete_packets_size = (len(combined_data) // 188) * 188
|
||||
|
||||
if complete_packets_size == 0:
|
||||
# Not enough data for a complete packet
|
||||
self._partial_packet = combined_data
|
||||
return True
|
||||
|
||||
# Split into complete packets and remainder
|
||||
complete_packets = combined_data[:complete_packets_size]
|
||||
self._partial_packet = combined_data[complete_packets_size:]
|
||||
|
||||
# Add completed packets to write buffer
|
||||
self._write_buffer.extend(complete_packets)
|
||||
|
||||
# Only write to Redis when we have enough data for an optimized chunk
|
||||
writes_done = 0
|
||||
with self.lock:
|
||||
# Increment index atomically
|
||||
if self.redis_client:
|
||||
# Use Redis to store and track chunks
|
||||
try:
|
||||
chunk_index = self.redis_client.incr(self.buffer_index_key)
|
||||
chunk_key = f"{self.buffer_prefix}{chunk_index}"
|
||||
setex_result = self.redis_client.setex(chunk_key, self.chunk_ttl, chunk)
|
||||
|
||||
if not setex_result:
|
||||
logging.error(f"Redis SETEX failed for chunk {chunk_index}")
|
||||
return False
|
||||
|
||||
# Update local tracking
|
||||
self.index = chunk_index
|
||||
|
||||
logging.debug(f"Added chunk {chunk_index} ({len(chunk)} bytes) to buffer")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"Redis operation failed in add_chunk: {e}")
|
||||
return False
|
||||
else:
|
||||
logging.error("Redis not available, cannot store chunks")
|
||||
return False
|
||||
while len(self._write_buffer) >= self.target_chunk_size:
|
||||
# Extract a full chunk
|
||||
chunk_data = self._write_buffer[:self.target_chunk_size]
|
||||
self._write_buffer = self._write_buffer[self.target_chunk_size:]
|
||||
|
||||
# Write optimized chunk to Redis
|
||||
if self.redis_client:
|
||||
try:
|
||||
chunk_index = self.redis_client.incr(self.buffer_index_key)
|
||||
chunk_key = f"{self.buffer_prefix}{chunk_index}"
|
||||
setex_result = self.redis_client.setex(chunk_key, self.chunk_ttl, bytes(chunk_data))
|
||||
|
||||
if not setex_result:
|
||||
logging.error(f"Redis SETEX failed for chunk {chunk_index}")
|
||||
continue
|
||||
|
||||
# Update local tracking
|
||||
self.index = chunk_index
|
||||
writes_done += 1
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"Redis operation failed in add_chunk: {e}")
|
||||
return False
|
||||
|
||||
if writes_done > 0:
|
||||
logging.debug(f"Added {writes_done} optimized chunks ({self.target_chunk_size} bytes each) to Redis")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"Error adding chunk to buffer: {e}")
|
||||
return False
|
||||
|
|
@ -534,6 +620,37 @@ class StreamBuffer:
|
|||
logging.error(f"Error getting exact chunks: {e}", exc_info=True)
|
||||
return []
|
||||
|
||||
def stop(self):
|
||||
"""Stop the buffer and flush any remaining data"""
|
||||
try:
|
||||
# Flush any remaining data in the write buffer
|
||||
if hasattr(self, '_write_buffer') and len(self._write_buffer) > 0:
|
||||
# Ensure remaining data is aligned to TS packets
|
||||
complete_size = (len(self._write_buffer) // 188) * 188
|
||||
|
||||
if complete_size > 0:
|
||||
final_chunk = self._write_buffer[:complete_size]
|
||||
|
||||
# Write final chunk to Redis
|
||||
with self.lock:
|
||||
if self.redis_client:
|
||||
try:
|
||||
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, bytes(final_chunk))
|
||||
self.index = chunk_index
|
||||
logging.info(f"Flushed final chunk of {len(final_chunk)} bytes to Redis")
|
||||
except Exception as e:
|
||||
logging.error(f"Error flushing final chunk: {e}")
|
||||
|
||||
# Clear buffers
|
||||
self._write_buffer = bytearray()
|
||||
if hasattr(self, '_partial_packet'):
|
||||
self._partial_packet = bytearray()
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"Error during buffer stop: {e}")
|
||||
|
||||
class ClientManager:
|
||||
"""Manages connected clients for a channel with cross-worker visibility"""
|
||||
|
||||
|
|
|
|||
|
|
@ -276,42 +276,36 @@ def stream_ts(request, channel_id):
|
|||
|
||||
# Main streaming loop
|
||||
while True:
|
||||
# Get chunks at client's position
|
||||
chunks = buffer.get_chunks_exact(local_index, Config.CHUNK_BATCH_SIZE)
|
||||
# Get chunks at client's position using improved strategy
|
||||
chunks, next_index = get_client_data(buffer, local_index)
|
||||
|
||||
if chunks:
|
||||
empty_reads = 0
|
||||
consecutive_empty = 0
|
||||
|
||||
# Track and send chunks
|
||||
chunk_sizes = [len(c) for c in chunks]
|
||||
total_size = sum(chunk_sizes)
|
||||
start_idx = local_index + 1
|
||||
end_idx = local_index + len(chunks)
|
||||
# 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}")
|
||||
|
||||
logger.debug(f"[{client_id}] Retrieved {len(chunks)} chunks ({total_size} bytes) from index {start_idx} to {end_idx}")
|
||||
|
||||
# SIMPLIFIED: Just send chunks directly without pacing
|
||||
# CRITICAL FIX: Actually send the chunks to the client
|
||||
for chunk in chunks:
|
||||
bytes_sent += len(chunk)
|
||||
chunks_sent += 1
|
||||
|
||||
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.info(f"[{client_id}] Client disconnected while yielding data: {e}")
|
||||
if channel_id in proxy_server.client_managers:
|
||||
proxy_server.client_managers[channel_id].remove_client(client_id)
|
||||
return # Exit the generator
|
||||
logger.error(f"[{client_id}] Error sending chunk to client: {e}")
|
||||
raise # Re-raise to exit the generator
|
||||
|
||||
# Log progress periodically
|
||||
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")
|
||||
|
||||
# Update local index and last yield time
|
||||
local_index = end_idx
|
||||
# Update index after successfully sending all chunks
|
||||
local_index = next_index
|
||||
last_yield_time = time.time()
|
||||
else:
|
||||
# No chunks available
|
||||
|
|
@ -562,4 +556,46 @@ def change_stream(request, channel_id):
|
|||
return JsonResponse({'error': 'Invalid JSON'}, status=400)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to change stream: {e}", exc_info=True)
|
||||
return JsonResponse({'error': str(e)}, status=500)
|
||||
return JsonResponse({'error': str(e)}, status=500)
|
||||
|
||||
def get_client_data(buffer, local_index):
|
||||
"""Get optimal amount of data for client"""
|
||||
# Define limits
|
||||
MIN_CHUNKS = 3 # Minimum chunks to read for efficiency
|
||||
MAX_CHUNKS = 20 # Safety limit to prevent memory spikes
|
||||
TARGET_SIZE = 1024 * 1024 # Target ~1MB per response (typical media buffer)
|
||||
MAX_SIZE = 2 * 1024 * 1024 # Hard cap at 2MB
|
||||
|
||||
# Calculate how far behind we are
|
||||
chunks_behind = buffer.index - local_index
|
||||
|
||||
# Determine optimal chunk count
|
||||
if chunks_behind <= MIN_CHUNKS:
|
||||
# Not much data, retrieve what's available
|
||||
chunk_count = max(1, chunks_behind)
|
||||
elif chunks_behind <= MAX_CHUNKS:
|
||||
# Reasonable amount behind, catch up completely
|
||||
chunk_count = chunks_behind
|
||||
else:
|
||||
# Way behind, retrieve MAX_CHUNKS to avoid memory pressure
|
||||
chunk_count = MAX_CHUNKS
|
||||
|
||||
# Retrieve chunks
|
||||
chunks = buffer.get_chunks_exact(local_index, chunk_count)
|
||||
|
||||
# Check total size
|
||||
total_size = sum(len(c) for c in chunks)
|
||||
|
||||
# If we're under target and have more chunks available, get more
|
||||
if total_size < TARGET_SIZE and chunks_behind > chunk_count:
|
||||
# Calculate how many more chunks we can get
|
||||
additional = min(MAX_CHUNKS - chunk_count, chunks_behind - chunk_count)
|
||||
more_chunks = buffer.get_chunks_exact(local_index + chunk_count, additional)
|
||||
|
||||
# Check if adding more would exceed MAX_SIZE
|
||||
additional_size = sum(len(c) for c in more_chunks)
|
||||
if total_size + additional_size <= MAX_SIZE:
|
||||
chunks.extend(more_chunks)
|
||||
chunk_count += additional
|
||||
|
||||
return chunks, local_index + chunk_count
|
||||
Loading…
Add table
Add a link
Reference in a new issue