StreamHandler::add_progress() - Code Metrics - joenilson/residentes - Measure and Improve Code Quality continuously with Scrutinizer

StreamHandler::add_progress() A
last analyzed 2024-04-17 04:56 UTC

↳ Parent: StreamHandler

Complexity

Conditions	2
Paths	1

Size

Total Lines	9
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	1	Features	0

Metric	Value
eloc	5
c	1
b	1
f	0
dl	0
loc	9
rs	10
cc	2
nc	1
nop	4

<?php

namespace GuzzleHttp\Handler;

use GuzzleHttp\Exception\ConnectException;
use GuzzleHttp\Exception\RequestException;
use GuzzleHttp\Promise as P;
use GuzzleHttp\Promise\FulfilledPromise;
use GuzzleHttp\Promise\PromiseInterface;
use GuzzleHttp\Psr7;
use GuzzleHttp\TransferStats;
use GuzzleHttp\Utils;
use Psr\Http\Message\RequestInterface;
use Psr\Http\Message\ResponseInterface;
use Psr\Http\Message\StreamInterface;
use Psr\Http\Message\UriInterface;

/**
 * HTTP handler that uses PHP's HTTP stream wrapper.
 *
 * @final
 */
class StreamHandler
{
    /**
     * @var array
     */
    private $lastHeaders = [];

    /**
     * Sends an HTTP request.
     *
     * @param RequestInterface $request Request to send.
     * @param array            $options Request transfer options.
     */
    public function __invoke(RequestInterface $request, array $options): PromiseInterface
    {
        // Sleep if there is a delay specified.
        if (isset($options['delay'])) {
            \usleep($options['delay'] * 1000);
        }

        $startTime = isset($options['on_stats']) ? Utils::currentTime() : null;

        try {
            // Does not support the expect header.
            $request = $request->withoutHeader('Expect');

            // Append a content-length header if body size is zero to match
            // cURL's behavior.
            if (0 === $request->getBody()->getSize()) {
                $request = $request->withHeader('Content-Length', '0');
            }

            return $this->createResponse(
                $request,
                $options,
                $this->createStream($request, $options),
                $startTime
            );
        } catch (\InvalidArgumentException $e) {
            throw $e;
        } catch (\Exception $e) {
            // Determine if the error was a networking error.
            $message = $e->getMessage();
            // This list can probably get more comprehensive.
            if (false !== \strpos($message, 'getaddrinfo') // DNS lookup failed
                || false !== \strpos($message, 'Connection refused')
                || false !== \strpos($message, "couldn't connect to host") // error on HHVM
                || false !== \strpos($message, "connection attempt failed")
            ) {
                $e = new ConnectException($e->getMessage(), $request, $e);
            } else {
                $e = RequestException::wrapException($request, $e);
            }
            $this->invokeStats($options, $request, $startTime, null, $e);

            return P\Create::rejectionFor($e);
        }
    }

    private function invokeStats(
        array $options,
        RequestInterface $request,
        ?float $startTime,
        ResponseInterface $response = null,
        \Throwable $error = null
    ): void {
        if (isset($options['on_stats'])) {
            $stats = new TransferStats($request, $response, Utils::currentTime() - $startTime, $error, []);
            ($options['on_stats'])($stats);
        }
    }

    /**
     * @param resource $stream
     */
    private function createResponse(RequestInterface $request, array $options, $stream, ?float $startTime): PromiseInterface
    {
        $hdrs = $this->lastHeaders;
        $this->lastHeaders = [];

        try {
            [$ver, $status, $reason, $headers] = HeaderProcessor::parseHeaders($hdrs);
        } catch (\Exception $e) {
            return P\Create::rejectionFor(
                new RequestException('An error was encountered while creating the response', $request, null, $e)
            );
        }

        [$stream, $headers] = $this->checkDecode($options, $headers, $stream);
        $stream = Psr7\Utils::streamFor($stream);
        $sink = $stream;

        if (\strcasecmp('HEAD', $request->getMethod())) {
            $sink = $this->createSink($stream, $options);
        }

        try {
            $response = new Psr7\Response($status, $headers, $sink, $ver, $reason);
        } catch (\Exception $e) {
            return P\Create::rejectionFor(
                new RequestException('An error was encountered while creating the response', $request, null, $e)
            );
        }

        if (isset($options['on_headers'])) {
            try {
                $options['on_headers']($response);
            } catch (\Exception $e) {
                return P\Create::rejectionFor(
                    new RequestException('An error was encountered during the on_headers event', $request, $response, $e)
                );
            }
        }

        // Do not drain when the request is a HEAD request because they have
        // no body.
        if ($sink !== $stream) {
            $this->drain($stream, $sink, $response->getHeaderLine('Content-Length'));
        }

        $this->invokeStats($options, $request, $startTime, $response, null);

        return new FulfilledPromise($response);
    }

