Dereku/snappymail
1
0
Fork 0
mirror of https://github.com/the-djmaze/snappymail.git synced 2026-08-28 19:49:20 +03:00
Code Issues Projects Releases Packages Wiki Activity Actions

Predis to v2.2.2

Browse source
This commit is contained in:
the-djmaze 2024-04-02 20:15:23 +02:00
parent f6b440adef
commit 84ffe1e552
259 changed files with 2407 additions and 9937 deletions
Download patch file Download diff file Expand all files Collapse all files

70
plugins/cache-redis/Predis/Connection/AbstractConnection.php
View file

@ -3,7 +3,8 @@
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
@ -11,15 +12,15 @@
namespace Predis\Connection;
use InvalidArgumentException;
use Predis\Command\CommandInterface;
use Predis\Command\RawCommand;
use Predis\CommunicationException;
use Predis\Protocol\ProtocolException;
/**
* Base class with the common logic used by connection classes to communicate
* with Redis.
*
* @author Daniele Alessandri <suppakilla@gmail.com>
*/
abstract class AbstractConnection implements NodeConnectionInterface
{
@ -27,7 +28,11 @@ abstract class AbstractConnection implements NodeConnectionInterface
private $cachedId;
protected $parameters;
protected $initCommands = array();
/**
* @var RawCommand[]
*/
protected $initCommands = [];
/**
* @param ParametersInterface $parameters Initialization parameters for the connection.
@ -51,24 +56,10 @@ abstract class AbstractConnection implements NodeConnectionInterface
*
* @param ParametersInterface $parameters Initialization parameters for the connection.
*
* @throws \InvalidArgumentException
*
* @return ParametersInterface
* @throws InvalidArgumentException
*/
protected function assertParameters(ParametersInterface $parameters)
{
switch ($parameters->scheme) {
case 'tcp':
case 'redis':
case 'unix':
break;
default:
throw new \InvalidArgumentException("Invalid scheme: '$parameters->scheme'.");
}
return $parameters;
}
abstract protected function assertParameters(ParametersInterface $parameters);
/**
* Creates the underlying resource used to communicate with Redis.
@ -115,6 +106,14 @@ abstract class AbstractConnection implements NodeConnectionInterface
$this->initCommands[] = $command;
}
/**
* {@inheritdoc}
*/
public function getInitCommands(): array
{
return $this->initCommands;
}
/**
* {@inheritdoc}
*/
@ -133,39 +132,16 @@ abstract class AbstractConnection implements NodeConnectionInterface
return $this->read();
}
/**
* Helper method that returns an exception message augmented with useful
* details from the connection parameters.
*
* @param string $message Error message.
*
* @return string
*/
private function createExceptionMessage($message)
{
$parameters = $this->parameters;
if ($parameters->scheme === 'unix') {
return "$message [$parameters->scheme:$parameters->path]";
}
if (filter_var($parameters->host, FILTER_VALIDATE_IP, FILTER_FLAG_IPV6)) {
return "$message [$parameters->scheme://[$parameters->host]:$parameters->port]";
}
return "$message [$parameters->scheme://$parameters->host:$parameters->port]";
}
/**
* Helper method to handle connection errors.
*
* @param string $message Error message.
* @param int $code Error code.
*/
protected function onConnectionError($message, $code = null)
protected function onConnectionError($message, $code = 0)
{
CommunicationException::handle(
new ConnectionException($this, static::createExceptionMessage($message), $code)
new ConnectionException($this, "$message [{$this->getParameters()}]", $code)
);
}
@ -177,7 +153,7 @@ abstract class AbstractConnection implements NodeConnectionInterface
protected function onProtocolError($message)
{
CommunicationException::handle(
new ProtocolException($this, static::createExceptionMessage($message))
new ProtocolException($this, "$message [{$this->getParameters()}]")
);
}
@ -234,6 +210,6 @@ abstract class AbstractConnection implements NodeConnectionInterface
*/
public function __sleep()
{
return array('parameters', 'initCommands');
return ['parameters', 'initCommands'];
}
}

24
plugins/cache-redis/Predis/Connection/Aggregate/ClusterInterface.php
View file

@ -1,24 +0,0 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Connection\Aggregate;
use Predis\Connection\AggregateConnectionInterface;
/**
* Defines a cluster of Redis servers formed by aggregating multiple connection
* instances to single Redis nodes.
*
* @author Daniele Alessandri <suppakilla@gmail.com>
*/
interface ClusterInterface extends AggregateConnectionInterface
{
}

264
plugins/cache-redis/Predis/Connection/Aggregate/MasterSlaveReplication.php
View file

@ -1,264 +0,0 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Connection\Aggregate;
use Predis\Command\CommandInterface;
use Predis\Connection\NodeConnectionInterface;
use Predis\Replication\ReplicationStrategy;
/**
* Aggregate connection handling replication of Redis nodes configured in a
* single master / multiple slaves setup.
*
* @author Daniele Alessandri <suppakilla@gmail.com>
*/
class MasterSlaveReplication implements ReplicationInterface
{
protected $strategy;
protected $master;
protected $slaves;
protected $current;
/**
* {@inheritdoc}
*/
public function __construct(ReplicationStrategy $strategy = null)
{
$this->slaves = array();
$this->strategy = $strategy ?: new ReplicationStrategy();
}
/**
* Checks if one master and at least one slave have been defined.
*/
protected function check()
{
if (!isset($this->master) || !$this->slaves) {
throw new \RuntimeException('Replication needs one master and at least one slave.');
}
}
/**
* Resets the connection state.
*/
protected function reset()
{
$this->current = null;
}
/**
* {@inheritdoc}
*/
public function add(NodeConnectionInterface $connection)
{
$alias = $connection->getParameters()->alias;
if ($alias === 'master') {
$this->master = $connection;
} else {
$this->slaves[$alias ?: count($this->slaves)] = $connection;
}
$this->reset();
}
/**
* {@inheritdoc}
*/
public function remove(NodeConnectionInterface $connection)
{
if ($connection->getParameters()->alias === 'master') {
$this->master = null;
$this->reset();
return true;
} else {
if (($id = array_search($connection, $this->slaves, true)) !== false) {
unset($this->slaves[$id]);
$this->reset();
return true;
}
}
return false;
}
/**
* {@inheritdoc}
*/
public function getConnection(CommandInterface $command)
{
if ($this->current === null) {
$this->check();
$this->current = $this->strategy->isReadOperation($command)
? $this->pickSlave()
: $this->master;
return $this->current;
}
if ($this->current === $this->master) {
return $this->current;
}
if (!$this->strategy->isReadOperation($command)) {
$this->current = $this->master;
}
return $this->current;
}
/**
* {@inheritdoc}
*/
public function getConnectionById($connectionId)
{
if ($connectionId === 'master') {
return $this->master;
}
if (isset($this->slaves[$connectionId])) {
return $this->slaves[$connectionId];
}
return;
}
/**
* {@inheritdoc}
*/
public function switchTo($connection)
{
$this->check();
if (!$connection instanceof NodeConnectionInterface) {
$connection = $this->getConnectionById($connection);
}
if ($connection !== $this->master && !in_array($connection, $this->slaves, true)) {
throw new \InvalidArgumentException('Invalid connection or connection not found.');
}
$this->current = $connection;
}
/**
* {@inheritdoc}
*/
public function getCurrent()
{
return $this->current;
}
/**
* {@inheritdoc}
*/
public function getMaster()
{
return $this->master;
}
/**
* {@inheritdoc}
*/
public function getSlaves()
{
return array_values($this->slaves);
}
/**
* Returns the underlying replication strategy.
*
* @return ReplicationStrategy
*/
public function getReplicationStrategy()
{
return $this->strategy;
}
/**
* Returns a random slave.
*
* @return NodeConnectionInterface
*/
protected function pickSlave()
{
return $this->slaves[array_rand($this->slaves)];
}
/**
* {@inheritdoc}
*/
public function isConnected()
{
return $this->current ? $this->current->isConnected() : false;
}
/**
* {@inheritdoc}
*/
public function connect()
{
if ($this->current === null) {
$this->check();
$this->current = $this->pickSlave();
}
$this->current->connect();
}
/**
* {@inheritdoc}
*/
public function disconnect()
{
if ($this->master) {
$this->master->disconnect();
}
foreach ($this->slaves as $connection) {
$connection->disconnect();
}
}
/**
* {@inheritdoc}
*/
public function writeRequest(CommandInterface $command)
{
$this->getConnection($command)->writeRequest($command);
}
/**
* {@inheritdoc}
*/
public function readResponse(CommandInterface $command)
{
return $this->getConnection($command)->readResponse($command);
}
/**
* {@inheritdoc}
*/
public function executeCommand(CommandInterface $command)
{
return $this->getConnection($command)->executeCommand($command);
}
/**
* {@inheritdoc}
*/
public function __sleep()
{
return array('master', 'slaves', 'strategy');
}
}

