mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-21 09:09:22 +00:00
Stream switching is working fairly consistently need to investigate some issues with it.
This commit is contained in:
SergeantPanda
parent
f164f22a4a
commit
f6aab5bc3c
2 changed files with 192 additions and 78 deletions
|
|
@ -96,27 +96,40 @@ class StreamManager:
|
|||
logging.info("Stream manager resources released")
|
||||
|
||||
def _process_complete_packets(self):
|
||||
"""Process TS packets with detailed logging"""
|
||||
"""Process TS packets with improved resync capability"""
|
||||
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
|
||||
# Enhanced sync byte detection with re-sync capability
|
||||
if (not self.sync_found or
|
||||
(len(self.recv_buffer) >= 188 and self.recv_buffer[0] != 0x47)):
|
||||
|
||||
# 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
|
||||
# Need to find sync pattern if we haven't found it yet or lost sync
|
||||
if len(self.recv_buffer) >= 376: # Need at least 2 packet lengths
|
||||
sync_found = False
|
||||
|
||||
# Look for at least two sync bytes (0x47) at 188-byte intervals
|
||||
for i in range(min(188, len(self.recv_buffer) - 188)):
|
||||
if (self.recv_buffer[i] == 0x47 and
|
||||
self.recv_buffer[i + 188] == 0x47):
|
||||
|
||||
# If already had sync but lost it, log the issue
|
||||
if self.sync_found:
|
||||
logging.warning(f"Re-syncing TS stream at position {i} (lost sync)")
|
||||
else:
|
||||
logging.debug(f"TS sync found at position {i}")
|
||||
|
||||
# Trim buffer to start at first sync byte
|
||||
self.recv_buffer = self.recv_buffer[i:]
|
||||
self.sync_found = True
|
||||
sync_found = True
|
||||
break
|
||||
|
||||
# If we couldn't find sync in this buffer, discard partial data
|
||||
if not sync_found:
|
||||
logging.warning(f"Failed to find sync pattern - discarding {len(self.recv_buffer) - 188} bytes")
|
||||
if len(self.recv_buffer) > 188:
|
||||
self.recv_buffer = self.recv_buffer[-188:] # Keep last chunk for next attempt
|
||||
return False
|
||||
|
||||
# If we don't have a complete packet yet, wait for more data
|
||||
if len(self.recv_buffer) < 188:
|
||||
return False
|
||||
|
|
@ -127,8 +140,17 @@ class StreamManager:
|
|||
if packet_count == 0:
|
||||
return False
|
||||
|
||||
# Log packet processing
|
||||
logging.debug(f"Processing {packet_count} TS packets ({packet_count * 188} bytes)")
|
||||
# Verify all packets have sync bytes
|
||||
all_synced = True
|
||||
for i in range(0, packet_count):
|
||||
if self.recv_buffer[i * 188] != 0x47:
|
||||
all_synced = False
|
||||
break
|
||||
|
||||
# If not all packets are synced, re-scan for sync
|
||||
if not all_synced:
|
||||
self.sync_found = False # Force re-sync on next call
|
||||
return False
|
||||
|
||||
# Extract complete packets
|
||||
packets = self.recv_buffer[:packet_count * 188]
|
||||
|
|
@ -144,6 +166,8 @@ class StreamManager:
|
|||
|
||||
if first_sync != 0x47 or last_sync != 0x47:
|
||||
logging.warning(f"TS packet alignment issue: first_sync=0x{first_sync:02x}, last_sync=0x{last_sync:02x}")
|
||||
# Don't process misaligned packets
|
||||
return False
|
||||
|
||||
before_index = self.buffer.index
|
||||
success = self.buffer.add_chunk(bytes(packets))
|
||||
|
|
@ -1123,10 +1147,29 @@ class ProxyServer:
|
|||
if self.redis_client:
|
||||
# Store stream metadata - can be done regardless of ownership
|
||||
metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
|
||||
if url: # Only update URL if one is provided
|
||||
self.redis_client.hset(metadata_key, "url", url)
|
||||
|
||||
# FIXED: Use hash operations consistently - don't mix string and hash operations
|
||||
# Create a dictionary of values to set in the hash
|
||||
metadata = {}
|
||||
if url: # Only include URL if provided
|
||||
metadata["url"] = url
|
||||
if user_agent:
|
||||
self.redis_client.hset(metadata_key, "user_agent", user_agent)
|
||||
metadata["user_agent"] = user_agent
|
||||
|
||||
# Initialize state
|
||||
metadata["state"] = "initializing"
|
||||
metadata["init_time"] = str(time.time())
|
||||
|
||||
# Set the hash fields all at once
|
||||
if metadata:
|
||||
self.redis_client.hset(metadata_key, mapping=metadata)
|
||||
|
||||
# Set expiration on the hash
|
||||
self.redis_client.expire(metadata_key, 3600) # 1 hour TTL
|
||||
|
||||
# Set activity key as a separate key
|
||||
activity_key = f"ts_proxy:active_channel:{channel_id}"
|
||||
self.redis_client.setex(activity_key, 300, "1") # 5 min TTL
|
||||
|
||||
# If no url was passed, try to get from Redis
|
||||
if not url:
|
||||
|
|
@ -1222,6 +1265,21 @@ class ProxyServer:
|
|||
|
||||
logging.info(f"Channel {channel_id} in connecting state - will start grace period after connection")
|
||||
|
||||
# SIMPLE CHANNEL REGISTRY - register this channel as active
|
||||
if self.redis_client:
|
||||
# Use a simple key for active channel registration
|
||||
registry_key = f"ts_proxy:active_channels:{channel_id}"
|
||||
# Store basic info and set a longer TTL (5 minutes)
|
||||
channel_info = {
|
||||
"url": url if url else "",
|
||||
"init_time": str(time.time()),
|
||||
"owner": self.worker_id
|
||||
}
|
||||
self.redis_client.hset(registry_key, mapping=channel_info)
|
||||
self.redis_client.expire(registry_key, 300) # 5 minute TTL
|
||||
|
||||
logging.info(f"Registered channel {channel_id} in active channels registry")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
|
|
@ -1231,27 +1289,16 @@ class ProxyServer:
|
|||
return False
|
||||
|
||||
def check_if_channel_exists(self, channel_id):
|
||||
"""Check if a channel exists in Redis or locally"""
|
||||
# Check local memory first
|
||||
"""Simple check if a channel exists using the registry"""
|
||||
# Check local memory first (quick check)
|
||||
if channel_id in self.stream_managers or channel_id in self.stream_buffers:
|
||||
return True
|
||||
|
||||
# Check Redis for channel metadata
|
||||
# Simple registry check in Redis
|
||||
if self.redis_client:
|
||||
metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
|
||||
if self.redis_client.exists(metadata_key):
|
||||
return True
|
||||
|
||||
# Also check for active channel marker
|
||||
activity_key = f"ts_proxy:active_channel:{channel_id}"
|
||||
if self.redis_client.exists(activity_key):
|
||||
return True
|
||||
|
||||
# Check for any buffer index
|
||||
buffer_key = f"ts_proxy:buffer:{channel_id}:index"
|
||||
if self.redis_client.exists(buffer_key):
|
||||
return True
|
||||
|
||||
registry_key = f"ts_proxy:active_channels:{channel_id}"
|
||||
return bool(self.redis_client.exists(registry_key))
|
||||
|
||||
return False
|
||||
|
||||
def stop_channel(self, channel_id):
|
||||
|
|
@ -1412,6 +1459,9 @@ class ProxyServer:
|
|||
|
||||
# Rest of the cleanup thread...
|
||||
|
||||
# Refresh active channel registry
|
||||
self.refresh_channel_registry()
|
||||
|
||||
except Exception as e:
|
||||
logging.error(f"Error in cleanup thread: {e}", exc_info=True)
|
||||
|
||||
|
|
@ -1491,3 +1541,16 @@ class ProxyServer:
|
|||
except Exception as e:
|
||||
logging.error(f"Error cleaning Redis keys for channel {channel_id}: {e}")
|
||||
|
||||
def refresh_channel_registry(self):
|
||||
"""Refresh TTL for active channels in registry"""
|
||||
if not self.redis_client:
|
||||
return
|
||||
|
||||
# Refresh registry entries for channels we own
|
||||
for channel_id in self.stream_managers.keys():
|
||||
registry_key = f"ts_proxy:active_channels:{channel_id}"
|
||||
if self.redis_client.exists(registry_key):
|
||||
# Update last_active timestamp and extend TTL
|
||||
self.redis_client.hset(registry_key, "last_active", str(time.time()))
|
||||
self.redis_client.expire(registry_key, 300) # 5 minute TTL
|
||||
|
||||
|
|
|
|||
|
|
@ -307,61 +307,113 @@ def stream_ts(request, channel_id):
|
|||
@csrf_exempt
|
||||
@require_http_methods(["POST"])
|
||||
def change_stream(request, channel_id):
|
||||
"""Change stream URL for existing channel with improved multi-worker support"""
|
||||
"""Change stream URL for existing channel with enhanced diagnostics"""
|
||||
try:
|
||||
data = json.loads(request.body)
|
||||
new_url = data.get('url')
|
||||
user_agent = data.get('user_agent') # Optional user agent
|
||||
user_agent = data.get('user_agent')
|
||||
|
||||
if not new_url:
|
||||
return JsonResponse({'error': 'No URL provided'}, status=400)
|
||||
|
||||
# Check if channel exists using the proper Redis check
|
||||
if not proxy_server.check_if_channel_exists(channel_id):
|
||||
logger.error(f"Channel {channel_id} not found in any worker or Redis")
|
||||
return JsonResponse({'error': 'Channel not found'}, status=404)
|
||||
|
||||
logger.info(f"Channel {channel_id} found, checking if current worker is owner")
|
||||
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
|
||||
|
||||
# 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}")
|
||||
|
||||
# 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"Changing stream URL for channel {channel_id} (owner worker)")
|
||||
logger.info(f"This worker is the owner, changing stream URL for channel {channel_id}")
|
||||
manager = proxy_server.stream_managers[channel_id]
|
||||
|
||||
# Update metadata in Redis first
|
||||
if proxy_server.redis_client:
|
||||
metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
|
||||
proxy_server.redis_client.hset(metadata_key, "url", new_url)
|
||||
if user_agent:
|
||||
proxy_server.redis_client.hset(metadata_key, "user_agent", user_agent)
|
||||
old_url = manager.url
|
||||
|
||||
# Update the stream
|
||||
old_url = manager.url
|
||||
manager.url = new_url
|
||||
manager.connected = False
|
||||
manager._close_socket() # Close existing connection
|
||||
|
||||
logger.info(f"Stream URL changed from {old_url} to {new_url}")
|
||||
result = manager.update_url(new_url)
|
||||
logger.info(f"Stream URL changed from {old_url} to {new_url}, result: {result}")
|
||||
return JsonResponse({
|
||||
'message': 'Stream URL updated',
|
||||
'channel': channel_id,
|
||||
'url': new_url,
|
||||
'owner': True
|
||||
'owner': True,
|
||||
'worker_id': proxy_server.worker_id
|
||||
})
|
||||
|
||||
# If we're not the owner, use Redis to request the change
|
||||
# If we're not the owner, publish an event for the owner to pick up
|
||||
else:
|
||||
logger.info(f"Requesting stream URL change for channel {channel_id} (non-owner worker)")
|
||||
|
||||
if not proxy_server.redis_client:
|
||||
return JsonResponse({'error': 'Redis not available, cannot request stream switch'}, status=500)
|
||||
|
||||
# Update metadata in Redis first (all workers can do this)
|
||||
metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
|
||||
proxy_server.redis_client.hset(metadata_key, "url", new_url)
|
||||
if user_agent:
|
||||
proxy_server.redis_client.hset(metadata_key, "user_agent", user_agent)
|
||||
|
||||
logger.info(f"This worker is not the owner, requesting URL change via Redis PubSub")
|
||||
# Publish switch request event
|
||||
switch_request = {
|
||||
"event": "stream_switch",
|
||||
|
|
@ -377,13 +429,12 @@ def change_stream(request, channel_id):
|
|||
json.dumps(switch_request)
|
||||
)
|
||||
|
||||
logger.info(f"Published stream switch request to {new_url}")
|
||||
|
||||
return JsonResponse({
|
||||
'message': 'Stream URL change requested',
|
||||
'channel': channel_id,
|
||||
'url': new_url,
|
||||
'owner': False
|
||||
'owner': False,
|
||||
'worker_id': proxy_server.worker_id
|
||||
})
|
||||
|
||||
except json.JSONDecodeError:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue