[v2.x] Relay support (#1263)

src/Client.php

@@ -25,9 +25,15 @@ use Predis\Configuration\OptionsInterface;
 use Predis\Connection\ConnectionInterface;
 use Predis\Connection\Parameters;
 use Predis\Connection\ParametersInterface;
+use Predis\Connection\RelayConnection;
 use Predis\Monitor\Consumer as MonitorConsumer;
+use Predis\Pipeline\Atomic;
+use Predis\Pipeline\FireAndForget;
 use Predis\Pipeline\Pipeline;
+use Predis\Pipeline\RelayAtomic;
+use Predis\Pipeline\RelayPipeline;
 use Predis\PubSub\Consumer as PubSubConsumer;
+use Predis\PubSub\RelayConsumer as RelayPubSubConsumer;
 use Predis\Response\ErrorInterface as ErrorResponseInterface;
 use Predis\Response\ResponseInterface;
 use Predis\Response\ServerException;
@@ -262,6 +268,32 @@ class Client implements ClientInterface, IteratorAggregate
         return $this->connection;
     }
 
+    /**
+     * Applies the configured serializer and compression to given value.
+     *
+     * @param  mixed  $value
+     * @return string
+     */
+    public function pack($value)
+    {
+        return $this->connection instanceof RelayConnection
+            ? $this->connection->pack($value)
+            : $value;
+    }
+
+    /**
+     * Deserializes and decompresses to given value.
+     *
+     * @param  mixed  $value
+     * @return string
+     */
+    public function unpack($value)
+    {
+        return $this->connection instanceof RelayConnection
+            ? $this->connection->unpack($value)
+            : $value;
+    }
+
     /**
      * Executes a command without filtering its arguments, parsing the response,
      * applying any prefix to keys or throwing exceptions on Redis errors even
@@ -435,19 +467,29 @@ class Client implements ClientInterface, IteratorAggregate
     /**
      * Actual pipeline context initializer method.
      *
-     * @param array $options  Options for the context.
-     * @param mixed $callable Optional callable used to execute the context.
+     * @param array|null $options  Options for the context.
+     * @param mixed      $callable Optional callable used to execute the context.
      *
      * @return Pipeline|array
      */
     protected function createPipeline(array $options = null, $callable = null)
     {
         if (isset($options['atomic']) && $options['atomic']) {
-            $class = 'Predis\Pipeline\Atomic';
+            $class = Atomic::class;
         } elseif (isset($options['fire-and-forget']) && $options['fire-and-forget']) {
-            $class = 'Predis\Pipeline\FireAndForget';
+            $class = FireAndForget::class;
         } else {
-            $class = 'Predis\Pipeline\Pipeline';
+            $class = Pipeline::class;
+        }
+
+        if ($this->connection instanceof RelayConnection) {
+            if (isset($options['atomic']) && $options['atomic']) {
+                $class = RelayAtomic::class;
+            } elseif (isset($options['fire-and-forget']) && $options['fire-and-forget']) {
+                throw new NotSupportedException('The "relay" extension does not support fire-and-forget pipelines.');
+            } else {
+                $class = RelayPipeline::class;
+            }
         }
 
         /*
@@ -517,7 +559,11 @@ class Client implements ClientInterface, IteratorAggregate
      */
     protected function createPubSub(array $options = null, $callable = null)
     {
-        $pubsub = new PubSubConsumer($this, $options);
+        if ($this->connection instanceof RelayConnection) {
+            $pubsub = new RelayPubSubConsumer($this, $options);
+        } else {
+            $pubsub = new PubSubConsumer($this, $options);
+        }
 
         if (!isset($callable)) {
             return $pubsub;