Add psubscribe and punsubscribe support

Author: huangdijia

src/CommandInvoker.php

@@ -113,6 +113,33 @@ public function receive(Connection $connection)
                 continue;
             }
 
+            if ($type == 'psubscribe' && count($buffer) == 6) {
+                $this->resultChannel->push($buffer);
+                $buffer = null;
+                continue;
+            }
+
+            if ($type == 'punsubscribe' && count($buffer) == 6) {
+                $this->resultChannel->push($buffer);
+                $buffer = null;
+                continue;
+            }
+
+            if ($type == 'pmessage' && count($buffer) == 9) {
+                $message = new Message();
+                $message->pattern = $buffer[4];
+                $message->channel = $buffer[6];
+                $message->payload = $buffer[8];
+                $timerID = Timer::after(30 * 1000, function () use ($message) {
+                    static::error(sprintf('Message channel (%s) is 30 seconds full, disconnected', $message->channel));
+                    $this->interrupt();
+                });
+                $this->messageChannel->push($message);
+                Timer::clear($timerID);
+                $buffer = null;
+                continue;
+            }
+
             if ($type == 'pong' && count($buffer) == 5) {
                 $this->pingChannel->push('pong');
                 $buffer = null;

src/Message.php

@@ -9,6 +9,11 @@
 class Message
 {
 
+    /**
+     * @var string
+     */
+    public $pattern;
+
     /**
      * @var string
      */

src/Subscriber.php

@@ -125,6 +125,44 @@ public function unsubscribe(string ...$channels)
         }
     }
 
+    /**
+     * PSubscribe.
+     * @throws \Swoole\Exception
+     * @throws Throwable
+     */
+    public function psubscribe(string ...$channels)
+    {
+        $channels = array_map(function ($channel) {
+            return $this->prefix . $channel;
+        }, $channels);
+        $result = $this->commandInvoker->invoke(['psubscribe', ...$channels], count($channels));
+        foreach ($result as $value) {
+            if ($value === false) {
+                $this->commandInvoker->interrupt();
+                throw new SubscribeException('Psubscribe failed');
+            }
+        }
+    }
+
+    /**
+     * PUnsubscribe.
+     * @throws \Swoole\Exception
+     * @throws Throwable
+     */
+    public function punsubscribe(string ...$channels)
+    {
+        $channels = array_map(function ($channel) {
+            return $this->prefix . $channel;
+        }, $channels);
+        $result = $this->commandInvoker->invoke(['punsubscribe', ...$channels], count($channels));
+        foreach ($result as $value) {
+            if ($value === false) {
+                $this->commandInvoker->interrupt();
+                throw new UnsubscribeException('Punsubscribe failed');
+            }
+        }
+    }
+
     /**
      * Channel
      * @return \Swoole\Coroutine\Channel