Automatically send unsubscribe/punsubscribe events for PubSub channels

Author: clue

README.md

@@ -233,6 +233,13 @@ outstanding commands and will return to the initial "idle" state. This
 means that you can keep sending additional commands at a later time which
 will again try to open the underlying connection.
 
+If the underlying database connection drops while using PubSub channels
+(see `SUBSCRIBE` and `PSUBSCRIBE` commands), it will automatically send the
+appropriate `unsubscribe` and `punsubscribe` events for all currently active
+channel and pattern subscriptions. This allows you to react to these
+events and restore your subscriptions by creating a new underlying
+connection with the above commands.
+
 Note that creating the underlying connection will be deferred until the
 first request is invoked. Accordingly, any eventual connection issues
 will be detected once this instance is first used. You can use the

examples/subscribe.php

@@ -12,10 +12,26 @@
 $client = $factory->createLazyClient('localhost');
 $client->subscribe($channel)->then(function () {
     echo 'Now subscribed to channel ' . PHP_EOL;
+}, function (Exception $e) {
+    echo 'Unable to subscribe: ' . $e->getMessage() . PHP_EOL;
 });
 
 $client->on('message', function ($channel, $message) {
     echo 'Message on ' . $channel . ': ' . $message . PHP_EOL;
 });
 
+// automatically re-subscribe to channel on connection issues
+$client->on('unsubscribe', function ($channel) use ($client, $loop) {
+    echo 'Unsubscribed from ' . $channel . PHP_EOL;
+
+    $loop->addPeriodicTimer(2.0, function ($timer) use ($client, $channel, $loop){
+        $client->subscribe($channel)->then(function () use ($timer, $loop) {
+            echo 'Now subscribed again' . PHP_EOL;
+            $loop->cancelTimer($timer);
+        }, function (Exception $e) {
+            echo 'Unable to subscribe again: ' . $e->getMessage() . PHP_EOL;
+        });
+    });
+});
+
 $loop->run();

src/LazyClient.php

@@ -35,8 +35,34 @@ private function client()
         $pending =& $this->promise;
         return $pending = $this->factory->createClient($this->target)->then(function (Client $client) use ($self, &$pending) {
             // connection completed => remember only until closed
-            $client->on('close', function () use (&$pending) {
+            $subscribed = array();
+            $psubscribed = array();
+            $client->on('close', function () use (&$pending, $self, &$subscribed, &$psubscribed) {
                 $pending = null;
+
+                // foward unsubscribe/punsubscribe events when underlying connection closes
+                $n = count($subscribed);
+                foreach ($subscribed as $channel => $_) {
+                    $self->emit('unsubscribe', array($channel, --$n));
+                }
+                $n = count($psubscribed);
+                foreach ($psubscribed as $pattern => $_) {
+                    $self->emit('punsubscribe', array($pattern, --$n));
+                }
+            });
+
+            // keep track of all channels and patterns this connection is subscribed to
+            $client->on('subscribe', function ($channel) use (&$subscribed) {
+                $subscribed[$channel] = true;
+            });
+            $client->on('psubscribe', function ($pattern) use (&$psubscribed) {
+                $psubscribed[$pattern] = true;
+            });
+            $client->on('unsubscribe', function ($channel) use (&$subscribed) {
+                unset($subscribed[$channel]);
+            });
+            $client->on('punsubscribe', function ($pattern) use (&$psubscribed) {
+                unset($psubscribed[$pattern]);
             });
 
             Util::forwardEvents(

tests/LazyClientTest.php

@@ -272,4 +272,34 @@ public function testEmitsMessageEventWhenUnderlyingClientEmitsMessageForPubSubCh
         $this->client->on('message', $this->expectCallableOnce());
         $client->emit('message', array('foo', 'bar'));
     }
+
+    public function testEmitsUnsubscribeAndPunsubscribeEventsWhenUnderlyingClientClosesWhileUsingPubSubChannel()
+    {
+        $client = $this->getMockBuilder('Clue\React\Redis\StreamingClient')->disableOriginalConstructor()->setMethods(array('__call'))->getMock();
+        $client->expects($this->exactly(6))->method('__call')->willReturn(\React\Promise\resolve());
+
+        $this->factory->expects($this->once())->method('createClient')->willReturn(\React\Promise\resolve($client));
+
+        $this->client->subscribe('foo');
+        $client->emit('subscribe', array('foo', 1));
+
+        $this->client->subscribe('bar');
+        $client->emit('subscribe', array('bar', 2));
+
+        $this->client->unsubscribe('bar');
+        $client->emit('unsubscribe', array('bar', 1));
+
+        $this->client->psubscribe('foo*');
+        $client->emit('psubscribe', array('foo*', 1));
+
+        $this->client->psubscribe('bar*');
+        $client->emit('psubscribe', array('bar*', 2));
+
+        $this->client->punsubscribe('bar*');
+        $client->emit('punsubscribe', array('bar*', 1));
+
+        $this->client->on('unsubscribe', $this->expectCallableOnce());
+        $this->client->on('punsubscribe', $this->expectCallableOnce());
+        $client->emit('close');
+    }
 }