235
plugins/cache-redis/Predis/Connection/Aggregate/PredisCluster.php
View file

@ -1,235 +0,0 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Connection\Aggregate;
use Predis\Cluster\PredisStrategy;
use Predis\Cluster\StrategyInterface;
use Predis\Command\CommandInterface;
use Predis\Connection\NodeConnectionInterface;
use Predis\NotSupportedException;
/**
* Abstraction for a cluster of aggregate connections to various Redis servers
* implementing client-side sharding based on pluggable distribution strategies.
*
* @author Daniele Alessandri <suppakilla@gmail.com>
*
* @todo Add the ability to remove connections from pool.
*/
class PredisCluster implements ClusterInterface, \IteratorAggregate, \Countable
{
private $pool;
private $strategy;
private $distributor;
/**
* @param StrategyInterface $strategy Optional cluster strategy.
*/
public function __construct(StrategyInterface $strategy = null)
{
$this->pool = array();
$this->strategy = $strategy ?: new PredisStrategy();
$this->distributor = $this->strategy->getDistributor();
}
/**
* {@inheritdoc}
*/
public function isConnected()
{
foreach ($this->pool as $connection) {
if ($connection->isConnected()) {
return true;
}
}
return false;
}
/**
* {@inheritdoc}
*/
public function connect()
{
foreach ($this->pool as $connection) {
$connection->connect();
}
}
/**
* {@inheritdoc}
*/
public function disconnect()
{
foreach ($this->pool as $connection) {
$connection->disconnect();
}
}
/**
* {@inheritdoc}
*/
public function add(NodeConnectionInterface $connection)
{
$parameters = $connection->getParameters();
if (isset($parameters->alias)) {
$this->pool[$parameters->alias] = $connection;
} else {
$this->pool[] = $connection;
}
$weight = isset($parameters->weight) ? $parameters->weight : null;
$this->distributor->add($connection, $weight);
}
/**
* {@inheritdoc}
*/
public function remove(NodeConnectionInterface $connection)
{
if (($id = array_search($connection, $this->pool, true)) !== false) {
unset($this->pool[$id]);
$this->distributor->remove($connection);
return true;
}
return false;
}
/**
* Removes a connection instance using its alias or index.
*
* @param string $connectionID Alias or index of a connection.
*
* @return bool Returns true if the connection was in the pool.
*/
public function removeById($connectionID)
{
if ($connection = $this->getConnectionById($connectionID)) {
return $this->remove($connection);
}
return false;
}
/**
* {@inheritdoc}
*/
public function getConnection(CommandInterface $command)
{
$slot = $this->strategy->getSlot($command);
if (!isset($slot)) {
throw new NotSupportedException(
"Cannot use '{$command->getId()}' over clusters of connections."
);
}
$node = $this->distributor->getBySlot($slot);
return $node;
}
/**
* {@inheritdoc}
*/
public function getConnectionById($connectionID)
{
return isset($this->pool[$connectionID]) ? $this->pool[$connectionID] : null;
}
/**
* Retrieves a connection instance from the cluster using a key.
*
* @param string $key Key string.
*
* @return NodeConnectionInterface
*/
public function getConnectionByKey($key)
{
$hash = $this->strategy->getSlotByKey($key);
$node = $this->distributor->getBySlot($hash);
return $node;
}
/**
* Returns the underlying command hash strategy used to hash commands by
* using keys found in their arguments.
*
* @return StrategyInterface
*/
public function getClusterStrategy()
{
return $this->strategy;
}
/**
* {@inheritdoc}
*/
public function count()
{
return count($this->pool);
}
/**
* {@inheritdoc}
*/
public function getIterator()
{
return new \ArrayIterator($this->pool);
}
/**
* {@inheritdoc}
*/
public function writeRequest(CommandInterface $command)
{
$this->getConnection($command)->writeRequest($command);
}
/**
* {@inheritdoc}
*/
public function readResponse(CommandInterface $command)
{
return $this->getConnection($command)->readResponse($command);
}
/**
* {@inheritdoc}
*/
public function executeCommand(CommandInterface $command)
{
return $this->getConnection($command)->executeCommand($command);
}
/**
* Executes the specified Redis command on all the nodes of a cluster.
*
* @param CommandInterface $command A Redis command.
*
* @return array
*/
public function executeCommandOnNodes(CommandInterface $command)
{
$responses = array();
foreach ($this->pool as $connection) {
$responses[] = $connection->executeCommand($command);
}
return $responses;
}
}

553
plugins/cache-redis/Predis/Connection/Aggregate/RedisCluster.php
View file