    private function createSink(StreamInterface $stream, array $options): StreamInterface
    {
        if (!empty($options['stream'])) {
            return $stream;
        }

        $sink = $options['sink'] ?? Psr7\Utils::tryFopen('php://temp', 'r+');

        return \is_string($sink) ? new Psr7\LazyOpenStream($sink, 'w+') : Psr7\Utils::streamFor($sink);
    }

    /**
     * @param resource $stream
     */
    private function checkDecode(array $options, array $headers, $stream): array
    {
        // Automatically decode responses when instructed.
        if (!empty($options['decode_content'])) {
            $normalizedKeys = Utils::normalizeHeaderKeys($headers);
            if (isset($normalizedKeys['content-encoding'])) {
                $encoding = $headers[$normalizedKeys['content-encoding']];
                if ($encoding[0] === 'gzip' || $encoding[0] === 'deflate') {
                    $stream = new Psr7\InflateStream(Psr7\Utils::streamFor($stream));
                    $headers['x-encoded-content-encoding'] = $headers[$normalizedKeys['content-encoding']];

                    // Remove content-encoding header
                    unset($headers[$normalizedKeys['content-encoding']]);

                    // Fix content-length header
                    if (isset($normalizedKeys['content-length'])) {
                        $headers['x-encoded-content-length'] = $headers[$normalizedKeys['content-length']];
                        $length = (int) $stream->getSize();
                        if ($length === 0) {
                            unset($headers[$normalizedKeys['content-length']]);
                        } else {
                            $headers[$normalizedKeys['content-length']] = [$length];
                        }
                    }
                }
            }
        }

        return [$stream, $headers];
    }

    /**
     * Drains the source stream into the "sink" client option.
     *
     * @param string $contentLength Header specifying the amount of
     *                              data to read.
     *
     * @throws \RuntimeException when the sink option is invalid.
     */
    private function drain(StreamInterface $source, StreamInterface $sink, string $contentLength): StreamInterface
    {
        // If a content-length header is provided, then stop reading once
        // that number of bytes has been read. This can prevent infinitely
        // reading from a stream when dealing with servers that do not honor
        // Connection: Close headers.
        Psr7\Utils::copyToStream(
            $source,
            $sink,
            (\strlen($contentLength) > 0 && (int) $contentLength > 0) ? (int) $contentLength : -1
        );

        $sink->seek(0);
        $source->close();

        return $sink;
    }

    /**
     * Create a resource and check to ensure it was created successfully
     *
     * @param callable $callback Callable that returns stream resource
     *
     * @return resource
     *
     * @throws \RuntimeException on error
     */
    private function createResource(callable $callback)
    {
        $errors = [];
        \set_error_handler(static function ($_, $msg, $file, $line) use (&$errors): bool {
            $errors[] = [
                'message' => $msg,
                'file'    => $file,
                'line'    => $line
            ];
            return true;
        });

        try {
            $resource = $callback();
        } finally {
            \restore_error_handler();
        }

        if (!$resource) {
            $message = 'Error creating resource: ';
            foreach ($errors as $err) {
                foreach ($err as $key => $value) {
                    $message .= "[$key] $value" . \PHP_EOL;
                }
            }
            throw new \RuntimeException(\trim($message));
        }

        return $resource;
    }

    /**
     * @return resource
     */
    private function createStream(RequestInterface $request, array $options)
    {
        static $methods;
        if (!$methods) {
            $methods = \array_flip(\get_class_methods(__CLASS__));
        }

        if (!\in_array($request->getUri()->getScheme(), ['http', 'https'])) {
            throw new RequestException(\sprintf("The scheme '%s' is not supported.", $request->getUri()->getScheme()), $request);
        }

        // HTTP/1.1 streams using the PHP stream wrapper require a
        // Connection: close header
        if ($request->getProtocolVersion() == '1.1'
            && !$request->hasHeader('Connection')
        ) {
            $request = $request->withHeader('Connection', 'close');
        }

        // Ensure SSL is verified by default
        if (!isset($options['verify'])) {
            $options['verify'] = true;
        }

        $params = [];
        $context = $this->getDefaultContext($request);

        if (isset($options['on_headers']) && !\is_callable($options['on_headers'])) {
            throw new \InvalidArgumentException('on_headers must be callable');
        }

        if (!empty($options)) {
            foreach ($options as $key => $value) {
                $method = "add_{$key}";
                if (isset($methods[$method])) {
                    $this->{$method}($request, $context, $value, $params);
                }
            }
        }

        if (isset($options['stream_context'])) {
            if (!\is_array($options['stream_context'])) {
                throw new \InvalidArgumentException('stream_context must be an array');
            }
            $context = \array_replace_recursive($context, $options['stream_context']);
        }

        // Microsoft NTLM authentication only supported with curl handler
        if (isset($options['auth'][2]) && 'ntlm' === $options['auth'][2]) {
            throw new \InvalidArgumentException('Microsoft NTLM authentication only supported with curl handler');
        }

        $uri = $this->resolveHost($request, $options);

        $contextResource = $this->createResource(
            static function () use ($context, $params) {
                return \stream_context_create($context, $params);
            }
        );

        return $this->createResource(
            function () use ($uri, &$http_response_header, $contextResource, $context, $options, $request) {
                $resource = @\fopen((string) $uri, 'r', false, $contextResource);
                $this->lastHeaders = $http_response_header ?? [];

                if (false === $resource) {
                    throw new ConnectException(sprintf('Connection refused for URI %s', $uri), $request, null, $context);
                }

                if (isset($options['read_timeout'])) {
                    $readTimeout = $options['read_timeout'];
                    $sec = (int) $readTimeout;
                    $usec = ($readTimeout - $sec) * 100000;
                    \stream_set_timeout($resource, $sec, $usec);
                }

                return $resource;
            }
        );
    }

