Skeleton for forwarding pub/sub events

Author: clue

src/Client.php

@@ -11,6 +11,14 @@
  *
  * @event data(ModelInterface $messageModel, Client $thisClient)
  * @event close()
+ *
+ * @event message($channel, $message)
+ * @event subscribe($channel, $numberOfChannels)
+ * @event unsubscribe($channel, $numberOfChannels)
+ *
+ * @event pmessage($pattern, $channel, $message)
+ * @event psubscribe($channel, $numberOfChannels)
+ * @event punsubscribe($channel, $numberOfChannels)
  */
 interface Client extends EventEmitterInterface
 {

src/StreamingClient.php

@@ -14,6 +14,7 @@
 use React\Promise\Deferred;
 use Clue\Redis\Protocol\Model\ErrorReply;
 use Clue\Redis\Protocol\Model\ModelInterface;
+use Clue\Redis\Protocol\Model\MultiBulkReply;
 
 class StreamingClient extends EventEmitter implements Client
 {
@@ -24,6 +25,9 @@ class StreamingClient extends EventEmitter implements Client
     private $ending = false;
     private $closed = false;
 
+    private $subscribed = 0;
+    private $psubscribed = 0;
+
     public function __construct(Stream $stream, ParserInterface $parser = null, SerializerInterface $serializer = null)
     {
         if ($parser === null || $serializer === null) {
@@ -85,6 +89,21 @@ public function handleMessage(ModelInterface $message)
     {
         $this->emit('data', array($message, $this));
 
+        if (/*($this->subscribed !== 0 || $this->psubscribed !== 0) &&*/ $message instanceof MultiBulkReply) {
+            $array = $message->getValueNative();
+            $first = array_shift($array);
+
+            // pub/sub events are to be forwarded
+            if (in_array($first, array('message', 'subscribe', 'unsubscribe', 'pmessage', 'psubscribe', 'punsubscribe'))) {
+                $this->emit($first, $array);
+            }
+
+            // pub/sub message events should not be processed as request responses
+            if (in_array($first, array('message', 'pmessage'))) {
+                return;
+            }
+        }
+
         if (!$this->requests) {
             throw new UnderflowException('Unexpected reply received, no matching request found');
         }

tests/StreamingClientTest.php

@@ -5,6 +5,8 @@
 use Clue\Redis\Protocol\Model\IntegerReply;
 use Clue\Redis\Protocol\Model\BulkReply;
 use Clue\Redis\Protocol\Model\ErrorReply;
+use Clue\Redis\Protocol\Model\MultiBulkReply;
+use Clue\React\Redis\Client;
 
 class ClientTest extends TestCase
 {
@@ -163,4 +165,25 @@ public function testReceivingUnexpectedMessageThrowsException()
         $this->setExpectedException('UnderflowException');
         $this->client->handleMessage(new BulkReply('PONG'));
     }
+
+    public function testPubsubSubscribe()
+    {
+        $promise = $this->client->subscribe('test');
+        $this->expectPromiseResolve($promise);
+
+        $this->client->on('subscribe', $this->expectCallableOnce());
+        $this->client->handleMessage(new MultiBulkReply(array(new BulkReply('subscribe'), new BulkReply('test'), new IntegerReply(1))));
+
+        return $this->client;
+    }
+
+    /**
+     * @depends testPubsubSubscribe
+     * @param Client $client
+     */
+    public function testPubsubMessage(Client $client)
+    {
+        $client->on('message', $this->expectCallableOnce());
+        $client->handleMessage(new MultiBulkReply(array(new BulkReply('message'), new BulkReply('test'), new BulkReply('payload'))));
+    }
 }