mirror of
https://github.com/predis/predis.git
synced 2026-11-13 05:46:15 +00:00
Improved Relay integration (#1423)
* Fixed Relay objects structure (#1419) * Added OSS cluster tests (#1421) --------- Co-authored-by: Till Krüss <tillkruss@users.noreply.github.com>
This commit is contained in:
Vladyslav Vildanov
committed by
GitHub
GitHub
parent
87b0d2fa15
commit
d56e32715b
@@ -16,6 +16,7 @@ use InvalidArgumentException;
|
||||
use Predis\ClientException;
|
||||
use Predis\Command\CommandInterface;
|
||||
use Predis\NotSupportedException;
|
||||
use Predis\Response\ErrorInterface as ErrorResponseInterface;
|
||||
use Predis\Response\ServerException;
|
||||
use Relay\Exception as RelayException;
|
||||
use Relay\Relay;
|
||||
@@ -49,7 +50,7 @@ use Relay\Relay;
|
||||
*
|
||||
* @see https://github.com/cachewerk/relay
|
||||
*/
|
||||
class RelayConnection extends StreamConnection
|
||||
class RelayConnection extends AbstractConnection
|
||||
{
|
||||
use RelayMethods;
|
||||
|
||||
@@ -89,12 +90,10 @@ class RelayConnection extends StreamConnection
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
public function __construct(ParametersInterface $parameters)
|
||||
public function __construct(ParametersInterface $parameters, Relay $client)
|
||||
{
|
||||
$this->assertExtensions();
|
||||
|
||||
$this->parameters = $this->assertParameters($parameters);
|
||||
$this->client = $this->createClient();
|
||||
$this->client = $client;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -115,76 +114,6 @@ class RelayConnection extends StreamConnection
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.'
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
protected function assertParameters(ParametersInterface $parameters)
|
||||
{
|
||||
if (!in_array($parameters->scheme, ['tcp', 'tls', 'unix', 'redis', 'rediss'])) {
|
||||
throw new InvalidArgumentException("Invalid scheme: '{$parameters->scheme}'.");
|
||||
}
|
||||
|
||||
if (!in_array($parameters->serializer, [null, 'php', 'igbinary', 'msgpack', 'json'])) {
|
||||
throw new InvalidArgumentException("Invalid serializer: '{$parameters->serializer}'.");
|
||||
}
|
||||
|
||||
if (!in_array($parameters->compression, [null, 'lzf', 'lz4', 'zstd'])) {
|
||||
throw new InvalidArgumentException("Invalid compression algorithm: '{$parameters->compression}'.");
|
||||
}
|
||||
|
||||
return $parameters;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new instance of the client.
|
||||
*
|
||||
* @return Relay
|
||||
*/
|
||||
private function createClient()
|
||||
{
|
||||
$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);
|
||||
|
||||
// disable Relay's command/connection retry
|
||||
$client->setOption(Relay::OPT_MAX_RETRIES, 0);
|
||||
|
||||
// whether to use in-memory caching
|
||||
$client->setOption(Relay::OPT_USE_CACHE, $this->parameters->cache ?? true);
|
||||
|
||||
// set data serializer
|
||||
$client->setOption(Relay::OPT_SERIALIZER, constant(sprintf(
|
||||
'%s::SERIALIZER_%s',
|
||||
Relay::class,
|
||||
strtoupper($this->parameters->serializer ?? 'none')
|
||||
)));
|
||||
|
||||
// set data compression algorithm
|
||||
$client->setOption(Relay::OPT_COMPRESSION, constant(sprintf(
|
||||
'%s::COMPRESSION_%s',
|
||||
Relay::class,
|
||||
strtoupper($this->parameters->compression ?? 'none')
|
||||
)));
|
||||
|
||||
return $client;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the underlying client.
|
||||
*
|
||||
@@ -196,17 +125,12 @@ class RelayConnection extends StreamConnection
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
* @param ParametersInterface $parameters
|
||||
* @param $address
|
||||
* @param $flags
|
||||
* @return Relay
|
||||
*/
|
||||
protected function getIdentifier()
|
||||
{
|
||||
return $this->client->endpointId();
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
protected function createStreamSocket(ParametersInterface $parameters, $address, $flags)
|
||||
protected function connectWithConfiguration(ParametersInterface $parameters, $address, $flags)
|
||||
{
|
||||
$timeout = isset($parameters->timeout) ? (float) $parameters->timeout : 5.0;
|
||||
|
||||
@@ -310,14 +234,6 @@ class RelayConnection extends StreamConnection
|
||||
throw new NotSupportedException('The "relay" extension does not support writing requests.');
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
public function readResponse(CommandInterface $command)
|
||||
{
|
||||
throw new NotSupportedException('The "relay" extension does not support reading responses.');
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
@@ -329,9 +245,128 @@ class RelayConnection extends StreamConnection
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
public function __wakeup()
|
||||
protected function createResource()
|
||||
{
|
||||
$this->assertExtensions();
|
||||
$this->client = $this->createClient();
|
||||
switch ($this->parameters->scheme) {
|
||||
case 'tcp':
|
||||
case 'redis':
|
||||
return $this->initializeTcpConnection($this->parameters);
|
||||
|
||||
case 'unix':
|
||||
return $this->initializeUnixConnection($this->parameters);
|
||||
|
||||
default:
|
||||
throw new InvalidArgumentException("Invalid scheme: '{$this->parameters->scheme}'.");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Initializes a TCP connection via client.
|
||||
*
|
||||
* @param ParametersInterface $parameters Initialization parameters for the connection.
|
||||
*
|
||||
* @return Relay
|
||||
*/
|
||||
protected function initializeTcpConnection(ParametersInterface $parameters)
|
||||
{
|
||||
if (!filter_var($parameters->host, FILTER_VALIDATE_IP, FILTER_FLAG_IPV6)) {
|
||||
$address = "tcp://$parameters->host:$parameters->port";
|
||||
} else {
|
||||
$address = "tcp://[$parameters->host]:$parameters->port";
|
||||
}
|
||||
|
||||
$flags = STREAM_CLIENT_CONNECT;
|
||||
|
||||
if (isset($parameters->async_connect) && $parameters->async_connect) {
|
||||
$flags |= STREAM_CLIENT_ASYNC_CONNECT;
|
||||
}
|
||||
|
||||
if (isset($parameters->persistent)) {
|
||||
if (false !== $persistent = filter_var($parameters->persistent, FILTER_VALIDATE_BOOLEAN, FILTER_NULL_ON_FAILURE)) {
|
||||
$flags |= STREAM_CLIENT_PERSISTENT;
|
||||
|
||||
if ($persistent === null) {
|
||||
$address = "{$address}/{$parameters->persistent}";
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return $this->connectWithConfiguration($parameters, $address, $flags);
|
||||
}
|
||||
|
||||
/**
|
||||
* Initializes a UNIX connection via client.
|
||||
*
|
||||
* @param ParametersInterface $parameters Initialization parameters for the connection.
|
||||
*
|
||||
* @return Relay
|
||||
*/
|
||||
protected function initializeUnixConnection(ParametersInterface $parameters)
|
||||
{
|
||||
if (!isset($parameters->path)) {
|
||||
throw new InvalidArgumentException('Missing UNIX domain socket path.');
|
||||
}
|
||||
|
||||
$flags = STREAM_CLIENT_CONNECT;
|
||||
|
||||
if (isset($parameters->persistent)) {
|
||||
if (false !== $persistent = filter_var($parameters->persistent, FILTER_VALIDATE_BOOLEAN, FILTER_NULL_ON_FAILURE)) {
|
||||
$flags |= STREAM_CLIENT_PERSISTENT;
|
||||
|
||||
if ($persistent === null) {
|
||||
throw new InvalidArgumentException(
|
||||
'Persistent connection IDs are not supported when using UNIX domain sockets.'
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return $this->connectWithConfiguration($parameters, "unix://{$parameters->path}", $flags);
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
public function connect()
|
||||
{
|
||||
if (parent::connect() && $this->initCommands) {
|
||||
foreach ($this->initCommands as $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);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
public function read()
|
||||
{
|
||||
throw new NotSupportedException('The "relay" extension does not support reading responses.');
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritdoc}
|
||||
*/
|
||||
protected function assertParameters(ParametersInterface $parameters)
|
||||
{
|
||||
if (!in_array($parameters->scheme, ['tcp', 'tls', 'unix', 'redis', 'rediss'])) {
|
||||
throw new InvalidArgumentException("Invalid scheme: '{$parameters->scheme}'.");
|
||||
}
|
||||
|
||||
if (!in_array($parameters->serializer, [null, 'php', 'igbinary', 'msgpack', 'json'])) {
|
||||
throw new InvalidArgumentException("Invalid serializer: '{$parameters->serializer}'.");
|
||||
}
|
||||
|
||||
if (!in_array($parameters->compression, [null, 'lzf', 'lz4', 'zstd'])) {
|
||||
throw new InvalidArgumentException("Invalid compression algorithm: '{$parameters->compression}'.");
|
||||
}
|
||||
|
||||
return $parameters;
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user