EVOLUTION-NINJA

Edit File: MultiExec.php

<?php

/*
 * This file is part of the Predis package.
 *
 * (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.
 */

namespace Predis\Transaction;

use Exception;
use InvalidArgumentException;
use Predis\ClientContextInterface;
use Predis\ClientException;
use Predis\ClientInterface;
use Predis\Command\CommandInterface;
use Predis\CommunicationException;
use Predis\Connection\Cluster\ClusterInterface;
use Predis\Connection\RelayConnection;
use Predis\NotSupportedException;
use Predis\Protocol\ProtocolException;
use Predis\Response\Error;
use Predis\Response\ErrorInterface as ErrorResponseInterface;
use Predis\Response\ServerException;
use Predis\Response\Status as StatusResponse;
use Relay\Exception as RelayException;
use Relay\Relay;
use SplQueue;

/**
 * Client-side abstraction of a Redis transaction based on MULTI / EXEC.
 *
 * {@inheritdoc}
 */
class MultiExec implements ClientContextInterface
{
    private $state;

protected $client;
    protected $commands;
    protected $exceptions = true;
    protected $attempts = 0;
    protected $watchKeys = [];
    protected $modeCAS = false;

/**
     * @param ClientInterface $client  Client instance used by the transaction.
     * @param array           $options Initialization options.
     */
    public function __construct(ClientInterface $client, array $options = null)
    {
        $this->assertClient($client);

$this->client = $client;
        $this->state = new MultiExecState();

$this->configure($client, $options ?: []);
        $this->reset();
    }

/**
     * Checks if the passed client instance satisfies the required conditions
     * needed to initialize the transaction object.
     *
     * @param ClientInterface $client Client instance used by the transaction object.
     *
     * @throws NotSupportedException
     */
    private function assertClient(ClientInterface $client)
    {
        if ($client->getConnection() instanceof ClusterInterface) {
            throw new NotSupportedException(
                'Cannot initialize a MULTI/EXEC transaction over cluster connections.'
            );
        }

if (!$client->getCommandFactory()->supports('MULTI', 'EXEC', 'DISCARD')) {
            throw new NotSupportedException(
                'MULTI, EXEC and DISCARD are not supported by the current command factory.'
            );
        }
    }

/**
     * Configures the transaction using the provided options.
     *
     * @param ClientInterface $client  Underlying client instance.
     * @param array           $options Array of options for the transaction.
     **/
    protected function configure(ClientInterface $client, array $options)
    {
        if (isset($options['exceptions'])) {
            $this->exceptions = (bool) $options['exceptions'];
        } else {
            $this->exceptions = $client->getOptions()->exceptions;
        }

if (isset($options['cas'])) {
            $this->modeCAS = (bool) $options['cas'];
        }

if (isset($options['watch']) && $keys = $options['watch']) {
            $this->watchKeys = $keys;
        }

if (isset($options['retry'])) {
            $this->attempts = (int) $options['retry'];
        }
    }

/**
     * Resets the state of the transaction.
     */
    protected function reset()
    {
        $this->state->reset();
        $this->commands = new SplQueue();
    }

/**
     * Initializes the transaction context.
     */
    protected function initialize()
    {
        if ($this->state->isInitialized()) {
            return;
        }

if ($this->modeCAS) {
            $this->state->flag(MultiExecState::CAS);
        }

if ($this->watchKeys) {
            $this->watch($this->watchKeys);
        }

$cas = $this->state->isCAS();
        $discarded = $this->state->isDiscarded();

if (!$cas || ($cas && $discarded)) {
            $this->call('MULTI');

if ($discarded) {
                $this->state->unflag(MultiExecState::CAS);
            }
        }

$this->state->unflag(MultiExecState::DISCARDED);
        $this->state->flag(MultiExecState::INITIALIZED);
    }

/**
     * Dynamically invokes a Redis command with the specified arguments.
     *
     * @param string $method    Command ID.
     * @param array  $arguments Arguments for the command.
     *
     * @return mixed
     */
    public function __call($method, $arguments)
    {
        return $this->executeCommand(
            $this->client->createCommand($method, $arguments)
        );
    }

/**
     * Executes a Redis command bypassing the transaction logic.
     *
     * @param string $commandID Command ID.
     * @param array  $arguments Arguments for the command.
     *
     * @return mixed
     * @throws ServerException
     */
    protected function call($commandID, array $arguments = [])
    {
        try {
            $response = $this->client->executeCommand(
                $this->client->createCommand($commandID, $arguments)
            );
        } catch (ServerException $exception) {
            if (!$this->client->getConnection() instanceof RelayConnection) {
                throw $exception;
            }

if (strcasecmp($commandID, 'EXEC') != 0) {
                throw $exception;
            }

if (!strpos($exception->getMessage(), 'RELAY_ERR_REDIS')) {
                throw $exception;
            }

return null;
        }

if ($response instanceof ErrorResponseInterface) {
            throw new ServerException($response->getMessage());
        }

return $response;
    }

/**
     * Executes the specified Redis command.
     *
     * @param CommandInterface $command Command instance.
     *
     * @return $this|mixed
     * @throws AbortedMultiExecException
     * @throws CommunicationException
     */
    public function executeCommand(CommandInterface $command)
    {
        $this->initialize();

if ($this->state->isCAS()) {
            return $this->client->executeCommand($command);
        }

$response = $this->client->getConnection()->executeCommand($command);

if ($response instanceof StatusResponse && $response == 'QUEUED') {
            $this->commands->enqueue($command);
        } elseif ($response instanceof Relay) {
            $this->commands->enqueue($command);
        } elseif ($response instanceof ErrorResponseInterface) {
            throw new AbortedMultiExecException($this, $response->getMessage());
        } else {
            $this->onProtocolError('The server did not return a +QUEUED status response.');
        }

return $this;
    }

/**
     * Executes WATCH against one or more keys.
     *
     * @param string|array $keys One or more keys.
     *
     * @return mixed
     * @throws NotSupportedException
     * @throws ClientException
     */
    public function watch($keys)
    {
        if (!$this->client->getCommandFactory()->supports('WATCH')) {
            throw new NotSupportedException('WATCH is not supported by the current command factory.');
        }

if ($this->state->isWatchAllowed()) {
            throw new ClientException('Sending WATCH after MULTI is not allowed.');
        }

$response = $this->call('WATCH', is_array($keys) ? $keys : [$keys]);
        $this->state->flag(MultiExecState::WATCH);

return $response;
    }

/**
     * Finalizes the transaction by executing MULTI on the server.
     *
     * @return MultiExec
     */
    public function multi()
    {
        if ($this->state->check(MultiExecState::INITIALIZED | MultiExecState::CAS)) {
            $this->state->unflag(MultiExecState::CAS);
            $this->call('MULTI');
        } else {
            $this->initialize();
        }

return $this;
    }

/**
     * Executes UNWATCH.
     *
     * @return MultiExec
     * @throws NotSupportedException
     */
    public function unwatch()
    {
        if (!$this->client->getCommandFactory()->supports('UNWATCH')) {
            throw new NotSupportedException(
                'UNWATCH is not supported by the current command factory.'
            );
        }

$this->state->unflag(MultiExecState::WATCH);
        $this->__call('UNWATCH', []);

return $this;
    }

/**
     * Resets the transaction by UNWATCH-ing the keys that are being WATCHed and
     * DISCARD-ing pending commands that have been already sent to the server.
     *
     * @return MultiExec
     */
    public function discard()
    {
        if ($this->state->isInitialized()) {
            $this->call($this->state->isCAS() ? 'UNWATCH' : 'DISCARD');

$this->reset();
            $this->state->flag(MultiExecState::DISCARDED);
        }

return $this;
    }

/**
     * Executes the whole transaction.
     *
     * @return mixed
     */
    public function exec()
    {
        return $this->execute();
    }

/**
     * Checks the state of the transaction before execution.
     *
     * @param mixed $callable Callback for execution.
     *
     * @throws InvalidArgumentException
     * @throws ClientException
     */
    private function checkBeforeExecution($callable)
    {
        if ($this->state->isExecuting()) {
            throw new ClientException(
                'Cannot invoke "execute" or "exec" inside an active transaction context.'
            );
        }

if ($callable) {
            if (!is_callable($callable)) {
                throw new InvalidArgumentException('The argument must be a callable object.');
            }

if (!$this->commands->isEmpty()) {
                $this->discard();

throw new ClientException(
                    'Cannot execute a transaction block after using fluent interface.'
                );
            }
        } elseif ($this->attempts) {
            $this->discard();

throw new ClientException(
                'Automatic retries are supported only when a callable block is provided.'
            );
        }
    }

/**
     * Handles the actual execution of the whole transaction.
     *
     * @param mixed $callable Optional callback for execution.
     *
     * @return array
     * @throws CommunicationException
     * @throws AbortedMultiExecException
     * @throws ServerException
     */
    public function execute($callable = null)
    {
        $this->checkBeforeExecution($callable);

$execResponse = null;
        $attempts = $this->attempts;

do {
            if ($callable) {
                $this->executeTransactionBlock($callable);
            }

if ($this->commands->isEmpty()) {
                if ($this->state->isWatching()) {
                    $this->discard();
                }

return;
            }

$execResponse = $this->call('EXEC');

// The additional `false` check is needed for Relay,
            // let's hope it won't break anything
            if ($execResponse === null || $execResponse === false) {
                if ($attempts === 0) {
                    throw new AbortedMultiExecException(
                        $this, 'The current transaction has been aborted by the server.'
                    );
                }

$this->reset();

continue;
            }

break;
        } while ($attempts-- > 0);

$response = [];
        $commands = $this->commands;
        $size = count($execResponse);

if ($size !== count($commands)) {
            $this->onProtocolError('EXEC returned an unexpected number of response items.');
        }

for ($i = 0; $i < $size; ++$i) {
            $cmdResponse = $execResponse[$i];

if ($this->exceptions && $cmdResponse instanceof ErrorResponseInterface) {
                throw new ServerException($cmdResponse->getMessage());
            }

if ($cmdResponse instanceof RelayException) {
                if ($this->exceptions) {
                    throw new ServerException($cmdResponse->getMessage(), $cmdResponse->getCode(), $cmdResponse);
                }

$commands->dequeue();
                $response[$i] = new Error($cmdResponse->getMessage());
                continue;
            }

$response[$i] = $commands->dequeue()->parseResponse($cmdResponse);
        }

return $response;
    }

/**
     * Passes the current transaction object to a callable block for execution.
     *
     * @param mixed $callable Callback.
     *
     * @throws CommunicationException
     * @throws ServerException
     */
    protected function executeTransactionBlock($callable)
    {
        $exception = null;
        $this->state->flag(MultiExecState::INSIDEBLOCK);

try {
            call_user_func($callable, $this);
        } catch (CommunicationException $exception) {
            // NOOP
        } catch (ServerException $exception) {
            // NOOP
        } catch (Exception $exception) {
            $this->discard();
        }

$this->state->unflag(MultiExecState::INSIDEBLOCK);

if ($exception) {
            throw $exception;
        }
    }

/**
     * Helper method for protocol errors encountered inside the transaction.
     *
     * @param string $message Error message.
     */
    private function onProtocolError($message)
    {
        // Since a MULTI/EXEC block cannot be initialized when using aggregate
        // connections we can safely assume that Predis\Client::getConnection()
        // will return a Predis\Connection\NodeConnectionInterface instance.
        CommunicationException::handle(new ProtocolException(
            $this->client->getConnection(), $message
        ));
    }
}