TcpConnection.php 文件495行和500行bug

问题描述

我使用AsyncTcpConnection方式服务器通过流方式代理远程视频图片文件，实时通过服务器中转下发。。
vendor/workerman/workerman/src/Connection/TcpConnection.php 文件500行 和495行
$len = @fwrite($this->socket, $sendBuffer);
if ($len === strlen($sendBuffer)) {} 会提示参数类型不对，sendBuffer需要一个字符串类型。这里$sendBuffer不是一个字符串。
我改为了
$len = @fwrite($this->socket, (string)$sendBuffer);
if ($len === strlen((string)$sendBuffer)) {}

程序代码或配置

从download()运行

        <?php

namespace app\api\services;

use app\common\services\BaseService;
use GuzzleHttp\Client;
use GuzzleHttp\Exception\GuzzleException;
use GuzzleHttp\Psr7\Uri;
use GuzzleHttp\Psr7\UriResolver;
use InvalidArgumentException;
use Psr\Http\Message\ResponseInterface;
use support\Log;
use support\Request;
use support\Response;
use Throwable;
use Workerman\Connection\AsyncTcpConnection;
use Workerman\Connection\TcpConnection;
use Workerman\Protocols\Http\Chunk;

/**
 * 媒体下载代理服务。
 */
class MediaDownloadProxyService extends BaseService
{
    /**
     * 上游允许转发的媒体请求头。
     */
    protected const ALLOW_UPSTREAM_HEADERS = [
        'user-agent'      => 'User-Agent',
        'referer'         => 'Referer',
        'origin'          => 'Origin',
        'accept'          => 'Accept',
        'accept-language' => 'Accept-Language',
    ];

    /**
     * 支持的上游跳转状态码。
     */
    protected const REDIRECT_STATUS_CODES = [301, 302, 303, 307, 308];

    /**
     * 最大跳转次数。
     */
    protected const MAX_REDIRECTS = 5;

    /**
     * 上游响应头最大长度。
     */
    protected const MAX_HEADER_SIZE = 65536;

    /**
     * 媒体资源实时下载转发。
     *
     * @param Request $request
     * @return Response
     */
    public function download(Request $request): Response
    {
        $urlHash = '';
        try {
            // 解析请求，可能抛出 InvalidArgumentException
            $download = $this->resolveRequest($request);
            $urlHash = md5($download['url']);

            $clientConnection = $request->connection;
            if (!$clientConnection instanceof TcpConnection) {
                throw new InvalidArgumentException('下载连接异常');
            }

            // ----- 预检上游 -----
            try {
                $probe = $this->probeUpstream($download['url'], $download['headers']);
            } catch (Throwable $e) {
                // 预检失败降级，记录详细错误
                Log::channel('media_downloader_response')->warning('预检失败，降级为直接代理', [
                    'url_hash'  => $urlHash,
                    'error'     => $e->getMessage(),
                    'code'      => $e->getCode(),
                    'exception' => get_class($e),
                ]);
                // 根据 URL 猜测 Content-Type
                $contentType = $this->guessContentTypeFromUrl($download['url']) ?: 'application/octet-stream';
                $probe = [
                    'status'              => 200,
                    'total_bytes'         => 0,
                    'content_type'        => $contentType,
                    'content_disposition' => '',
                    'content_range'       => '',
                    'accept_ranges'       => '',
                    'etag'                => '',
                    'last_modified'       => '',
                ];
            }

            // 启动流式代理（异步）
            $this->startStreamProxy($clientConnection, $download['url'], $download['headers']);

            // 构建响应头，立即返回给客户端
            $isRangeDownload = isset($download['headers']['Range']);
            $statusCode      = $isRangeDownload && ($probe['status'] ?? 0) === 206 ? 206 : 200;
            $headers = $this->buildStreamResponseHeaders($download['url'], $probe, $isRangeDownload);

            return response('', $statusCode, $headers);
        } catch (InvalidArgumentException $exception) {
            $statusCode = $exception->getCode() >= 400 && $exception->getCode() <= 599 ? $exception->getCode() : 400;
            $msg = $exception->getMessage();
            Log::channel('media_downloader_response')->error('下载参数错误，返回文本错误', [
                'url_hash'   => $urlHash,
                'message'    => $msg,
                'statusCode' => $statusCode,
            ]);
            return response($msg, $statusCode, ['Content-Type' => 'text/plain;charset=utf-8']);
        } catch (GuzzleException $exception) {
            Log::channel('media_downloader_response')->error('下载过程 Guzzle 异常（非预检）', [
                'url_hash'  => $urlHash,
                'exception' => get_class($exception),
                'code'      => $exception->getCode(),
                'message'   => $exception->getMessage(),
            ]);
            return response('资源下载失败', 502, ['Content-Type' => 'text/plain;charset=utf-8']);
        } catch (Throwable $throwable) {
            Log::channel('media_downloader_response')->error('下载过程未知异常，返回500', [
                'url_hash'  => $urlHash,
                'exception' => get_class($throwable),
                'code'      => $throwable->getCode(),
                'message'   => $throwable->getMessage(),
                'trace'     => $throwable->getTraceAsString(), // 调试时可保留
            ]);
            return response('资源下载失败', 500, ['Content-Type' => 'text/plain;charset=utf-8']);
        }
    }