@ -1,553 +0,0 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Connection\Aggregate;
use Predis\Cluster\RedisStrategy as RedisClusterStrategy;
use Predis\Cluster\StrategyInterface;
use Predis\Command\CommandInterface;
use Predis\Command\RawCommand;
use Predis\Connection\FactoryInterface;
use Predis\Connection\NodeConnectionInterface;
use Predis\NotSupportedException;
use Predis\Response\ErrorInterface as ErrorResponseInterface;
/**
* Abstraction for a Redis-backed cluster of nodes (Redis >= 3.0.0).
*
* This connection backend offers smart support for redis-cluster by handling
* automatic slots map (re)generation upon -MOVED or -ASK responses returned by
* Redis when redirecting a client to a different node.
*
* The cluster can be pre-initialized using only a subset of the actual nodes in
* the cluster, Predis will do the rest by adjusting the slots map and creating
* the missing underlying connection instances on the fly.
*
* It is possible to pre-associate connections to a slots range with the "slots"
* parameter in the form "$first-$last". This can greatly reduce runtime node
* guessing and redirections.
*
* It is also possible to ask for the full and updated slots map directly to one
* of the nodes and optionally enable such a behaviour upon -MOVED redirections.
* Asking for the cluster configuration to Redis is actually done by issuing a
* CLUSTER SLOTS command to a random node in the pool.
*
* @author Daniele Alessandri <suppakilla@gmail.com>
*/
class RedisCluster implements ClusterInterface, \IteratorAggregate, \Countable
{
private $useClusterSlots = true;
private $defaultParameters = array();
private $pool = array();
private $slots = array();
private $slotsMap;
private $strategy;
private $connections;
/**
* @param FactoryInterface $connections Optional connection factory.
* @param StrategyInterface $strategy Optional cluster strategy.
*/
public function __construct(
FactoryInterface $connections,
StrategyInterface $strategy = null
) {
$this->connections = $connections;
$this->strategy = $strategy ?: new RedisClusterStrategy();
}
/**
* {@inheritdoc}
*/
public function isConnected()
{
foreach ($this->pool as $connection) {
if ($connection->isConnected()) {
return true;
}
}
return false;
}
/**
* {@inheritdoc}
*/
public function connect()
{
if ($connection = $this->getRandomConnection()) {
$connection->connect();
}
}
/**
* {@inheritdoc}
*/
public function disconnect()
{
foreach ($this->pool as $connection) {
$connection->disconnect();
}
}
/**
* {@inheritdoc}
*/
public function add(NodeConnectionInterface $connection)
{
$this->pool[(string) $connection] = $connection;
unset($this->slotsMap);
}
/**
* {@inheritdoc}
*/
public function remove(NodeConnectionInterface $connection)
{
if (false !== $id = array_search($connection, $this->pool, true)) {
unset(
$this->pool[$id],
$this->slotsMap
);
return true;
}
return false;
}
/**
* Removes a connection instance by using its identifier.
*
* @param string $connectionID Connection identifier.
*
* @return bool True if the connection was in the pool.
*/
public function removeById($connectionID)
{
if (isset($this->pool[$connectionID])) {
unset(
$this->pool[$connectionID],
$this->slotsMap
);
return true;
}
return false;
}
/**
* Generates the current slots map by guessing the cluster configuration out
* of the connection parameters of the connections in the pool.
*
* Generation is based on the same algorithm used by Redis to generate the
* cluster, so it is most effective when all of the connections supplied on
* initialization have the "slots" parameter properly set accordingly to the
* current cluster configuration.
*/
public function buildSlotsMap()
{
$this->slotsMap = array();
foreach ($this->pool as $connectionID => $connection) {
$parameters = $connection->getParameters();
if (!isset($parameters->slots)) {
continue;
}
$slots = explode('-', $parameters->slots, 2);
$this->setSlots($slots[0], $slots[1], $connectionID);
}
}
/**
* Generates an updated slots map fetching the cluster configuration using
* the CLUSTER SLOTS command against the specified node or a random one from
* the pool.
*
* @param NodeConnectionInterface $connection Optional connection instance.
*
* @return array
*/
public function askSlotsMap(NodeConnectionInterface $connection = null)
{
if (!$connection && !$connection = $this->getRandomConnection()) {
return array();
}
$command = RawCommand::create('CLUSTER', 'SLOTS');
$response = $connection->executeCommand($command);
foreach ($response as $slots) {
// We only support master servers for now, so we ignore subsequent
// elements in the $slots array identifying slaves.
list($start, $end, $master) = $slots;
if ($master[0] === '') {
$this->setSlots($start, $end, (string) $connection);
} else {
$this->setSlots($start, $end, "{$master[0]}:{$master[1]}");
}
}
return $this->slotsMap;
}
/**
* Returns the current slots map for the cluster.
*
* @return array
*/
public function getSlotsMap()
{
if (!isset($this->slotsMap)) {
$this->slotsMap = array();
}
return $this->slotsMap;
}
/**
* Pre-associates a connection to a slots range to avoid runtime guessing.
*
* @param int $first Initial slot of the range.
* @param int $last Last slot of the range.
* @param NodeConnectionInterface|string $connection ID or connection instance.
*
* @throws \OutOfBoundsException
*/
public function setSlots($first, $last, $connection)
{
if ($first < 0x0000 || $first > 0x3FFF ||
$last < 0x0000 || $last > 0x3FFF ||
$last < $first
) {
throw new \OutOfBoundsException(
"Invalid slot range for $connection: [$first-$last]."
);
}
$slots = array_fill($first, $last - $first + 1, (string) $connection);
$this->slotsMap = $this->getSlotsMap() + $slots;
}
/**
* Guesses the correct node associated to a given slot using a precalculated
* slots map, falling back to the same logic used by Redis to initialize a
* cluster (best-effort).
*
* @param int $slot Slot index.
*
* @return string Connection ID.
*/
protected function guessNode($slot)
{
if (!isset($this->slotsMap)) {
$this->buildSlotsMap();
}
if (isset($this->slotsMap[$slot])) {
return $this->slotsMap[$slot];
}
$count = count($this->pool);
$index = min((int) ($slot / (int) (16384 / $count)), $count - 1);
$nodes = array_keys($this->pool);
return $nodes[$index];
}
/**
* Creates a new connection instance from the given connection ID.
*
* @param string $connectionID Identifier for the connection.
*
* @return NodeConnectionInterface
*/
protected function createConnection($connectionID)
{
$separator = strrpos($connectionID, ':');
$parameters = array_merge($this->defaultParameters, array(
'host' => substr($connectionID, 0, $separator),
'port' => substr($connectionID, $separator + 1),
));
$connection = $this->connections->create($parameters);
return $connection;
}
/**
* {@inheritdoc}
*/
public function getConnection(CommandInterface $command)
{
$slot = $this->strategy->getSlot($command);
if (!isset($slot)) {
throw new NotSupportedException(
"Cannot use '{$command->getId()}' with redis-cluster."
);
}
if (isset($this->slots[$slot])) {
return $this->slots[$slot];
} else {
return $this->getConnectionBySlot($slot);
}
}
/**
* Returns the connection currently associated to a given slot.
*
* @param int $slot Slot index.
*
* @throws \OutOfBoundsException
*
* @return NodeConnectionInterface
*/
public function getConnectionBySlot($slot)
{
if ($slot < 0x0000 || $slot > 0x3FFF) {
throw new \OutOfBoundsException("Invalid slot [$slot].");
}
if (isset($this->slots[$slot])) {
return $this->slots[$slot];
}
$connectionID = $this->guessNode($slot);
if (!$connection = $this->getConnectionById($connectionID)) {
$connection = $this->createConnection($connectionID);
$this->pool[$connectionID] = $connection;
}
return $this->slots[$slot] = $connection;
}
/**
* {@inheritdoc}
*/
public function getConnectionById($connectionID)
{
if (isset($this->pool[$connectionID])) {
return $this->pool[$connectionID];
}
}
/**
* Returns a random connection from the pool.
*
* @return NodeConnectionInterface|null
*/
protected function getRandomConnection()
{
if ($this->pool) {
return $this->pool[array_rand($this->pool)];
}
}
/**
* Permanently associates the connection instance to a new slot.
* The connection is added to the connections pool if not yet included.
*
* @param NodeConnectionInterface $connection Connection instance.
* @param int $slot Target slot index.
*/
protected function move(NodeConnectionInterface $connection, $slot)
{
$this->pool[(string) $connection] = $connection;
$this->slots[(int) $slot] = $connection;
}
/**
* Handles -ERR responses returned by Redis.
*
* @param CommandInterface $command Command that generated the -ERR response.
* @param ErrorResponseInterface $error Redis error response object.
*
* @return mixed
*/
protected function onErrorResponse(CommandInterface $command, ErrorResponseInterface $error)
{
$details = explode(' ', $error->getMessage(), 2);
switch ($details[0]) {
case 'MOVED':
return $this->onMovedResponse($command, $details[1]);
case 'ASK':
return $this->onAskResponse($command, $details[1]);
default:
return $error;
}
}
/**
* Handles -MOVED responses by executing again the command against the node
* indicated by the Redis response.
*
* @param CommandInterface $command Command that generated the -MOVED response.
* @param string $details Parameters of the -MOVED response.
*
* @return mixed
*/
protected function onMovedResponse(CommandInterface $command, $details)
{
list($slot, $connectionID) = explode(' ', $details, 2);
if (!$connection = $this->getConnectionById($connectionID)) {
$connection = $this->createConnection($connectionID);
}
if ($this->useClusterSlots) {
$this->askSlotsMap($connection);
}
$this->move($connection, $slot);
$response = $this->executeCommand($command);
return $response;
}
/**
* Handles -ASK responses by executing again the command against the node
* indicated by the Redis response.
*
* @param CommandInterface $command Command that generated the -ASK response.
* @param string $details Parameters of the -ASK response.
*
* @return mixed
*/
protected function onAskResponse(CommandInterface $command, $details)
{
list($slot, $connectionID) = explode(' ', $details, 2);
if (!$connection = $this->getConnectionById($connectionID)) {
$connection = $this->createConnection($connectionID);
}
$connection->executeCommand(RawCommand::create('ASKING'));
$response = $connection->executeCommand($command);
return $response;
}
/**
* {@inheritdoc}
*/
public function writeRequest(CommandInterface $command)
{
$this->getConnection($command)->writeRequest($command);
}
/**
* {@inheritdoc}
*/
public function readResponse(CommandInterface $command)
{
return $this->getConnection($command)->readResponse($command);
}
/**
* {@inheritdoc}
*/
public function executeCommand(CommandInterface $command)
{
$connection = $this->getConnection($command);
$response = $connection->executeCommand($command);
if ($response instanceof ErrorResponseInterface) {
return $this->onErrorResponse($command, $response);
}
return $response;
}
/**
* {@inheritdoc}
*/
public function count()
{
return count($this->pool);
}
/**
* {@inheritdoc}
*/
public function getIterator()
{
return new \ArrayIterator(array_values($this->pool));
}
/**
* Returns the underlying command hash strategy used to hash commands by
* using keys found in their arguments.
*
* @return StrategyInterface
*/
public function getClusterStrategy()
{
return $this->strategy;
}
/**
* Returns the underlying connection factory used to create new connection
* instances to Redis nodes indicated by redis-cluster.
*
* @return FactoryInterface
*/
public function getConnectionFactory()
{
return $this->connections;
}
/**
* Enables automatic fetching of the current slots map from one of the nodes
* using the CLUSTER SLOTS command. This option is disabled by default but
* asking the current slots map to Redis upon -MOVED responses may reduce
* overhead by eliminating the trial-and-error nature of the node guessing
* procedure, mostly when targeting many keys that would end up in a lot of
* redirections.
*
* The slots map can still be manually fetched using the askSlotsMap()
* method whether or not this option is enabled.
*
* @param bool $value Enable or disable the use of CLUSTER SLOTS.
*/
public function useClusterSlots($value)
{
$this->useClusterSlots = (bool) $value;
}
/**
* Sets a default array of connection parameters to be applied when creating
* new connection instances on the fly when they are not part of the initial
* pool supplied upon cluster initialization.
*
* These parameters are not applied to connections added to the pool using
* the add() method.
*
* @param array $parameters Array of connection parameters.
*/
public function setDefaultParameters(array $parameters)
{
$this->defaultParameters = array_merge(
$this->defaultParameters,
$parameters ?: array()
);
}
}

