16929978125

Committed 13 Aug 2025 06:59AM UTC coverage: 93.114% (+0.004%) from 93.11%

Build # 16929978125

Build Type

push

github

Committed by

web-flow

Commit Message

Refactor pipeline data writing depends on connection type (#1586)

* Refactor pipeline data writing depends on connection type

* Updated CHANGELOG.md

Run Details

17 of 20 new or added lines in 4 files covered. (85.0%)

3 existing lines in 2 files now uncovered.

7681 of 8249 relevant lines covered (93.11%)

112.6 hits per line

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

0.0

/src/Pipeline/ConnectionErrorProof.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\Pipeline;

use Predis\CommunicationException;
use Predis\Connection\Cluster\ClusterInterface;
use Predis\Connection\ConnectionInterface;
use Predis\Connection\NodeConnectionInterface;
use Predis\NotSupportedException;
use SplQueue;

/**
 * Command pipeline that does not throw exceptions on connection errors, but
 * returns the exception instances as the rest of the response elements.
 */
class ConnectionErrorProof extends Pipeline
{
    /**
     * {@inheritdoc}
     */
    protected function getConnection()
    {
        return $this->getClient()->getConnection();
    }

    /**
     * {@inheritdoc}
     */
    protected function executePipeline(ConnectionInterface $connection, SplQueue $commands)
    {
        if ($connection instanceof NodeConnectionInterface) {
            return $this->executeSingleNode($connection, $commands);
        } elseif ($connection instanceof ClusterInterface) {
            return $this->executeCluster($connection, $commands);
        } else {
            $class = get_class($connection);

            throw new NotSupportedException("The connection class '$class' is not supported.");
        }
    }

    /**
     * {@inheritdoc}
     */
    protected function executeSingleNode(NodeConnectionInterface $connection, SplQueue $commands)
    {
        $responses = [];
        $sizeOfPipe = count($commands);
        $buffer = '';

        foreach ($commands as $command) {
            $buffer .= $command->serializeCommand();
        }

        try {
            $connection->write($buffer);
        } catch (CommunicationException $exception) {
            return array_fill(0, $sizeOfPipe, $exception);
        }

        for ($i = 0; $i < $sizeOfPipe; ++$i) {
            $command = $commands->dequeue();

            try {
                $responses[$i] = $connection->readResponse($command);
            } catch (CommunicationException $exception) {
                $add = count($commands) - count($responses);
                $responses = array_merge($responses, array_fill(0, $add, $exception));

                break;
            }
        }

        return $responses;
    }

    /**
     * {@inheritdoc}
     */
    protected function executeCluster(ClusterInterface $connection, SplQueue $commands)
    {
        $responses = [];
        $sizeOfPipe = count($commands);
        $exceptions = [];

        foreach ($commands as $command) {
            $nodeConnection = $connection->getConnectionByCommand($command);
            $nodeConnection->write($command->serializeCommand());
        }

        for ($i = 0; $i < $sizeOfPipe; ++$i) {
            $command = $commands->dequeue();

            $cmdConnection = $connection->getConnectionByCommand($command);
            $connectionHash = spl_object_hash($cmdConnection);

            if (isset($exceptions[$connectionHash])) {
                $responses[$i] = $exceptions[$connectionHash];
                continue;
            }

            try {
                $responses[$i] = $cmdConnection->readResponse($command);
            } catch (CommunicationException $exception) {
                $responses[$i] = $exception;
                $exceptions[$connectionHash] = $exception;
            }
        }

        return $responses;
    }
}

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\Pipeline;
14
15	use Predis\CommunicationException;
16	use Predis\Connection\Cluster\ClusterInterface;
17	use Predis\Connection\ConnectionInterface;
18	use Predis\Connection\NodeConnectionInterface;
19	use Predis\NotSupportedException;
20	use SplQueue;
21
22	/**
23	* Command pipeline that does not throw exceptions on connection errors, but
24	* returns the exception instances as the rest of the response elements.
25	*/
26	class ConnectionErrorProof extends Pipeline
27	{
28	/**
29	* {@inheritdoc}
30	*/
31	protected function getConnection()	×
32	{
33	return $this->getClient()->getConnection();	×
34	}
35
36	/**
37	* {@inheritdoc}
38	*/
39	protected function executePipeline(ConnectionInterface $connection, SplQueue $commands)	×
40	{
41	if ($connection instanceof NodeConnectionInterface) {	×
42	return $this->executeSingleNode($connection, $commands);	×
43	} elseif ($connection instanceof ClusterInterface) {	×
44	return $this->executeCluster($connection, $commands);	×
45	} else {
46	$class = get_class($connection);	×
47
48	throw new NotSupportedException("The connection class '$class' is not supported.");	×
49	}
50	}
51
52	/**
53	* {@inheritdoc}
54	*/
55	protected function executeSingleNode(NodeConnectionInterface $connection, SplQueue $commands)	×
56	{
57	$responses = [];	×
58	$sizeOfPipe = count($commands);	×
59	$buffer = '';	×
60
61	foreach ($commands as $command) {	×
62	$buffer .= $command->serializeCommand();	×
63	}
64
65	try {
66	$connection->write($buffer);	×
67	} catch (CommunicationException $exception) {	×
68	return array_fill(0, $sizeOfPipe, $exception);	×
69	}
70
71	for ($i = 0; $i < $sizeOfPipe; ++$i) {	×
72	$command = $commands->dequeue();	×
73
74	try {
75	$responses[$i] = $connection->readResponse($command);	×
76	} catch (CommunicationException $exception) {	×
77	$add = count($commands) - count($responses);	×
78	$responses = array_merge($responses, array_fill(0, $add, $exception));	×
79
80	break;	×
81	}
82	}
83
84	return $responses;	×
85	}
86
87	/**
88	* {@inheritdoc}
89	*/
90	protected function executeCluster(ClusterInterface $connection, SplQueue $commands)	×
91	{
92	$responses = [];	×
93	$sizeOfPipe = count($commands);	×
94	$exceptions = [];	×
95
UNCOV 96	foreach ($commands as $command) {	×
NEW 97	$nodeConnection = $connection->getConnectionByCommand($command);	×
NEW 98	$nodeConnection->write($command->serializeCommand());	×
99	}
100
101	for ($i = 0; $i < $sizeOfPipe; ++$i) {	×
UNCOV 102	$command = $commands->dequeue();	×
103
104	$cmdConnection = $connection->getConnectionByCommand($command);	×
105	$connectionHash = spl_object_hash($cmdConnection);	×
106
107	if (isset($exceptions[$connectionHash])) {	×
108	$responses[$i] = $exceptions[$connectionHash];	×
109	continue;	×
110	}
111
112	try {
113	$responses[$i] = $cmdConnection->readResponse($command);	×
114	} catch (CommunicationException $exception) {	×
115	$responses[$i] = $exception;	×
116	$exceptions[$connectionHash] = $exception;	×
117	}
118	}
119
120	return $responses;	×
121	}
122	}

predis / predis / 16929978125

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