    /**
     * 获取媒体资源下载信息。
     *
     * @param Request $request
     * @return array
     */
    public function info(Request $request): array
    {
        $urlHash = '';
        try {
            $download = $this->resolveRequest($request);
            $urlHash = md5($download['url']);

            // 优先 HEAD
            $upstream = $this->openUpstream('HEAD', $download['url'], $download['headers'], false, 15);

            if (!in_array($upstream->getStatusCode(), [200, 206], true)) {
                $upstream->getBody()->close();
                Log::channel('media_downloader_response')->info('HEAD失败，回退GET Range=0-0', ['url_hash' => $urlHash]);
                $headers = $download['headers'];
                Log::channel('media_downloader_response')->info('HEAD失败，回退GET Range=0-0', ['url_hash' => $urlHash]);
                $headers['Range'] = 'bytes=0-0';
                Log::channel('media_downloader_response')->info('HEAD失败，回退GET Range=0-0', ['url_hash' => $urlHash]);
                $upstream = $this->openUpstream('GET', $download['url'], $headers, true, 15);
            }

            $status = $upstream->getStatusCode();
            if (!in_array($status, [200, 206], true)) {
                $this->logUpstreamError($download['url'], $upstream, '媒体下载信息上游响应异常');
                $upstream->getBody()->close();
                throw new InvalidArgumentException('资源信息获取失败');
            }

            try {
                $totalBytes   = $this->resolveTotalBytes($upstream);
                $contentType  = $upstream->getHeaderLine('Content-Type') ?: 'application/octet-stream';
                $contentRange = $upstream->getHeaderLine('Content-Range');
                $acceptRanges = $upstream->getHeaderLine('Accept-Ranges');

                $info = [
                    'total_bytes'    => $totalBytes,
                    'content_length' => $totalBytes,
                    'content_type'   => $contentType,
                    'mime_type'      => strtolower(trim(explode(';', $contentType)[0] ?? '')),
                    'support_range'  => strtolower($acceptRanges) === 'bytes' || $contentRange !== '',
                    'accept_ranges'  => $acceptRanges,
                    'content_range'  => $contentRange,

                ];
                //Log::channel('media_downloader_response')->info('媒体信息获取成功', ['url_hash' => $urlHash, 'info' => $info]);
                return $info;
            } finally {
                $upstream->getBody()->close();
            }
        } catch (InvalidArgumentException $exception) {
            throw $exception;
        } catch (GuzzleException $exception) {
            Log::channel('media_downloader_response')->error('媒体信息请求 Guzzle 异常', [
                'url_hash'  => $urlHash,
                'exception' => get_class($exception),
                'code'      => $exception->getCode(),
                'message'   => $exception->getMessage(),
            ]);
            throw new InvalidArgumentException('资源信息获取失败');
        } catch (Throwable $throwable) {
            Log::channel('media_downloader_response')->error('媒体信息未知异常', [
                'url_hash'  => $urlHash,
                'exception' => get_class($throwable),
                'code'      => $throwable->getCode(),
                'message'   => $throwable->getMessage(),
            ]);
            throw new InvalidArgumentException('资源信息获取失败');
        }
    }

