init & add agent & add job

predis/src/Pipeline/Atomic.php

@@ -0,0 +1,119 @@
+<?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\Pipeline;
+
+use Predis\ClientException;
+use Predis\ClientInterface;
+use Predis\Connection\ConnectionInterface;
+use Predis\Connection\NodeConnectionInterface;
+use Predis\Response\ErrorInterface as ErrorResponseInterface;
+use Predis\Response\ResponseInterface;
+use Predis\Response\ServerException;
+
+/**
+ * Command pipeline wrapped into a MULTI / EXEC transaction.
+ *
+ * @author Daniele Alessandri <suppakilla@gmail.com>
+ */
+class Atomic extends Pipeline
+{
+    /**
+     * {@inheritdoc}
+     */
+    public function __construct(ClientInterface $client)
+    {
+        if (!$client->getProfile()->supportsCommands(array('multi', 'exec', 'discard'))) {
+            throw new ClientException(
+                "The current profile does not support 'MULTI', 'EXEC' and 'DISCARD'."
+            );
+        }
+
+        parent::__construct($client);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function getConnection()
+    {
+        $connection = $this->getClient()->getConnection();
+
+        if (!$connection instanceof NodeConnectionInterface) {
+            $class = __CLASS__;
+
+            throw new ClientException("The class '$class' does not support aggregate connections.");
+        }
+
+        return $connection;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function executePipeline(ConnectionInterface $connection, \SplQueue $commands)
+    {
+        $profile = $this->getClient()->getProfile();
+        $connection->executeCommand($profile->createCommand('multi'));
+
+        foreach ($commands as $command) {
+            $connection->writeRequest($command);
+        }
+
+        foreach ($commands as $command) {
+            $response = $connection->readResponse($command);
+
+            if ($response instanceof ErrorResponseInterface) {
+                $connection->executeCommand($profile->createCommand('discard'));
+                throw new ServerException($response->getMessage());
+            }
+        }
+
+        $executed = $connection->executeCommand($profile->createCommand('exec'));
+
+        if (!isset($executed)) {
+            // TODO: should be throwing a more appropriate exception.
+            throw new ClientException(
+                'The underlying transaction has been aborted by the server.'
+            );
+        }
+
+        if (count($executed) !== count($commands)) {
+            $expected = count($commands);
+            $received = count($executed);
+
+            throw new ClientException(
+                "Invalid number of responses [expected $expected, received $received]."
+            );
+        }
+
+        $responses = array();
+        $sizeOfPipe = count($commands);
+        $exceptions = $this->throwServerExceptions();
+
+        for ($i = 0; $i < $sizeOfPipe; ++$i) {
+            $command = $commands->dequeue();
+            $response = $executed[$i];
+
+            if (!$response instanceof ResponseInterface) {
+                $responses[] = $command->parseResponse($response);
+            } elseif ($response instanceof ErrorResponseInterface && $exceptions) {
+                $this->exception($connection, $response);
+            } else {
+                $responses[] = $response;
+            }
+
+            unset($executed[$i]);
+        }
+
+        return $responses;
+    }
+}

predis/src/Pipeline/ConnectionErrorProof.php

@@ -0,0 +1,130 @@
+<?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\Pipeline;
+
+use Predis\CommunicationException;
+use Predis\Connection\Aggregate\ClusterInterface;
+use Predis\Connection\ConnectionInterface;
+use Predis\Connection\NodeConnectionInterface;
+use Predis\NotSupportedException;
+
+/**
+ * Command pipeline that does not throw exceptions on connection errors, but
+ * returns the exception instances as the rest of the response elements.
+ *
+ * @todo Awful naming!
+ *
+ * @author Daniele Alessandri <suppakilla@gmail.com>
+ */
+class ConnectionErrorProof extends Pipeline
+{
+    /**
+     * {@inheritdoc}
+     */
+    protected function getConnection()
+    {
+        return $this->getClient()->getConnection();
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function executePipeline(ConnectionInterface $connection, \SplQueue $commands)
+    {
+        if ($connection instanceof NodeConnectionInterface) {
+            return $this->executeSingleNode($connection, $commands);
+        } elseif ($connection instanceof ClusterInterface) {
+            return $this->executeCluster($connection, $commands);
+        } else {
+            $class = get_class($connection);
+
+            throw new NotSupportedException("The connection class '$class' is not supported.");
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function executeSingleNode(NodeConnectionInterface $connection, \SplQueue $commands)
+    {
+        $responses = array();
+        $sizeOfPipe = count($commands);
+
+        foreach ($commands as $command) {
+            try {
+                $connection->writeRequest($command);
+            } catch (CommunicationException $exception) {
+                return array_fill(0, $sizeOfPipe, $exception);
+            }
+        }
+
+        for ($i = 0; $i < $sizeOfPipe; ++$i) {
+            $command = $commands->dequeue();
+
+            try {
+                $responses[$i] = $connection->readResponse($command);
+            } catch (CommunicationException $exception) {
+                $add = count($commands) - count($responses);
+                $responses = array_merge($responses, array_fill(0, $add, $exception));
+
+                break;
+            }
+        }
+
+        return $responses;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function executeCluster(ClusterInterface $connection, \SplQueue $commands)
+    {
+        $responses = array();
+        $sizeOfPipe = count($commands);
+        $exceptions = array();
+
+        foreach ($commands as $command) {
+            $cmdConnection = $connection->getConnection($command);
+
+            if (isset($exceptions[spl_object_hash($cmdConnection)])) {
+                continue;
+            }
+
+            try {
+                $cmdConnection->writeRequest($command);
+            } catch (CommunicationException $exception) {
+                $exceptions[spl_object_hash($cmdConnection)] = $exception;
+            }
+        }
+
+        for ($i = 0; $i < $sizeOfPipe; ++$i) {
+            $command = $commands->dequeue();
+
+            $cmdConnection = $connection->getConnection($command);
+            $connectionHash = spl_object_hash($cmdConnection);
+
+            if (isset($exceptions[$connectionHash])) {
+                $responses[$i] = $exceptions[$connectionHash];
+                continue;
+            }
+
+            try {
+                $responses[$i] = $cmdConnection->readResponse($command);
+            } catch (CommunicationException $exception) {
+                $responses[$i] = $exception;
+                $exceptions[$connectionHash] = $exception;
+            }
+        }
+
+        return $responses;
+    }
+}

predis/src/Pipeline/FireAndForget.php

@@ -0,0 +1,36 @@
+<?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\Pipeline;
+
+use Predis\Connection\ConnectionInterface;
+
+/**
+ * Command pipeline that writes commands to the servers but discards responses.
+ *
+ * @author Daniele Alessandri <suppakilla@gmail.com>
+ */
+class FireAndForget extends Pipeline
+{
+    /**
+     * {@inheritdoc}
+     */
+    protected function executePipeline(ConnectionInterface $connection, \SplQueue $commands)
+    {
+        while (!$commands->isEmpty()) {
+            $connection->writeRequest($commands->dequeue());
+        }
+
+        $connection->disconnect();
+
+        return array();
+    }
+}

predis/src/Pipeline/Pipeline.php

@@ -0,0 +1,247 @@
+<?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\Pipeline;
+
+use Predis\ClientContextInterface;
+use Predis\ClientException;
+use Predis\ClientInterface;
+use Predis\Command\CommandInterface;
+use Predis\Connection\Aggregate\ReplicationInterface;
+use Predis\Connection\ConnectionInterface;
+use Predis\Response\ErrorInterface as ErrorResponseInterface;
+use Predis\Response\ResponseInterface;
+use Predis\Response\ServerException;
+
+/**
+ * Implementation of a command pipeline in which write and read operations of
+ * Redis commands are pipelined to alleviate the effects of network round-trips.
+ *
+ * {@inheritdoc}
+ *
+ * @author Daniele Alessandri <suppakilla@gmail.com>
+ */
+class Pipeline implements ClientContextInterface
+{
+    private $client;
+    private $pipeline;
+
+    private $responses = array();
+    private $running = false;
+
+    /**
+     * @param ClientInterface $client Client instance used by the context.
+     */
+    public function __construct(ClientInterface $client)
+    {
+        $this->client = $client;
+        $this->pipeline = new \SplQueue();
+    }
+
+    /**
+     * Queues a command into the pipeline buffer.
+     *
+     * @param string $method    Command ID.
+     * @param array  $arguments Arguments for the command.
+     *
+     * @return $this
+     */
+    public function __call($method, $arguments)
+    {
+        $command = $this->client->createCommand($method, $arguments);
+        $this->recordCommand($command);
+
+        return $this;
+    }
+
+    /**
+     * Queues a command instance into the pipeline buffer.
+     *
+     * @param CommandInterface $command Command to be queued in the buffer.
+     */
+    protected function recordCommand(CommandInterface $command)
+    {
+        $this->pipeline->enqueue($command);
+    }
+
+    /**
+     * Queues a command instance into the pipeline buffer.
+     *
+     * @param CommandInterface $command Command instance to be queued in the buffer.
+     *
+     * @return $this
+     */
+    public function executeCommand(CommandInterface $command)
+    {
+        $this->recordCommand($command);
+
+        return $this;
+    }
+
+    /**
+     * Throws an exception on -ERR responses returned by Redis.
+     *
+     * @param ConnectionInterface    $connection Redis connection that returned the error.
+     * @param ErrorResponseInterface $response   Instance of the error response.
+     *
+     * @throws ServerException
+     */
+    protected function exception(ConnectionInterface $connection, ErrorResponseInterface $response)
+    {
+        $connection->disconnect();
+        $message = $response->getMessage();
+
+        throw new ServerException($message);
+    }
+
+    /**
+     * Returns the underlying connection to be used by the pipeline.
+     *
+     * @return ConnectionInterface
+     */
+    protected function getConnection()
+    {
+        $connection = $this->getClient()->getConnection();
+
+        if ($connection instanceof ReplicationInterface) {
+            $connection->switchTo('master');
+        }
+
+        return $connection;
+    }
+
+    /**
+     * Implements the logic to flush the queued commands and read the responses
+     * from the current connection.
+     *
+     * @param ConnectionInterface $connection Current connection instance.
+     * @param \SplQueue           $commands   Queued commands.
+     *
+     * @return array
+     */
+    protected function executePipeline(ConnectionInterface $connection, \SplQueue $commands)
+    {
+        foreach ($commands as $command) {
+            $connection->writeRequest($command);
+        }
+
+        $responses = array();
+        $exceptions = $this->throwServerExceptions();
+
+        while (!$commands->isEmpty()) {
+            $command = $commands->dequeue();
+            $response = $connection->readResponse($command);
+
+            if (!$response instanceof ResponseInterface) {
+                $responses[] = $command->parseResponse($response);
+            } elseif ($response instanceof ErrorResponseInterface && $exceptions) {
+                $this->exception($connection, $response);
+            } else {
+                $responses[] = $response;
+            }
+        }
+
+        return $responses;
+    }
+
+    /**
+     * Flushes the buffer holding all of the commands queued so far.
+     *
+     * @param bool $send Specifies if the commands in the buffer should be sent to Redis.
+     *
+     * @return $this
+     */
+    public function flushPipeline($send = true)
+    {
+        if ($send && !$this->pipeline->isEmpty()) {
+            $responses = $this->executePipeline($this->getConnection(), $this->pipeline);
+            $this->responses = array_merge($this->responses, $responses);
+        } else {
+            $this->pipeline = new \SplQueue();
+        }
+
+        return $this;
+    }
+
+    /**
+     * Marks the running status of the pipeline.
+     *
+     * @param bool $bool Sets the running status of the pipeline.
+     *
+     * @throws ClientException
+     */
+    private function setRunning($bool)
+    {
+        if ($bool && $this->running) {
+            throw new ClientException('The current pipeline context is already being executed.');
+        }
+
+        $this->running = $bool;
+    }
+
+    /**
+     * Handles the actual execution of the whole pipeline.
+     *
+     * @param mixed $callable Optional callback for execution.
+     *
+     * @throws \Exception
+     * @throws \InvalidArgumentException
+     *
+     * @return array
+     */
+    public function execute($callable = null)
+    {
+        if ($callable && !is_callable($callable)) {
+            throw new \InvalidArgumentException('The argument must be a callable object.');
+        }
+
+        $exception = null;
+        $this->setRunning(true);
+
+        try {
+            if ($callable) {
+                call_user_func($callable, $this);
+            }
+
+            $this->flushPipeline();
+        } catch (\Exception $exception) {
+            // NOOP
+        }
+
+        $this->setRunning(false);
+
+        if ($exception) {
+            throw $exception;
+        }
+
+        return $this->responses;
+    }
+
+    /**
+     * Returns if the pipeline should throw exceptions on server errors.
+     *
+     * @return bool
+     */
+    protected function throwServerExceptions()
+    {
+        return (bool) $this->client->getOptions()->exceptions;
+    }
+
+    /**
+     * Returns the underlying client instance used by the pipeline object.
+     *
+     * @return ClientInterface
+     */
+    public function getClient()
+    {
+        return $this->client;
+    }
+}