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

74.24

/src/Connection/RelayFactory.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;

use InvalidArgumentException;
use Predis\Command\RawCommand;
use Predis\NotSupportedException;
use Predis\Retry\Strategy\EqualBackoff;
use Predis\Retry\Strategy\ExponentialBackoff;
use Relay\Relay;

class RelayFactory extends Factory
{
    /**
     * @var string[]
     */
    protected $schemes = [
        'tcp' => RelayConnection::class,
        'tls' => RelayConnection::class,
        'unix' => RelayConnection::class,
        'redis' => RelayConnection::class,
        'rediss' => RelayConnection::class,
    ];

    /**
     * {@inheritDoc}
     */
    public function define($scheme, $initializer)
    {
        throw new NotSupportedException('Does not allow to override existing initializer.');
    }

    /**
     * {@inheritDoc}
     */
    public function undefine($scheme)
    {
        throw new NotSupportedException('Does not allow to override existing initializer.');
    }

    /**
     * {@inheritDoc}
     */
    public function create($parameters): NodeConnectionInterface
    {
        $this->assertExtensions();

        if (!$parameters instanceof ParametersInterface) {
            $parameters = $this->createParameters($parameters);
        }

        $scheme = $parameters->scheme;

        if (!isset($this->schemes[$scheme])) {
            throw new InvalidArgumentException("Unknown connection scheme: '$scheme'.");
        }

        $initializer = $this->schemes[$scheme];
        $client = $this->createClient($parameters);

        $connection = new $initializer($parameters, $client);

        $this->prepareConnection($connection);

        return $connection;
    }

    /**
     * Checks if the Relay extension is loaded in PHP.
     */
    private function assertExtensions()
    {
        if (!extension_loaded('relay')) {
            throw new NotSupportedException(
                'The "relay" extension is required by this connection backend.'
            );
        }
    }

    /**
     * Creates a new instance of the client.
     *
     * @return Relay
     */
    private function createClient(ParametersInterface $parameters)
    {
        $client = new Relay();

        // throw when errors occur and return `null` for non-existent keys
        $client->setOption(Relay::OPT_PHPREDIS_COMPATIBILITY, false);

        // use reply literals
        $client->setOption(Relay::OPT_REPLY_LITERAL, true);

        // whether to use in-memory caching
        $client->setOption(Relay::OPT_USE_CACHE, $parameters->cache ?? true);

        // set data serializer
        $client->setOption(Relay::OPT_SERIALIZER, constant(sprintf(
            '%s::SERIALIZER_%s',
            Relay::class,
            strtoupper($parameters->serializer ?? 'none')
        )));

        // set data compression algorithm
        $client->setOption(Relay::OPT_COMPRESSION, constant(sprintf(
            '%s::COMPRESSION_%s',
            Relay::class,
            strtoupper($parameters->compression ?? 'none')
        )));

        if ($parameters->isDisabledRetry()) {
            $client->setOption(Relay::OPT_MAX_RETRIES, 0);
        } else {
            $client->setOption(Relay::OPT_MAX_RETRIES, $parameters->retry->getRetries());

            $retryStrategy = $parameters->retry->getStrategy();

            if ($retryStrategy instanceof ExponentialBackoff) {
                $algorithm = Relay::BACKOFF_ALGORITHM_FULL_JITTER;
                $base = $retryStrategy->getBase();
                $cap = $retryStrategy->getCap();
            } else {
                $algorithm = Relay::BACKOFF_ALGORITHM_DEFAULT;

                if ($retryStrategy instanceof EqualBackoff) {
                    $base = $cap = $retryStrategy->compute(0);
                } else {
                    $base = $retryStrategy::DEFAULT_BASE;
                    $cap = $retryStrategy::DEFAULT_CAP;
                }
            }

            $client->setOption(Relay::OPT_BACKOFF_ALGORITHM, $algorithm);
            $client->setOption(Relay::OPT_BACKOFF_BASE, $base / 1000);
            $client->setOption(Relay::OPT_BACKOFF_CAP, $cap / 1000);
        }

        return $client;
    }

