Move the handling of pipeline options inside the pipeline class.

lib/Predis/Client.php

@@ -178,22 +178,8 @@ class Client {
     }
 
     private function initPipeline(Array $options = null, $pipelineBlock = null) {
-        $pipeline = null;
-        if (isset($options)) {
-            if (isset($options['safe']) && $options['safe'] == true) {
-                $connection = $this->_connection;
-                $pipeline = new PipelineContext($this,
-                    Helpers::isCluster($connection)
-                        ? new Pipeline\SafeClusterExecutor()
-                        : new Pipeline\SafeExecutor()
-                );
-            }
-            else {
-                $pipeline = new PipelineContext($this);
-            }
-        }
         return $this->pipelineExecute(
-            $pipeline ?: new PipelineContext($this), $pipelineBlock
+            new PipelineContext($this, $options), $pipelineBlock
         );
     }
 

lib/Predis/Pipeline/PipelineContext.php

@@ -3,19 +3,38 @@
 namespace Predis\Pipeline;
 
 use Predis\Client;
+use Predis\Helpers;
 use Predis\ClientException;
 use Predis\Commands\ICommand;
 
 class PipelineContext {
     private $_client, $_pipelineBuffer, $_returnValues, $_running, $_executor;
 
-    public function __construct(Client $client, IPipelineExecutor $executor = null) {
+    public function __construct(Client $client, Array $options = null) {
         $this->_client         = $client;
-        $this->_executor       = $executor ?: new StandardExecutor();
+        $this->_executor       = $this->getExecutor($client, $options ?: array());
         $this->_pipelineBuffer = array();
         $this->_returnValues   = array();
     }
 
+    protected function getExecutor(Client $client, Array $options) {
+        if (!$options) {
+            return new StandardExecutor();
+        }
+        if (isset($options['executor'])) {
+            $executor = $options['executor'];
+            if (!$executor instanceof IPipelineExecutor) {
+                throw new \ArgumentException();
+            }
+            return $executor;
+        }
+        if (isset($options['safe']) && $options['safe'] == true) {
+            $isCluster = Helpers::isCluster($client->getConnection());
+            return $isCluster ? new SafeClusterExecutor() : new SafeExecutor();
+        }
+        return new StandardExecutor();
+    }
+
     public function __call($method, $arguments) {
         $command = $this->_client->createCommand($method, $arguments);
         $this->recordCommand($command);