Drop support for streamable multibulk responses.

Supporting this feature has been problematic and leaded to some ugly
code to make abstractions such as pipelines and transactions aware of
these kind of response objects. Furthermore, it was not possible to
add them to all the connection classes due to implementation limits.

For such reasons Predis do not support them globally anymore, but the
actual classes are still shipped within the library so that they can
be used to build custom stuff at a level lower than client (that is,
unless we decide to remove them for good before going stable).

CHANGELOG.md

@@ -3,6 +3,10 @@ v0.9.0 (201x-xx-xx)
 
 - The default server profile for Redis is now `2.8`.
 
+- Dropped support for streamable multibulk responses. Actually we still ship the
+  iterator response classes just in case anyone would want to build custom stuff
+  at a level lower than the client abstraction.
+
 - The `Predis\Option` namespace is now known as `Predis\Configuration` and have
   a fully-reworked `Options` class with the ability to lazily initialize values
   using objects that responds to `__invoke()` (not all the kinds of callables)

examples/MultiBulkReplyIterators.php

@@ -1,56 +0,0 @@
-<?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.
- */
-
-require 'SharedConfigurations.php';
-
-// Operations such as LRANGE, ZRANGE and others can potentially generate replies
-// containing a huge number of items. In some corner cases, such replies might
-// end up exhausting the maximum allowed memory allocated for a PHP process.
-// Multibulk iterators can be handy because they allow you to stream multibulk
-// replies using plain old PHP iterators, making it possible to iterate them with
-// a classic `foreach` loop and avoiding to consume an excessive amount of memory.
-//
-// PS: please note that multibulk iterators are supported only by the standard
-// connection backend class (Predis\Connection\StreamConnection) and not the
-// phpiredis-based one (Predis\Connection\PhpiredisConnection).
-
-// Create a client and force the connection to use iterable multibulk responses.
-$client = new Predis\Client($single_server + array('iterable_multibulk' => true));
-
-// Prepare an hash with some fields and their respective values.
-$client->hmset('metavars', array('foo' => 'bar', 'hoge' => 'piyo', 'lol' => 'wut'));
-
-// By default multibulk iterators iterate over the reply as a list of items...
-foreach ($client->hgetall('metavars') as $index => $item) {
-    echo "[$index] $item\n";
-}
-
-/* OUTPUT:
-[0] foo
-[1] bar
-[2] hoge
-[3] piyo
-[4] lol
-[5] wut
-*/
-
-// ... but certain multibulk replies are better represented as lists of tuples.
-foreach ($client->hgetall('metavars')->asTuple() as $index => $kv) {
-    list($key, $value) = $kv;
-
-    echo "[$index] $key => $value\n";
-}
-
-/* OUTPUT:
-[0] foo => bar
-[1] hoge => piyo
-[2] lol => wut
-*/

lib/Predis/Connection/ComposableStreamConnection.php

@@ -130,6 +130,6 @@ class ComposableStreamConnection extends StreamConnection implements ComposableC
      */
     public function __sleep()
     {
-        return array_diff(array_merge(parent::__sleep(), array('protocol')), array('mbiterable'));
+        return array_merge(parent::__sleep(), array('protocol'));
     }
 }

lib/Predis/Connection/ConnectionParameters.php

@@ -64,7 +64,6 @@ class ConnectionParameters implements ConnectionParametersInterface
             'persistent' => 'self::castBoolean',
             'timeout' => 'self::castFloat',
             'read_write_timeout' => 'self::castFloat',
-            'iterable_multibulk' => 'self::castBoolean',
         );
     }
 

lib/Predis/Connection/PhpiredisConnection.php

@@ -89,9 +89,6 @@ class PhpiredisConnection extends AbstractConnection
      */
     protected function checkParameters(ConnectionParametersInterface $parameters)
     {
-        if (isset($parameters->iterable_multibulk)) {
-            $this->onInvalidOption('iterable_multibulk', $parameters);
-        }
         if (isset($parameters->persistent)) {
             $this->onInvalidOption('persistent', $parameters);
         }

lib/Predis/Connection/PhpiredisStreamConnection.php

@@ -83,18 +83,6 @@ class PhpiredisStreamConnection extends StreamConnection
         }
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    protected function checkParameters(ConnectionParametersInterface $parameters)
-    {
-        if (isset($parameters->iterable_multibulk)) {
-            $this->onInvalidOption('iterable_multibulk', $parameters);
-        }
-
-        return parent::checkParameters($parameters);
-    }
-
     /**
      * Initializes the protocol reader resource.
      */
