Mirrors/Dispatcharr
1
0
Fork 1
mirror of https://github.com/Dispatcharr/Dispatcharr.git synced 2026-08-03 07:12:30 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Final refactoring and minor bug fix for transcode stream switching.

Browse source
This commit is contained in:
SergeantPanda 2025-03-19 20:22:13 -05:00
parent 569846a687
commit 86e6c942d7
5 changed files with 36 additions and 37 deletions
Download patch file Download diff file Expand all files Collapse all files

35
apps/proxy/ts_proxy/channel_status.py
View file

@ -2,6 +2,8 @@ import logging
import time
import re
from . import proxy_server
from .redis_keys import RedisKeys
from .constants import TS_PACKET_SIZE
logger = logging.getLogger("ts_proxy")
@ -9,22 +11,15 @@ class ChannelStatus:
def get_detailed_channel_info(channel_id):
# Get channel metadata
metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
metadata = proxy_server.redis_client.hgetall(metadata_key)
if not metadata:
return None
# Get detailed info - existing implementation
# Get channel metadata
metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
metadata_key = RedisKeys.channel_metadata(channel_id)
metadata = proxy_server.redis_client.hgetall(metadata_key)
if not metadata:
return None
# Basic channel info
buffer_index_value = proxy_server.redis_client.get(f"ts_proxy:channel:{channel_id}:buffer:index")
buffer_index_key = RedisKeys.buffer_index(channel_id)
buffer_index_value = proxy_server.redis_client.get(buffer_index_key)
info = {
'channel_id': channel_id,
@ -33,8 +28,6 @@ class ChannelStatus:
'profile': metadata.get(b'profile', b'unknown').decode('utf-8'),
'started_at': metadata.get(b'init_time', b'0').decode('utf-8'),
'owner': metadata.get(b'owner', b'unknown').decode('utf-8'),
# Properly decode the buffer index value
'buffer_index': int(buffer_index_value.decode('utf-8')) if buffer_index_value else 0,
}
@ -50,7 +43,7 @@ class ChannelStatus:
info['uptime'] = time.time() - created_at
# Get client information
client_set_key = f"ts_proxy:channel:{channel_id}:clients"
client_set_key = RedisKeys.clients(channel_id)
client_ids = proxy_server.redis_client.smembers(client_set_key)
clients = []
@ -97,7 +90,7 @@ class ChannelStatus:
# Check if the keys exist before getting
for i in range(info['buffer_index']-sample_chunks+1, info['buffer_index']+1):
chunk_key = f"ts_proxy:channel:{channel_id}:buffer:chunk:{i}"
chunk_key = RedisKeys.buffer_chunk(channel_id, i)
# Check if key exists first
if proxy_server.redis_client.exists(chunk_key):
@ -135,9 +128,9 @@ class ChannelStatus:
buffer_stats['total_sample_bytes'] = total_data
# Add TS packet analysis
total_ts_packets = total_data // 188
total_ts_packets = total_data // TS_PACKET_SIZE
buffer_stats['estimated_ts_packets'] = total_ts_packets
buffer_stats['is_ts_aligned'] = all(size % 188 == 0 for size in chunk_sizes)
buffer_stats['is_ts_aligned'] = all(size % TS_PACKET_SIZE == 0 for size in chunk_sizes)
else:
# If no chunks found, scan for keys to help debug
all_buffer_keys = []
@ -161,7 +154,7 @@ class ChannelStatus:
buffer_stats['diagnostics']['exception'] = str(e)
# Add TTL information to see if chunks are expiring
chunk_ttl_key = f"ts_proxy:channel:{channel_id}:buffer:chunk:{info['buffer_index']}"
chunk_ttl_key = RedisKeys.buffer_chunk(channel_id, info['buffer_index'])
chunk_ttl = proxy_server.redis_client.ttl(chunk_ttl_key)
buffer_stats['latest_chunk_ttl'] = chunk_ttl
@ -182,17 +175,18 @@ class ChannelStatus:
# Function for basic channel info (used for all channels summary)
def get_basic_channel_info(channel_id):
# Get channel metadata
metadata_key = f"ts_proxy:channel:{channel_id}:metadata"
metadata_key = RedisKeys.channel_metadata(channel_id)
metadata = proxy_server.redis_client.hgetall(metadata_key)
if not metadata:
return None
# Basic channel info only - omit diagnostics and details
buffer_index_value = proxy_server.redis_client.get(f"ts_proxy:channel:{channel_id}:buffer:index")
buffer_index_key = RedisKeys.buffer_index(channel_id)
buffer_index_value = proxy_server.redis_client.get(buffer_index_key)
# Count clients (using efficient count method)
client_set_key = f"ts_proxy:channel:{channel_id}:clients"
client_set_key = RedisKeys.clients(channel_id)
client_count = proxy_server.redis_client.scard(client_set_key) or 0
# Calculate uptime
@ -218,7 +212,6 @@ class ChannelStatus:
# Get concise client information
clients = []
client_set_key = f"ts_proxy:channel:{channel_id}:clients"
client_ids = proxy_server.redis_client.smembers(client_set_key)
# Process only if we have clients and keep it limited

