15269011225

Committed 27 May 2025 07:17AM UTC coverage: 92.653% (-0.04%) from 92.689%

Build # 15269011225

Build Type

push

github

Committed by

web-flow

Commit Message

Backport #1503 (#1545)

* Backport #1503

* Update CHANGELOG.md

---------

Co-authored-by: Till Krüss <tillkruss@users.noreply.github.com>

Run Details

0 of 4 new or added lines in 1 file covered. (0.0%)

1 existing line in 1 file now uncovered.

7150 of 7717 relevant lines covered (92.65%)

113.59 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

60.0

/src/Connection/Resource/StreamFactory.php

<?php

/*
 * This file is part of the Predis package.
 *
 * (c) 2009-2020 Daniele Alessandri
 * (c) 2021-2025 Till Krüss
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Predis\Connection\Resource;

use InvalidArgumentException;
use Predis\Connection\ParametersInterface;
use Predis\Connection\Resource\Exception\StreamInitException;
use Psr\Http\Message\StreamInterface;

class StreamFactory implements StreamFactoryInterface
{
    /**
     * {@inheritDoc}
     * @throws StreamInitException
     */
    public function createStream(ParametersInterface $parameters): StreamInterface
    {
        $parameters = $this->assertParameters($parameters);

        switch ($parameters->scheme) {
            case 'tcp':
            case 'redis':
                $stream = $this->tcpStreamInitializer($parameters);
                break;

            case 'unix':
                $stream = $this->unixStreamInitializer($parameters);
                break;

            case 'tls':
            case 'rediss':
                $stream = $this->tlsStreamInitializer($parameters);
                break;

            default:
                throw new InvalidArgumentException("Invalid scheme: '{$parameters->scheme}'.");
        }

        return new Stream($stream);
    }

    /**
     * Checks some parameters used to initialize the connection.
     *
     * @param ParametersInterface $parameters Initialization parameters for the connection.
     *
     * @return ParametersInterface
     * @throws InvalidArgumentException
     */
    protected function assertParameters(ParametersInterface $parameters): ParametersInterface
    {
        switch ($parameters->scheme) {
            case 'tcp':
            case 'redis':
            case 'unix':
            case 'tls':
            case 'rediss':
                break;

            default:
                throw new InvalidArgumentException("Invalid scheme: '$parameters->scheme'.");
        }

        return $parameters;
    }

    /**
     * Initializes a TCP stream resource.
     *
     * @param ParametersInterface $parameters Initialization parameters for the connection.
     *
     * @return resource
     * @throws StreamInitException
     */
    protected function tcpStreamInitializer(ParametersInterface $parameters)
    {
        if (!filter_var($parameters->host, FILTER_VALIDATE_IP, FILTER_FLAG_IPV6)) {
            $address = "tcp://$parameters->host:$parameters->port";
        } else {
            $address = "tcp://[$parameters->host]:$parameters->port";
        }

        $flags = STREAM_CLIENT_CONNECT;

        if (isset($parameters->async_connect) && $parameters->async_connect) {
            $flags |= STREAM_CLIENT_ASYNC_CONNECT;
        }

        if (isset($parameters->persistent)) {
            if (false !== $persistent = filter_var($parameters->persistent, FILTER_VALIDATE_BOOLEAN, FILTER_NULL_ON_FAILURE)) {
                $flags |= STREAM_CLIENT_PERSISTENT;

                if ($persistent === null) {
                    $address = "{$address}/{$parameters->persistent}";
                }
            }
        }

        return $this->createStreamSocket($parameters, $address, $flags);
    }

    /**
     * Initializes a UNIX stream resource.
     *
     * @param ParametersInterface $parameters Initialization parameters for the connection.
     *
     * @return resource
     * @throws StreamInitException
     */
    protected function unixStreamInitializer(ParametersInterface $parameters)
    {
        if (!isset($parameters->path)) {
            throw new InvalidArgumentException('Missing UNIX domain socket path.');
        }

        $flags = STREAM_CLIENT_CONNECT;

        if (isset($parameters->persistent)) {
            if (false !== $persistent = filter_var($parameters->persistent, FILTER_VALIDATE_BOOLEAN, FILTER_NULL_ON_FAILURE)) {
                $flags |= STREAM_CLIENT_PERSISTENT;

                if ($persistent === null) {
                    throw new InvalidArgumentException(
                        'Persistent connection IDs are not supported when using UNIX domain sockets.'
                    );
                }
            }
        }

        return $this->createStreamSocket($parameters, "unix://{$parameters->path}", $flags);
    }

    /**
     * Initializes a SSL-encrypted TCP stream resource.
     *
     * @param ParametersInterface $parameters Initialization parameters for the connection.
     *
     * @return resource
     * @throws StreamInitException
     */
    protected function tlsStreamInitializer(ParametersInterface $parameters)
    {
        $resource = $this->tcpStreamInitializer($parameters);
        $metadata = stream_get_meta_data($resource);

        // Detect if crypto mode is already enabled for this stream (PHP >= 7.0.0).
        if (isset($metadata['crypto'])) {
            return $resource;
        }

        if (isset($parameters->ssl) && is_array($parameters->ssl)) {
            $options = $parameters->ssl;
        } else {
            $options = [];
        }

        if (!isset($options['crypto_type'])) {
            $options['crypto_type'] = STREAM_CRYPTO_METHOD_TLS_CLIENT;
        }

        $context_options = function_exists('stream_context_set_options')
            ? stream_context_set_options($resource, ['ssl' => $options])
            : stream_context_set_option($resource, ['ssl' => $options]);

        if (!$context_options) {
            $this->onInitializationError($resource, $parameters, 'Error while setting SSL context options');
        }

        if (!stream_socket_enable_crypto($resource, true, $options['crypto_type'])) {
            $this->onInitializationError($resource, $parameters, 'Error while switching to encrypted communication');
        }

        return $resource;
    }

    /**
     * Creates a connected stream socket resource.
     *
     * @param ParametersInterface $parameters Connection parameters.
     * @param string              $address    Address for stream_socket_client().
     * @param int                 $flags      Flags for stream_socket_client().
     *
     * @return resource
     * @throws StreamInitException
     */
    protected function createStreamSocket(ParametersInterface $parameters, $address, $flags)
    {
        $timeout = (isset($parameters->timeout) ? (float) $parameters->timeout : 5.0);
        $context = stream_context_create(['socket' => ['tcp_nodelay' => (bool) $parameters->tcp_nodelay]]);

        if (
            (isset($parameters->persistent) && $parameters->persistent)
            && (isset($parameters->conn_uid) && $parameters->conn_uid)
        ) {
            $conn_uid = '/' . $parameters->conn_uid;
        } else {
            $conn_uid = '';
        }

        // Needs to create multiple persistent connections to the same resource
        $address = $address . $conn_uid;

        if (!$resource = @stream_socket_client($address, $errno, $errstr, $timeout, $flags, $context)) {
            $this->onInitializationError($resource, $parameters, trim($errstr), $errno);
        }

        if (isset($parameters->read_write_timeout)) {
            $rwtimeout = (float) $parameters->read_write_timeout;
            $rwtimeout = $rwtimeout > 0 ? $rwtimeout : -1;
            $timeoutSeconds = floor($rwtimeout);
            $timeoutUSeconds = ($rwtimeout - $timeoutSeconds) * 1000000;
            stream_set_timeout($resource, $timeoutSeconds, $timeoutUSeconds);
        }

        return $resource;
    }

    /**
     * Helper method to handle connection errors.
     *
     * @param  string              $message Error message.
     * @param  int                 $code    Error code.
     * @throws StreamInitException
     */
    protected function onInitializationError($stream, ParametersInterface $parameters, string $message, int $code = 0): void
    {
        if (is_resource($stream)) {
            fclose($stream);
        }

        throw new StreamInitException("$message [{$parameters}]", $code);
    }
}

1	<?php
2
3	/*
4	* This file is part of the Predis package.
5	*
6	* (c) 2009-2020 Daniele Alessandri
7	* (c) 2021-2025 Till Krüss
8	*
9	* For the full copyright and license information, please view the LICENSE
10	* file that was distributed with this source code.
11	*/
12
13	namespace Predis\Connection\Resource;
14
15	use InvalidArgumentException;
16	use Predis\Connection\ParametersInterface;
17	use Predis\Connection\Resource\Exception\StreamInitException;
18	use Psr\Http\Message\StreamInterface;
19
20	class StreamFactory implements StreamFactoryInterface
21	{
22	/**
23	* {@inheritDoc}
24	* @throws StreamInitException
25	*/
26	public function createStream(ParametersInterface $parameters): StreamInterface	1,525✔
27	{
28	$parameters = $this->assertParameters($parameters);	1,525✔
29
30	switch ($parameters->scheme) {	1,525✔
31	case 'tcp':	1,525✔
32	case 'redis':	1✔
33	$stream = $this->tcpStreamInitializer($parameters);	1,524✔
34	break;	1,522✔
35
36	case 'unix':	1✔
37	$stream = $this->unixStreamInitializer($parameters);	1✔
38	break;	×
39
40	case 'tls':	×
41	case 'rediss':	×
42	$stream = $this->tlsStreamInitializer($parameters);	×
43	break;	×
44
45	default:
46	throw new InvalidArgumentException("Invalid scheme: '{$parameters->scheme}'.");	×
47	}
48
49	return new Stream($stream);	1,522✔
50	}
51
52	/**
53	* Checks some parameters used to initialize the connection.
54	*
55	* @param ParametersInterface $parameters Initialization parameters for the connection.
56	*
57	* @return ParametersInterface
58	* @throws InvalidArgumentException
59	*/
60	protected function assertParameters(ParametersInterface $parameters): ParametersInterface	1,525✔
61	{
62	switch ($parameters->scheme) {	1,525✔
63	case 'tcp':	1,525✔
64	case 'redis':	1✔
65	case 'unix':	1✔
66	case 'tls':	×
67	case 'rediss':	×
68	break;	1,525✔
69
70	default:
71	throw new InvalidArgumentException("Invalid scheme: '$parameters->scheme'.");	×
72	}
73
74	return $parameters;	1,525✔
75	}
76
77	/**
78	* Initializes a TCP stream resource.
79	*
80	* @param ParametersInterface $parameters Initialization parameters for the connection.
81	*
82	* @return resource
83	* @throws StreamInitException
84	*/
85	protected function tcpStreamInitializer(ParametersInterface $parameters)	1,524✔
86	{
87	if (!filter_var($parameters->host, FILTER_VALIDATE_IP, FILTER_FLAG_IPV6)) {	1,524✔
88	$address = "tcp://$parameters->host:$parameters->port";	1,523✔
89	} else {
90	$address = "tcp://[$parameters->host]:$parameters->port";	1✔
91	}
92
93	$flags = STREAM_CLIENT_CONNECT;	1,524✔
94
95	if (isset($parameters->async_connect) && $parameters->async_connect) {	1,524✔
96	$flags \|= STREAM_CLIENT_ASYNC_CONNECT;	×
97	}
98
99	if (isset($parameters->persistent)) {	1,524✔
100	if (false !== $persistent = filter_var($parameters->persistent, FILTER_VALIDATE_BOOLEAN, FILTER_NULL_ON_FAILURE)) {	10✔
101	$flags \|= STREAM_CLIENT_PERSISTENT;	8✔
102
103	if ($persistent === null) {	8✔
104	$address = "{$address}/{$parameters->persistent}";	2✔
105	}
106	}
107	}
108
109	return $this->createStreamSocket($parameters, $address, $flags);	1,524✔
110	}
111
112	/**
113	* Initializes a UNIX stream resource.
114	*
115	* @param ParametersInterface $parameters Initialization parameters for the connection.
116	*
117	* @return resource
118	* @throws StreamInitException
119	*/
120	protected function unixStreamInitializer(ParametersInterface $parameters)	1✔
121	{
122	if (!isset($parameters->path)) {	1✔
123	throw new InvalidArgumentException('Missing UNIX domain socket path.');	×
124	}
125
126	$flags = STREAM_CLIENT_CONNECT;	1✔
127
128	if (isset($parameters->persistent)) {	1✔
129	if (false !== $persistent = filter_var($parameters->persistent, FILTER_VALIDATE_BOOLEAN, FILTER_NULL_ON_FAILURE)) {	×
130	$flags \|= STREAM_CLIENT_PERSISTENT;	×
131
132	if ($persistent === null) {	×
133	throw new InvalidArgumentException(	×
134	'Persistent connection IDs are not supported when using UNIX domain sockets.'	×
135	);	×
136	}
137	}
138	}
139
140	return $this->createStreamSocket($parameters, "unix://{$parameters->path}", $flags);	1✔
141	}
142
143	/**
144	* Initializes a SSL-encrypted TCP stream resource.
145	*
146	* @param ParametersInterface $parameters Initialization parameters for the connection.
147	*
148	* @return resource
149	* @throws StreamInitException
150	*/
151	protected function tlsStreamInitializer(ParametersInterface $parameters)	×
152	{
153	$resource = $this->tcpStreamInitializer($parameters);	×
154	$metadata = stream_get_meta_data($resource);	×
155
156	// Detect if crypto mode is already enabled for this stream (PHP >= 7.0.0).
157	if (isset($metadata['crypto'])) {	×
158	return $resource;	×
159	}
160
161	if (isset($parameters->ssl) && is_array($parameters->ssl)) {	×
162	$options = $parameters->ssl;	×
163	} else {
164	$options = [];	×
165	}
166
167	if (!isset($options['crypto_type'])) {	×
168	$options['crypto_type'] = STREAM_CRYPTO_METHOD_TLS_CLIENT;	×
169	}
170
NEW 171	$context_options = function_exists('stream_context_set_options')	×
NEW 172	? stream_context_set_options($resource, ['ssl' => $options])	×
NEW 173	: stream_context_set_option($resource, ['ssl' => $options]);	×
174
NEW 175	if (!$context_options) {	×
UNCOV 176	$this->onInitializationError($resource, $parameters, 'Error while setting SSL context options');	×
177	}
178
179	if (!stream_socket_enable_crypto($resource, true, $options['crypto_type'])) {	×
180	$this->onInitializationError($resource, $parameters, 'Error while switching to encrypted communication');	×
181	}
182
183	return $resource;	×
184	}
185
186	/**
187	* Creates a connected stream socket resource.
188	*
189	* @param ParametersInterface $parameters Connection parameters.
190	* @param string $address Address for stream_socket_client().
191	* @param int $flags Flags for stream_socket_client().
192	*
193	* @return resource
194	* @throws StreamInitException
195	*/
196	protected function createStreamSocket(ParametersInterface $parameters, $address, $flags)	1,525✔
197	{
198	$timeout = (isset($parameters->timeout) ? (float) $parameters->timeout : 5.0);	1,525✔
199	$context = stream_context_create(['socket' => ['tcp_nodelay' => (bool) $parameters->tcp_nodelay]]);	1,525✔
200
201	if (
202	(isset($parameters->persistent) && $parameters->persistent)	1,525✔
203	&& (isset($parameters->conn_uid) && $parameters->conn_uid)	1,525✔
204	) {
205	$conn_uid = '/' . $parameters->conn_uid;	10✔
206	} else {
207	$conn_uid = '';	1,518✔
208	}
209
210	// Needs to create multiple persistent connections to the same resource
211	$address = $address . $conn_uid;	1,525✔
212
213	if (!$resource = @stream_socket_client($address, $errno, $errstr, $timeout, $flags, $context)) {	1,525✔
214	$this->onInitializationError($resource, $parameters, trim($errstr), $errno);	3✔
215	}
216
217	if (isset($parameters->read_write_timeout)) {	1,522✔
218	$rwtimeout = (float) $parameters->read_write_timeout;	63✔
219	$rwtimeout = $rwtimeout > 0 ? $rwtimeout : -1;	63✔
220	$timeoutSeconds = floor($rwtimeout);	63✔
221	$timeoutUSeconds = ($rwtimeout - $timeoutSeconds) * 1000000;	63✔
222	stream_set_timeout($resource, $timeoutSeconds, $timeoutUSeconds);	63✔
223	}
224
225	return $resource;	1,522✔
226	}
227
228	/**
229	* Helper method to handle connection errors.
230	*
231	* @param string $message Error message.
232	* @param int $code Error code.
233	* @throws StreamInitException
234	*/
235	protected function onInitializationError($stream, ParametersInterface $parameters, string $message, int $code = 0): void	3✔
236	{
237	if (is_resource($stream)) {	3✔
238	fclose($stream);	×
239	}
240
241	throw new StreamInitException("$message [{$parameters}]", $code);	3✔
242	}
243	}

predis / predis / 15269011225

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous