diff --git a/src/Cli/Commands/ProxyCommand.php b/src/Cli/Commands/ProxyCommand.php deleted file mode 100644 index b54edf43..00000000 --- a/src/Cli/Commands/ProxyCommand.php +++ /dev/null @@ -1,313 +0,0 @@ - - * @copyright 2025-2026 Vateron Media - * @link https://github.com/Vateron-Media/XC_VM - * @license AGPL-3.0 https://www.gnu.org/licenses/agpl-3.0.html - */ - -class ProxyCommand implements CommandInterface { - - public function getName(): string { - return 'proxy'; - } - - public function getDescription(): string { - return 'Proxy — MPEG-TS stream proxying via sockets'; - } - - public function execute(array $rArgs): int { - if (posix_getpwuid(posix_geteuid())['name'] != 'xc_vm') { - echo "Please run as \XC_VM!\n"; - return 1; - } - - if (empty($rArgs[0])) { - return 0; - } - - $rStreamID = intval($rArgs[0]); - - global $db; - - $this->checkRunning($rStreamID); - - $rFP = null; - - register_shutdown_function(function () use ($rStreamID, &$rFP) { - @unlink(STREAMS_PATH . $rStreamID . '_.monitor'); - @unlink(STREAMS_PATH . $rStreamID . '_.pid'); - shell_exec('rm -rf ' . CONS_TMP_PATH . $rStreamID . '/'); - if (is_resource($rFP)) { - @fclose($rFP); - } - }); - - set_time_limit(0); - cli_set_process_title('XC_VMProxy[' . $rStreamID . ']'); - - $db->query('SELECT * FROM `streams` t1 INNER JOIN `streams_servers` t2 ON t2.stream_id = t1.id AND t2.server_id = ? WHERE t1.id = ?', SERVER_ID, $rStreamID); - if ($db->num_rows() <= 0) { - StreamProcess::stopStream($rStreamID); - return 0; - } - - file_put_contents(STREAMS_PATH . $rStreamID . '_.monitor', getmypid()); - @unlink(STREAMS_PATH . $rStreamID . '_.pid'); - $rStreamInfo = $db->get_row(); - $db->query('SELECT t1.*, t2.* FROM `streams_options` t1, `streams_arguments` t2 WHERE t1.stream_id = ? AND t1.argument_id = t2.id', $rStreamID); - $rStreamArguments = $db->get_rows(true, 'argument_key'); - - define('PAT_HEADER', "�\r"); - define('PACKET_SIZE', 188); - define('BUFFER_SIZE', 12032); - define('PAT_PERIOD', 2); - define('TIMEOUT', 20); - define('CLOSE_EMPTY', 3000); - define('STORE_PREBUFFER', 1128000); - define('MAX_PREBUFFER', 10528000); - - $this->startProxy($rStreamID, $rStreamInfo, $rStreamArguments, $rFP); - - return 0; - } - - private function startProxy($rStreamID, $rStreamInfo, $rStreamArguments, &$rFP): void { - global $db; - if (!file_exists(CONS_TMP_PATH . $rStreamID . '/')) { - mkdir(CONS_TMP_PATH . $rStreamID); - } - $rUserAgent = (isset($rStreamArguments['user_agent']) ? ($rStreamArguments['user_agent']['value'] ?: $rStreamArguments['user_agent']['argument_default_value']) : 'Mozilla/5.0'); - $rOptions = array('ssl' => array('verify_peer' => false, 'verify_peer_name' => false, 'allow_self_signed' => true), 'http' => array('method' => 'GET', 'user_agent' => $rUserAgent, 'timeout' => TIMEOUT, 'header' => '')); - if (isset($rStreamArguments['proxy'])) { - $rOptions['http']['proxy'] = 'tcp://' . $rStreamArguments['proxy']['value']; - $rOptions['http']['request_fulluri'] = true; - } - if (isset($rStreamArguments['cookie'])) { - $rOptions['http']['header'] .= 'Cookie: ' . $rStreamArguments['cookie']['value'] . "\r\n"; - } - if (SettingsManager::getAll()['request_prebuffer']) { - $rOptions['http']['header'] .= 'X-\XC_VM-Prebuffer: 1' . "\r\n"; - } - $rContext = stream_context_create($rOptions); - $rURLs = json_decode($rStreamInfo['stream_source'], true); - $rFP = $this->getActiveStream($rURLs, $rContext); - if (!is_resource($rFP)) { - $rHeaders = (!empty($rOptions['http']['header']) ? '-headers ' . escapeshellarg($rOptions['http']['header']) : ''); - $rProxy = (!empty($rStreamArguments['proxy']) ? '-http_proxy ' . escapeshellarg($rStreamArguments['proxy']) : ''); - $rCommand = FfmpegPaths::cpu() . ' -copyts -vsync 0 -nostats -nostdin -hide_banner -loglevel quiet -y -user_agent ' . escapeshellarg($rUserAgent) . ' ' . $rHeaders . ' ' . $rProxy . ' -i ' . escapeshellarg($rFP) . ' -map 0 -c copy -mpegts_flags +initial_discontinuity -pat_period ' . PAT_PERIOD . ' -f mpegts -'; - $rFP = popen($rCommand, 'rb'); - } - if ($rFP) { - $db->query('UPDATE `streams_servers` SET `monitor_pid` = ?, `pid` = ?, `stream_started` = ?, `stream_status` = 0, `to_analyze` = 0 WHERE `server_stream_id` = ?', getmypid(), getmypid(), time(), $rStreamInfo['server_stream_id']); - if (SettingsManager::getAll()['enable_cache']) { - StreamProcess::updateStream($rStreamID); - } - shell_exec('rm -f ' . STREAMS_PATH . intval($rStreamID) . '_*.ts'); - file_put_contents(STREAMS_PATH . $rStreamID . '_.pid', getmypid()); - $db->close_mysql(); - $rLastSocket = null; - stream_set_blocking($rFP, false); - $rExcessBuffer = $rAnalyseBuffer = $rPrebuffer = $rBuffer = $rPacket = ''; - $rHasPrebuffer = $rPATHeaders = array(); - $rAnalysed = $rPAT = false; - $rFirstKeyframe = false; - while (!feof($rFP)) { - stream_set_timeout($rFP, TIMEOUT); - $rBuffer = $rBuffer . $rExcessBuffer . fread($rFP, BUFFER_SIZE - strlen($rBuffer . $rExcessBuffer)); - $rExcessBuffer = ''; - $rPacketNum = floor(strlen($rBuffer) / PACKET_SIZE); - if (0 < $rPacketNum) { - if (strlen($rBuffer) != $rPacketNum * PACKET_SIZE) { - $rExcessBuffer = substr($rBuffer, $rPacketNum * PACKET_SIZE, strlen($rBuffer) - $rPacketNum * PACKET_SIZE); - $rBuffer = substr($rBuffer, 0, $rPacketNum * PACKET_SIZE); - } - foreach (str_split($rBuffer, PACKET_SIZE) as $rPacket) { - list(, $rHeader) = unpack('N', substr($rPacket, 0, 4)); - $rSync = $rHeader >> 24 & 255; - if ($rSync == 71) { - if (substr($rPacket, 6, 4) == PAT_HEADER) { - $rPAT = true; - $rPATHeaders = array(); - } else { - $rAdaptationField = $rHeader >> 4 & 3; - if (($rAdaptationField & 2) === 2) { - if (0 < count($rPATHeaders) && unpack('C', $rPacket[4])[1] == 7 && substr($rPacket, 4, 2) == "\x07" . 'P') { - if (!$rPrebuffer || STORE_PREBUFFER <= strlen($rPrebuffer)) { - $rPrebuffer = implode('', $rPATHeaders) . $rPacket; - } - $rFirstKeyframe = true; - $rPAT = false; - $rPATHeaders = array(); - } - } - } - if ($rPAT && count($rPATHeaders) < 10) { - $rPATHeaders[] = $rPacket; - } - if (strlen($rPrebuffer) < MAX_PREBUFFER && $rFirstKeyframe) { - $rPrebuffer .= $rPacket; - } - if (!$rAnalysed) { - $rAnalyseBuffer .= $rPacket; - if (3000 * PACKET_SIZE <= strlen($rAnalyseBuffer)) { - echo 'Write analysis buffer' . "\n"; - file_put_contents(STREAMS_PATH . $rStreamID . '.analyse', $rAnalyseBuffer); - $rAnalyseBuffer = null; - $rAnalysed = true; - } - } - } - } - $rSockets = $this->getSockets($rStreamID); - if (0 < count($rSockets)) { - $rLastSocket = round(microtime(true) * 1000); - foreach ($rSockets as $rSocketID) { - $rSocketFile = CONS_TMP_PATH . $rStreamID . '/' . $rSocketID; - if (file_exists($rSocketFile) && (!isset($rHasPrebuffer[$rSocketID]) || !empty($rBuffer))) { - $rSocket = socket_create(AF_UNIX, SOCK_DGRAM, 0); - socket_set_nonblock($rSocket); - if (!isset($rHasPrebuffer[$rSocketID])) { - if (!empty($rPrebuffer)) { - echo 'Send prebuffer: ' . strlen($rPrebuffer) . ' bytes' . "\n"; - $rHasPrebuffer[$rSocketID] = true; - foreach (str_split($rPrebuffer, BUFFER_SIZE) as $rChunk) { - socket_sendto($rSocket, $rChunk, BUFFER_SIZE, 0, $rSocketFile); - } - } - } else { - if (!empty($rBuffer)) { - socket_sendto($rSocket, $rBuffer, BUFFER_SIZE, 0, $rSocketFile); - } - } - socket_close($rSocket); - } - } - } else { - if (!$rLastSocket) { - $rLastSocket = round(microtime(true) * 1000); - } - if (CLOSE_EMPTY <= round(microtime(true) * 1000) - $rLastSocket) { - echo 'No sockets waiting, close stream' . "\n"; - } - } - $rBuffer = ''; - break; - } - if (!$rLastSocket || 100000 < round(microtime(true) * 1000) - $rLastSocket) { - $rSockets = $this->getSockets($rStreamID); - if (0 < count($rSockets)) { - $rLastSocket = round(microtime(true) * 1000); - if (!empty($rPrebuffer)) { - foreach ($rSockets as $rSocketID) { - if (!isset($rHasPrebuffer[$rSocketID])) { - $rSocket = socket_create(AF_UNIX, SOCK_DGRAM, 0); - socket_set_nonblock($rSocket); - echo 'Send prebuffer: ' . strlen($rPrebuffer) . ' bytes' . "\n"; - $rHasPrebuffer[$rSocketID] = true; - foreach (str_split($rPrebuffer, BUFFER_SIZE) as $rChunk) { - socket_sendto($rSocket, $rChunk, BUFFER_SIZE, 0, CONS_TMP_PATH . $rStreamID . '/' . $rSocketID); - } - socket_close($rSocket); - } - } - } - } else { - if (!$rLastSocket) { - $rLastSocket = round(microtime(true) * 1000); - } - if (CLOSE_EMPTY <= round(microtime(true) * 1000) - $rLastSocket) { - echo 'No sockets waiting, close stream' . "\n"; - } - } - } - } - fclose($rFP); - $db->db_connect(); - $db->query('UPDATE `streams_servers` SET `monitor_pid` = null, `pid` = null, `stream_status` = 1 WHERE `server_stream_id` = ?;', $rStreamInfo['server_stream_id']); - if (SettingsManager::getAll()['enable_cache']) { - StreamProcess::updateStream($rStreamID); - } - return; - } - - echo 'Failed!' . "\n"; - StreamProcess::streamLog($rStreamID, SERVER_ID, 'STREAM_START_FAIL'); - $db->query('UPDATE `streams_servers` SET `monitor_pid` = null, `pid` = null, `stream_status` = 1 WHERE `server_stream_id` = ?;', $rStreamInfo['server_stream_id']); - if (SettingsManager::getAll()['enable_cache']) { - StreamProcess::updateStream($rStreamID); - } - } - - private function getSockets($rStreamID): array { - $rSockets = array(); - $rHandle = opendir(CONS_TMP_PATH . $rStreamID . '/'); - if ($rHandle) { - while (false !== ($rFilename = readdir($rHandle))) { - if ($rFilename != '.' && $rFilename != '..') { - $rSockets[] = $rFilename; - } - } - closedir($rHandle); - } - return $rSockets; - } - - private function getActiveStream($rURLs, $rContext) { - foreach ($rURLs as $rURL) { - $rURL = StreamUtils::parseStreamURL($rURL); - $rFP = @fopen($rURL, 'rb', false, $rContext); - if ($rFP) { - $rMetadata = stream_get_meta_data($rFP); - $rHeaders = array(); - foreach ($rMetadata['wrapper_data'] as $rLine) { - if (strpos($rLine, 'HTTP') !== 0) { - list($rKey, $rValue) = explode(': ', $rLine); - $rHeaders[$rKey] = $rValue; - } else { - $rHeaders[0] = $rLine; - } - } - $rContentType = (is_array($rHeaders['Content-Type']) ? $rHeaders['Content-Type'][count($rHeaders['Content-Type']) - 1] : $rHeaders['Content-Type']); - if (strtolower($rContentType) == 'video/mp2t') { - return $rFP; - } - fclose($rFP); - if (in_array(strtolower($rContentType), array('application/x-mpegurl', 'application/vnd.apple.mpegurl', 'audio/x-mpegurl'))) { - return $rURL; - } - } - } - return null; - } - - private function checkRunning($rStreamID): void { - clearstatcache(true); - $rPID = null; - if (file_exists(STREAMS_PATH . $rStreamID . '_.monitor')) { - $rPID = intval(file_get_contents(STREAMS_PATH . $rStreamID . '_.monitor')); - } - if (empty($rPID)) { - shell_exec("kill -9 `ps -ef | grep 'XC_VMProxy\\[" . intval($rStreamID) . "\\]' | grep -v grep | awk '{print \$2}'`;"); - } else { - if (file_exists('/proc/' . $rPID)) { - $rCommand = trim(file_get_contents('/proc/' . $rPID . '/cmdline')); - if ($rCommand == 'XC_VMProxy[' . $rStreamID . ']' && is_numeric($rPID) && 0 < $rPID) { - posix_kill($rPID, 9); - } - } - } - } -} diff --git a/src/Core/Process/ProcessManager.php b/src/Core/Process/ProcessManager.php index 8c62137b..4d38dc31 100644 --- a/src/Core/Process/ProcessManager.php +++ b/src/Core/Process/ProcessManager.php @@ -507,18 +507,6 @@ class ProcessManager { return true; } - /** - * Start a proxy process in background. - * - * Extracted from ProcessManager::startProxy(). - * - * @param int $streamID - * @return bool - */ - public static function startProxy($streamID) { - shell_exec(PHP_BIN . ' ' . MAIN_HOME . 'console.php proxy ' . intval($streamID) . ' >/dev/null 2>/dev/null &'); - return true; - } // ─────────────────────────────────────────────────────────── // Utility diff --git a/src/Domain/Stream/StreamProcess.php b/src/Domain/Stream/StreamProcess.php index 66218de7..ab0ae6eb 100644 --- a/src/Domain/Stream/StreamProcess.php +++ b/src/Domain/Stream/StreamProcess.php @@ -95,16 +95,6 @@ class StreamProcess { return true; } - /** - * Start the proxy process for a stream. - * - * @param int $rStreamID Stream id. - * @return mixed Start result. - */ - public static function startProxy($rStreamID) { - shell_exec(PHP_BIN . ' ' . MAIN_HOME . 'console.php proxy ' . intval($rStreamID) . ' >/dev/null 2>/dev/null &'); - return true; - } /** * Start thumbnail generation for a stream. diff --git a/src/Public/stream/live.php b/src/Public/stream/live.php index f292c867..abde0e2f 100644 --- a/src/Public/stream/live.php +++ b/src/Public/stream/live.php @@ -178,24 +178,12 @@ if ($rChannelInfo) { OffAirHandler::showNotOnAir($rExtension, $rUserInfo, $rIP, $rCountryCode, $rServerID, $rProxyID); } } else { - if (!empty($rChannelInfo["proxy"])) { - if (!($rChannelInfo["monitor_pid"] && ProcessManager::isMonitorAlive($rChannelInfo["monitor_pid"], $rStreamID))) { - @unlink(STREAMS_PATH . $rStreamID . "_.pid"); - ProcessManager::startProxy($rStreamID); - - if (AsyncFileOperations::awaitFileExists(STREAMS_PATH . $rStreamID . "_.monitor", 300, 10)) { - $rChannelInfo["monitor_pid"] = intval(AsyncFileOperations::readFile(STREAMS_PATH . $rStreamID . "_.monitor")); - } - } - - if (!$rChannelInfo["monitor_pid"]) { - OffAirHandler::showNotOnAir($rExtension, $rUserInfo, $rIP, $rCountryCode, $rServerID, $rProxyID); - } - - $rChannelInfo["pid"] = $rChannelInfo["monitor_pid"]; - } else { - OffAirHandler::showNotOnAir($rExtension, $rUserInfo, $rIP, $rCountryCode, $rServerID, $rProxyID); - } + // Non-on-demand: proxy streams are daemon-only now (ADR 0003, Phase E + // — the legacy ProxyCommand producer was removed). If we reach here + // $rFanout is false, i.e. the daemon is unreachable, so show + // not-on-air (the keepalive brings the daemon back in ~2s). A dead + // non-proxy stream is likewise not-on-air. + OffAirHandler::showNotOnAir($rExtension, $rUserInfo, $rIP, $rCountryCode, $rServerID, $rProxyID); } } @@ -415,59 +403,19 @@ if ($rChannelInfo) { if ($rChannelInfo["proxy"]) { // ──────────────────────────────────────────────────────────────── - // Fanout hand-off (ADR 0002/0003, P2). Auth is done and the source - // was already registered with the daemon above (that is what set - // $rFanout). Instead of pinning this PHP-FPM worker in the socket - // relay for the whole session, hand the byte path to nginx via - // X-Accel-Redirect (same pattern as P1 segment delivery): the - // worker is freed the moment we return and nginx proxies the viewer - // straight to the daemon's /live/. + // Proxy streams are daemon-only (ADR 0003, Phase E — the legacy + // ProxyCommand producer + socket relay were removed). Auth is done + // and the source was already registered + probed with the daemon + // above (that set $rFanout; an unreachable/dead daemon already went + // not-on-air). Hand the byte path to nginx → daemon via + // X-Accel-Redirect; the FPM worker is freed the instant we return. // ──────────────────────────────────────────────────────────────── - if ($rFanout) { - header("Content-Type: video/mp2t"); - header("X-Accel-Buffering: no"); - header("X-Accel-Redirect: /xc_fanout/" . rawurlencode((string) $rStreamID) . "?c=" . rawurlencode($rTokenData["uuid"])); - exit; + if (!$rFanout) { + OffAirHandler::showNotOnAir($rExtension, $rUserInfo, $rIP, $rCountryCode, $rServerID, $rProxyID); } - - // ──────────────────────────────────────────────────────────────── - // Legacy proxy relay (daemon unreachable): relay ffmpeg's - // unix-socket datagrams straight to the client. The producer was - // started by startProxy above because $rFanout is false. - // ──────────────────────────────────────────────────────────────── - header("Content-type: video/mp2t"); - - if (!file_exists(CONS_TMP_PATH . $rStreamID . "/")) { - mkdir(CONS_TMP_PATH . $rStreamID); - } - - $rSocketFile = CONS_TMP_PATH . $rStreamID . "/" . $rTokenData["uuid"]; - $rSocket = socket_create(AF_UNIX, SOCK_DGRAM, 0); - @unlink($rSocketFile); - socket_bind($rSocket, $rSocketFile); - socket_set_option($rSocket, SOL_SOCKET, SO_RCVTIMEO, array("sec" => 20, "usec" => 0)); - socket_set_nonblock($rSocket); - $rTotalFails = 200; - $rFails = 0; - - while ($rFails <= $rTotalFails) { - // MPEG-TS packet size = 188 bytes - // 64 packets per read: - // 188 * 64 = 12032 bytes (~12 KB) - $rBuffer = socket_read($rSocket, 188 * 64); - - if ($rBuffer !== false && $rBuffer !== '') { - $rFails = 0; - echo $rBuffer; - flush(); - } else { - $rFails++; - usleep(80000); // 80ms backoff when no data - } - } - // cleanup - socket_close($rSocket); - @unlink($rSocketFile); + header("Content-Type: video/mp2t"); + header("X-Accel-Buffering: no"); + header("X-Accel-Redirect: /xc_fanout/" . rawurlencode((string) $rStreamID) . "?c=" . rawurlencode($rTokenData["uuid"])); exit; }