Implement support for master / slave replication configurations.

We used a new kind of virtual connection that, just like with cluster, groups
multiple connection objects and handles the logic needed to switch among them
depending on the kind of operation performed by commands..

Our default implementation starts by picking up a random slave and switches to
the master as soon as a command performing a write operation against a key is
detected. The master server will then be used for all the subsequent requests
unless a manual switch to a different connection object is performed.

Redis transactions are always performed on the master server, which means that
the switch to master is done as soon as the client issues a WATCH, MULTI or
any other command related to transactions.

See also https://github.com/nrk/predis/issues/21 for more details.

lib/Predis/Network/IConnectionReplication.php

@@ -0,0 +1,81 @@
+<?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\Network;
+
+use Predis\Commands\ICommand;
+
+/**
+ * Defines a group of Redis servers in a master/slave replication configuration.
+ *
+ * @author Daniele Alessandri <suppakilla@gmail.com>
+ */
+interface IConnectionReplication extends IConnection
+{
+    /**
+     * Adds a connection instance to the cluster.
+     *
+     * @param IConnectionSingle $connection Instance of a connection.
+     */
+    public function add(IConnectionSingle $connection);
+
+    /**
+     * Removes the specified connection instance from the cluster.
+     *
+     * @param IConnectionSingle $connection Instance of a connection.
+     * @return Boolean Returns true if the connection was in the pool.
+     */
+    public function remove(IConnectionSingle $connection);
+
+    /**
+     * Gets the actual connection instance in charge of the specified command.
+     *
+     * @param ICommand $command Instance of a Redis command.
+     * @return IConnectionSingle
+     */
+    public function getConnection(ICommand $command);
+
+    /**
+     * Retrieves a connection instance from the cluster using an alias.
+     *
+     * @param string $connectionId Alias of a connection
+     * @return IConnectionSingle
+     */
+    public function getConnectionById($connectionId);
+
+    /**
+     * Switches the internal connection object being used.
+     *
+     * @param string $connection Alias of a connection
+     */
+    public function switchTo($connection);
+
+    /**
+     * Retrieves the connection object currently being used.
+     *
+     * @return IConnectionSingle
+     */
+    public function getCurrent();
+
+    /**
+     * Retrieves the connection object to the master Redis server.
+     *
+     * @return IConnectionSingle
+     */
+    public function getMaster();
+
+    /**
+     * Retrieves a list of connection objects to slaves Redis servers.
+     *
+     * @return IConnectionSingle
+     */
+    public function getSlaves();
+}

lib/Predis/Network/PredisReplication.php

@@ -0,0 +1,319 @@
+<?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\Network;
+
+use Predis\Commands\ICommand;
+
+/**
+ * Defines the standard virtual connection class that is used
+ * by Predis to handle replication with a group of servers in
+ * a master/slave configuration.
+ *
+ * @author Daniele Alessandri <suppakilla@gmail.com>
+ */
+class PredisReplication implements IConnectionReplication
+{
+    private $readonly = array();
+    private $current = null;
+    private $master = null;
+    private $slaves = array();
+
+    /**
+     *
+     */
+    public function __construct()
+    {
+        $this->readonly = $this->getReadOnlyOperations();
+    }
+
+    /**
+     * Returns if the specified command performs a read-only operation
+     * against a key stored on Redis.
+     *
+     * @param ICommand $command Instance of Redis command.
+     * @return Boolean
+     */
+    protected function isReadOperation(ICommand $command)
+    {
+        if (isset($this->readonly[$id = $command->getId()])) {
+            if (true === $readonly = $this->readonly[$id]) {
+                return true;
+            }
+
+            return $readonly($command);
+        }
+
+        return false;
+    }
+
+    /**
+     * 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 a master and at least one slave.');
+        }
+    }
+
+    /**
+     * Resets the connection state.
+     */
+    protected function reset()
+    {
+        $this->current = null;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function add(IConnectionSingle $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(IConnectionSingle $connection)
+    {
+        if ($connection->getParameters()->alias === 'master') {
+            $this->master = null;
+            $this->reset();
+        }
+        else {
+            if (($id = array_search($connection, $this->slaves, true)) !== false) {
+                unset($this->slaves[$id]);
+                $this->reset();
+            }
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getConnection(ICommand $command)
+    {
+        if ($this->current === null) {
+            $this->check();
+            $this->current = $this->isReadOperation($command) ? $this->pickSlave() : $this->master;
+
+            return $this->current;
+        }
+
+        if ($this->current === $this->master) {
+            return $this->current;
+        }
+
+        if (!$this->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 null;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function switchTo($connection)
+    {
+        $this->check();
+
+        if (!$connection instanceof IConnectionSingle) {
+            $connection = $this->getConnectionById($connection);
+        }
+        if ($connection !== $this->master && !in_array($connection, $this->slaves, true)) {
+            throw new \InvalidArgumentException('The specified connection is not valid.');
+        }
+
+        $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 a random slave.
+     *
+     * @return IConnectionSingle
+     */
+    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 writeCommand(ICommand $command)
+    {
+        $this->getConnection($command)->writeCommand($command);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function readResponse(ICommand $command)
+    {
+        return $this->getConnection($command)->readResponse($command);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function executeCommand(ICommand $command)
+    {
+        return $this->getConnection($command)->executeCommand($command);
+    }
+
+    /**
+     * Returns a list of commands that perform read-only operations.
+     *
+     * @return array
+     */
+    protected function getReadOnlyOperations()
+    {
+        return array(
+            'EXISTS'            => true,
+            'TYPE'              => true,
+            'KEYS'              => true,
+            'RANDOMKEY'         => true,
+            'TTL'               => true,
+            'GET'               => true,
+            'MGET'              => true,
+            'SUBSTR'            => true,
+            'STRLEN'            => true,
+            'GETRANGE'          => true,
+            'GETBIT'            => true,
+            'LLEN'              => true,
+            'LRANGE'            => true,
+            'LINDEX'            => true,
+            'SCARD'             => true,
+            'SISMEMBER'         => true,
+            'SINTER'            => true,
+            'SUNION'            => true,
+            'SDIFF'             => true,
+            'SMEMBERS'          => true,
+            'SRANDMEMBER'       => true,
+            'ZRANGE'            => true,
+            'ZREVRANGE'         => true,
+            'ZRANGEBYSCORE'     => true,
+            'ZREVRANGEBYSCORE'  => true,
+            'ZCARD'             => true,
+            'ZSCORE'            => true,
+            'ZCOUNT'            => true,
+            'ZRANK'             => true,
+            'ZREVRANK'          => true,
+            'HGET'              => true,
+            'HMGET'             => true,
+            'HEXISTS'           => true,
+            'HLEN'              => true,
+            'HKEYS'             => true,
+            'HVELS'             => true,
+            'HGETALL'           => true,
+            'PING'              => true,
+            'AUTH'              => true,
+            'SELECT'            => true,
+            'ECHO'              => true,
+            'QUIT'              => true,
+            'INFO'              => true,
+            'DBSIZE'            => true,
+            'MONITOR'           => true,
+            'LASTSAVE'          => true,
+            'SHUTDOWN'          => true,
+            'OBJECT'            => true,
+            'SORT'              => function(ICommand $command) {
+                $arguments = $command->getArguments();
+                return ($c = count($arguments)) === 1 ? true : $arguments[$c - 2] !== 'STORE';
+            },
+        );
+    }
+}