ProtocolProcessor.php 3.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122
  1. <?php
  2. /*
  3. * This file is part of the Predis package.
  4. *
  5. * (c) Daniele Alessandri <suppakilla@gmail.com>
  6. *
  7. * For the full copyright and license information, please view the LICENSE
  8. * file that was distributed with this source code.
  9. */
  10. namespace Predis\Protocol\Text;
  11. use Predis\Command\CommandInterface;
  12. use Predis\CommunicationException;
  13. use Predis\Connection\CompositeConnectionInterface;
  14. use Predis\Protocol\ProtocolException;
  15. use Predis\Protocol\ProtocolProcessorInterface;
  16. use Predis\Response\Error as ErrorResponse;
  17. use Predis\Response\Iterator\MultiBulk as MultiBulkIterator;
  18. use Predis\Response\Status as StatusResponse;
  19. /**
  20. * Protocol processor for the standard Redis wire protocol.
  21. *
  22. * @link http://redis.io/topics/protocol
  23. *
  24. * @author Daniele Alessandri <suppakilla@gmail.com>
  25. */
  26. class ProtocolProcessor implements ProtocolProcessorInterface
  27. {
  28. protected $mbiterable;
  29. protected $serializer;
  30. /**
  31. *
  32. */
  33. public function __construct()
  34. {
  35. $this->mbiterable = false;
  36. $this->serializer = new RequestSerializer();
  37. }
  38. /**
  39. * {@inheritdoc}
  40. */
  41. public function write(CompositeConnectionInterface $connection, CommandInterface $command)
  42. {
  43. $request = $this->serializer->serialize($command);
  44. $connection->writeBuffer($request);
  45. }
  46. /**
  47. * {@inheritdoc}
  48. */
  49. public function read(CompositeConnectionInterface $connection)
  50. {
  51. $chunk = $connection->readLine();
  52. $prefix = $chunk[0];
  53. $payload = substr($chunk, 1);
  54. switch ($prefix) {
  55. case '+':
  56. return new StatusResponse($payload);
  57. case '$':
  58. $size = (int) $payload;
  59. if ($size === -1) {
  60. return;
  61. }
  62. return substr($connection->readBuffer($size + 2), 0, -2);
  63. case '*':
  64. $count = (int) $payload;
  65. if ($count === -1) {
  66. return;
  67. }
  68. if ($this->mbiterable) {
  69. return new MultiBulkIterator($connection, $count);
  70. }
  71. $multibulk = array();
  72. for ($i = 0; $i < $count; ++$i) {
  73. $multibulk[$i] = $this->read($connection);
  74. }
  75. return $multibulk;
  76. case ':':
  77. return (int) $payload;
  78. case '-':
  79. return new ErrorResponse($payload);
  80. default:
  81. CommunicationException::handle(new ProtocolException(
  82. $connection, "Unknown response prefix: '$prefix'."
  83. ));
  84. return;
  85. }
  86. }
  87. /**
  88. * Enables or disables returning multibulk responses as specialized PHP
  89. * iterators used to stream bulk elements of a multibulk response instead
  90. * returning a plain array.
  91. *
  92. * Streamable multibulk responses are not globally supported by the
  93. * abstractions built-in into Predis, such as transactions or pipelines.
  94. * Use them with care!
  95. *
  96. * @param bool $value Enable or disable streamable multibulk responses.
  97. */
  98. public function useIterableMultibulk($value)
  99. {
  100. $this->mbiterable = (bool) $value;
  101. }
  102. }