20455852140

Committed 23 Dec 2025 08:40AM UTC coverage: 92.615% (-0.2%) from 92.788%

Build # 20455852140

Build Type

push

github

Committed by

web-flow

Commit Message

Added retry support (#1616)

* Initial work on retries

* Added retry class and test coverage

* Added support for standalone and cluster

* Make TimeoutException instance of CommunicationException

* Added pipeline, trasnaction, replication support

* Fixed broken test

* Marked test as relay-incompatible

* Marked test as relay-incompatible

* Fixed analysis errors, added missing tests

* Codestyle fixes

* Fixed test

* Update README.md

* Update README.md

* Update README.md

* Updated README.md

* Refactor retry on read and write

* Added check for timeout value

* Updated README.md

* Fixed README.md

* Codestyle changes

* Added missing coverage

* Added missing test coverage

* Removed comments

* Added retry support for Relay connection (#1620)

* Added integration test case with mocked retry

* Changed client initialisation in tests

* Marked test as relay-incompatible

---------

Co-authored-by: Pavlo Yatsukhnenko <yatsukhnenko@users.noreply.github.com>

Run Details

301 of 358 new or added lines in 18 files covered. (84.08%)

2 existing lines in 2 files now uncovered.

8214 of 8869 relevant lines covered (92.61%)

112.31 hits per line

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

84.87

/src/Connection/Resource/Stream.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 Psr\Http\Message\StreamInterface;
use RuntimeException;

class Stream implements StreamInterface
{
    /**
     * @see https://www.php.net/manual/en/function.fopen.php
     * @see https://www.php.net/manual/en/function.gzopen.php
     */
    private const READABLE_MODES = '/r|a\+|ab\+|w\+|wb\+|x\+|xb\+|c\+|cb\+/';
    private const WRITABLE_MODES = '/a|w|r\+|rb\+|rw|x|c/';

    /**
     * @var resource
     */
    private $stream;

    /**
     * @var bool
     */
    private $seekable;

    /**
     * @var bool
     */
    private $readable;

    /**
     * @var bool
     */
    private $writable;

    /**
     * @param  resource                 $stream
     * @throws InvalidArgumentException if stream is not a valid resource.
     */
    public function __construct($stream)
    {
        if (!is_resource($stream)) {
            throw new InvalidArgumentException('Given stream is not a valid resource');
        }

        $this->stream = $stream;
        $metadata = stream_get_meta_data($this->stream);
        $this->seekable = $metadata['seekable'];
        $this->readable = (bool) preg_match(self::READABLE_MODES, $metadata['mode']);
        $this->writable = (bool) preg_match(self::WRITABLE_MODES, $metadata['mode']);
    }

    /**
     * {@inheritDoc}
     */
    public function __toString(): string
    {
        if ($this->isSeekable()) {
            $this->seek(0);
        }

        return $this->getContents();
    }

    /**
     * {@inheritDoc}
     */
    public function close(): void
    {
        if (isset($this->stream)) {
            fclose($this->stream);
        }

        $this->detach();
    }

    /**
     * {@inheritDoc}
     */
    public function detach()
    {
        if (!isset($this->stream)) {
            return null;
        }

        $result = $this->stream;
        unset($this->stream);
        $this->readable = $this->writable = $this->seekable = false;

        return $result;
    }

    /**
     * {@inheritDoc}
     */
    public function getSize(): ?int
    {
        if (!isset($this->stream)) {
            return null;
        }

        $stats = fstat($this->stream);
        if (is_array($stats) && isset($stats['size'])) {
            return $stats['size'];
        }

        return null;
    }

    /**
     * {@inheritDoc}
     */
    public function tell(): int
    {
        if (!isset($this->stream)) {
            throw new RuntimeException('Stream is detached');
        }

        $result = ftell($this->stream);

        if ($result === false) {
            throw new RuntimeException('Unable to determine stream position');
        }

        return $result;
    }

    /**
     * {@inheritDoc}
     */
    public function eof(): bool
    {
        if (!isset($this->stream)) {
            throw new RuntimeException('Stream is detached');
        }

        return feof($this->stream);
    }

    /**
     * {@inheritDoc}
     */
    public function isSeekable(): bool
    {
        return $this->seekable;
    }

    /**
     * {@inheritDoc}
     */
    public function seek(int $offset, int $whence = SEEK_SET): void
    {
        if (!isset($this->stream)) {
            throw new RuntimeException('Stream is detached');
        }

        if (!$this->isSeekable()) {
            throw new RuntimeException('Stream is not seekable');
        }

        if (fseek($this->stream, $offset, $whence) === -1) {
            throw new RuntimeException("Unable to seek stream from offset {$offset} to whence {$whence}");
        }
    }

    /**
     * {@inheritDoc}
     */
    public function rewind(): void
    {
        $this->seek(0);
    }

    /**
     * {@inheritDoc}
     */
    public function isWritable(): bool
    {
        return $this->writable;
    }

    /**
     * {@inheritDoc}
     * @throws RuntimeException
     */
    public function write(string $string): int
    {
        if (!isset($this->stream)) {
            throw new RuntimeException('Stream is detached');
        }

        if (!$this->isWritable()) {
            throw new RuntimeException('Cannot write to a non-writable stream');
        }

        $result = fwrite($this->stream, $string);

        if ($result === false || $result === 0) {
            $metadata = $this->getMetadata();

            if ($this->eof()) {
                throw new RuntimeException('Connection closed by peer during write', 1);
            }

            if (!is_resource($this->stream)) {
                throw new RuntimeException(
                    'Stream resource is no longer valid',
                    1
                );
            }

            if (array_key_exists('timed_out', $metadata) && $metadata['timed_out']) {
                throw new RuntimeException(
                    'Stream has been timed out',
                    2
                );
            }

            throw new RuntimeException('Unable to write to stream', 1);
        }

        return $result;
    }

    /**
     * {@inheritDoc}
     */
    public function isReadable(): bool
    {
        return $this->readable;
    }

    /**
     * {@inheritDoc}
     * @param  int              $length If length = -1, reads a stream line by line (e.g fgets())
     * @throws RuntimeException
     */
    public function read(int $length): string
    {
        if (!isset($this->stream)) {
            throw new RuntimeException('Stream is detached');
        }

        if (!$this->isReadable()) {
            throw new RuntimeException('Cannot read from non-readable stream');
        }

        if ($length < -1) {
            throw new RuntimeException('Length parameter cannot be negative');
        }

        if (0 === $length) {
            return '';
        }

        if ($length === -1) {
            $string = fgets($this->stream);
        } else {
            $string = fread($this->stream, $length);
        }

        if (false === $string) {
            $metadata = $this->getMetadata();

            if ($this->eof()) {
                throw new RuntimeException('Connection closed by peer during read', 1);
            }

            if (!is_resource($this->stream)) {
                throw new RuntimeException(
                    'Stream resource is no longer valid',
                    1
                );
            }

            if (array_key_exists('timed_out', $metadata) && $metadata['timed_out']) {
                throw new RuntimeException(
                    'Stream has been timed out',
                    2
                );
            }

            throw new RuntimeException('Unable to read from stream', 1);
        }

        return $string;
    }

    /**
     * {@inheritDoc}
     */
    public function getContents(): string
    {
        if (!isset($this->stream)) {
            throw new RuntimeException('Stream is detached');
        }

        if (!$this->isReadable()) {
            throw new RuntimeException('Cannot read from non-readable stream');
        }

        return stream_get_contents($this->stream);
    }

    /**
     * {@inheritDoc}
     * @return mixed
     */
    public function getMetadata(?string $key = null)
    {
        if (!isset($this->stream)) {
            return null;
        }

        if (!$key) {
            return stream_get_meta_data($this->stream);
        }

        $metadata = stream_get_meta_data($this->stream);

        return $metadata[$key] ?? null;
    }
}

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 Psr\Http\Message\StreamInterface;
17	use RuntimeException;
18
19	class Stream implements StreamInterface
20	{
21	/**
22	* @see https://www.php.net/manual/en/function.fopen.php
23	* @see https://www.php.net/manual/en/function.gzopen.php
24	*/
25	private const READABLE_MODES = '/r\|a\+\|ab\+\|w\+\|wb\+\|x\+\|xb\+\|c\+\|cb\+/';
26	private const WRITABLE_MODES = '/a\|w\|r\+\|rb\+\|rw\|x\|c/';
27
28	/**
29	* @var resource
30	*/
31	private $stream;
32
33	/**
34	* @var bool
35	*/
36	private $seekable;
37
38	/**
39	* @var bool
40	*/
41	private $readable;
42
43	/**
44	* @var bool
45	*/
46	private $writable;
47
48	/**
49	* @param resource $stream
50	* @throws InvalidArgumentException if stream is not a valid resource.
51	*/
52	public function __construct($stream)	1,668✔
53	{
54	if (!is_resource($stream)) {	1,668✔
55	throw new InvalidArgumentException('Given stream is not a valid resource');	1✔
56	}
57
58	$this->stream = $stream;	1,667✔
59	$metadata = stream_get_meta_data($this->stream);	1,667✔
60	$this->seekable = $metadata['seekable'];	1,667✔
61	$this->readable = (bool) preg_match(self::READABLE_MODES, $metadata['mode']);	1,667✔
62	$this->writable = (bool) preg_match(self::WRITABLE_MODES, $metadata['mode']);	1,667✔
63	}
64
65	/**
66	* {@inheritDoc}
67	*/
68	public function __toString(): string	1✔
69	{
70	if ($this->isSeekable()) {	1✔
71	$this->seek(0);	1✔
72	}
73
74	return $this->getContents();	1✔
75	}
76
77	/**
78	* {@inheritDoc}
79	*/
80	public function close(): void	1,605✔
81	{
82	if (isset($this->stream)) {	1,605✔
83	fclose($this->stream);	1,596✔
84	}
85
86	$this->detach();	1,605✔
87	}
88
89	/**
90	* {@inheritDoc}
91	*/
92	public function detach()	1,616✔
93	{
94	if (!isset($this->stream)) {	1,616✔
95	return null;	10✔
96	}
97
98	$result = $this->stream;	1,616✔
99	unset($this->stream);	1,616✔
100	$this->readable = $this->writable = $this->seekable = false;	1,616✔
101
102	return $result;	1,616✔
103	}
104
105	/**
106	* {@inheritDoc}
107	*/
108	public function getSize(): ?int	1✔
109	{
110	if (!isset($this->stream)) {	1✔
111	return null;	1✔
112	}
113
114	$stats = fstat($this->stream);	1✔
115	if (is_array($stats) && isset($stats['size'])) {	1✔
116	return $stats['size'];	1✔
117	}
118
119	return null;	×
120	}
121
122	/**
123	* {@inheritDoc}
124	*/
125	public function tell(): int	2✔
126	{
127	if (!isset($this->stream)) {	2✔
128	throw new RuntimeException('Stream is detached');	1✔
129	}
130
131	$result = ftell($this->stream);	1✔
132
133	if ($result === false) {	1✔
134	throw new RuntimeException('Unable to determine stream position');	×
135	}
136
137	return $result;	1✔
138	}
139
140	/**
141	* {@inheritDoc}
142	*/
143	public function eof(): bool	1,608✔
144	{
145	if (!isset($this->stream)) {	1,608✔
146	throw new RuntimeException('Stream is detached');	1✔
147	}
148
149	return feof($this->stream);	1,607✔
150	}
151
152	/**
153	* {@inheritDoc}
154	*/
155	public function isSeekable(): bool	8✔
156	{
157	return $this->seekable;	8✔
158	}
159
160	/**
161	* {@inheritDoc}
162	*/
163	public function seek(int $offset, int $whence = SEEK_SET): void	7✔
164	{
165	if (!isset($this->stream)) {	7✔
166	throw new RuntimeException('Stream is detached');	1✔
167	}
168
169	if (!$this->isSeekable()) {	6✔
170	throw new RuntimeException('Stream is not seekable');	×
171	}
172
173	if (fseek($this->stream, $offset, $whence) === -1) {	6✔
174	throw new RuntimeException("Unable to seek stream from offset {$offset} to whence {$whence}");	1✔
175	}
176	}
177
178	/**
179	* {@inheritDoc}
180	*/
181	public function rewind(): void	4✔
182	{
183	$this->seek(0);	4✔
184	}
185
186	/**
187	* {@inheritDoc}
188	*/
189	public function isWritable(): bool	1,630✔
190	{
191	return $this->writable;	1,630✔
192	}
193
194	/**
195	* {@inheritDoc}
196	* @throws RuntimeException
197	*/
198	public function write(string $string): int	1,612✔
199	{
200	if (!isset($this->stream)) {	1,612✔
201	throw new RuntimeException('Stream is detached');	1✔
202	}
203
204	if (!$this->isWritable()) {	1,611✔
205	throw new RuntimeException('Cannot write to a non-writable stream');	1✔
206	}
207
208	$result = fwrite($this->stream, $string);	1,610✔
209
210	if ($result === false \|\| $result === 0) {	1,610✔
211	$metadata = $this->getMetadata();	1✔
212
213	if ($this->eof()) {	1✔
NEW 214	throw new RuntimeException('Connection closed by peer during write', 1);	×
215	}
216
217	if (!is_resource($this->stream)) {	1✔
NEW 218	throw new RuntimeException(	×
NEW 219	'Stream resource is no longer valid',	×
NEW 220	1	×
NEW 221	);	×
222	}
223
224	if (array_key_exists('timed_out', $metadata) && $metadata['timed_out']) {	1✔
NEW 225	throw new RuntimeException(	×
NEW 226	'Stream has been timed out',	×
NEW 227	2	×
NEW 228	);	×
229	}
230
231	throw new RuntimeException('Unable to write to stream', 1);	1✔
232	}
233
234	return $result;	1,609✔
235	}
236
237	/**
238	* {@inheritDoc}
239	*/
240	public function isReadable(): bool	1,631✔
241	{
242	return $this->readable;	1,631✔
243	}
244
245	/**
246	* {@inheritDoc}
247	* @param int $length If length = -1, reads a stream line by line (e.g fgets())
248	* @throws RuntimeException
249	*/
250	public function read(int $length): string	1,611✔
251	{
252	if (!isset($this->stream)) {	1,611✔
253	throw new RuntimeException('Stream is detached');	1✔
254	}
255
256	if (!$this->isReadable()) {	1,610✔
257	throw new RuntimeException('Cannot read from non-readable stream');	1✔
258	}
259
260	if ($length < -1) {	1,609✔
261	throw new RuntimeException('Length parameter cannot be negative');	1✔
262	}
263
264	if (0 === $length) {	1,608✔
265	return '';	1✔
266	}
267
268	if ($length === -1) {	1,607✔
269	$string = fgets($this->stream);	1,605✔
270	} else {
271	$string = fread($this->stream, $length);	1,607✔
272	}
273
274	if (false === $string) {	1,607✔
275	$metadata = $this->getMetadata();	7✔
276
277	if ($this->eof()) {	7✔
NEW 278	throw new RuntimeException('Connection closed by peer during read', 1);	×
279	}
280
281	if (!is_resource($this->stream)) {	7✔
NEW 282	throw new RuntimeException(	×
NEW 283	'Stream resource is no longer valid',	×
NEW 284	1	×
NEW 285	);	×
286	}
287
288	if (array_key_exists('timed_out', $metadata) && $metadata['timed_out']) {	7✔
289	throw new RuntimeException(	7✔
290	'Stream has been timed out',	7✔
291	2	7✔
292	);	7✔
293	}
294
UNCOV 295	throw new RuntimeException('Unable to read from stream', 1);	×
296	}
297
298	return $string;	1,607✔
299	}
300
301	/**
302	* {@inheritDoc}
303	*/
304	public function getContents(): string	4✔
305	{
306	if (!isset($this->stream)) {	4✔
307	throw new RuntimeException('Stream is detached');	1✔
308	}
309
310	if (!$this->isReadable()) {	3✔
311	throw new RuntimeException('Cannot read from non-readable stream');	1✔
312	}
313
314	return stream_get_contents($this->stream);	2✔
315	}
316
317	/**
318	* {@inheritDoc}
319	* @return mixed
320	*/
321	public function getMetadata(?string $key = null)	9✔
322	{
323	if (!isset($this->stream)) {	9✔
324	return null;	1✔
325	}
326
327	if (!$key) {	9✔
328	return stream_get_meta_data($this->stream);	9✔
329	}
330
331	$metadata = stream_get_meta_data($this->stream);	1✔
332
333	return $metadata[$key] ?? null;	1✔
334	}
335	}

predis / predis / 20455852140

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