    private function resolveHost(RequestInterface $request, array $options): UriInterface
    {
        $uri = $request->getUri();

        if (isset($options['force_ip_resolve']) && !\filter_var($uri->getHost(), \FILTER_VALIDATE_IP)) {
            if ('v4' === $options['force_ip_resolve']) {
                $records = \dns_get_record($uri->getHost(), \DNS_A);
                if (false === $records || !isset($records[0]['ip'])) {
                    throw new ConnectException(\sprintf("Could not resolve IPv4 address for host '%s'", $uri->getHost()), $request);
                }
                return $uri->withHost($records[0]['ip']);
            }
            if ('v6' === $options['force_ip_resolve']) {
                $records = \dns_get_record($uri->getHost(), \DNS_AAAA);
                if (false === $records || !isset($records[0]['ipv6'])) {
                    throw new ConnectException(\sprintf("Could not resolve IPv6 address for host '%s'", $uri->getHost()), $request);
                }
                return $uri->withHost('[' . $records[0]['ipv6'] . ']');
            }
        }

        return $uri;
    }

    private function getDefaultContext(RequestInterface $request): array
    {
        $headers = '';
        foreach ($request->getHeaders() as $name => $value) {
            foreach ($value as $val) {
                $headers .= "$name: $val\r\n";
            }
        }

        $context = [
            'http' => [
                'method'           => $request->getMethod(),
                'header'           => $headers,
                'protocol_version' => $request->getProtocolVersion(),
                'ignore_errors'    => true,
                'follow_location'  => 0,
            ],
            'ssl' => [
                'peer_name' => $request->getUri()->getHost(),
            ],
        ];

        $body = (string) $request->getBody();

        if (!empty($body)) {
            $context['http']['content'] = $body;
            // Prevent the HTTP handler from adding a Content-Type header.
            if (!$request->hasHeader('Content-Type')) {
                $context['http']['header'] .= "Content-Type:\r\n";
            }
        }

        $context['http']['header'] = \rtrim($context['http']['header']);

        return $context;
    }

    /**
     * @param mixed $value as passed via Request transfer options.
     */
    private function add_proxy(RequestInterface $request, array &$options, $value, array &$params): void
    {
        $uri = null;

        if (!\is_array($value)) {
            $uri = $value;
        } else {
            $scheme = $request->getUri()->getScheme();
            if (isset($value[$scheme])) {
                if (!isset($value['no']) || !Utils::isHostInNoProxy($request->getUri()->getHost(), $value['no'])) {
                    $uri = $value[$scheme];
                }
            }
        }

        if (!$uri) {
            return;
        }

        $parsed = $this->parse_proxy($uri);
        $options['http']['proxy'] = $parsed['proxy'];

        if ($parsed['auth']) {
            if (!isset($options['http']['header'])) {
                $options['http']['header'] = [];
            }
            $options['http']['header'] .= "\r\nProxy-Authorization: {$parsed['auth']}";
        }
    }

    /**
     * Parses the given proxy URL to make it compatible with the format PHP's stream context expects.
     */
    private function parse_proxy(string $url): array
    {
        $parsed = \parse_url($url);

        if ($parsed !== false && isset($parsed['scheme']) && $parsed['scheme'] === 'http') {
            if (isset($parsed['host']) && isset($parsed['port'])) {
                $auth = null;
                if (isset($parsed['user']) && isset($parsed['pass'])) {
                    $auth = \base64_encode("{$parsed['user']}:{$parsed['pass']}");
                }

                return [
                    'proxy' => "tcp://{$parsed['host']}:{$parsed['port']}",
                    'auth' => $auth ? "Basic {$auth}" : null,
                ];
            }
        }

        // Return proxy as-is.
        return [
            'proxy' => $url,
            'auth' => null,
        ];
    }