    /**
     * 解析下载地址和上游请求头。
     *
     * @param Request $request
     * @return array
     */
    private function resolveRequest(Request $request): array
    {
        $url = trim((string)$request->input('url', ''));
        if ($url === '') {
            throw new InvalidArgumentException('下载地址不能为空');
        }
        $this->validateDownloadUrl($url);

        $headers = [
            'Accept'          => '*/*',
            'Accept-Encoding' => 'identity',
            'User-Agent'      => trim((string)$request->header('user-agent', '')) ?: 'Mozilla/5.0',
        ];

        $inputHeaders = trim((string)$request->input('headers', ''));
        if ($inputHeaders !== '') {
            $decodedHeaders = json_decode($inputHeaders, true);
            if (!is_array($decodedHeaders) || json_last_error() !== JSON_ERROR_NONE) {
                throw new InvalidArgumentException('headers格式不正确');
            }

            foreach ($decodedHeaders as $name => $value) {
                $headerName = strtolower(trim((string)$name));
                if (!isset(self::ALLOW_UPSTREAM_HEADERS[$headerName]) || is_array($value) || is_object($value)) {
                    continue;
                }

                $normalizedName  = self::ALLOW_UPSTREAM_HEADERS[$headerName];
                $normalizedValue = trim((string)$value);
                if ($normalizedName === 'User-Agent' && $normalizedValue === '') {
                    $normalizedValue = trim((string)$request->header('user-agent', '')) ?: 'Mozilla/5.0';
                }
                if ($normalizedValue !== '') {
                    $headers[$normalizedName] = $normalizedValue;
                }
            }
        }

        foreach (['range' => 'Range', 'if-range' => 'If-Range'] as $requestHeader => $forwardHeader) {
            $value = trim((string)$request->header($requestHeader, ''));
            if ($value !== '') {
                $headers[$forwardHeader] = $value;
            }
        }

        return [
            'url'     => $url,
            'headers' => $headers,
        ];
    }

    /**
     * 校验下载地址。
     *
     * @param string $url
     * @return array
     */
    private function validateDownloadUrl(string $url): array
    {
        if (strlen($url) > 8000) {
            throw new InvalidArgumentException('下载地址过长');
        }

        $parts  = parse_url($url);
        $scheme = strtolower((string)($parts['scheme'] ?? ''));
        $host   = (string)($parts['host'] ?? '');
        if (!is_array($parts) || !in_array($scheme, ['http', 'https'], true) || $host === '') {
            throw new InvalidArgumentException('下载地址格式不正确');
        }
        if (isset($parts['user']) || isset($parts['pass']) || $this->isBlockedHost($host)) {
            throw new InvalidArgumentException('下载地址不允许访问');
        }

        return $parts;
    }

    /**
     * 打开上游响应。
     *
     * @param string $method
     * @param string $url
     * @param array $headers
     * @param bool $stream
     * @param int|null $timeout
     * @return ResponseInterface
     * @throws GuzzleException
     */
    private function openUpstream(string $method, string $url, array $headers, bool $stream = false, ?int $timeout = null): ResponseInterface
    {
        return (new Client([
            'connect_timeout' => 10,
            'timeout'         => $timeout ?? ($stream ? 0 : 15),
            'http_errors'     => false,
        ]))->request(strtoupper($method), $url, [
            'headers'         => $headers,
            'stream'          => $stream,
            'decode_content'  => false,
            'allow_redirects' => [
                'max'             => self::MAX_REDIRECTS,
                'strict'          => false,
                'referer'         => true,
                'protocols'       => ['http', 'https'],
                'track_redirects' => false,
                'on_redirect'     => function ($request, $response, $uri): void {
                    $this->validateDownloadUrl((string)$uri);
                },
            ],
        ]);
    }

    /**
     * 预检上游资源。
     *
     * @param string $url
     * @param array $headers
     * @return array
     * @throws GuzzleException
     */
    private function probeUpstream(string $url, array $headers): array
    {
        $upstream = null;
        try {
            $upstream = isset($headers['Range'])
                ? $this->openUpstream('GET', $url, $headers, true, 15)
                : $this->openUpstream('HEAD', $url, $headers, false, 15);

            if (!in_array($upstream->getStatusCode(), [200, 206], true)) {
                $upstream->getBody()->close();
                $headers['Range'] = 'bytes=0-0';
                $upstream = $this->openUpstream('GET', $url, $headers, true, 15);
            }

            $status = $upstream->getStatusCode();
            if (!in_array($status, [200, 206], true)) {
                $this->logUpstreamError($url, $upstream, '媒体下载转发预检上游响应异常');
                throw new InvalidArgumentException('资源下载失败', 502);
            }

            return [
                'status'              => $status,
                'total_bytes'         => $this->resolveTotalBytes($upstream),
                'content_type'        => $upstream->getHeaderLine('Content-Type') ?: 'application/octet-stream',
                'content_disposition' => $upstream->getHeaderLine('Content-Disposition'),
                'content_range'       => $upstream->getHeaderLine('Content-Range'),
                'accept_ranges'       => $upstream->getHeaderLine('Accept-Ranges'),
                'etag'                => $upstream->getHeaderLine('ETag'),
                'last_modified'       => $upstream->getHeaderLine('Last-Modified'),
            ];
        } finally {
            if ($upstream instanceof ResponseInterface) {
                $upstream->getBody()->close();
            }
        }
    }

