mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-28 20:41:19 +00:00
Moved stream generator out of stream_ts and into stream_generator.py
This commit is contained in:
SergeantPanda
parent
245215eb6f
commit
8f1c233d64
2 changed files with 336 additions and 280 deletions
329
apps/proxy/ts_proxy/stream_generator.py
Normal file
329
apps/proxy/ts_proxy/stream_generator.py
Normal file
|
|
@ -0,0 +1,329 @@
|
|||
"""
|
||||
Stream generation and client-side handling for TS streams.
|
||||
This module handles generating and delivering video streams to clients.
|
||||
"""
|
||||
|
||||
import time
|
||||
import logging
|
||||
from apps.proxy.config import TSConfig as Config
|
||||
from . import proxy_server
|
||||
|
||||
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
|
||||
|
||||
# Keep track of initialization state
|
||||
initialization_start = time.time()
|
||||
max_init_wait = getattr(Config, 'CLIENT_WAIT_TIMEOUT', 30)
|
||||
channel_ready = not self.channel_initializing
|
||||
keepalive_interval = 0.5
|
||||
last_keepalive = 0
|
||||
|
||||
try:
|
||||
logger.info(f"[{self.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:{self.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"[{self.client_id}] Channel {self.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"[{self.client_id}] Channel {self.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"[{self.client_id}] Sending keepalive packet during initialization, state={state}")
|
||||
yield bytes(keepalive_packet)
|
||||
self.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"[{self.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 - 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()
|
||||
|
||||
# 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
|
||||
|
||||
# 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"[{self.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
|
||||
self.local_index = local_index
|
||||
|
||||
# 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"[{self.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(self.channel_id) if hasattr(proxy_server, 'am_i_owner') else True
|
||||
|
||||
# Main streaming loop
|
||||
while True:
|
||||
# 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")
|
||||
break
|
||||
|
||||
if self.channel_id not in proxy_server.client_managers:
|
||||
logger.info(f"[{self.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:{self.channel_id}:stopping"
|
||||
if proxy_server.redis_client.exists(stop_key):
|
||||
logger.info(f"[{self.client_id}] Detected channel stop signal, terminating stream")
|
||||
break
|
||||
|
||||
# Client stop check
|
||||
client_stop_key = f"ts_proxy:channel:{self.channel_id}:client:{self.client_id}:stop"
|
||||
if proxy_server.redis_client.exists(client_stop_key):
|
||||
logger.info(f"[{self.client_id}] Detected client stop signal, terminating stream")
|
||||
break
|
||||
|
||||
# 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")
|
||||
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"[{self.client_id}] Retrieved {len(chunks)} chunks ({total_size} bytes) from index {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
|
||||
|
||||
# Update index after successfully sending all chunks
|
||||
local_index = next_index
|
||||
self.local_index = local_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"[{self.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"[{self.client_id}] Waiting for chunks beyond {local_index} (buffer at {buffer.index}, stream: {stream_status})")
|
||||
|
||||
# Check for client disconnect during wait periods
|
||||
if consecutive_empty > 100 and buffer.index > local_index + 50:
|
||||
logger.warning(f"[{self.client_id}] Possible ghost client: buffer has advanced {buffer.index - local_index} chunks ahead but client stuck at {local_index}")
|
||||
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"[{self.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"[{self.client_id}] Non-owner worker with no data for {Config.STREAM_TIMEOUT}s, disconnecting")
|
||||
break
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[{self.client_id}] Stream error: {e}", exc_info=True)
|
||||
finally:
|
||||
# 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, {bytes_sent/1024:.1f}KB in {chunks_sent} chunks (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")
|
||||
|
||||
import threading
|
||||
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
|
||||
|
|
@ -5,11 +5,11 @@ 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
|
||||
import logging
|
||||
from apps.channels.models import Channel, Stream
|
||||
from apps.m3u.models import M3UAccount, M3UAccountProfile
|
||||
|
|
@ -168,291 +168,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)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue