Add close() method to force-close the connection

README.md

@@ -19,6 +19,8 @@ It is written in pure PHP and does not require any extensions.
     * [query()](#query)
     * [queryStream()](#querystream)
     * [ping()](#ping)
+    * [quit()](#quit)
+    * [close()](#close)
 * [Install](#install)
 * [Tests](#tests)
 * [License](#license)
@@ -275,6 +277,22 @@ $connection->query('CREATE TABLE test ...');
 $connection->quit();
 ```
 
+#### close()
+
+The `close(): void` method can be used to
+force-close the connection.
+
+Unlike the `quit()` method, this method will immediately force-close the
+connection and reject all oustanding commands.
+
+```php
+$connection->close();
+```
+
+Forcefully closing the connection will yield a warning in the server logs
+and should generally only be used as a last resort. See also
+[`quit()`](#quit) as a safe alternative.
+
 ## Install
 
 The recommended way to install this library is [through Composer](https://getcomposer.org).

src/ConnectionInterface.php

@@ -163,4 +163,22 @@ public function ping();
      * @return PromiseInterface Returns a Promise<void,Exception>
      */
     public function quit();
+
+    /**
+     * Force-close the connection.
+     *
+     * Unlike the `quit()` method, this method will immediately force-close the
+     * connection and reject all oustanding commands.
+     *
+     * ```php
+     * $connection->close();
+     * ```
+     *
+     * Forcefully closing the connection will yield a warning in the server logs
+     * and should generally only be used as a last resort. See also
+     * [`quit()`](#quit) as a safe alternative.
+     *
+     * @return void
+     */
+    public function close();
 }

src/Io/Connection.php

@@ -167,6 +167,27 @@ public function quit()
         });
     }
 
+    public function close()
+    {
+        if ($this->state === self::STATE_CLOSED) {
+            return;
+        }
+
+        $this->state = self::STATE_CLOSED;
+        $this->stream->close();
+
+        // reject all pending commands if connection is closed
+        while (!$this->executor->isIdle()) {
+            $command = $this->executor->dequeue();
+            $command->emit('error', array(
+                new \RuntimeException('Connection lost')
+            ));
+        }
+
+        $this->emit('close');
+        $this->removeAllListeners();
+    }
+
     /**
      * @param Exception $err Error from socket.
      *
@@ -185,17 +206,10 @@ public function handleConnectionError($err)
     public function handleConnectionClosed()
     {
         if ($this->state < self::STATE_CLOSEING) {
-            $this->state = self::STATE_CLOSED;
             $this->emit('error', [new \RuntimeException('mysql server has gone away'), $this]);
         }
 
-        // reject all pending commands if connection is closed
-        while (!$this->executor->isIdle()) {
-            $command = $this->executor->dequeue();
-            $command->emit('error', array(
-                new \RuntimeException('Connection lost')
-            ));
-        }
+        $this->close();
     }
 
     /**

tests/FactoryTest.php

@@ -177,4 +177,57 @@ public function testConnectWithValidAuthQuitOnlyOnce()
 
         $loop->run();
     }
+
+    public function testConnectWithValidAuthCanCloseOnlyOnce()
+    {
+        $this->expectOutputString('connected.closed.');
+
+        $loop = \React\EventLoop\Factory::create();
+        $factory = new Factory($loop);
+
+        $uri = $this->getConnectionString();
+        $factory->createConnection($uri)->then(function (ConnectionInterface $connection) {
+            echo 'connected.';
+            $connection->on('close', function () {
+                echo 'closed.';
+            });
+            $connection->on('error', function () {
+                echo 'error?';
+            });
+
+            $connection->close();
+            $connection->close();
+        }, 'printf')->then(null, 'printf');
+
+        $loop->run();
+    }
+
+    public function testConnectWithValidAuthCanCloseAndAbortPing()
+    {
+        $this->expectOutputString('connected.aborted pending (Connection lost).aborted queued (Connection lost).closed.');
+
+        $loop = \React\EventLoop\Factory::create();
+        $factory = new Factory($loop);
+
+        $uri = $this->getConnectionString();
+        $factory->createConnection($uri)->then(function (ConnectionInterface $connection) {
+            echo 'connected.';
+            $connection->on('close', function () {
+                echo 'closed.';
+            });
+            $connection->on('error', function () {
+                echo 'error?';
+            });
+
+            $connection->ping()->then(null, function ($e) {
+                echo 'aborted pending (' . $e->getMessage() .').';
+            });
+            $connection->ping()->then(null, function ($e) {
+                echo 'aborted queued (' . $e->getMessage() . ').';
+            });
+            $connection->close();
+        }, 'printf')->then(null, 'printf');
+
+        $loop->run();
+    }
 }