    /**
     * 构建实时流响应头。
     *
     * @param string $url
     * @param array $probe
     * @param bool $isRangeDownload
     * @return array
     */
    private function buildStreamResponseHeaders(string $url, array $probe, bool $isRangeDownload = false): array
    {
        $contentType = (string)($probe['content_type'] ?? 'application/octet-stream');
        $headers     = [
            'Content-Type'           => $contentType,
            'Transfer-Encoding'      => 'chunked',
            'Content-Disposition'    => $this->resolveContentDisposition($url, $contentType),
            'Cache-Control'          => 'no-store',
            'X-Accel-Buffering'      => 'no',
            'X-Content-Type-Options' => 'nosniff',
            'Connection'             => 'close',
        ];

        $forwardHeaders = [
            'Accept-Ranges' => 'accept_ranges',
            'ETag'          => 'etag',
            'Last-Modified' => 'last_modified',
        ];
        if ($isRangeDownload) {
            $forwardHeaders['Content-Range'] = 'content_range';
        }

        foreach ($forwardHeaders as $header => $key) {
            $value = trim((string)($probe[$key] ?? ''));
            if ($value !== '') {
                $headers[$header] = $value;
            }
        }

        return $headers;
    }

    /**
     * 启动上游实时流代理。
     *
     * @param TcpConnection $clientConnection
     * @param string $url
     * @param array $headers
     * @param int $redirects
     * @return void
     */
    private function startStreamProxy(TcpConnection $clientConnection, string $url, array $headers, int $redirects = 0): void
    {
        $parts      = $this->validateDownloadUrl($url);
        $connection = new AsyncTcpConnection($this->buildAsyncAddress($parts), $this->buildSocketContext($parts));
        $state      = [
            'url'              => $url,
            'headers'          => $headers,
            'redirects'        => $redirects,
            'ignore_close'     => false,
            'header_parsed'    => false,
            'header_buffer'    => '',
            'upstream_chunked' => false,
            'chunk_buffer'     => '',
            'chunk_remaining'  => null,
        ];

        $this->bindClientCallbacks($clientConnection, $connection);

        $connection->onConnect = function (AsyncTcpConnection $connection) use ($parts, $headers): void {
            $connection->send($this->buildRawHttpRequest($parts, $headers), true);
        };

        $connection->onMessage = function (AsyncTcpConnection $connection, string $buffer) use ($clientConnection, &$state): void {
            $this->handleUpstreamBuffer($connection, $clientConnection, $buffer, $state);
        };

        $connection->onClose = function () use ($clientConnection, &$state): void {
            if ($state['ignore_close']) {
                return;
            }
            if (!$this->isClientStreamFinished($clientConnection)) {
                Log::channel('media_downloader_response')->warning('上游连接关闭，客户端流未完成，主动结束', [
                    'url_hash' => md5($state['url']),
                ]);
                $this->finishClientStream($clientConnection);
            }
        };

        $connection->onError = function (AsyncTcpConnection $connection, int $code, string $message) use ($clientConnection, $url): void {
            $this->logStreamError('媒体下载转发上游连接异常', $url, [
                'code'    => $code,
                'message' => $message,
            ]);
            if (!$this->isClientStreamFinished($clientConnection)) {
                $this->finishClientStream($clientConnection);
            }
        };

        $connection->connect();
    }

