<?php
* This file is part of the Predis package.
*
* (c) 2009-2020 Daniele Alessandri
* (c) 2021-2026 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 Predis\Command\CommandInterface;
use Predis\Command\RawCommand;
use Predis\CommunicationException;
use Predis\Connection\Resource\Exception\StreamInitException;
use Predis\Protocol\Parser\ParserStrategyResolver;
use Predis\Protocol\Parser\Strategy\ParserStrategyInterface;
use Predis\Protocol\ProtocolException;
use Predis\TimeoutException;
* Base class with the common logic used by connection classes to communicate
* with Redis.
*/
abstract class AbstractConnection implements NodeConnectionInterface
{
* @var ParserStrategyInterface
*/
protected $parserStrategy;
* @var int|null
*/
protected $clientId;
protected $resource;
private $cachedId;
protected $parameters;
* @var RawCommand[]
*/
protected $initCommands = [];
* Commands that restore server-side session state (e.g. HIMPORT fieldsets)
* on the physical connection. Unlike init commands, these are keyed so they
* can be replaced and removed, are replayed best-effort after init commands
* on every (re)connect, and are intentionally excluded from serialization: a
* deserialized or freshly created connection is a new server session.
*
* @var array<string, CommandInterface>
*/
protected $sessionCommands = [];
* @param ParametersInterface $parameters Initialization parameters for the connection.
*/
public function __construct(ParametersInterface $parameters)
{
$this->parameters = $parameters;
$this->setParserStrategy();
}
* Disconnects from the server and destroys the underlying resource when
* PHP's garbage collector kicks in.
*/
public function __destruct()
{
$this->disconnect();
}
* {@inheritdoc}
*/
public function isConnected()
{
return isset($this->resource);
}
* {@inheritdoc}
*/
public function hasDataToRead(): bool
{
return true;
}
* Creates a stream resource to communicate with Redis.
*
* @return mixed
* @throws StreamInitException
*/
abstract protected function createResource();
* {@inheritdoc}
*/
public function connect()
{
if (!$this->isConnected()) {
$this->resource = $this->createResource();
return true;
}
return false;
}
* {@inheritdoc}
*/
public function disconnect()
{
unset($this->resource);
}
* {@inheritdoc}
*/
public function addConnectCommand(CommandInterface $command)
{
$this->initCommands[] = $command;
}
* {@inheritdoc}
*/
public function getInitCommands(): array
{
return $this->initCommands;
}
* Registers (or replaces) a session command under the given key so it is
* replayed on the next (re)connect. Keying makes replacement idempotent.
*
* @param string $key Identifier for the session command.
* @param CommandInterface $command Command replayed to restore session state.
*/
public function addSessionCommand(string $key, CommandInterface $command): void
{
$this->sessionCommands[$key] = $command;
}
* Removes a single session command previously registered under the key.
*
* @param string $key Identifier for the session command.
*/
public function removeSessionCommand(string $key): void
{
unset($this->sessionCommands[$key]);
}
* Removes every session command whose key starts with the given prefix.
*
* @param string $prefix Key prefix (e.g. "himport:") to match.
*/
public function removeSessionCommandsByPrefix(string $prefix): void
{
foreach (array_keys($this->sessionCommands) as $key) {
if (strpos($key, $prefix) === 0) {
unset($this->sessionCommands[$key]);
}
}
}
* @return array<string, CommandInterface>
*/
public function getSessionCommands(): array
{
return $this->sessionCommands;
}
* {@inheritdoc}
*/
public function executeCommand(CommandInterface $command)
{
$this->writeRequest($command);
return $this->readResponse($command);
}
* {@inheritdoc}
*/
public function readResponse(CommandInterface $command)
{
return $this->read();
}
* Helper method to handle connection errors.
*
* @param string $message Error message.
* @param int $code Error code.
* @throws CommunicationException
*/
protected function onConnectionError($message, $code = 0): void
{
CommunicationException::handle(
new ConnectionException($this, "$message [{$this->getParameters()}]", $code)
);
}
* Helper method to handle timeout errors.
*
* @param int $code
* @return void
* @throws CommunicationException
*/
protected function onTimeoutError(int $code = 0): void
{
CommunicationException::handle(
new TimeoutException($this, $code)
);
}
* Helper method to handle protocol errors.
*
* @param string $message Error message.
* @throws CommunicationException
*/
protected function onProtocolError($message)
{
CommunicationException::handle(
new ProtocolException($this, "$message [{$this->getParameters()}]")
);
}
* {@inheritdoc}
*/
public function getResource()
{
if (isset($this->resource)) {
return $this->resource;
}
$this->connect();
return $this->resource;
}
* {@inheritdoc}
*/
public function getParameters()
{
return $this->parameters;
}
* Gets an identifier for the connection.
*
* @return string
*/
protected function getIdentifier()
{
if ($this->parameters->scheme === 'unix') {
return $this->parameters->path;
}
return "{$this->parameters->host}:{$this->parameters->port}";
}
* {@inheritDoc}
*/
public function getClientId(): ?int
{
return $this->clientId;
}
* {@inheritdoc}
*/
public function __toString()
{
if (!isset($this->cachedId)) {
$this->cachedId = $this->getIdentifier();
}
return $this->cachedId;
}
* {@inheritdoc}
*/
public function __sleep()
{
return ['parameters', 'initCommands'];
}
* Set parser strategy for given connection.
*
* @return void
*/
protected function setParserStrategy(): void
{
$strategyResolver = new ParserStrategyResolver();
$this->parserStrategy = $strategyResolver->resolve((int) $this->parameters->protocol);
}
}