Improve Predis\Pipeline\MultiExecExecutor.

lib/Predis/Pipeline/MultiExecExecutor.php

@@ -13,8 +13,9 @@ namespace Predis\Pipeline;
 
 use SplQueue;
 use Predis\ClientException;
-use Predis\ResponseQueued;
 use Predis\ResponseErrorInterface;
+use Predis\ResponseObjectInterface;
+use Predis\ResponseQueued;
 use Predis\ServerException;
 use Predis\Connection\ConnectionInterface;
 use Predis\Connection\SingleConnectionInterface;
@@ -35,9 +36,9 @@ class MultiExecExecutor implements PipelineExecutorInterface
     /**
      *
      */
-    public function __construct()
+    public function __construct(ServerProfileInterface $profile = null)
     {
-        $this->setProfile(ServerProfile::getDefault());
+        $this->setProfile($profile ?: ServerProfile::getDefault());
     }
 
     /**
@@ -95,10 +96,19 @@ class MultiExecExecutor implements PipelineExecutorInterface
         }
 
         for ($i = 0; $i < $size; $i++) {
-            if ($response = $responses[$i] instanceof \Iterator) {
-                $response = iterator_to_array($response);
+            $commandReply = $responses[$i];
+
+            if ($commandReply instanceof ResponseObjectInterface) {
+                $values[$i] = $commandReply;
+                $commands->dequeue();
+            } else {
+                if ($commandReply instanceof \Iterator) {
+                    $commandReply = iterator_to_array($commandReply);
+                }
+
+                $values[$i] = $commands->dequeue()->parseResponse($commandReply);
             }
-            $values[$i] = $commands->dequeue()->parseResponse($responses[$i]);
+
             unset($responses[$i]);
         }
 

tests/Predis/Pipeline/MultiExecExecutorTest.php

@@ -92,6 +92,29 @@ class MultiExecExecutorTest extends StandardTestCase
         $executor->execute($connection, $pipeline);
     }
 
+    /**
+     * @group disconnected
+     */
+    public function testExecutorWithErrorInCommandResponse()
+    {
+        $executor = new MultiExecExecutor();
+        $pipeline = $this->getCommandsQueue();
+        $queued = new ResponseQueued();
+        $error = new ResponseError('ERR Test error');
+
+        $connection = $this->getMock('Predis\Connection\SingleConnectionInterface');
+        $connection->expects($this->exactly(3))
+                   ->method('readResponse')
+                   ->will($this->onConsecutiveCalls($queued, $queued, $queued));
+        $connection->expects($this->at(7))
+                   ->method('executeCommand')
+                   ->will($this->returnValue(array('PONG', 'PONG', $error)));
+
+        $replies = $executor->execute($connection, $pipeline);
+
+        $this->assertSame(array(true, true, $error), $replies);
+    }
+
     /**
      * @group disconnected
      * @expectedException Predis\ClientException