Merge pull request #88 from clue-labs/idle

Implement "idle" timeout for LazyConnection to close underlying connection when unused and automatically create new underlying connection on demand again

Author: WyriHaximus

README.md

@@ -171,24 +171,35 @@ $connection->query(…);
 This method immediately returns a "virtual" connection implementing the
 [`ConnectionInterface`](#connectioninterface) that can be used to
 interface with your MySQL database. Internally, it lazily creates the
-underlying database connection (which may take some time) only once the
-first request is invoked on this instance and will queue all outstanding
-requests until the underlying connection is ready.
+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 only keep this
+underlying connection in an "idle" state for 60s by default and will
+automatically end the underlying connection when it is no longer needed.
 
 From a consumer side this means that you can start sending queries to the
-database right away while the actual connection may still be outstanding.
-It will ensure that all commands will be executed in the order they are
-enqueued once the connection is ready. If the database connection fails,
-it will emit an `error` event, reject all outstanding commands and `close`
-the connection as described in the `ConnectionInterface`. In other words,
-it behaves just like a real connection and frees you from having to deal
-with its async resolution.
+database right away while the underlying connection may still be
+outstanding. Because creating this underlying connection may take some
+time, it will enqueue all oustanding commands and will ensure that all
+commands will be executed in correct order once the connection is ready.
+In other words, this "virtual" connection behaves just like a "real"
+connection as described in the `ConnectionInterface` and frees you from
+having 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 a new underlying connection. Note that this may
+require special care if you're using transactions that are kept open for
+longer than the idle period.
 
 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. Similarly, calling
-`quit()` on this instance before invoking any requests will succeed
-immediately and will not wait for an actual underlying connection.
+will be detected once this instance is first used. You can use the
+`quit()` method to ensure that the "virtual" connection will be soft-closed
+and no further commands can be enqueued. Similarly, calling `quit()` on
+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 `createConnection()` which resolves with a promise. For many
@@ -225,6 +236,19 @@ in seconds (or use a negative number to not apply a timeout) like this:
 $factory->createLazyConnection('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->createLazyConnection('localhost?idle=0.1');
+```
+
 ### ConnectionInterface
 
 The `ConnectionInterface` represents a connection that is responsible for
@@ -426,7 +450,7 @@ $connecion->on('close', function () {
 });
 ```
 
-See also the [#close](#close) method.
+See also the [`close()`](#close) method.
 
 ## Install
 

composer.json

@@ -6,7 +6,7 @@
     "require": {
         "php": ">=5.4.0",
         "evenement/evenement": "^3.0 || ^2.1 || ^1.1",
-        "react/event-loop": "^1.0 || ^0.5 || ^0.4",
+        "react/event-loop": "^1.0 || ^0.5",
         "react/promise": "^2.7",
         "react/promise-stream": "^1.1",
         "react/promise-timer": "^1.5",

src/ConnectionInterface.php

@@ -38,7 +38,7 @@
  *     });
  *     ```
  *
- *     See also the [#close](#close) method.
+ *     See also the [`close()`](#close) method.
  */
 interface ConnectionInterface extends EventEmitterInterface
 {

src/Factory.php

@@ -211,24 +211,35 @@ public function createConnection($uri)
      * This method immediately returns a "virtual" connection implementing the
      * [`ConnectionInterface`](#connectioninterface) that can be used to
      * interface with your MySQL database. Internally, it lazily creates the
-     * underlying database connection (which may take some time) only once the
-     * first request is invoked on this instance and will queue all outstanding
-     * requests until the underlying connection is ready.
+     * 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 only keep this
+     * underlying connection in an "idle" state for 60s by default and will
+     * automatically end the underlying connection when it is no longer needed.
      *
      * From a consumer side this means that you can start sending queries to the
-     * database right away while the actual connection may still be outstanding.
-     * It will ensure that all commands will be executed in the order they are
-     * enqueued once the connection is ready. If the database connection fails,
-     * it will emit an `error` event, reject all outstanding commands and `close`
-     * the connection as described in the `ConnectionInterface`. In other words,
-     * it behaves just like a real connection and frees you from having to deal
-     * with its async resolution.
+     * database right away while the underlying connection may still be
+     * outstanding. Because creating this underlying connection may take some
+     * time, it will enqueue all oustanding commands and will ensure that all
+     * commands will be executed in correct order once the connection is ready.
+     * In other words, this "virtual" connection behaves just like a "real"
+     * connection as described in the `ConnectionInterface` and frees you from
+     * having 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 a new underlying connection. Note that this may
+     * require special care if you're using transactions that are kept open for
+     * longer than the idle period.
      *
      * 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. Similarly, calling
-     * `quit()` on this instance before invoking any requests will succeed
-     * immediately and will not wait for an actual underlying connection.
+     * will be detected once this instance is first used. You can use the
+     * `quit()` method to ensure that the "virtual" connection will be soft-closed
+     * and no further commands can be enqueued. Similarly, calling `quit()` on
+     * 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 `createConnection()` which resolves with a promise. For many
@@ -265,11 +276,24 @@ public function createConnection($uri)
      * $factory->createLazyConnection('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->createLazyConnection('localhost?idle=0.1');
+     * ```
+     *
      * @param string $uri
      * @return ConnectionInterface
      */
     public function createLazyConnection($uri)
     {
-        return new LazyConnection($this, $uri);
+        return new LazyConnection($this, $uri, $this->loop);
     }
 }

src/Io/LazyConnection.php

@@ -6,6 +6,8 @@
 use Evenement\EventEmitter;
 use React\MySQL\Exception;
 use React\MySQL\Factory;
+use React\EventLoop\LoopInterface;
+use React\MySQL\QueryResult;
 
 /**
  * @internal
@@ -19,37 +21,94 @@ class LazyConnection extends EventEmitter implements ConnectionInterface
     private $closed = false;
     private $busy = false;
 
-    public function __construct(Factory $factory, $uri)
+    /**
+     * @var ConnectionInterface|null
+     */
+    private $disconnecting;
+
+    private $loop;
+    private $idlePeriod = 60.0;
+    private $idleTimer;
+    private $pending = 0;
+
+    public function __construct(Factory $factory, $uri, LoopInterface $loop)
     {
+        $args = array();
+        \parse_str(\parse_url($uri, \PHP_URL_QUERY), $args);
+        if (isset($args['idle'])) {
+            $this->idlePeriod = (float)$args['idle'];
+        }
+
         $this->factory = $factory;
         $this->uri = $uri;
+        $this->loop = $loop;
     }
 
     private function connecting()
     {
-        if ($this->connecting === null) {
-            $this->connecting = $this->factory->createConnection($this->uri);
+        if ($this->connecting !== null) {
+            return $this->connecting;
+        }
 
-            $this->connecting->then(function (ConnectionInterface $connection) {
-                // connection completed => forward error and close events
-                $connection->on('error', function ($e) {
-                    $this->emit('error', [$e]);
-                });
-                $connection->on('close', function () {
-                    $this->close();
-                });
-            }, function (\Exception $e) {
-                // connection failed => emit error if connection is not already closed
-                if ($this->closed) {
-                    return;
-                }
+        // force-close connection if still waiting for previous disconnection
+        if ($this->disconnecting !== null) {
+            $this->disconnecting->close();
+            $this->disconnecting = null;
+        }
 
-                $this->emit('error', [$e]);
-                $this->close();
+        $this->connecting = $connecting = $this->factory->createConnection($this->uri);
+        $this->connecting->then(function (ConnectionInterface $connection) {
+            // connection completed => remember only until closed
+            $connection->on('close', function () {
+                $this->connecting = null;
+
+                if ($this->idleTimer !== null) {
+                    $this->loop->cancelTimer($this->idleTimer);
+                    $this->idleTimer = null;
+                }
             });
+        }, function () {
+            // connection failed => discard connection attempt
+            $this->connecting = null;
+        });
+
+        return $connecting;
+    }
+
+    private function awake()
+    {
+        ++$this->pending;
+
+        if ($this->idleTimer !== null) {
+            $this->loop->cancelTimer($this->idleTimer);
+            $this->idleTimer = null;
         }
+    }
 
-        return $this->connecting;
+    private function idle()
+    {
+        --$this->pending;
+
+        if ($this->pending < 1 && $this->idlePeriod >= 0) {
+            $this->idleTimer = $this->loop->addTimer($this->idlePeriod, function () {
+                $this->connecting->then(function (ConnectionInterface $connection) {
+                    $this->disconnecting = $connection;
+                    $connection->quit()->then(
+                        function () {
+                            // successfully disconnected => remove reference
+                            $this->disconnecting = null;
+                        },
+                        function () use ($connection) {
+                            // soft-close failed => force-close connection
+                            $connection->close();
+                            $this->disconnecting = null;
+                        }
+                    );
+                });
+                $this->connecting = null;
+                $this->idleTimer = null;
+            });
+        }
     }
 
     public function query($sql, array $params = [])
@@ -59,7 +118,17 @@ public function query($sql, array $params = [])
         }
 
         return $this->connecting()->then(function (ConnectionInterface $connection) use ($sql, $params) {
-            return $connection->query($sql, $params);
+            $this->awake();
+            return $connection->query($sql, $params)->then(
+                function (QueryResult $result) {
+                    $this->idle();
+                    return $result;
+                },
+                function (\Exception $e) {
+                    $this->idle();
+                    throw $e;
+                }
+            );
         });
     }
 
@@ -71,7 +140,14 @@ public function queryStream($sql, $params = [])
 
         return \React\Promise\Stream\unwrapReadable(
             $this->connecting()->then(function (ConnectionInterface $connection) use ($sql, $params) {
-                return $connection->queryStream($sql, $params);
+                $stream = $connection->queryStream($sql, $params);
+
+                $this->awake();
+                $stream->on('close', function () {
+                    $this->idle();
+                });
+
+                return $stream;
             })
         );
     }
@@ -83,7 +159,16 @@ public function ping()
         }
 
         return $this->connecting()->then(function (ConnectionInterface $connection) {
-            return $connection->ping();
+            $this->awake();
+            return $connection->ping()->then(
+                function () {
+                    $this->idle();
+                },
+                function (\Exception $e) {
+                    $this->idle();
+                    throw $e;
+                }
+            );
         });
     }
 
@@ -100,7 +185,16 @@ public function quit()
         }
 
         return $this->connecting()->then(function (ConnectionInterface $connection) {
-            return $connection->quit();
+            $this->awake();
+            return $connection->quit()->then(
+                function () {
+                    $this->close();
+                },
+                function (\Exception $e) {
+                    $this->close();
+                    throw $e;
+                }
+            );
         });
     }
 
@@ -112,6 +206,12 @@ public function close()
 
         $this->closed = true;
 
+        // force-close connection if still waiting for previous disconnection
+        if ($this->disconnecting !== null) {
+            $this->disconnecting->close();
+            $this->disconnecting = null;
+        }
+
         // either close active connection or cancel pending connection attempt
         if ($this->connecting !== null) {
             $this->connecting->then(function (ConnectionInterface $connection) {
@@ -121,6 +221,11 @@ public function close()
             $this->connecting = null;
         }
 
+        if ($this->idleTimer !== null) {
+            $this->loop->cancelTimer($this->idleTimer);
+            $this->idleTimer = null;
+        }
+
         $this->emit('close');
         $this->removeAllListeners();
     }