Implement "idle" timeout to close underlying connection when unused

Author: clue

README.md

@@ -215,9 +215,9 @@ This method immediately returns a "virtual" connection implementing the
 Internally, it lazily creates the underlying database connection only on
 demand once the first request is invoked on this instance and will queue
 all outstanding requests until the underlying connection is ready.
-Additionally, it will keep track of this underlying connection and will
-create a new underlying connection on demand when the current connection
-is lost.
+Additionally, it will only keep this underlying connection in an "idle" state
+for 60s by default and will automatically close the underlying connection when
+it is no longer needed.
 
 From a consumer side this means that you can start sending commands to the
 database right away while the underlying connection may still be
@@ -231,7 +231,9 @@ to deal with its async resolution.
 If the underlying database connection fails, it will reject all
 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.
+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
@@ -245,8 +247,8 @@ first request is invoked. Accordingly, any eventual connection issues
 will be detected once this instance is first used. You can use the
 `end()` method to ensure that the "virtual" connection will be soft-closed
 and no further commands can be enqueued. Similarly, calling `end()` on
-this instance before invoking any requests will succeed immediately and
-will not wait for an actual underlying connection.
+this instance when not currently connected will succeed immediately and
+will not have to wait for an actual underlying connection.
 
 Depending on your particular use case, you may prefer this method or the
 underlying `createClient()` which resolves with a promise. For many
@@ -312,6 +314,19 @@ in seconds (or use a negative number to not apply a timeout) like this:
 $factory->createLazyClient('localhost?timeout=0.5');
 ```
 
+By default, this method will keep "idle" connection open for 60s and will
+then end the underlying connection. The next request after an "idle"
+connection ended will automatically create a new underlying connection.
+This ensure you always get a "fresh" connection and as such should not be
+confused with a "keepalive" or "heartbeat" mechanism, as this will not
+actively try to probe the connection. You can explicitly pass a custom
+idle timeout value in seconds (or use a negative number to not apply a
+timeout) like this:
+
+```php
+$factory->createLazyClient('localhost?idle=0.1');
+```
+
 ### Client
 
 The `Client` is responsible for exchanging messages with Redis

composer.json

@@ -14,7 +14,7 @@
         "php": ">=5.3",
         "clue/redis-protocol": "0.3.*",
         "evenement/evenement": "^3.0 || ^2.0 || ^1.0",
-        "react/event-loop": "^1.0 || ^0.5 || ^0.4 || ^0.3",
+        "react/event-loop": "^1.0 || ^0.5",
         "react/promise": "^2.0 || ^1.1",
         "react/promise-timer": "^1.5",
         "react/socket": "^1.1"

src/Factory.php

@@ -123,7 +123,7 @@ function ($error) use ($client) {
      */
     public function createLazyClient($target)
     {
-        return new LazyClient($target, $this);
+        return new LazyClient($target, $this, $this->loop);
     }
 
     /**

src/LazyClient.php

@@ -4,6 +4,7 @@
 
 use Evenement\EventEmitter;
 use React\Stream\Util;
+use React\EventLoop\LoopInterface;
 
 /**
  * @internal
@@ -16,13 +17,25 @@ class LazyClient extends EventEmitter implements Client
     private $closed = false;
     private $promise;
 
+    private $loop;
+    private $idlePeriod = 60.0;
+    private $idleTimer;
+    private $pending = 0;
+
     /**
      * @param $target
      */
-    public function __construct($target, Factory $factory)
+    public function __construct($target, Factory $factory, LoopInterface $loop)
     {
+        $args = array();
+        \parse_str(\parse_url($target, \PHP_URL_QUERY), $args);
+        if (isset($args['idle'])) {
+            $this->idlePeriod = (float)$args['idle'];
+        }
+
         $this->target = $target;
         $this->factory = $factory;
+        $this->loop = $loop;
     }
 
     private function client()
@@ -33,11 +46,13 @@ private function client()
 
         $self = $this;
         $pending =& $this->promise;
-        return $pending = $this->factory->createClient($this->target)->then(function (Client $client) use ($self, &$pending) {
+        $idleTimer=& $this->idleTimer;
+        $loop = $this->loop;
+        return $pending = $this->factory->createClient($this->target)->then(function (Client $client) use ($self, &$pending, &$idleTimer, $loop) {
             // connection completed => remember only until closed
             $subscribed = array();
             $psubscribed = array();
-            $client->on('close', function () use (&$pending, $self, &$subscribed, &$psubscribed) {
+            $client->on('close', function () use (&$pending, $self, &$subscribed, &$psubscribed, &$idleTimer, $loop) {
                 $pending = null;
 
                 // foward unsubscribe/punsubscribe events when underlying connection closes
@@ -49,6 +64,11 @@ private function client()
                 foreach ($psubscribed as $pattern => $_) {
                     $self->emit('punsubscribe', array($pattern, --$n));
                 }
+
+                if ($idleTimer !== null) {
+                    $loop->cancelTimer($idleTimer);
+                    $idleTimer = null;
+                }
             });
 
             // keep track of all channels and patterns this connection is subscribed to
@@ -93,8 +113,19 @@ public function __call($name, $args)
             return \React\Promise\reject(new \RuntimeException('Connection closed'));
         }
 
-        return $this->client()->then(function (Client $client) use ($name, $args) {
-            return \call_user_func_array(array($client, $name), $args);
+        $that = $this;
+        return $this->client()->then(function (Client $client) use ($name, $args, $that) {
+            $that->awake();
+            return \call_user_func_array(array($client, $name), $args)->then(
+                function ($result) use ($that) {
+                    $that->idle();
+                    return $result;
+                },
+                function ($error) use ($that) {
+                    $that->idle();
+                    throw $error;
+                }
+            );
         });
     }
 
@@ -134,7 +165,45 @@ public function close()
             $this->promise = null;
         }
 
+        if ($this->idleTimer !== null) {
+            $this->loop->cancelTimer($this->idleTimer);
+            $this->idleTimer = null;
+        }
+
         $this->emit('close');
         $this->removeAllListeners();
     }
+
+    /**
+     * @internal
+     */
+    public function awake()
+    {
+        ++$this->pending;
+
+        if ($this->idleTimer !== null) {
+            $this->loop->cancelTimer($this->idleTimer);
+            $this->idleTimer = null;
+        }
+    }
+
+    /**
+     * @internal
+     */
+    public function idle()
+    {
+        --$this->pending;
+
+        if ($this->pending < 1 && $this->idlePeriod >= 0) {
+            $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;
+            });
+        }
+    }
 }

tests/FunctionalTest.php

@@ -51,6 +51,30 @@ public function testPingLazy()
         $this->assertEquals('PONG', $ret);
     }
 
+    /**
+     * @doesNotPerformAssertions
+     */
+    public function testPingLazyWillNotBlockLoopWhenIdleTimeIsSmall()
+    {
+        $client = $this->factory->createLazyClient($this->uri . '?idle=0');
+
+        $client->ping();
+
+        $this->loop->run();
+    }
+
+    /**
+     * @doesNotPerformAssertions
+     */
+    public function testLazyClientWithoutCommandsWillNotBlockLoop()
+    {
+        $client = $this->factory->createLazyClient($this->uri);
+
+        $this->loop->run();
+
+        unset($client);
+    }
+
     public function testMgetIsNotInterpretedAsSubMessage()
     {
         $client = $this->createClient($this->uri);