Keep track of (p)subscribe responses to toggle PubSub state

The (p)(un)subscribe commands only accept a single topic name for now.

Author: clue

src/StreamingClient.php

@@ -11,6 +11,7 @@
 use Clue\Redis\Protocol\Factory as ProtocolFactory;
 use UnderflowException;
 use RuntimeException;
+use InvalidArgumentException;
 use React\Promise\Deferred;
 use Clue\Redis\Protocol\Model\ErrorReply;
 use Clue\Redis\Protocol\Model\ModelInterface;
@@ -77,18 +78,43 @@ public function __call($name, $args)
     {
         $request = new Deferred();
 
+        $name = strtolower($name);
+
+        // special (p)(un)subscribe commands only accept a single parameter and have custom response logic applied
+        static $pubsubs = array('subscribe', 'unsubscribe', 'psubscribe', 'punsubscribe');
+
         if ($this->ending) {
             $request->reject(new RuntimeException('Connection closed'));
+        } elseif (count($args) !== 1 && in_array($name, $pubsubs)) {
+            $request->reject(new InvalidArgumentException('PubSub commands limited to single argument'));
         } else {
             $this->stream->write($this->serializer->getRequestMessage($name, $args));
             $this->requests []= $request;
         }
 
-        if (strtolower($name) === 'monitor') {
+        if ($name === 'monitor') {
             $monitoring =& $this->monitoring;
             $request->then(function () use (&$monitoring) {
                 $monitoring = true;
             });
+        } elseif (in_array($name, $pubsubs)) {
+            $that = $this;
+            $subscribed =& $this->subscribed;
+            $psubscribed =& $this->psubscribed;
+
+            $request->then(function ($array) use ($that, &$subscribed, &$psubscribed) {
+                $first = array_shift($array);
+
+                // (p)(un)subscribe messages are to be forwarded
+                $that->emit($first, $array);
+
+                // remember number of (p)subscribe topics
+                if ($first === 'subscribe' || $first === 'unsubscribe') {
+                    $subscribed = $array[1];
+                } else {
+                    $psubscribed = $array[1];
+                }
+            });
         }
 
         return $request->promise();
@@ -103,17 +129,13 @@ public function handleMessage(ModelInterface $message)
             return;
         }
 
-        if (/*($this->subscribed !== 0 || $this->psubscribed !== 0) &&*/ $message instanceof MultiBulkReply) {
+        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
+            // pub/sub messages are to be forwarded and should not be processed as request responses
             if (in_array($first, array('message', 'pmessage'))) {
+                $this->emit($first, $array);
                 return;
             }
         }

tests/FunctionalTest.php

@@ -36,8 +36,6 @@ public function testPing()
 
     public function testMgetIsNotInterpretedAsSubMessage()
     {
-        $this->markTestIncomplete();
-
         $client = $this->createClient();
 
         $client->mset('message', 'message', 'channel', 'channel', 'payload', 'payload');

tests/StreamingClientTest.php

@@ -228,27 +228,31 @@ public function testPubsubSubscribe()
      * @depends testPubsubSubscribe
      * @param Client $client
      */
+    public function testPubsubPatternSubscribe(Client $client)
+    {
+         $promise = $client->psubscribe('demo_*');
+         $this->expectPromiseResolve($promise);
+
+         $client->on('psubscribe', $this->expectCallableOnce());
+         $client->handleMessage(new MultiBulkReply(array(new BulkReply('psubscribe'), new BulkReply('demo_*'), new IntegerReply(1))));
+
+        return $client;
+    }
+
+    /**
+     * @depends testPubsubPatternSubscribe
+     * @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'))));
     }
 
-    public function testPubsubSubscribeMultiple()
+    public function testPubsubSubscribeSingleOnly()
     {
-        $this->markTestIncomplete();
-
-        $promise = $this->client->subscribe('first', 'second');
-        $this->expectPromiseResolve($promise);
-
-        // expect two "subscribe" events
-        $mock = $this->createCallableMock();
-        $mock->expects($this->exactly(2))->method('__invoke');
-        $this->client->on('subscribe', $mock);
-
-        $this->client->handleMessage(new MultiBulkReply(array(new BulkReply('subscribe'), new BulkReply('first'), new IntegerReply(1))));
-        $this->client->handleMessage(new MultiBulkReply(array(new BulkReply('subscribe'), new BulkReply('second'), new IntegerReply(2))));
-
-        return $this->client;
+        $this->expectPromiseReject($this->client->subscribe('a', 'b'));
+        $this->expectPromiseReject($this->client->unsubscribe('a', 'b'));
+        $this->expectPromiseReject($this->client->unsubscribe());
     }
 }