    /**
     * 绑定客户端背压和断开处理。
     *
     * @param TcpConnection $clientConnection
     * @param AsyncTcpConnection $upstreamConnection
     * @return void
     */
    private function bindClientCallbacks(TcpConnection $clientConnection, AsyncTcpConnection $upstreamConnection): void
    {
        if (!isset($clientConnection->context->mediaDownloadProxy)) {
            $clientConnection->context->mediaDownloadProxy = (object)[
                'upstream'          => null,
                'finished'          => false,
                'callbacks_bound'   => false,
                'previous_on_close' => $clientConnection->onClose,
            ];
        }

        $clientConnection->context->mediaDownloadProxy->upstream = $upstreamConnection;
        if ($clientConnection->context->mediaDownloadProxy->callbacks_bound) {
            return;
        }

        $clientConnection->context->mediaDownloadProxy->callbacks_bound = true;
        $previousOnClose = $clientConnection->context->mediaDownloadProxy->previous_on_close;

        $clientConnection->onClose = function (TcpConnection $connection) use ($previousOnClose): void {
            $proxy = $connection->context->mediaDownloadProxy ?? null;
            if ($proxy && $proxy->upstream instanceof AsyncTcpConnection) {
                $proxy->upstream->close();
            }
            if (is_callable($previousOnClose)) {
                $previousOnClose($connection);
            }
        };

        $clientConnection->onBufferFull = function (TcpConnection $connection): void {
            $proxy = $connection->context->mediaDownloadProxy ?? null;
            if ($proxy && $proxy->upstream instanceof AsyncTcpConnection) {
                $proxy->upstream->pauseRecv();
            }
        };

        $clientConnection->onBufferDrain = function (TcpConnection $connection): void {
            $proxy = $connection->context->mediaDownloadProxy ?? null;
            if ($proxy && $proxy->upstream instanceof AsyncTcpConnection) {
                $proxy->upstream->resumeRecv();
            }
        };
    }

    /**
     * 处理上游原始响应数据。
     *
     * @param AsyncTcpConnection $upstreamConnection
     * @param TcpConnection $clientConnection
     * @param string $buffer
     * @param array $state
     * @return void
     */
    private function handleUpstreamBuffer(AsyncTcpConnection $upstreamConnection, TcpConnection $clientConnection, string $buffer, array &$state): void
    {
        if ($this->isClientStreamFinished($clientConnection)) {
            $upstreamConnection->close();
            return;
        }

        if (!$state['header_parsed']) {
            $state['header_buffer'] .= $buffer;
            if (strlen($state['header_buffer']) > self::MAX_HEADER_SIZE) {
                $this->logStreamError('媒体下载转发上游响应头过大', $state['url']);
                $upstreamConnection->close();
                $this->finishClientStream($clientConnection);
                return;
            }

            $headerEnd = strpos($state['header_buffer'], "\r\n\r\n");
            if ($headerEnd === false) {
                return;
            }

            $rawHeader              = substr($state['header_buffer'], 0, $headerEnd + 4);
            $body                   = substr($state['header_buffer'], $headerEnd + 4);
            $state['header_buffer'] = '';
            $state['header_parsed'] = true;
            $response               = $this->parseRawResponseHeaders($rawHeader);
            $status                 = $response['status'];

            if (in_array($status, self::REDIRECT_STATUS_CODES, true)) {
                $state['ignore_close'] = true;
                $upstreamConnection->close();
                $this->handleStreamRedirect($clientConnection, $state, $response);
                return;
            }

            if (!in_array($status, [200, 206], true)) {
                $this->logStreamError('媒体下载转发上游响应异常', $state['url'], [
                    'status_code'   => $status,
                    'reason_phrase' => $response['reason'],
                ]);
                $upstreamConnection->close();
                $this->finishClientStream($clientConnection);
                return;
            }

            $state['upstream_chunked'] = $this->hasHeaderToken($response['headers'], 'transfer-encoding', 'chunked');
            if ($body === '') {
                return;
            }

            $buffer = $body;
        }

        $this->sendUpstreamBody($upstreamConnection, $clientConnection, $buffer, $state);
    }