@@ -178,14 +166,6 @@ class PhpiredisStreamConnection extends StreamConnection
         $this->writeBytes(phpiredis_format_command($cmdargs));
     }
 
-    /**
-     * {@inheritdoc}
-     */
-    public function __sleep()
-    {
-        return array_diff(parent::__sleep(), array('mbiterable'));
-    }
-
     /**
      * {@inheritdoc}
      */

lib/Predis/Connection/StreamConnection.php

@@ -14,8 +14,6 @@ namespace Predis\Connection;
 use Predis\ResponseError;
 use Predis\ResponseQueued;
 use Predis\Command\CommandInterface;
-use Predis\Iterator\MultiBulkResponseSimple;
-
 /**
  * Standard connection to Redis servers implemented on top of PHP's streams.
  * The connection parameters supported by this class are:
@@ -28,24 +26,11 @@ use Predis\Iterator\MultiBulkResponseSimple;
  *  - async_connect: performs the connection asynchronously.
  *  - tcp_nodelay: enables or disables Nagle's algorithm for coalescing.
  *  - persistent: the connection is left intact after a GC collection.
- *  - iterable_multibulk: multibulk replies treated as iterable objects.
  *
  * @author Daniele Alessandri <suppakilla@gmail.com>
  */
 class StreamConnection extends AbstractConnection
 {
-    private $mbiterable;
-
-    /**
-     * {@inheritdoc}
-     */
-    public function __construct(ConnectionParametersInterface $parameters)
-    {
-        $this->mbiterable = (bool) $parameters->iterable_multibulk;
-
-        parent::__construct($parameters);
-    }
-
     /**
      * Disconnects from the server and destroys the underlying resource when
      * PHP's garbage collector kicks in only if the connection has not been
@@ -249,9 +234,6 @@ class StreamConnection extends AbstractConnection
                 if ($count === -1) {
                     return null;
                 }
-                if ($this->mbiterable) {
-                    return new MultiBulkResponseSimple($this, $count);
-                }
 
                 $multibulk = array();
 
@@ -293,12 +275,4 @@ class StreamConnection extends AbstractConnection
 
         $this->writeBytes($buffer);
     }
-
-    /**
-     * {@inheritdoc}
-     */
-    public function __sleep()
-    {
-        return array_merge(parent::__sleep(), array('mbiterable'));
-    }
 }

lib/Predis/Pipeline/MultiExecExecutor.php

@@ -11,7 +11,6 @@
 
 namespace Predis\Pipeline;
 
-use Iterator;
 use SplQueue;
 use Predis\ClientException;
 use Predis\ResponseErrorInterface;
@@ -92,38 +91,7 @@ class MultiExecExecutor implements PipelineExecutorInterface
             throw new ClientException("Invalid number of replies [expected: ".count($commands)." - actual: ".count($responses)."]");
         }
 
-        $consumer = $responses instanceof Iterator ? 'consumeIteratorResponse' : 'consumeArrayResponse';
-
-        return $this->$consumer($commands, $responses);
-    }
-
-    /**
-     * Consumes an iterator response returned by EXEC.
-     *
-     * @param SplQueue $commands Pipelined commands
-     * @param Iterator $responses Responses returned by EXEC.
-     * @return array
-     */
-    protected function consumeIteratorResponse(SplQueue $commands, Iterator $responses)
-    {
-        $values = array();
-
-        foreach ($responses as $response) {
-            $command = $commands->dequeue();
-
-            if ($response instanceof ResponseObjectInterface) {
-                if ($response instanceof Iterator) {
-                    $response = iterator_to_array($response);
-                    $values[] = $command->parseResponse($response);
-                } else {
-                    $values[] = $response;
-                }
-            } else {
-                $values[] = $command->parseResponse($response);
-            }
-        }
-
-        return $values;
+        return $this->consumeArrayResponse($commands, $responses);
     }
 
     /**

lib/Predis/Pipeline/SafeClusterExecutor.php

@@ -59,8 +59,7 @@ class SafeClusterExecutor implements PipelineExecutorInterface
             }
 
             try {
-                $response = $cmdConnection->readResponse($command);
-                $values[$i] = $response instanceof \Iterator ? iterator_to_array($response) : $response;
+                $values[$i] = $cmdConnection->readResponse($command);
             } catch (CommunicationException $exception) {
                 $values[$i] = $exception;
                 $connectionExceptions[$connectionObjectHash] = $exception;

lib/Predis/Pipeline/SafeExecutor.php

@@ -43,8 +43,7 @@ class SafeExecutor implements PipelineExecutorInterface
             $command = $commands->dequeue();
 
             try {
-                $response = $connection->readResponse($command);
-                $values[$i] = $response instanceof \Iterator ? iterator_to_array($response) : $response;
+                $values[$i] = $connection->readResponse($command);
             } catch (CommunicationException $exception) {
                 $toAdd = count($commands) - count($values);
                 $values = array_merge($values, array_fill(0, $toAdd, $exception));

lib/Predis/Pipeline/StandardExecutor.php

@@ -11,7 +11,6 @@
 
 namespace Predis\Pipeline;
 
-use Iterator;
 use SplQueue;
 use Predis\ResponseErrorInterface;
 use Predis\ResponseObjectInterface;
@@ -67,10 +66,6 @@ class StandardExecutor implements PipelineExecutorInterface
             return $this->onResponseError($connection, $response);
         }
 
-        if ($response instanceof Iterator) {
-            return $command->parseResponse(iterator_to_array($response));
-        }
-
         return $response;
     }
 

lib/Predis/Transaction/MultiExecContext.php

@@ -375,10 +375,9 @@ class MultiExecContext implements BasicClientInterface, ExecutableContextInterfa
             break;
         } while ($attempts-- > 0);
 
-        $exec = $reply instanceof \Iterator ? iterator_to_array($reply) : $reply;
         $commands = $this->commands;
+        $size = count($reply);
 
-        $size = count($exec);
         if ($size !== count($commands)) {
             $this->onProtocolError("EXEC returned an unexpected number of replies");
         }
@@ -387,17 +386,13 @@ class MultiExecContext implements BasicClientInterface, ExecutableContextInterfa
         $useExceptions = isset($clientOpts->exceptions) ? $clientOpts->exceptions : true;
 
         for ($i = 0; $i < $size; $i++) {
-            $commandReply = $exec[$i];
+            $commandReply = $reply[$i];
 
             if ($commandReply instanceof ResponseErrorInterface && $useExceptions) {
                 $message = $commandReply->getMessage();
                 throw new ServerException($message);
             }
 
-            if ($commandReply instanceof \Iterator) {
-                $commandReply = iterator_to_array($commandReply);
-            }
-
             $values[$i] = $commands->dequeue()->parseResponse($commandReply);
         }
 

tests/Predis/Connection/ConnectionParametersTest.php

@@ -74,7 +74,6 @@ class ParametersTest extends StandardTestCase
         $overrides = array(
             'port' => 7000,
             'database' => 5,
-            'iterable_multibulk' => false,
             'custom' => 'foobar',
         );
 
@@ -85,7 +84,6 @@ class ParametersTest extends StandardTestCase
         $this->assertEquals($overrides['port'], $parameters->port);
 
         $this->assertEquals($overrides['database'], $parameters->database);
-        $this->assertEquals($overrides['iterable_multibulk'], $parameters->iterable_multibulk);
 
         $this->assertTrue(isset($parameters->custom));
         $this->assertEquals($overrides['custom'], $parameters->custom);

tests/Predis/Connection/StreamConnectionTest.php

@@ -87,20 +87,6 @@ class StreamConnectionTest extends ConnectionTestCase
         $this->assertTrue($connection->isConnected());
     }
 
-    /**
-     * @group connected
-     */
-    public function testReadsMultibulkRepliesAsIterators()
-    {
-        $connection = $this->getConnection($profile, true, array('iterable_multibulk' => true));
-
-        $connection->executeCommand($profile->createCommand('rpush', array('metavars', 'foo', 'hoge', 'lol')));
-        $connection->writeCommand($profile->createCommand('lrange', array('metavars', 0, -1)));
-
-        $this->assertInstanceOf('Predis\Iterator\MultiBulkResponse', $iterator = $connection->read());
-        $this->assertSame(array('foo', 'hoge', 'lol'), iterator_to_array($iterator));
-    }
-
     /**
      * @group connected
      * @expectedException Predis\Protocol\ProtocolException

tests/Predis/Iterator/MultiBulkResponseSimpleTest.php

@@ -14,6 +14,9 @@ namespace Predis\Iterator;
 use \PHPUnit_Framework_TestCase as StandardTestCase;
 
 use Predis\Client;
+use Predis\Connection\ComposableStreamConnection;
+use Predis\Connection\ConnectionParameters;
+use Predis\Protocol\Text\TextProtocol;
 
 /**
  * @group realm-iterators
@@ -117,18 +120,22 @@ class MultiBulkResponseSimpleTest extends StandardTestCase
      */
     protected function getClient()
     {
-        $parameters = array(
+        $parameters = new ConnectionParameters(array(
             'host' => REDIS_SERVER_HOST,
             'port' => REDIS_SERVER_PORT,
-            'iterable_multibulk' => true,
             'read_write_timeout' => 2,
-        );
+        ));
 
         $options = array(
             'profile' => REDIS_SERVER_VERSION,
         );
 
-        $client = new Client($parameters, $options);
+        $protocol = new TextProtocol();
+        $protocol->setOption('iterable_multibulk', true);
+
+        $connection = new ComposableStreamConnection($parameters, $protocol);
+
+        $client = new Client($connection, $options);
         $client->connect();
         $client->select(REDIS_SERVER_DBNUM);
         $client->flushdb();

tests/Predis/Iterator/MultiBulkResponseTupleTest.php

@@ -14,6 +14,9 @@ namespace Predis\Iterator;
 use \PHPUnit_Framework_TestCase as StandardTestCase;
 
 use Predis\Client;
+use Predis\Connection\ComposableStreamConnection;
+use Predis\Connection\ConnectionParameters;
+use Predis\Protocol\Text\TextProtocol;
 
 /**
  * @group realm-iterators
@@ -103,18 +106,22 @@ class MultiBulkResponseTupleTest extends StandardTestCase
      */
     protected function getClient()
     {
-        $parameters = array(
+        $parameters = new ConnectionParameters(array(
             'host' => REDIS_SERVER_HOST,
             'port' => REDIS_SERVER_PORT,
-            'iterable_multibulk' => true,
             'read_write_timeout' => 2,
-        );
+        ));
 
         $options = array(
             'profile' => REDIS_SERVER_VERSION,
         );
 
-        $client = new Client($parameters, $options);
+        $protocol = new TextProtocol();
+        $protocol->setOption('iterable_multibulk', true);
+
+        $connection = new ComposableStreamConnection($parameters, $protocol);
+
+        $client = new Client($connection, $options);
         $client->connect();
         $client->select(REDIS_SERVER_DBNUM);
         $client->flushdb();

tests/Predis/Pipeline/MultiExecExecutorTest.php

@@ -13,20 +13,12 @@ namespace Predis\Pipeline;
 
 use \PHPUnit_Framework_TestCase as StandardTestCase;
 
-use ArrayIterator;
 use SplQueue;
 use Predis\ResponseError;
 use Predis\ResponseObjectInterface;
 use Predis\ResponseQueued;
 use Predis\Profile\ServerProfile;
 
-/**
- *
- */
-class ResponseIteratorStub extends ArrayIterator implements ResponseObjectInterface
-{
-}
-
 /**
  *
  */
@@ -57,32 +49,6 @@ class MultiExecExecutorTest extends StandardTestCase
         $this->assertSame(array(true, true, true), $replies);
     }
 
-    /**
-     * @group disconnected
-     */
-    public function testExecutorWithSingleConnectionReturningIterator()
-    {
-        $executor = new MultiExecExecutor();
-        $pipeline = $this->getCommandsQueue();
-        $queued = new ResponseQueued();
-        $execResponse = new ResponseIteratorStub(array('PONG', 'PONG', 'PONG'));
-
-        $connection = $this->getMock('Predis\Connection\SingleConnectionInterface');
-        $connection->expects($this->exactly(2))
-                   ->method('executeCommand')
-                   ->will($this->onConsecutiveCalls(true, $execResponse));
-        $connection->expects($this->exactly(3))
-                   ->method('writeCommand');
-        $connection->expects($this->at(3))
-                   ->method('readResponse')
-                   ->will($this->onConsecutiveCalls($queued, $queued, $queued));
-
-        $replies = $executor->execute($connection, $pipeline);
-
-        $this->assertTrue($pipeline->isEmpty());
-        $this->assertSame(array(true, true, true), $replies);
-    }
-
     /**
      * @group disconnected
      * @expectedException Predis\ClientException