    /**
     * @param mixed $value as passed via Request transfer options.
     */
    private function add_timeout(RequestInterface $request, array &$options, $value, array &$params): void
    {
        if ($value > 0) {
            $options['http']['timeout'] = $value;
        }
    }

    /**
     * @param mixed $value as passed via Request transfer options.
     */
    private function add_verify(RequestInterface $request, array &$options, $value, array &$params): void
    {
        if ($value === false) {
            $options['ssl']['verify_peer'] = false;
            $options['ssl']['verify_peer_name'] = false;

            return;
        }

        if (\is_string($value)) {
            $options['ssl']['cafile'] = $value;
            if (!\file_exists($value)) {
                throw new \RuntimeException("SSL CA bundle not found: $value");
            }
        } elseif ($value !== true) {
            throw new \InvalidArgumentException('Invalid verify request option');
        }

        $options['ssl']['verify_peer'] = true;
        $options['ssl']['verify_peer_name'] = true;
        $options['ssl']['allow_self_signed'] = false;
    }

    /**
     * @param mixed $value as passed via Request transfer options.
     */
    private function add_cert(RequestInterface $request, array &$options, $value, array &$params): void
    {
        if (\is_array($value)) {
            $options['ssl']['passphrase'] = $value[1];
            $value = $value[0];
        }

        if (!\file_exists($value)) {
            throw new \RuntimeException("SSL certificate not found: {$value}");
        }

        $options['ssl']['local_cert'] = $value;
    }

    /**
     * @param mixed $value as passed via Request transfer options.
     */
    private function add_progress(RequestInterface $request, array &$options, $value, array &$params): void
    {
        self::addNotification(
            $params,
            static function ($code, $a, $b, $c, $transferred, $total) use ($value) {
                if ($code == \STREAM_NOTIFY_PROGRESS) {
                    // The upload progress cannot be determined. Use 0 for cURL compatibility:
                    // https://curl.se/libcurl/c/CURLOPT_PROGRESSFUNCTION.html
                    $value($total, $transferred, 0, 0);
                }
            }
        );
    }

    /**
     * @param mixed $value as passed via Request transfer options.
     */
    private function add_debug(RequestInterface $request, array &$options, $value, array &$params): void
    {
        if ($value === false) {
            return;
        }

        static $map = [
            \STREAM_NOTIFY_CONNECT       => 'CONNECT',
            \STREAM_NOTIFY_AUTH_REQUIRED => 'AUTH_REQUIRED',
            \STREAM_NOTIFY_AUTH_RESULT   => 'AUTH_RESULT',
            \STREAM_NOTIFY_MIME_TYPE_IS  => 'MIME_TYPE_IS',
            \STREAM_NOTIFY_FILE_SIZE_IS  => 'FILE_SIZE_IS',
            \STREAM_NOTIFY_REDIRECTED    => 'REDIRECTED',
            \STREAM_NOTIFY_PROGRESS      => 'PROGRESS',
            \STREAM_NOTIFY_FAILURE       => 'FAILURE',
            \STREAM_NOTIFY_COMPLETED     => 'COMPLETED',
            \STREAM_NOTIFY_RESOLVE       => 'RESOLVE',
        ];
        static $args = ['severity', 'message', 'message_code', 'bytes_transferred', 'bytes_max'];

        $value = Utils::debugResource($value);
        $ident = $request->getMethod() . ' ' . $request->getUri()->withFragment('');
        self::addNotification(
            $params,
            static function (int $code, ...$passed) use ($ident, $value, $map, $args): void {
                \fprintf($value, '<%s> [%s] ', $ident, $map[$code]);
                foreach (\array_filter($passed) as $i => $v) {
                    \fwrite($value, $args[$i] . ': "' . $v . '" ');
                }
                \fwrite($value, "\n");
            }
        );
    }

    private static function addNotification(array &$params, callable $notify): void
    {
        // Wrap the existing function if needed.
        if (!isset($params['notification'])) {
            $params['notification'] = $notify;
        } else {
            $params['notification'] = self::callArray([
                $params['notification'],
                $notify
            ]);
        }
    }

    private static function callArray(array $functions): callable
    {
        return static function (...$args) use ($functions) {
            foreach ($functions as $fn) {
                $fn(...$args);
            }
        };
    }
}


1			<?php
2
3			namespace GuzzleHttp\Handler;
4
5			use GuzzleHttp\Exception\ConnectException;
6			use GuzzleHttp\Exception\RequestException;
7			use GuzzleHttp\Promise as P;
8			use GuzzleHttp\Promise\FulfilledPromise;
9			use GuzzleHttp\Promise\PromiseInterface;
10			use GuzzleHttp\Psr7;
11			use GuzzleHttp\TransferStats;
12			use GuzzleHttp\Utils;
13			use Psr\Http\Message\RequestInterface;
14			use Psr\Http\Message\ResponseInterface;
15			use Psr\Http\Message\StreamInterface;
16			use Psr\Http\Message\UriInterface;
17
18			/**
19			* HTTP handler that uses PHP's HTTP stream wrapper.
20			*
21			* @final
22			*/
23			class StreamHandler
24			{
25			/**
26			* @var array
27			*/
28			private $lastHeaders = [];
29
30			/**
31			* Sends an HTTP request.
32			*
33			* @param RequestInterface $request Request to send.
34			* @param array $options Request transfer options.
35			*/
36			public function __invoke(RequestInterface $request, array $options): PromiseInterface
37			{
38			// Sleep if there is a delay specified.
39			if (isset($options['delay'])) {
40			\usleep($options['delay'] * 1000);
41			}
42
43			$startTime = isset($options['on_stats']) ? Utils::currentTime() : null;
44
45			try {
46			// Does not support the expect header.
47			$request = $request->withoutHeader('Expect');
48
49			// Append a content-length header if body size is zero to match
50			// cURL's behavior.
51			if (0 === $request->getBody()->getSize()) {
52			$request = $request->withHeader('Content-Length', '0');
53			}
54
55			return $this->createResponse(
56			$request,
57			$options,
58			$this->createStream($request, $options),
59			$startTime
60			);
61			} catch (\InvalidArgumentException $e) {
62			throw $e;
63			} catch (\Exception $e) {
64			// Determine if the error was a networking error.
65			$message = $e->getMessage();
66			// This list can probably get more comprehensive.
67			if (false !== \strpos($message, 'getaddrinfo') // DNS lookup failed
68			\|\| false !== \strpos($message, 'Connection refused')
69			\|\| false !== \strpos($message, "couldn't connect to host") // error on HHVM
70			\|\| false !== \strpos($message, "connection attempt failed")
71			) {
72			$e = new ConnectException($e->getMessage(), $request, $e);
73			} else {
74			$e = RequestException::wrapException($request, $e);
75			}
76			$this->invokeStats($options, $request, $startTime, null, $e);
77
78			return P\Create::rejectionFor($e);
79			}
80			}
81
82			private function invokeStats(
83			array $options,
84			RequestInterface $request,
85			?float $startTime,
86			ResponseInterface $response = null,
87			\Throwable $error = null
88			): void {
89			if (isset($options['on_stats'])) {
90			$stats = new TransferStats($request, $response, Utils::currentTime() - $startTime, $error, []);
91			($options['on_stats'])($stats);
92			}
93			}
94
95			/**
96			* @param resource $stream
97			*/
98			private function createResponse(RequestInterface $request, array $options, $stream, ?float $startTime): PromiseInterface
99			{
100			$hdrs = $this->lastHeaders;
101			$this->lastHeaders = [];
102
103			try {
104			[$ver, $status, $reason, $headers] = HeaderProcessor::parseHeaders($hdrs);
105			} catch (\Exception $e) {
106			return P\Create::rejectionFor(
107			new RequestException('An error was encountered while creating the response', $request, null, $e)
108			);
109			}
110
111			[$stream, $headers] = $this->checkDecode($options, $headers, $stream);
112			$stream = Psr7\Utils::streamFor($stream);
113			$sink = $stream;
114
115			if (\strcasecmp('HEAD', $request->getMethod())) {
116			$sink = $this->createSink($stream, $options);
117			}
118
119			try {
120			$response = new Psr7\Response($status, $headers, $sink, $ver, $reason);
121			} catch (\Exception $e) {
122			return P\Create::rejectionFor(
123			new RequestException('An error was encountered while creating the response', $request, null, $e)
124			);
125			}
126
127			if (isset($options['on_headers'])) {
128			try {
129			$options['on_headers']($response);
130			} catch (\Exception $e) {
131			return P\Create::rejectionFor(
132			new RequestException('An error was encountered during the on_headers event', $request, $response, $e)
133			);
134			}
135			}
136
137			// Do not drain when the request is a HEAD request because they have
138			// no body.
139			if ($sink !== $stream) {
140			$this->drain($stream, $sink, $response->getHeaderLine('Content-Length'));
141			}
142
143			$this->invokeStats($options, $request, $startTime, $response, null);
144
145			return new FulfilledPromise($response);
146			}
147
148			private function createSink(StreamInterface $stream, array $options): StreamInterface
149			{
150			if (!empty($options['stream'])) {
151			return $stream;
152			}
153
154			$sink = $options['sink'] ?? Psr7\Utils::tryFopen('php://temp', 'r+');
155
156			return \is_string($sink) ? new Psr7\LazyOpenStream($sink, 'w+') : Psr7\Utils::streamFor($sink);
157			}
158
159			/**
160			* @param resource $stream
161			*/
162			private function checkDecode(array $options, array $headers, $stream): array
163			{
164			// Automatically decode responses when instructed.
165			if (!empty($options['decode_content'])) {
166			$normalizedKeys = Utils::normalizeHeaderKeys($headers);
167			if (isset($normalizedKeys['content-encoding'])) {
168			$encoding = $headers[$normalizedKeys['content-encoding']];
169			if ($encoding[0] === 'gzip' \|\| $encoding[0] === 'deflate') {
170			$stream = new Psr7\InflateStream(Psr7\Utils::streamFor($stream));
171			$headers['x-encoded-content-encoding'] = $headers[$normalizedKeys['content-encoding']];
172
173			// Remove content-encoding header
174			unset($headers[$normalizedKeys['content-encoding']]);
175
176			// Fix content-length header
177			if (isset($normalizedKeys['content-length'])) {
178			$headers['x-encoded-content-length'] = $headers[$normalizedKeys['content-length']];
179			$length = (int) $stream->getSize();
180			if ($length === 0) {
181			unset($headers[$normalizedKeys['content-length']]);
182			} else {
183			$headers[$normalizedKeys['content-length']] = [$length];
184			}
185			}
186			}
187			}
188			}
189
190			return [$stream, $headers];
191			}
192
193			/**
194			* Drains the source stream into the "sink" client option.
195			*
196			* @param string $contentLength Header specifying the amount of
197			* data to read.
198			*
199			* @throws \RuntimeException when the sink option is invalid.
200			*/
201			private function drain(StreamInterface $source, StreamInterface $sink, string $contentLength): StreamInterface
202			{
203			// If a content-length header is provided, then stop reading once
204			// that number of bytes has been read. This can prevent infinitely
205			// reading from a stream when dealing with servers that do not honor
206			// Connection: Close headers.
207			Psr7\Utils::copyToStream(
208			$source,
209			$sink,
210			(\strlen($contentLength) > 0 && (int) $contentLength > 0) ? (int) $contentLength : -1
211			);
212
213			$sink->seek(0);
214			$source->close();
215
216			return $sink;
217			}
218
219			/**
220			* Create a resource and check to ensure it was created successfully
221			*
222			* @param callable $callback Callable that returns stream resource
223			*
224			* @return resource
225			*
226			* @throws \RuntimeException on error
227			*/
228			private function createResource(callable $callback)
229			{
230			$errors = [];
231			\set_error_handler(static function ($_, $msg, $file, $line) use (&$errors): bool {
232			$errors[] = [
233			'message' => $msg,
234			'file' => $file,
235			'line' => $line
236			];
237			return true;
238			});
239
240			try {
241			$resource = $callback();
242			} finally {
243			\restore_error_handler();
244			}
245
246			if (!$resource) {
247			$message = 'Error creating resource: ';
248			foreach ($errors as $err) {
249			foreach ($err as $key => $value) {
250			$message .= "[$key] $value" . \PHP_EOL;
251			}
252			}
253			throw new \RuntimeException(\trim($message));
254			}
255
256			return $resource;
257			}
258
259			/**
260			* @return resource
261			*/
262			private function createStream(RequestInterface $request, array $options)
263			{
264			static $methods;
265			if (!$methods) {
266			$methods = \array_flip(\get_class_methods(__CLASS__));
267			}
268
269			if (!\in_array($request->getUri()->getScheme(), ['http', 'https'])) {
270			throw new RequestException(\sprintf("The scheme '%s' is not supported.", $request->getUri()->getScheme()), $request);
271			}
272
273			// HTTP/1.1 streams using the PHP stream wrapper require a
274			// Connection: close header
275			if ($request->getProtocolVersion() == '1.1'
276			&& !$request->hasHeader('Connection')
277			) {
278			$request = $request->withHeader('Connection', 'close');
279			}
280
281			// Ensure SSL is verified by default
282			if (!isset($options['verify'])) {
283			$options['verify'] = true;
284			}
285
286			$params = [];
287			$context = $this->getDefaultContext($request);
288
289			if (isset($options['on_headers']) && !\is_callable($options['on_headers'])) {
290			throw new \InvalidArgumentException('on_headers must be callable');
291			}
292
293			if (!empty($options)) {
294			foreach ($options as $key => $value) {
295			$method = "add_{$key}";
296			if (isset($methods[$method])) {
297			$this->{$method}($request, $context, $value, $params);
298			}
299			}
300			}
301
302			if (isset($options['stream_context'])) {
303			if (!\is_array($options['stream_context'])) {
304			throw new \InvalidArgumentException('stream_context must be an array');
305			}
306			$context = \array_replace_recursive($context, $options['stream_context']);
307			}
308
309			// Microsoft NTLM authentication only supported with curl handler
310			if (isset($options['auth'][2]) && 'ntlm' === $options['auth'][2]) {
311			throw new \InvalidArgumentException('Microsoft NTLM authentication only supported with curl handler');
312			}
313
314			$uri = $this->resolveHost($request, $options);
315
316			$contextResource = $this->createResource(
317			static function () use ($context, $params) {
318			return \stream_context_create($context, $params);
319			}
320			);
321
322			return $this->createResource(
323			function () use ($uri, &$http_response_header, $contextResource, $context, $options, $request) {
324			$resource = @\fopen((string) $uri, 'r', false, $contextResource);
325			$this->lastHeaders = $http_response_header ?? [];
326
327			if (false === $resource) {
328			throw new ConnectException(sprintf('Connection refused for URI %s', $uri), $request, null, $context);
329			}
330
331			if (isset($options['read_timeout'])) {
332			$readTimeout = $options['read_timeout'];
333			$sec = (int) $readTimeout;
334			$usec = ($readTimeout - $sec) * 100000;
335			\stream_set_timeout($resource, $sec, $usec);
336			}
337
338			return $resource;
339			}
340			);
341			}
342
343			private function resolveHost(RequestInterface $request, array $options): UriInterface
344			{
345			$uri = $request->getUri();
346
347			if (isset($options['force_ip_resolve']) && !\filter_var($uri->getHost(), \FILTER_VALIDATE_IP)) {
348			if ('v4' === $options['force_ip_resolve']) {
349			$records = \dns_get_record($uri->getHost(), \DNS_A);
350			if (false === $records \|\| !isset($records[0]['ip'])) {
351			throw new ConnectException(\sprintf("Could not resolve IPv4 address for host '%s'", $uri->getHost()), $request);
352			}
353			return $uri->withHost($records[0]['ip']);
354			}
355			if ('v6' === $options['force_ip_resolve']) {
356			$records = \dns_get_record($uri->getHost(), \DNS_AAAA);
357			if (false === $records \|\| !isset($records[0]['ipv6'])) {
358			throw new ConnectException(\sprintf("Could not resolve IPv6 address for host '%s'", $uri->getHost()), $request);
359			}
360			return $uri->withHost('[' . $records[0]['ipv6'] . ']');
361			}
362			}
363
364			return $uri;
365			}
366
367			private function getDefaultContext(RequestInterface $request): array
368			{
369			$headers = '';
370			foreach ($request->getHeaders() as $name => $value) {
371			foreach ($value as $val) {
372			$headers .= "$name: $val\r\n";
373			}
374			}
375
376			$context = [
377			'http' => [
378			'method' => $request->getMethod(),
379			'header' => $headers,
380			'protocol_version' => $request->getProtocolVersion(),
381			'ignore_errors' => true,
382			'follow_location' => 0,
383			],
384			'ssl' => [
385			'peer_name' => $request->getUri()->getHost(),
386			],
387			];
388
389			$body = (string) $request->getBody();
390
391			if (!empty($body)) {
392			$context['http']['content'] = $body;
393			// Prevent the HTTP handler from adding a Content-Type header.
394			if (!$request->hasHeader('Content-Type')) {
395			$context['http']['header'] .= "Content-Type:\r\n";
396			}
397			}
398
399			$context['http']['header'] = \rtrim($context['http']['header']);
400
401			return $context;
402			}
403
404			/**
405			* @param mixed $value as passed via Request transfer options.
406			*/
407			private function add_proxy(RequestInterface $request, array &$options, $value, array &$params): void
408			{
409			$uri = null;
410
411			if (!\is_array($value)) {
412			$uri = $value;
413			} else {
414			$scheme = $request->getUri()->getScheme();
415			if (isset($value[$scheme])) {
416			if (!isset($value['no']) \|\| !Utils::isHostInNoProxy($request->getUri()->getHost(), $value['no'])) {
417			$uri = $value[$scheme];
418			}
419			}
420			}
421
422			if (!$uri) {
423			return;
424			}
425
426			$parsed = $this->parse_proxy($uri);
427			$options['http']['proxy'] = $parsed['proxy'];
428
429			if ($parsed['auth']) {
430			if (!isset($options['http']['header'])) {
431			$options['http']['header'] = [];
432			}
433			$options['http']['header'] .= "\r\nProxy-Authorization: {$parsed['auth']}";
434			}
435			}
436
437			/**
438			* Parses the given proxy URL to make it compatible with the format PHP's stream context expects.
439			*/
440			private function parse_proxy(string $url): array
441			{
442			$parsed = \parse_url($url);
443
444			if ($parsed !== false && isset($parsed['scheme']) && $parsed['scheme'] === 'http') {
445			if (isset($parsed['host']) && isset($parsed['port'])) {
446			$auth = null;
447			if (isset($parsed['user']) && isset($parsed['pass'])) {
448			$auth = \base64_encode("{$parsed['user']}:{$parsed['pass']}");
449			}
450
451			return [
452			'proxy' => "tcp://{$parsed['host']}:{$parsed['port']}",
453			'auth' => $auth ? "Basic {$auth}" : null,
454			];
455			}
456			}
457
458			// Return proxy as-is.
459			return [
460			'proxy' => $url,
461			'auth' => null,
462			];
463			}
464
465			/**
466			* @param mixed $value as passed via Request transfer options.
467			*/
468			private function add_timeout(RequestInterface $request, array &$options, $value, array &$params): void
469			{
470			if ($value > 0) {
471			$options['http']['timeout'] = $value;
472			}
473			}
474
475			/**
476			* @param mixed $value as passed via Request transfer options.
477			*/
478			private function add_verify(RequestInterface $request, array &$options, $value, array &$params): void
479			{
480			if ($value === false) {
481			$options['ssl']['verify_peer'] = false;
482			$options['ssl']['verify_peer_name'] = false;
483
484			return;
485			}
486
487			if (\is_string($value)) {
488			$options['ssl']['cafile'] = $value;
489			if (!\file_exists($value)) {
490			throw new \RuntimeException("SSL CA bundle not found: $value");
491			}
492			} elseif ($value !== true) {
493			throw new \InvalidArgumentException('Invalid verify request option');
494			}
495
496			$options['ssl']['verify_peer'] = true;
497			$options['ssl']['verify_peer_name'] = true;
498			$options['ssl']['allow_self_signed'] = false;
499			}
500
501			/**
502			* @param mixed $value as passed via Request transfer options.
503			*/
504			private function add_cert(RequestInterface $request, array &$options, $value, array &$params): void
505			{
506			if (\is_array($value)) {
507			$options['ssl']['passphrase'] = $value[1];
508			$value = $value[0];
509			}
510
511			if (!\file_exists($value)) {
512			throw new \RuntimeException("SSL certificate not found: {$value}");
513			}
514
515			$options['ssl']['local_cert'] = $value;
516			}
517
518			/**
519			* @param mixed $value as passed via Request transfer options.
520			*/
521			private function add_progress(RequestInterface $request, array &$options, $value, array &$params): void
522			{
523			self::addNotification(
524			$params,
525			static function ($code, $a, $b, $c, $transferred, $total) use ($value) {
526			if ($code == \STREAM_NOTIFY_PROGRESS) {
527			// The upload progress cannot be determined. Use 0 for cURL compatibility:
528			// https://curl.se/libcurl/c/CURLOPT_PROGRESSFUNCTION.html
529			$value($total, $transferred, 0, 0);
530			}
531			}
532			);
533			}
534
535			/**
536			* @param mixed $value as passed via Request transfer options.
537			*/
538			private function add_debug(RequestInterface $request, array &$options, $value, array &$params): void
539			{
540			if ($value === false) {
541			return;
542			}
543
544			static $map = [
545			\STREAM_NOTIFY_CONNECT => 'CONNECT',
546			\STREAM_NOTIFY_AUTH_REQUIRED => 'AUTH_REQUIRED',
547			\STREAM_NOTIFY_AUTH_RESULT => 'AUTH_RESULT',
548			\STREAM_NOTIFY_MIME_TYPE_IS => 'MIME_TYPE_IS',
549			\STREAM_NOTIFY_FILE_SIZE_IS => 'FILE_SIZE_IS',
550			\STREAM_NOTIFY_REDIRECTED => 'REDIRECTED',
551			\STREAM_NOTIFY_PROGRESS => 'PROGRESS',
552			\STREAM_NOTIFY_FAILURE => 'FAILURE',
553			\STREAM_NOTIFY_COMPLETED => 'COMPLETED',
554			\STREAM_NOTIFY_RESOLVE => 'RESOLVE',
555			];
556			static $args = ['severity', 'message', 'message_code', 'bytes_transferred', 'bytes_max'];
557
558			$value = Utils::debugResource($value);
559			$ident = $request->getMethod() . ' ' . $request->getUri()->withFragment('');
560			self::addNotification(
561			$params,
562			static function (int $code, ...$passed) use ($ident, $value, $map, $args): void {
563			\fprintf($value, '<%s> [%s] ', $ident, $map[$code]);
564			foreach (\array_filter($passed) as $i => $v) {
565			\fwrite($value, $args[$i] . ': "' . $v . '" ');
566			}
567			\fwrite($value, "\n");
568			}
569			);
570			}
571
572			private static function addNotification(array &$params, callable $notify): void
573			{
574			// Wrap the existing function if needed.
575			if (!isset($params['notification'])) {
576			$params['notification'] = $notify;
577			} else {
578			$params['notification'] = self::callArray([
579			$params['notification'],
580			$notify
581			]);
582			}
583			}
584
585			private static function callArray(array $functions): callable
586			{
587			return static function (...$args) use ($functions) {
588			foreach ($functions as $fn) {
589			$fn(...$args);
590			}
591			};
592			}
593			}
594

joenilson / residentes

StreamHandler::add_progress() A last analyzed 2024-04-17 04:56 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

StreamHandler::add_progress() A
last analyzed 2024-04-17 04:56 UTC