    /**
     * 处理上游跳转。
     *
     * @param TcpConnection $clientConnection
     * @param array $state
     * @param array $response
     * @return void
     */
    private function handleStreamRedirect(TcpConnection $clientConnection, array $state, array $response): void
    {
        $location = trim((string)($response['headers']['location'][0] ?? ''));
        if ($location === '' || $state['redirects'] >= self::MAX_REDIRECTS) {
            $this->logStreamError('媒体下载转发跳转失败', $state['url'], [
                'status_code' => $response['status'],
                'location'    => $location === '' ? '' : md5($location),
            ]);
            $this->finishClientStream($clientConnection);
            return;
        }

        try {
            $redirectUrl = (string)UriResolver::resolve(new Uri($state['url']), new Uri($location));
            $this->validateDownloadUrl($redirectUrl);
            $this->startStreamProxy($clientConnection, $redirectUrl, $state['headers'], $state['redirects'] + 1);
        } catch (Throwable $throwable) {
            $this->logStreamError('媒体下载转发跳转地址异常', $state['url'], [
                'exception' => get_class($throwable),
                'code'      => $throwable->getCode(),
            ]);
            $this->finishClientStream($clientConnection);
        }
    }

    /**
     * 发送上游响应体。
     *
     * @param AsyncTcpConnection $upstreamConnection
     * @param TcpConnection $clientConnection
     * @param string $buffer
     * @param array $state
     * @return void
     */
    private function sendUpstreamBody(AsyncTcpConnection $upstreamConnection, TcpConnection $clientConnection, string $buffer, array &$state): void
    {
        if ($buffer === '') {
            return;
        }

        if ($state['upstream_chunked']) {
            $this->sendDecodedUpstreamChunks($upstreamConnection, $clientConnection, $buffer, $state);
            return;
        }

        if ($clientConnection->send(new Chunk($buffer), true) === false) {
            Log::channel('media_downloader_response')->warning('发送数据块失败，客户端缓冲区满', [
                'url_hash' => md5($state['url']),
            ]);
            $upstreamConnection->close();
            $this->finishClientStream($clientConnection);
        }
    }

    /**
     * 解码上游 chunked 响应后再按本接口的 chunked 响应下发。
     *
     * @param AsyncTcpConnection $upstreamConnection
     * @param TcpConnection $clientConnection
     * @param string $buffer
     * @param array $state
     * @return void
     */
    private function sendDecodedUpstreamChunks(AsyncTcpConnection $upstreamConnection, TcpConnection $clientConnection, string $buffer, array &$state): void
    {
        $state['chunk_buffer'] .= $buffer;

        while ($state['chunk_buffer'] !== '') {
            if ($state['chunk_remaining'] === null) {
                $lineEnd = strpos($state['chunk_buffer'], "\r\n");
                if ($lineEnd === false) {
                    return;
                }

                $line = substr($state['chunk_buffer'], 0, $lineEnd);
                $state['chunk_buffer'] = substr($state['chunk_buffer'], $lineEnd + 2);
                $hex = trim(explode(';', $line, 2)[0]);
                if ($hex === '' || !ctype_xdigit($hex)) {
                    $this->logStreamError('媒体下载转发上游 chunk 格式异常', $state['url']);
                    $upstreamConnection->close();
                    $this->finishClientStream($clientConnection);
                    return;
                }

                $state['chunk_remaining'] = hexdec($hex);
                if ($state['chunk_remaining'] === 0) {
                    $upstreamConnection->close();
                    $this->finishClientStream($clientConnection);
                    return;
                }
            }

            if (strlen($state['chunk_buffer']) < $state['chunk_remaining'] + 2) {
                return;
            }

            $data = substr($state['chunk_buffer'], 0, $state['chunk_remaining']);
            $state['chunk_buffer'] = substr($state['chunk_buffer'], $state['chunk_remaining'] + 2);
            $state['chunk_remaining'] = null;

            if ($data !== '' && $clientConnection->send(new Chunk($data), true) === false) {
                Log::channel('media_downloader_response')->warning('发送chunk数据失败', [
                    'url_hash' => md5($state['url']),
                ]);
                $upstreamConnection->close();
                $this->finishClientStream($clientConnection);
                return;
            }
        }
    }

    /**
     * 结束客户端 chunked 响应。
     *
     * @param TcpConnection $clientConnection
     * @return void
     */
    private function finishClientStream(TcpConnection $clientConnection): void
    {
        $proxy = $clientConnection->context->mediaDownloadProxy ?? null;
        if ($proxy && $proxy->finished) {
            return;
        }
        if ($proxy) {
            $proxy->finished = true;
            $proxy->upstream = null;
        }

        $clientConnection->onBufferFull = null;
        $clientConnection->onBufferDrain = null;
        $clientConnection->close(new Chunk(''), true);
    }

    /**
     * 判断客户端流是否已结束。
     *
     * @param TcpConnection $clientConnection
     * @return bool
     */
    private function isClientStreamFinished(TcpConnection $clientConnection): bool
    {
        return (bool)($clientConnection->context->mediaDownloadProxy->finished ?? false);
    }

    /**
     * 构建 AsyncTcpConnection 地址。
     *
     * @param array $parts
     * @return string
     */
    private function buildAsyncAddress(array $parts): string
    {
        $scheme = strtolower((string)$parts['scheme']);
        $host   = (string)$parts['host'];
        $port   = (int)($parts['port'] ?? ($scheme === 'https' ? 443 : 80));
        $host   = str_contains($host, ':') && $host[0] !== '[' ? '[' . $host . ']' : $host;

        return ($scheme === 'https' ? 'ssl' : 'tcp') . '://' . $host . ':' . $port;
    }

    /**
     * 构建上游连接上下文。
     *
     * @param array $parts
     * @return array
     */
    private function buildSocketContext(array $parts): array
    {
        if (strtolower((string)$parts['scheme']) !== 'https') {
            return [];
        }

        return [
            'ssl' => [
                'peer_name'        => (string)$parts['host'],
                'SNI_enabled'      => true,
                'verify_peer'      => true,
                'verify_peer_name' => true,
            ],
        ];
    }

    /**
     * 构建发送给上游的原始 HTTP 请求。
     *
     * @param array $parts
     * @param array $headers
     * @return string
     */
    private function buildRawHttpRequest(array $parts, array $headers): string
    {
        $path = (string)($parts['path'] ?? '/');
        $path = $path === '' ? '/' : $path;
        if (isset($parts['query']) && $parts['query'] !== '') {
            $path .= '?' . $parts['query'];
        }

        $host = (string)$parts['host'];
        $port = (int)($parts['port'] ?? 0);
        $defaultPort = strtolower((string)$parts['scheme']) === 'https' ? 443 : 80;
        $hostHeader = $port > 0 && $port !== $defaultPort ? $host . ':' . $port : $host;

        $lines = [
            'GET ' . $path . ' HTTP/1.1',
            'Host: ' . $hostHeader,
            'Connection: close',
        ];

        foreach ($headers as $name => $value) {
            if (strcasecmp((string)$name, 'Host') === 0 || strcasecmp((string)$name, 'Connection') === 0) {
                continue;
            }
            if (!$this->isSafeHeaderLine((string)$name, (string)$value)) {
                continue;
            }
            $lines[] = $name . ': ' . $value;
        }

        return implode("\r\n", $lines) . "\r\n\r\n";
    }

    /**
     * 解析原始响应头。
     *
     * @param string $rawHeader
     * @return array
     */
    private function parseRawResponseHeaders(string $rawHeader): array
    {
        $lines = preg_split("/\r\n/", trim($rawHeader));
        $statusLine = (string)array_shift($lines);
        if (!preg_match('#^HTTP/\d(?:\.\d)?\s+(\d{3})(?:\s+(.+))?#i', $statusLine, $matches)) {
            return [
                'status'  => 0,
                'reason'  => 'Invalid response',
                'headers' => [],
            ];
        }

        $headers = [];
        foreach ($lines as $line) {
            if ($line === '' || !str_contains($line, ':')) {
                continue;
            }
            [$name, $value] = explode(':', $line, 2);
            $headers[strtolower(trim($name))][] = trim($value);
        }

        return [
            'status'  => (int)$matches[1],
            'reason'  => trim((string)($matches[2] ?? '')),
            'headers' => $headers,
        ];
    }

    /**
     * 判断响应头中是否包含指定 token。
     *
     * @param array $headers
     * @param string $name
     * @param string $token
     * @return bool
     */
    private function hasHeaderToken(array $headers, string $name, string $token): bool
    {
        foreach ($headers[strtolower($name)] ?? [] as $value) {
            foreach (explode(',', strtolower((string)$value)) as $item) {
                if (trim($item) === strtolower($token)) {
                    return true;
                }
            }
        }

        return false;
    }

    /**
     * 判断请求头是否安全。
     *
     * @param string $name
     * @param string $value
     * @return bool
     */
    private function isSafeHeaderLine(string $name, string $value): bool
    {
        return $name !== '' && strpbrk($name, ":\r\n") === false && strpbrk($value, "\r\n") === false;
    }

