Do not enter "idle" state when using PubSub channels

Author: clue

README.md

@@ -235,12 +235,13 @@ will again try to open a new underlying connection. Note that this may
 require special care if you're using transactions (`MULTI`/`EXEC`) that are kept
 open for longer than the idle period.
 
-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.
+While using PubSub channels (see `SUBSCRIBE` and `PSUBSCRIBE` commands), this client
+will never reach an "idle" state and will keep pending forever (or until the
+underlying database connection is lost). Additionally, if the underlying
+database connection drops, 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 repeating the above commands again.
 
 Note that creating the underlying connection will be deferred until the
 first request is invoked. Accordingly, any eventual connection issues

src/LazyClient.php

@@ -22,6 +22,9 @@ class LazyClient extends EventEmitter implements Client
     private $idleTimer;
     private $pending = 0;
 
+    private $subscribed = array();
+    private $psubscribed = array();
+
     /**
      * @param $target
      */
@@ -47,11 +50,11 @@ private function client()
         $self = $this;
         $pending =& $this->promise;
         $idleTimer=& $this->idleTimer;
+        $subscribed =& $this->subscribed;
+        $psubscribed =& $this->psubscribed;
         $loop = $this->loop;
-        return $pending = $this->factory->createClient($this->target)->then(function (Client $client) use ($self, &$pending, &$idleTimer, $loop) {
+        return $pending = $this->factory->createClient($this->target)->then(function (Client $client) use ($self, &$pending, &$idleTimer, &$subscribed, &$psubscribed, $loop) {
             // connection completed => remember only until closed
-            $subscribed = array();
-            $psubscribed = array();
             $client->on('close', function () use (&$pending, $self, &$subscribed, &$psubscribed, &$idleTimer, $loop) {
                 $pending = null;
 
@@ -64,6 +67,8 @@ private function client()
                 foreach ($psubscribed as $pattern => $_) {
                     $self->emit('punsubscribe', array($pattern, --$n));
                 }
+                $subscribed = array();
+                $psubscribed = array();
 
                 if ($idleTimer !== null) {
                     $loop->cancelTimer($idleTimer);
@@ -194,15 +199,15 @@ public function idle()
     {
         --$this->pending;
 
-        if ($this->pending < 1 && $this->idlePeriod >= 0) {
+        if ($this->pending < 1 && $this->idlePeriod >= 0 && !$this->subscribed && !$this->psubscribed) {
             $idleTimer =& $this->idleTimer;
             $promise =& $this->promise;
             $idleTimer = $this->loop->addTimer($this->idlePeriod, function () use (&$idleTimer, &$promise) {
                 $promise->then(function (Client $client) {
                     $client->close();
                 });
-                    $promise = null;
-                    $idleTimer = null;
+                $promise = null;
+                $idleTimer = null;
             });
         }
     }

tests/LazyClientTest.php

@@ -444,4 +444,43 @@ public function testEmitsUnsubscribeAndPunsubscribeEventsWhenUnderlyingClientClo
         $this->client->on('punsubscribe', $this->expectCallableOnce());
         $client->emit('close');
     }
+
+    public function testSubscribeWillResolveWhenUnderlyingClientResolvesSubscribeAndNotStartIdleTimerWithIdleDueToSubscription()
+    {
+        $deferred = new Deferred();
+        $client = $this->getMockBuilder('Clue\React\Redis\StreamingClient')->disableOriginalConstructor()->setMethods(array('__call'))->getMock();
+        $client->expects($this->once())->method('__call')->with('subscribe')->willReturn($deferred->promise());
+
+        $this->factory->expects($this->once())->method('createClient')->willReturn(\React\Promise\resolve($client));
+
+        $this->loop->expects($this->never())->method('addTimer');
+
+        $promise = $this->client->subscribe('foo');
+        $client->emit('subscribe', array('foo', 1));
+        $deferred->resolve(array('subscribe', 'foo', 1));
+
+        $promise->then($this->expectCallableOnceWith(array('subscribe', 'foo', 1)));
+    }
+
+    public function testUnsubscribeAfterSubscribeWillResolveWhenUnderlyingClientResolvesUnsubscribeAndStartIdleTimerWhenSubscriptionStopped()
+    {
+        $deferredSubscribe = new Deferred();
+        $deferredUnsubscribe = new Deferred();
+        $client = $this->getMockBuilder('Clue\React\Redis\StreamingClient')->disableOriginalConstructor()->setMethods(array('__call'))->getMock();
+        $client->expects($this->exactly(2))->method('__call')->willReturnOnConsecutiveCalls($deferredSubscribe->promise(), $deferredUnsubscribe->promise());
+
+        $this->factory->expects($this->once())->method('createClient')->willReturn(\React\Promise\resolve($client));
+
+        $this->loop->expects($this->once())->method('addTimer');
+
+        $promise = $this->client->subscribe('foo');
+        $client->emit('subscribe', array('foo', 1));
+        $deferredSubscribe->resolve(array('subscribe', 'foo', 1));
+        $promise->then($this->expectCallableOnceWith(array('subscribe', 'foo', 1)));
+
+        $promise = $this->client->unsubscribe('foo');
+        $client->emit('unsubscribe', array('foo', 0));
+        $deferredUnsubscribe->resolve(array('unsubscribe', 'foo', 0));
+        $promise->then($this->expectCallableOnceWith(array('unsubscribe', 'foo', 0)));
+    }
 }