11
apps/proxy/ts_proxy/client_manager.py
View file

@ -146,7 +146,7 @@ class ClientManager:
self._registered_clients.add(client_id)
# FIX: Consistent key naming - note the 's' in 'clients'
# Use a function to get the client key
client_key = f"ts_proxy:channel:{self.channel_id}:clients:{client_id}"
# Prepare client data
@ -175,7 +175,8 @@ class ClientManager:
self.redis_client.expire(self.client_set_key, self.client_ttl)
# Clear any initialization timer
self.redis_client.delete(f"ts_proxy:channel:{self.channel_id}:init_time")
init_key = f"ts_proxy:channel:{self.channel_id}:init_time"
self.redis_client.delete(init_key)
self._notify_owner_of_activity()
@ -195,7 +196,7 @@ class ClientManager:
logger.debug(f"No user agent provided for client {client_id}")
self.redis_client.publish(
f"ts_proxy:events:{self.channel_id}",
RedisKeys.events_channel(self.channel_id), # Use RedisKeys instead of string
json.dumps(event_data)
)
@ -236,7 +237,7 @@ class ClientManager:
logger.warning(f"Last client removed: {client_id} - channel may shut down soon")
# Trigger disconnect time tracking even if we're not the owner
disconnect_key = f"ts_proxy:channel:{self.channel_id}:last_client_disconnect_time"
disconnect_key = RedisKeys.last_client_disconnect(self.channel_id)
self.redis_client.setex(disconnect_key, 60, str(time.time()))
self._notify_owner_of_activity()
@ -250,7 +251,7 @@ class ClientManager:
"timestamp": time.time(),
"remaining_clients": remaining
})
self.redis_client.publish(f"ts_proxy:events:{self.channel_id}", event_data)
self.redis_client.publish(RedisKeys.events_channel(self.channel_id), event_data)
total_clients = self.get_total_client_count()
logger.info(f"Client disconnected: {client_id} (local: {len(self.clients)}, total: {total_clients})")

4
apps/proxy/ts_proxy/server.py
View file

@ -596,8 +596,8 @@ class ProxyServer:
def _cleanup_channel(self, channel_id: str) -> None:
"""Remove channel resources"""
for collection in [self.stream_managers, self.stream_buffers,
self.client_managers, self.fetch_threads]:
# Removed reference to non-existent fetch_threads collection
for collection in [self.stream_managers, self.stream_buffers, self.client_managers]:
collection.pop(channel_id, None)
def shutdown(self) -> None:

7
apps/proxy/ts_proxy/services/channel_service.py
View file

@ -11,6 +11,7 @@ from apps.channels.models import Channel
from apps.proxy.config import TSConfig as Config
from .. import proxy_server
from ..redis_keys import RedisKeys
from ..constants import EventType
logger = logging.getLogger("ts_proxy")
@ -317,7 +318,7 @@ class ChannelService:
return False
switch_request = {
"event": "stream_switch",
"event": EventType.STREAM_SWITCH, # Use constant instead of string
"channel_id": channel_id,
"url": new_url,
"user_agent": user_agent,
@ -338,7 +339,7 @@ class ChannelService:
return False
stop_request = {
"event": "channel_stop",
"event": EventType.CHANNEL_STOP, # Use constant instead of string
"channel_id": channel_id,
"requester_worker_id": proxy_server.worker_id,
"timestamp": time.time()
@ -359,7 +360,7 @@ class ChannelService:
return False
stop_request = {
"event": "client_stop",
"event": EventType.CLIENT_STOP, # Use constant instead of string
"channel_id": channel_id,
"client_id": client_id,
"requester_worker_id": proxy_server.worker_id,

16
apps/proxy/ts_proxy/stream_manager.py
View file

@ -3,6 +3,7 @@
import threading
import logging
import time
import socket
import requests
import subprocess
from typing import Optional, List
@ -414,16 +415,19 @@ class StreamManager:
logger.debug(f"Error closing session: {e}")
self.current_session = None
# Keep backward compatibility - let's create an alias to the new method
def _close_socket(self):
"""Backward compatibility wrapper for _close_connection"""
if self.current_response:
return self._close_connection()
"""Close socket and transcode resources as needed"""
# First try to use _close_connection for HTTP resources
if self.current_response or self.current_session:
self._close_connection()
return
# Otherwise handle socket and transcode resources
if self.socket:
try:
self.socket.close()
except Exception as e:
logging.debug(f"Error closing socket: {e}")
logger.debug(f"Error closing socket: {e}")
pass
self.socket = None
@ -434,7 +438,7 @@ class StreamManager:
self.transcode_process.terminate()
self.transcode_process.wait()
except Exception as e:
logging.debug(f"Error terminating transcode process: {e}")
logger.debug(f"Error terminating transcode process: {e}")
pass
self.transcode_process = None