    /**
     * 构建下载文件名响应头。
     *
     * @param string $url
     * @param string $contentType
     * @return string
     */
    private function resolveContentDisposition(string $url, string $contentType): string
    {
        $extension = $this->guessExtension($url, $contentType) ?: '.bin';
        return 'attachment; filename="media' . $extension . '"';
    }

    /**
     * 从响应头解析资源总字节数。
     *
     * @param ResponseInterface $source
     * @return int
     */
    private function resolveTotalBytes(ResponseInterface $source): int
    {
        $contentRange = $source->getHeaderLine('Content-Range');
        if ($contentRange !== '' && preg_match('/\/(\d+)$/', $contentRange, $matches)) {
            return (int)$matches[1];
        }

        $contentLength = trim($source->getHeaderLine('Content-Length'));
        return ctype_digit($contentLength) ? (int)$contentLength : 0;
    }

    /**
     * 猜测媒体扩展名。
     *
     * @param string $url
     * @param string $contentType
     * @return string
     */
    private function guessExtension(string $url, string $contentType = ''): string
    {
        $path = strtolower((string)(parse_url($url, PHP_URL_PATH) ?: ''));
        if (preg_match('/\.(mp4|mov|m4v|jpg|jpeg|png|gif|webp)$/', $path, $matches)) {
            return $matches[1] === 'jpeg' ? '.jpg' : '.' . $matches[1];
        }

        $mimeType = strtolower(trim(explode(';', $contentType)[0] ?? ''));
        return match ($mimeType) {
            'video/mp4' => '.mp4',
            'video/quicktime' => '.mov',
            'image/jpeg' => '.jpg',
            'image/png' => '.png',
            'image/gif' => '.gif',
            'image/webp' => '.webp',
            default => '',
        };
    }

    /**
     * 从 URL 猜测 Content-Type
     *
     * @param string $url
     * @return string|null
     */
    private function guessContentTypeFromUrl(string $url): ?string
    {
        $ext = $this->guessExtension($url);
        if ($ext === '') {
            return null;
        }
        return match ($ext) {
            '.mp4'  => 'video/mp4',
            '.mov'  => 'video/quicktime',
            '.jpg'  => 'image/jpeg',
            '.jpeg' => 'image/jpeg',
            '.png'  => 'image/png',
            '.gif'  => 'image/gif',
            '.webp' => 'image/webp',
            default => null,
        };
    }

    /**
     * 判断下载域名是否禁止访问。
     *
     * @param string $host
     * @return bool
     */
    private function isBlockedHost(string $host): bool
    {
        $host = trim($host, '[]');
        if (in_array(strtolower($host), ['localhost', 'localhost.localdomain'], true)) {
            return true;
        }

        if (filter_var($host, FILTER_VALIDATE_IP)) {
            return !filter_var($host, FILTER_VALIDATE_IP, FILTER_FLAG_NO_PRIV_RANGE | FILTER_FLAG_NO_RES_RANGE);
        }

        $records = @dns_get_record($host, DNS_A + DNS_AAAA);
        if (!$records) {
            return false;
        }

        foreach ($records as $record) {
            $ip = $record['ip'] ?? $record['ipv6'] ?? '';
            if ($ip !== '' && !filter_var($ip, FILTER_VALIDATE_IP, FILTER_FLAG_NO_PRIV_RANGE | FILTER_FLAG_NO_RES_RANGE)) {
                return true;
            }
        }

        return false;
    }

    /**
     * 记录上游错误。
     *
     * @param string $url
     * @param ResponseInterface $response
     * @param string $message
     * @return void
     */
    private function logUpstreamError(string $url, ResponseInterface $response, string $message): void
    {
        Log::channel('media_downloader_response')->error($message, [
            'url_hash'      => md5($url),
            'status_code'   => $response->getStatusCode(),
            'reason_phrase' => $response->getReasonPhrase(),
        ]);
    }

    /**
     * 记录实时转发错误。
     *
     * @param string $message
     * @param string $url
     * @param array $context
     * @return void
     */
    private function logStreamError(string $message, string $url, array $context = []): void
    {
        Log::channel('media_downloader_response')->error($message, array_merge([
            'url_hash' => md5($url),
        ], $context));
    }
}

重现问题的步骤

操作系统环境及workerman/webman等具体版本

linux 最新版本