    /**
     * {@inheritdoc}
     */
    protected function prepareConnection(NodeConnectionInterface $connection)
    {
        $parameters = $connection->getParameters();

        if (isset($parameters->password) && strlen($parameters->password)) {
            $cmdAuthArgs = isset($parameters->username) && strlen($parameters->username)
                ? [$parameters->username, $parameters->password]
                : [$parameters->password];

            $connection->addConnectCommand(
                new RawCommand('AUTH', $cmdAuthArgs)
            );
        }

        if (isset($parameters->database) && strlen($parameters->database)) {
            $connection->addConnectCommand(
                new RawCommand('SELECT', [$parameters->database])
            );
        }
    }
}

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;
14
15	use InvalidArgumentException;
16	use Predis\Command\RawCommand;
17	use Predis\NotSupportedException;
18	use Predis\Retry\Strategy\EqualBackoff;
19	use Predis\Retry\Strategy\ExponentialBackoff;
20	use Relay\Relay;
21
22	class RelayFactory extends Factory
23	{
24	/**
25	* @var string[]
26	*/
27	protected $schemes = [
28	'tcp' => RelayConnection::class,
29	'tls' => RelayConnection::class,
30	'unix' => RelayConnection::class,
31	'redis' => RelayConnection::class,
32	'rediss' => RelayConnection::class,
33	];
34
35	/**
36	* {@inheritDoc}
37	*/
38	public function define($scheme, $initializer)	1✔
39	{
40	throw new NotSupportedException('Does not allow to override existing initializer.');	1✔
41	}
42
43	/**
44	* {@inheritDoc}
45	*/
46	public function undefine($scheme)	1✔
47	{
48	throw new NotSupportedException('Does not allow to override existing initializer.');	1✔
49	}
50
51	/**
52	* {@inheritDoc}
53	*/
54	public function create($parameters): NodeConnectionInterface	1,118✔
55	{
56	$this->assertExtensions();	1,118✔
57
58	if (!$parameters instanceof ParametersInterface) {	1,118✔
59	$parameters = $this->createParameters($parameters);	1,117✔
60	}
61
62	$scheme = $parameters->scheme;	1,118✔
63
64	if (!isset($this->schemes[$scheme])) {	1,118✔
65	throw new InvalidArgumentException("Unknown connection scheme: '$scheme'.");	1✔
66	}
67
68	$initializer = $this->schemes[$scheme];	1,117✔
69	$client = $this->createClient($parameters);	1,117✔
70
71	$connection = new $initializer($parameters, $client);	1,117✔
72
73	$this->prepareConnection($connection);	1,117✔
74
75	return $connection;	1,117✔
76	}
77
78	/**
79	* Checks if the Relay extension is loaded in PHP.
80	*/
81	private function assertExtensions()	1,118✔
82	{
83	if (!extension_loaded('relay')) {	1,118✔
84	throw new NotSupportedException(	×
85	'The "relay" extension is required by this connection backend.'	×
86	);	×
87	}
88	}
89
90	/**
91	* Creates a new instance of the client.
92	*
93	* @return Relay
94	*/
95	private function createClient(ParametersInterface $parameters)	1,117✔
96	{
97	$client = new Relay();	1,117✔
98
99	// throw when errors occur and return `null` for non-existent keys
100	$client->setOption(Relay::OPT_PHPREDIS_COMPATIBILITY, false);	1,117✔
101
102	// use reply literals
103	$client->setOption(Relay::OPT_REPLY_LITERAL, true);	1,117✔
104
105	// whether to use in-memory caching
106	$client->setOption(Relay::OPT_USE_CACHE, $parameters->cache ?? true);	1,117✔
107
108	// set data serializer
109	$client->setOption(Relay::OPT_SERIALIZER, constant(sprintf(	1,117✔
110	'%s::SERIALIZER_%s',	1,117✔
111	Relay::class,	1,117✔
112	strtoupper($parameters->serializer ?? 'none')	1,117✔
113	)));	1,117✔
114
115	// set data compression algorithm
116	$client->setOption(Relay::OPT_COMPRESSION, constant(sprintf(	1,117✔
117	'%s::COMPRESSION_%s',	1,117✔
118	Relay::class,	1,117✔
119	strtoupper($parameters->compression ?? 'none')	1,117✔
120	)));	1,117✔
121
122	if ($parameters->isDisabledRetry()) {	1,117✔
123	$client->setOption(Relay::OPT_MAX_RETRIES, 0);	1,117✔
124	} else {
NEW 125	$client->setOption(Relay::OPT_MAX_RETRIES, $parameters->retry->getRetries());	×
126
NEW 127	$retryStrategy = $parameters->retry->getStrategy();	×
128
NEW 129	if ($retryStrategy instanceof ExponentialBackoff) {	×
NEW 130	$algorithm = Relay::BACKOFF_ALGORITHM_FULL_JITTER;	×
NEW 131	$base = $retryStrategy->getBase();	×
NEW 132	$cap = $retryStrategy->getCap();	×
133	} else {
NEW 134	$algorithm = Relay::BACKOFF_ALGORITHM_DEFAULT;	×
135
NEW 136	if ($retryStrategy instanceof EqualBackoff) {	×
NEW 137	$base = $cap = $retryStrategy->compute(0);	×
138	} else {
NEW 139	$base = $retryStrategy::DEFAULT_BASE;	×
NEW 140	$cap = $retryStrategy::DEFAULT_CAP;	×
141	}
142	}
143
NEW 144	$client->setOption(Relay::OPT_BACKOFF_ALGORITHM, $algorithm);	×
NEW 145	$client->setOption(Relay::OPT_BACKOFF_BASE, $base / 1000);	×
NEW 146	$client->setOption(Relay::OPT_BACKOFF_CAP, $cap / 1000);	×
147	}
148
149	return $client;	1,117✔
150	}
151
152	/**
153	* {@inheritdoc}
154	*/
155	protected function prepareConnection(NodeConnectionInterface $connection)	1,117✔
156	{
157	$parameters = $connection->getParameters();	1,117✔
158
159	if (isset($parameters->password) && strlen($parameters->password)) {	1,117✔
160	$cmdAuthArgs = isset($parameters->username) && strlen($parameters->username)	1,115✔
161	? [$parameters->username, $parameters->password]	1✔
162	: [$parameters->password];	1,114✔
163
164	$connection->addConnectCommand(	1,115✔
165	new RawCommand('AUTH', $cmdAuthArgs)	1,115✔
166	);	1,115✔
167	}
168
169	if (isset($parameters->database) && strlen($parameters->database)) {	1,117✔
170	$connection->addConnectCommand(	1,088✔
171	new RawCommand('SELECT', [$parameters->database])	1,088✔
172	);	1,088✔
173	}
174	}
175	}

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