52
plugins/cache-redis/Predis/Connection/Aggregate/ReplicationInterface.php
View file

@ -1,52 +0,0 @@
<?php
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Predis\Connection\Aggregate;
use Predis\Connection\AggregateConnectionInterface;
use Predis\Connection\NodeConnectionInterface;
/**
* Defines a group of Redis nodes in a master / slave replication setup.
*
* @author Daniele Alessandri <suppakilla@gmail.com>
*/
interface ReplicationInterface extends AggregateConnectionInterface
{
/**
* Switches the internal connection instance in use.
*
* @param string $connection Alias of a connection
*/
public function switchTo($connection);
/**
* Returns the connection instance currently in use by the aggregate
* connection.
*
* @return NodeConnectionInterface
*/
public function getCurrent();
/**
* Returns the connection instance for the master Redis node.
*
* @return NodeConnectionInterface
*/
public function getMaster();
/**
* Returns a list of connection instances to slave nodes.
*
* @return NodeConnectionInterface
*/
public function getSlaves();
}

7
plugins/cache-redis/Predis/Connection/AggregateConnectionInterface.php
View file

@ -3,7 +3,8 @@
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
@ -16,8 +17,6 @@ use Predis\Command\CommandInterface;
/**
* Defines a virtual connection composed of multiple connection instances to
* single Redis nodes.
*
* @author Daniele Alessandri <suppakilla@gmail.com>
*/
interface AggregateConnectionInterface extends ConnectionInterface
{
@ -44,7 +43,7 @@ interface AggregateConnectionInterface extends ConnectionInterface
*
* @return NodeConnectionInterface
*/
public function getConnection(CommandInterface $command);
public function getConnectionByCommand(CommandInterface $command);
/**
* Returns a connection instance from the aggregate connection by its alias.

9
plugins/cache-redis/Predis/Connection/CompositeConnectionInterface.php
View file

@ -3,7 +3,8 @@
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
@ -14,8 +15,6 @@ namespace Predis\Connection;
/**
* Defines a connection to communicate with a single Redis server that leverages
* an external protocol processor to handle pluggable protocol handlers.
*
* @author Daniele Alessandri <suppakilla@gmail.com>
*/
interface CompositeConnectionInterface extends NodeConnectionInterface
{
@ -34,7 +33,7 @@ interface CompositeConnectionInterface extends NodeConnectionInterface
/**
* Reads the given number of bytes from the connection.
*
* @param int $length Number of bytes to read from the connection.
* @param int $length Number of bytes to read from the connection.
*
* @return string
*/
@ -43,7 +42,7 @@ interface CompositeConnectionInterface extends NodeConnectionInterface
/**
* Reads a line from the connection.
*
* @param string
* @return string
*/
public function readLine();
}

10
plugins/cache-redis/Predis/Connection/CompositeStreamConnection.php
View file

@ -3,7 +3,8 @@
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
@ -11,6 +12,7 @@
namespace Predis\Connection;
use InvalidArgumentException;
use Predis\Command\CommandInterface;
use Predis\Protocol\ProtocolProcessorInterface;
use Predis\Protocol\Text\ProtocolProcessor as TextProtocolProcessor;
@ -18,8 +20,6 @@ use Predis\Protocol\Text\ProtocolProcessor as TextProtocolProcessor;
/**
* Connection abstraction to Redis servers based on PHP's stream that uses an
* external protocol processor defining the protocol used for the communication.
*
* @author Daniele Alessandri <suppakilla@gmail.com>
*/
class CompositeStreamConnection extends StreamConnection implements CompositeConnectionInterface
{
@ -59,7 +59,7 @@ class CompositeStreamConnection extends StreamConnection implements CompositeCon
public function readBuffer($length)
{
if ($length <= 0) {
throw new \InvalidArgumentException('Length parameter must be greater than 0.');
throw new InvalidArgumentException('Length parameter must be greater than 0.');
}
$value = '';
@ -120,6 +120,6 @@ class CompositeStreamConnection extends StreamConnection implements CompositeCon
*/
public function __sleep()
{
return array_merge(parent::__sleep(), array('protocol'));
return array_merge(parent::__sleep(), ['protocol']);
}
}

5
plugins/cache-redis/Predis/Connection/ConnectionException.php
View file

@ -3,7 +3,8 @@
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
@ -15,8 +16,6 @@ use Predis\CommunicationException;
/**
* Exception class that identifies connection-related errors.
*
* @author Daniele Alessandri <suppakilla@gmail.com>
*/
class ConnectionException extends CommunicationException
{

5
plugins/cache-redis/Predis/Connection/ConnectionInterface.php
View file

@ -3,7 +3,8 @@
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
@ -16,8 +17,6 @@ use Predis\Command\CommandInterface;
/**
* Defines a connection object used to communicate with one or multiple
* Redis servers.
*
* @author Daniele Alessandri <suppakilla@gmail.com>
*/
interface ConnectionInterface
{

87
plugins/cache-redis/Predis/Connection/Factory.php
View file

@ -3,7 +3,8 @@
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
@ -11,21 +12,27 @@
namespace Predis\Connection;
use InvalidArgumentException;
use Predis\Client;
use Predis\Command\RawCommand;
use ReflectionClass;
use UnexpectedValueException;
/**
* Standard connection factory for creating connections to Redis nodes.
*
* @author Daniele Alessandri <suppakilla@gmail.com>
*/
class Factory implements FactoryInterface
{
protected $schemes = array(
private $defaults = [];
protected $schemes = [
'tcp' => 'Predis\Connection\StreamConnection',
'unix' => 'Predis\Connection\StreamConnection',
'tls' => 'Predis\Connection\StreamConnection',
'redis' => 'Predis\Connection\StreamConnection',
'rediss' => 'Predis\Connection\StreamConnection',
'http' => 'Predis\Connection\WebdisConnection',
);
];
/**
* Checks if the provided argument represents a valid connection class
@ -34,9 +41,8 @@ class Factory implements FactoryInterface
*
* @param mixed $initializer FQN of a connection class or a callable for lazy initialization.
*
* @throws \InvalidArgumentException
*
* @return mixed
* @throws InvalidArgumentException
*/
protected function checkInitializer($initializer)
{
@ -44,10 +50,10 @@ class Factory implements FactoryInterface
return $initializer;
}
$class = new \ReflectionClass($initializer);
$class = new ReflectionClass($initializer);
if (!$class->isSubclassOf('Predis\Connection\NodeConnectionInterface')) {
throw new \InvalidArgumentException(
throw new InvalidArgumentException(
'A connection initializer must be a valid connection class or a callable object.'
);
}
@ -83,7 +89,7 @@ class Factory implements FactoryInterface
$scheme = $parameters->scheme;
if (!isset($this->schemes[$scheme])) {
throw new \InvalidArgumentException("Unknown connection scheme: '$scheme'.");
throw new InvalidArgumentException("Unknown connection scheme: '$scheme'.");
}
$initializer = $this->schemes[$scheme];
@ -96,8 +102,8 @@ class Factory implements FactoryInterface
}
if (!$connection instanceof NodeConnectionInterface) {
throw new \UnexpectedValueException(
'Objects returned by connection initializers must implement '.
throw new UnexpectedValueException(
'Objects returned by connection initializers must implement ' .
"'Predis\Connection\NodeConnectionInterface'."
);
}
@ -106,13 +112,26 @@ class Factory implements FactoryInterface
}
/**
* {@inheritdoc}
* Assigns a default set of parameters applied to new connections.
*
* The set of parameters passed to create a new connection have precedence
* over the default values set for the connection factory.
*
* @param array $parameters Set of connection parameters.
*/
public function aggregate(AggregateConnectionInterface $connection, array $parameters)
public function setDefaultParameters(array $parameters)
{
foreach ($parameters as $node) {
$connection->add($node instanceof NodeConnectionInterface ? $node : $this->create($node));
}
$this->defaults = $parameters;
}
/**
* Returns the default set of parameters applied to new connections.
*
* @return array
*/
public function getDefaultParameters()
{
return $this->defaults;
}
/**
@ -124,7 +143,17 @@ class Factory implements FactoryInterface
*/
protected function createParameters($parameters)
{
return Parameters::create($parameters);
if (is_string($parameters)) {
$parameters = Parameters::parse($parameters);
} else {
$parameters = $parameters ?: [];
}
if ($this->defaults) {
$parameters += $this->defaults;
}
return new Parameters($parameters);
}
/**
@ -136,15 +165,29 @@ class Factory implements FactoryInterface
{
$parameters = $connection->getParameters();
if (isset($parameters->password)) {
if (isset($parameters->password) && strlen($parameters->password)) {
$cmdAuthArgs = isset($parameters->username) && strlen($parameters->username)
? [$parameters->username, $parameters->password]
: [$parameters->password];
$connection->addConnectCommand(
new RawCommand(array('AUTH', $parameters->password))
new RawCommand('AUTH', $cmdAuthArgs)
);
}
if (isset($parameters->database)) {
if ($parameters->client_info ?? false && !$connection instanceof RelayConnection) {
$connection->addConnectCommand(
new RawCommand(array('SELECT', $parameters->database))
new RawCommand('CLIENT', ['SETINFO', 'LIB-NAME', 'predis'])
);
$connection->addConnectCommand(
new RawCommand('CLIENT', ['SETINFO', 'LIB-VER', Client::VERSION])
);
}
if (isset($parameters->database) && strlen($parameters->database)) {
$connection->addConnectCommand(
new RawCommand('SELECT', [$parameters->database])
);
}
}

13
plugins/cache-redis/Predis/Connection/FactoryInterface.php
View file

@ -3,7 +3,8 @@
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
@ -13,8 +14,6 @@ namespace Predis\Connection;
/**
* Interface for classes providing a factory of connections to Redis nodes.
*
* @author Daniele Alessandri <suppakilla@gmail.com>
*/
interface FactoryInterface
{
@ -41,12 +40,4 @@ interface FactoryInterface
* @return NodeConnectionInterface
*/
public function create($parameters);
/**
* Aggregates single connections into an aggregate connection instance.
*
* @param AggregateConnectionInterface $aggregate Aggregate connection instance.
* @param array $parameters List of parameters for each connection.
*/
public function aggregate(AggregateConnectionInterface $aggregate, array $parameters);
}

5
plugins/cache-redis/Predis/Connection/NodeConnectionInterface.php
View file

@ -3,7 +3,8 @@
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
@ -15,8 +16,6 @@ use Predis\Command\CommandInterface;
/**
* Defines a connection used to communicate with a single Redis node.
*
* @author Daniele Alessandri <suppakilla@gmail.com>
*/
interface NodeConnectionInterface extends ConnectionInterface
{

89
plugins/cache-redis/Predis/Connection/Parameters.php
View file

@ -3,7 +3,8 @@
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
@ -11,40 +12,49 @@
namespace Predis\Connection;
use InvalidArgumentException;
/**
* Container for connection parameters used to initialize connections to Redis.
*
* {@inheritdoc}
*
* @author Daniele Alessandri <suppakilla@gmail.com>
*/
class Parameters implements ParametersInterface
{
private $parameters;
private static $defaults = array(
protected static $defaults = [
'scheme' => 'tcp',
'host' => '127.0.0.1',
'port' => 6379,
'timeout' => 5.0,
);
];
/**
* Set of connection parameters already filtered
* for NULL or 0-length string values.
*
* @var array
*/
protected $parameters;
/**
* @param array $parameters Named array of connection parameters.
*/
public function __construct(array $parameters = array())
public function __construct(array $parameters = [])
{
$this->parameters = $this->filter($parameters) + $this->getDefaults();
$this->parameters = $this->filter($parameters + static::$defaults);
}
/**
* Returns some default parameters with their values.
* Filters parameters removing entries with NULL or 0-length string values.
*
* @params array $parameters Array of parameters to be filtered
*
* @return array
*/
protected function getDefaults()
protected function filter(array $parameters)
{
return self::$defaults;
return array_filter($parameters, function ($value) {
return $value !== null && $value !== '';
});
}
/**
@ -61,7 +71,7 @@ class Parameters implements ParametersInterface
$parameters = static::parse($parameters);
}
return new static($parameters ?: array());
return new static($parameters ?: []);
}
/**
@ -73,24 +83,24 @@ class Parameters implements ParametersInterface
* database number in the "path" part these values override the values of
* "password" and "database" if they are present in the "query" part.
*
* @link http://www.iana.org/assignments/uri-schemes/prov/redis
* @link http://www.iana.org/assignments/uri-schemes/prov/redis
* @see http://www.iana.org/assignments/uri-schemes/prov/redis
* @see http://www.iana.org/assignments/uri-schemes/prov/rediss
*
* @param string $uri URI string.
*
* @throws \InvalidArgumentException
*
* @return array
* @throws InvalidArgumentException
*/
public static function parse($uri)
{
if (stripos($uri, 'unix') === 0) {
// Hack to support URIs for UNIX sockets with minimal effort.
$uri = str_ireplace('unix:///', 'unix://localhost/', $uri);
if (stripos($uri, 'unix://') === 0) {
// parse_url() can parse unix:/path/to/sock so we do not need the
// unix:///path/to/sock hack, we will support it anyway until 2.0.
$uri = str_ireplace('unix://', 'unix:', $uri);
}
if (!$parsed = parse_url($uri)) {
throw new \InvalidArgumentException("Invalid parameters URI: $uri");
throw new InvalidArgumentException("Invalid parameters URI: $uri");
}
if (
@ -109,8 +119,17 @@ class Parameters implements ParametersInterface
}
if (stripos($uri, 'redis') === 0) {
if (isset($parsed['user'])) {
if (strlen($parsed['user'])) {
$parsed['username'] = $parsed['user'];
}
unset($parsed['user']);
}
if (isset($parsed['pass'])) {
$parsed['password'] = $parsed['pass'];
if (strlen($parsed['pass'])) {
$parsed['password'] = $parsed['pass'];
}
unset($parsed['pass']);
}
@ -129,15 +148,11 @@ class Parameters implements ParametersInterface
}
/**
* Validates and converts each value of the connection parameters array.
*
* @param array $parameters Connection parameters.
*
* @return array
* {@inheritdoc}
*/
protected function filter(array $parameters)
public function toArray()
{
return $parameters ?: array();
return $this->parameters;
}
/**
@ -161,9 +176,17 @@ class Parameters implements ParametersInterface
/**
* {@inheritdoc}
*/
public function toArray()
public function __toString()
{
return $this->parameters;
if ($this->scheme === 'unix') {
return "$this->scheme:$this->path";
}
if (filter_var($this->host, FILTER_VALIDATE_IP, FILTER_FLAG_IPV6)) {
return "$this->scheme://[$this->host]:$this->port";
}
return "$this->scheme://$this->host:$this->port";
}
/**
@ -171,6 +194,6 @@ class Parameters implements ParametersInterface
*/
public function __sleep()
{
return array('parameters');
return ['parameters'];
}
}

40
plugins/cache-redis/Predis/Connection/ParametersInterface.php
View file

@ -3,7 +3,8 @@
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
@ -18,20 +19,22 @@ namespace Predis\Connection;
* each connection backend class (please refer to their specific documentation),
* but the most common parameters used through the library are:
*
* @property-read string scheme Connection scheme, such as 'tcp' or 'unix'.
* @property-read string host IP address or hostname of Redis.
* @property-read int port TCP port on which Redis is listening to.
* @property-read string path Path of a UNIX domain socket file.
* @property-read string alias Alias for the connection.
* @property-read float timeout Timeout for the connect() operation.
* @property-read float read_write_timeout Timeout for read() and write() operations.
* @property-read bool async_connect Performs the connect() operation asynchronously.
* @property-read bool tcp_nodelay Toggles the Nagle's algorithm for coalescing.
* @property-read bool persistent Leaves the connection open after a GC collection.
* @property-read string password Password to access Redis (see the AUTH command).
* @property-read string database Database index (see the SELECT command).
*
* @author Daniele Alessandri <suppakilla@gmail.com>
* @property string $scheme Connection scheme, such as 'tcp' or 'unix'.
* @property string $host IP address or hostname of Redis.
* @property int $port TCP port on which Redis is listening to.
* @property string $path Path of a UNIX domain socket file.
* @property string $alias Alias for the connection.
* @property float $timeout Timeout for the connect() operation.
* @property float $read_write_timeout Timeout for read() and write() operations.
* @property bool $persistent Leaves the connection open after a GC collection.
* @property string $password Password to access Redis (see the AUTH command).
* @property string $database Database index (see the SELECT command).
* @property bool $async_connect Performs the connect() operation asynchronously.
* @property bool $tcp_nodelay Toggles the Nagle's algorithm for coalescing.
* @property bool $client_info Whether to set LIB-NAME and LIB-VER when connecting.
* @property bool $cache (Relay only) Whether to use in-memory caching.
* @property string $serializer (Relay only) Serializer used for data serialization.
* @property string $compression (Relay only) Algorithm used for data compression.
*/
interface ParametersInterface
{
@ -53,6 +56,13 @@ interface ParametersInterface
*/
public function __get($parameter);
/**
* Returns basic connection parameters as a valid URI string.
*
* @return string
*/
public function __toString();
/**
* Returns an array representation of the connection parameters.
*

79
plugins/cache-redis/Predis/Connection/PhpiredisSocketConnection.php
View file

@ -3,7 +3,8 @@
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
@ -11,9 +12,12 @@
namespace Predis\Connection;
use Closure;
use InvalidArgumentException;
use Predis\Command\CommandInterface;
use Predis\NotSupportedException;
use Predis\Response\Error as ErrorResponse;
use Predis\Response\ErrorInterface as ErrorResponseInterface;
use Predis\Response\Status as StatusResponse;
/**
@ -36,12 +40,11 @@ use Predis\Response\Status as StatusResponse;
* - host: hostname or IP address of the server.
* - port: TCP port of the server.
* - path: path of a UNIX domain socket when scheme is 'unix'.
* - timeout: timeout to perform the connection.
* - timeout: timeout to perform the connection (default is 5 seconds).
* - read_write_timeout: timeout of read / write operations.
*
* @link http://github.com/nrk/phpiredis
*
* @author Daniele Alessandri <suppakilla@gmail.com>
* @see http://github.com/nrk/phpiredis
* @deprecated 2.1.2
*/
class PhpiredisSocketConnection extends AbstractConnection
{
@ -65,9 +68,9 @@ class PhpiredisSocketConnection extends AbstractConnection
*/
public function __destruct()
{
phpiredis_reader_destroy($this->reader);
parent::__destruct();
phpiredis_reader_destroy($this->reader);
}
/**
@ -93,7 +96,15 @@ class PhpiredisSocketConnection extends AbstractConnection
*/
protected function assertParameters(ParametersInterface $parameters)
{
parent::assertParameters($parameters);
switch ($parameters->scheme) {
case 'tcp':
case 'redis':
case 'unix':
break;
default:
throw new InvalidArgumentException("Invalid scheme: '$parameters->scheme'.");
}
if (isset($parameters->persistent)) {
throw new NotSupportedException(
@ -132,25 +143,37 @@ class PhpiredisSocketConnection extends AbstractConnection
/**
* Returns the handler used by the protocol reader for inline responses.
*
* @return \Closure
* @return Closure
*/
private function getStatusHandler()
protected function getStatusHandler()
{
return function ($payload) {
return StatusResponse::get($payload);
};
static $statusHandler;
if (!$statusHandler) {
$statusHandler = function ($payload) {
return StatusResponse::get($payload);
};
}
return $statusHandler;
}
/**
* Returns the handler used by the protocol reader for error responses.
*
* @return \Closure
* @return Closure
*/
protected function getErrorHandler()
{
return function ($payload) {
return new ErrorResponse($payload);
};
static $errorHandler;
if (!$errorHandler) {
$errorHandler = function ($errorMessage) {
return new ErrorResponse($errorMessage);
};
}
return $errorHandler;
}
/**
@ -206,9 +229,7 @@ class PhpiredisSocketConnection extends AbstractConnection
$protocol = SOL_TCP;
}
$socket = @socket_create($domain, SOCK_STREAM, $protocol);
if (!is_resource($socket)) {
if (false === $socket = @socket_create($domain, SOCK_STREAM, $protocol)) {
$this->emitSocketError();
}
@ -241,10 +262,10 @@ class PhpiredisSocketConnection extends AbstractConnection
$timeoutSec = floor($rwtimeout);
$timeoutUsec = ($rwtimeout - $timeoutSec) * 1000000;
$timeout = array(
$timeout = [
'sec' => $timeoutSec,
'usec' => $timeoutUsec,
);
];
if (!socket_set_option($socket, SOL_SOCKET, SO_SNDTIMEO, $timeout)) {
$this->emitSocketError();
@ -263,7 +284,7 @@ class PhpiredisSocketConnection extends AbstractConnection
* @param string $address IP address (DNS-resolved from hostname)
* @param ParametersInterface $parameters Parameters used to initialize the connection.
*
* @return string
* @return void
*/
private function connectWithTimeout($socket, $address, ParametersInterface $parameters)
{
@ -280,9 +301,9 @@ class PhpiredisSocketConnection extends AbstractConnection
socket_set_block($socket);
$null = null;
$selectable = array($socket);
$selectable = [$socket];
$timeout = (float) $parameters->timeout;
$timeout = (isset($parameters->timeout) ? (float) $parameters->timeout : 5.0);
$timeoutSecs = floor($timeout);
$timeoutUSecs = ($timeout - $timeoutSecs) * 1000000;
@ -308,7 +329,11 @@ class PhpiredisSocketConnection extends AbstractConnection
{
if (parent::connect() && $this->initCommands) {
foreach ($this->initCommands as $command) {
$this->executeCommand($command);
$response = $this->executeCommand($command);
if ($response instanceof ErrorResponseInterface) {
$this->onConnectionError("`{$command->getId()}` failed: {$response->getMessage()}", 0);
}
}
}
}
@ -319,7 +344,9 @@ class PhpiredisSocketConnection extends AbstractConnection
public function disconnect()
{
if ($this->isConnected()) {
phpiredis_reader_reset($this->reader);
socket_close($this->getResource());
parent::disconnect();
}
}

96
plugins/cache-redis/Predis/Connection/PhpiredisStreamConnection.php
View file

@ -3,7 +3,8 @@
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
@ -11,6 +12,8 @@
namespace Predis\Connection;
use Closure;
use InvalidArgumentException;
use Predis\Command\CommandInterface;
use Predis\NotSupportedException;
use Predis\Response\Error as ErrorResponse;
@ -42,9 +45,8 @@ use Predis\Response\Status as StatusResponse;
* - tcp_nodelay: enables or disables Nagle's algorithm for coalescing.
* - persistent: the connection is left intact after a GC collection.
*
* @link https://github.com/nrk/phpiredis
*
* @author Daniele Alessandri <suppakilla@gmail.com>
* @see https://github.com/nrk/phpiredis
* @deprecated 2.1.2
*/
class PhpiredisStreamConnection extends StreamConnection
{
@ -67,9 +69,19 @@ class PhpiredisStreamConnection extends StreamConnection
*/
public function __destruct()
{
phpiredis_reader_destroy($this->reader);
parent::__destruct();
phpiredis_reader_destroy($this->reader);
}
/**
* {@inheritdoc}
*/
public function disconnect()
{
phpiredis_reader_reset($this->reader);
parent::disconnect();
}
/**
@ -87,24 +99,34 @@ class PhpiredisStreamConnection extends StreamConnection
/**
* {@inheritdoc}
*/
protected function tcpStreamInitializer(ParametersInterface $parameters)
protected function assertParameters(ParametersInterface $parameters)
{
switch ($parameters->scheme) {
case 'tcp':
case 'redis':
case 'unix':
break;
case 'tls':
case 'rediss':
throw new InvalidArgumentException('SSL encryption is not supported by this connection backend.');
default:
throw new InvalidArgumentException("Invalid scheme: '$parameters->scheme'.");
}
return $parameters;
}
/**
* {@inheritdoc}
*/
protected function createStreamSocket(ParametersInterface $parameters, $address, $flags)
{
$uri = "tcp://[{$parameters->host}]:{$parameters->port}";
$flags = STREAM_CLIENT_CONNECT;
$socket = null;
$timeout = (isset($parameters->timeout) ? (float) $parameters->timeout : 5.0);
$context = stream_context_create(['socket' => ['tcp_nodelay' => (bool) $parameters->tcp_nodelay]]);
if (isset($parameters->async_connect) && (bool) $parameters->async_connect) {
$flags |= STREAM_CLIENT_ASYNC_CONNECT;
}
if (isset($parameters->persistent) && (bool) $parameters->persistent) {
$flags |= STREAM_CLIENT_PERSISTENT;
$uri .= strpos($path = $parameters->path, '/') === 0 ? $path : "/$path";
}
$resource = @stream_socket_client($uri, $errno, $errstr, (float) $parameters->timeout, $flags);
if (!$resource) {
if (!$resource = @stream_socket_client($address, $errno, $errstr, $timeout, $flags, $context)) {
$this->onConnectionError(trim($errstr), $errno);
}
@ -112,10 +134,10 @@ class PhpiredisStreamConnection extends StreamConnection
$rwtimeout = (float) $parameters->read_write_timeout;
$rwtimeout = $rwtimeout > 0 ? $rwtimeout : -1;
$timeout = array(
$timeout = [
'sec' => $timeoutSeconds = floor($rwtimeout),
'usec' => ($rwtimeout - $timeoutSeconds) * 1000000,
);
];
$socket = $socket ?: socket_import_stream($resource);
@socket_set_option($socket, SOL_SOCKET, SO_SNDTIMEO, $timeout);
@ -158,25 +180,37 @@ class PhpiredisStreamConnection extends StreamConnection
/**
* Returns the handler used by the protocol reader for inline responses.
*
* @return \Closure
* @return Closure
*/
protected function getStatusHandler()
{
return function ($payload) {
return StatusResponse::get($payload);
};
static $statusHandler;
if (!$statusHandler) {
$statusHandler = function ($payload) {
return StatusResponse::get($payload);
};
}
return $statusHandler;
}
/**
* Returns the handler used by the protocol reader for error responses.
*
* @return \Closure
* @return Closure
*/
protected function getErrorHandler()
{
return function ($errorMessage) {
return new ErrorResponse($errorMessage);
};
static $errorHandler;
if (!$errorHandler) {
$errorHandler = function ($errorMessage) {
return new ErrorResponse($errorMessage);
};
}
return $errorHandler;
}
/**

188
plugins/cache-redis/Predis/Connection/StreamConnection.php
View file

@ -3,7 +3,8 @@
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
@ -11,25 +12,26 @@
namespace Predis\Connection;
use InvalidArgumentException;
use Predis\Command\CommandInterface;
use Predis\Response\Error as ErrorResponse;
use Predis\Response\ErrorInterface as ErrorResponseInterface;
use Predis\Response\Status as StatusResponse;
/**
* Standard connection to Redis servers implemented on top of PHP's streams.
* The connection parameters supported by this class are:.
*
* - scheme: it can be either 'redis', 'tcp' or 'unix'.
* - scheme: it can be either 'redis', 'tcp', 'rediss', 'tls' or 'unix'.
* - host: hostname or IP address of the server.
* - port: TCP port of the server.
* - path: path of a UNIX domain socket when scheme is 'unix'.
* - timeout: timeout to perform the connection.
* - timeout: timeout to perform the connection (default is 5 seconds).
* - read_write_timeout: timeout of read / write operations.
* - async_connect: performs the connection asynchronously.
* - tcp_nodelay: enables or disables Nagle's algorithm for coalescing.
* - persistent: the connection is left intact after a GC collection.
*
* @author Daniele Alessandri <suppakilla@gmail.com>
* - ssl: context options array (see http://php.net/manual/en/context.ssl.php)
*/
class StreamConnection extends AbstractConnection
{
@ -47,6 +49,26 @@ class StreamConnection extends AbstractConnection
$this->disconnect();
}
/**
* {@inheritdoc}
*/
protected function assertParameters(ParametersInterface $parameters)
{
switch ($parameters->scheme) {
case 'tcp':
case 'redis':
case 'unix':
case 'tls':
case 'rediss':
break;
default:
throw new InvalidArgumentException("Invalid scheme: '$parameters->scheme'.");
}
return $parameters;
}
/**
* {@inheritdoc}
*/
@ -60,11 +82,44 @@ class StreamConnection extends AbstractConnection
case 'unix':
return $this->unixStreamInitializer($this->parameters);
case 'tls':
case 'rediss':
return $this->tlsStreamInitializer($this->parameters);
default:
throw new \InvalidArgumentException("Invalid scheme: '{$this->parameters->scheme}'.");
throw new InvalidArgumentException("Invalid scheme: '{$this->parameters->scheme}'.");
}
}
/**
* 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
*/
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 (!$resource = @stream_socket_client($address, $errno, $errstr, $timeout, $flags, $context)) {
$this->onConnectionError(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;
}
/**
* Initializes a TCP stream resource.
*
@ -75,42 +130,28 @@ class StreamConnection extends AbstractConnection
protected function tcpStreamInitializer(ParametersInterface $parameters)
{
if (!filter_var($parameters->host, FILTER_VALIDATE_IP, FILTER_FLAG_IPV6)) {
$uri = "tcp://$parameters->host:$parameters->port";
$address = "tcp://$parameters->host:$parameters->port";
} else {
$uri = "tcp://[$parameters->host]:$parameters->port";
$address = "tcp://[$parameters->host]:$parameters->port";
}
$flags = STREAM_CLIENT_CONNECT;
if (isset($parameters->async_connect) && (bool) $parameters->async_connect) {
if (isset($parameters->async_connect) && $parameters->async_connect) {
$flags |= STREAM_CLIENT_ASYNC_CONNECT;
}
if (isset($parameters->persistent) && (bool) $parameters->persistent) {
$flags |= STREAM_CLIENT_PERSISTENT;
$uri .= strpos($path = $parameters->path, '/') === 0 ? $path : "/$path";
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}";
}
}
}
$resource = @stream_socket_client($uri, $errno, $errstr, (float) $parameters->timeout, $flags);
if (!$resource) {
$this->onConnectionError(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);
}
if (isset($parameters->tcp_nodelay) && function_exists('socket_import_stream')) {
$socket = socket_import_stream($resource);
socket_set_option($socket, SOL_TCP, TCP_NODELAY, (int) $parameters->tcp_nodelay);
}
return $resource;
return $this->createStreamSocket($parameters, $address, $flags);
}
/**
@ -126,25 +167,56 @@ class StreamConnection extends AbstractConnection
throw new InvalidArgumentException('Missing UNIX domain socket path.');
}
$uri = "unix://{$parameters->path}";
$flags = STREAM_CLIENT_CONNECT;
if ((bool) $parameters->persistent) {
$flags |= STREAM_CLIENT_PERSISTENT;
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.'
);
}
}
}
$resource = @stream_socket_client($uri, $errno, $errstr, (float) $parameters->timeout, $flags);
return $this->createStreamSocket($parameters, "unix://{$parameters->path}", $flags);
}
if (!$resource) {
$this->onConnectionError(trim($errstr), $errno);
/**
* Initializes a SSL-encrypted TCP stream resource.
*
* @param ParametersInterface $parameters Initialization parameters for the connection.
*
* @return resource
*/
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->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);
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;
}
if (!stream_context_set_option($resource, ['ssl' => $options])) {
$this->onConnectionError('Error while setting SSL context options');
}
if (!stream_socket_enable_crypto($resource, true, $options['crypto_type'])) {
$this->onConnectionError('Error while switching to encrypted communication');
}
return $resource;
@ -157,7 +229,13 @@ class StreamConnection extends AbstractConnection
{
if (parent::connect() && $this->initCommands) {
foreach ($this->initCommands as $command) {
$this->executeCommand($command);
$response = $this->executeCommand($command);
if ($response instanceof ErrorResponseInterface && $command->getId() === 'CLIENT') {
// Do nothing on CLIENT SETINFO command failure
} elseif ($response instanceof ErrorResponseInterface) {
$this->onConnectionError("`{$command->getId()}` failed: {$response->getMessage()}", 0);
}
}
}
}
@ -168,7 +246,10 @@ class StreamConnection extends AbstractConnection
public function disconnect()
{
if ($this->isConnected()) {
fclose($this->getResource());
$resource = $this->getResource();
if (is_resource($resource)) {
fclose($resource);
}
parent::disconnect();
}
}
@ -184,7 +265,7 @@ class StreamConnection extends AbstractConnection
$socket = $this->getResource();
while (($length = strlen($buffer)) > 0) {
$written = @fwrite($socket, $buffer);
$written = is_resource($socket) ? @fwrite($socket, $buffer) : false;
if ($length === $written) {
return;
@ -228,7 +309,7 @@ class StreamConnection extends AbstractConnection
$bytesLeft = ($size += 2);
do {
$chunk = fread($socket, min($bytesLeft, 4096));
$chunk = is_resource($socket) ? fread($socket, min($bytesLeft, 4096)) : false;
if ($chunk === false || $chunk === '') {
$this->onConnectionError('Error while reading bytes from the server.');
@ -247,7 +328,7 @@ class StreamConnection extends AbstractConnection
return;
}
$multibulk = array();
$multibulk = [];
for ($i = 0; $i < $count; ++$i) {
$multibulk[$i] = $this->read();
@ -256,7 +337,9 @@ class StreamConnection extends AbstractConnection
return $multibulk;
case ':':
return (int) $payload;
$integer = (int) $payload;
return $integer == $payload ? $integer : $payload;
case '-':
return new ErrorResponse($payload);
@ -281,9 +364,8 @@ class StreamConnection extends AbstractConnection
$buffer = "*{$reqlen}\r\n\${$cmdlen}\r\n{$commandID}\r\n";
for ($i = 0, $reqlen--; $i < $reqlen; ++$i) {
$argument = $arguments[$i];
$arglen = strlen($argument);
foreach ($arguments as $argument) {
$arglen = strlen(strval($argument));
$buffer .= "\${$arglen}\r\n{$argument}\r\n";
}

69
plugins/cache-redis/Predis/Connection/WebdisConnection.php
View file

@ -3,7 +3,8 @@
/*
* This file is part of the Predis package.
*
* (c) Daniele Alessandri <suppakilla@gmail.com>
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2023 Till Krüss
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
@ -11,6 +12,8 @@
namespace Predis\Connection;
use Closure;
use InvalidArgumentException;
use Predis\Command\CommandInterface;
use Predis\NotSupportedException;
use Predis\Protocol\ProtocolException;
@ -33,15 +36,14 @@ use Predis\Response\Status as StatusResponse;
* - scheme: must be 'http'.
* - host: hostname or IP address of the server.
* - port: TCP port of the server.
* - timeout: timeout to perform the connection.
* - timeout: timeout to perform the connection (default is 5 seconds).
* - user: username for authentication.
* - pass: password for authentication.
*
* @link http://webd.is
* @link http://github.com/nicolasff/webdis
* @link http://github.com/seppo0010/phpiredis
*
* @author Daniele Alessandri <suppakilla@gmail.com>
* @see http://webd.is
* @see http://github.com/nicolasff/webdis
* @see http://github.com/seppo0010/phpiredis
* @deprecated 2.1.2
*/
class WebdisConnection implements NodeConnectionInterface
{
@ -52,14 +54,14 @@ class WebdisConnection implements NodeConnectionInterface
/**
* @param ParametersInterface $parameters Initialization parameters for the connection.
*
* @throws \InvalidArgumentException
* @throws InvalidArgumentException
*/
public function __construct(ParametersInterface $parameters)
{
$this->assertExtensions();
if ($parameters->scheme !== 'http') {
throw new \InvalidArgumentException("Invalid scheme: '{$parameters->scheme}'.");
throw new InvalidArgumentException("Invalid scheme: '{$parameters->scheme}'.");
}
$this->parameters = $parameters;
@ -117,19 +119,20 @@ class WebdisConnection implements NodeConnectionInterface
private function createCurl()
{
$parameters = $this->getParameters();
$timeout = (isset($parameters->timeout) ? (float) $parameters->timeout : 5.0) * 1000;
if (filter_var($host = $parameters->host, FILTER_VALIDATE_IP)) {
if (filter_var($host = $parameters->host, FILTER_VALIDATE_IP, FILTER_FLAG_IPV6)) {
$host = "[$host]";
}
$options = array(
$options = [
CURLOPT_FAILONERROR => true,
CURLOPT_CONNECTTIMEOUT_MS => $parameters->timeout * 1000,
CURLOPT_CONNECTTIMEOUT_MS => $timeout,
CURLOPT_URL => "$parameters->scheme://$host:$parameters->port",
CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
CURLOPT_POST => true,
CURLOPT_WRITEFUNCTION => array($this, 'feedReader'),
);
CURLOPT_WRITEFUNCTION => [$this, 'feedReader'],
];
if (isset($parameters->user, $parameters->pass)) {
$options[CURLOPT_USERPWD] = "{$parameters->user}:{$parameters->pass}";
@ -158,25 +161,37 @@ class WebdisConnection implements NodeConnectionInterface
/**
* Returns the handler used by the protocol reader for inline responses.
*
* @return \Closure
* @return Closure
*/
protected function getStatusHandler()
{
return function ($payload) {
return StatusResponse::get($payload);
};
static $statusHandler;
if (!$statusHandler) {
$statusHandler = function ($payload) {
return StatusResponse::get($payload);
};
}
return $statusHandler;
}
/**
* Returns the handler used by the protocol reader for error responses.
*
* @return \Closure
* @return Closure
*/
protected function getErrorHandler()
{
return function ($payload) {
return new ErrorResponse($payload);
};
static $errorHandler;
if (!$errorHandler) {
$errorHandler = function ($errorMessage) {
return new ErrorResponse($errorMessage);
};
}
return $errorHandler;
}
/**
@ -223,9 +238,8 @@ class WebdisConnection implements NodeConnectionInterface
*
* @param CommandInterface $command Command instance.
*
* @throws NotSupportedException
*
* @return string
* @throws NotSupportedException
*/
protected function getCommandId(CommandInterface $command)
{
@ -239,7 +253,6 @@ class WebdisConnection implements NodeConnectionInterface
case 'DISCARD':
case 'MONITOR':
throw new NotSupportedException("Command '$commandID' is not allowed by Webdis.");
default:
return $commandID;
}
@ -279,10 +292,10 @@ class WebdisConnection implements NodeConnectionInterface
curl_setopt($resource, CURLOPT_POSTFIELDS, $serializedCommand);
if (curl_exec($resource) === false) {
$error = curl_error($resource);
$error = trim(curl_error($resource));
$errno = curl_errno($resource);
throw new ConnectionException($this, trim($error), $errno);
throw new ConnectionException($this, "$error{$this->getParameters()}]", $errno);
}
if (phpiredis_reader_get_state($this->reader) !== PHPIREDIS_READER_STATE_COMPLETE) {
@ -337,7 +350,7 @@ class WebdisConnection implements NodeConnectionInterface
*/
public function __sleep()
{
return array('parameters');
return ['parameters'];
}
/**