Reject pending commands if connection is closed

Author: clue

src/Command.php

@@ -138,7 +138,7 @@ abstract class Command extends EventEmitter implements CommandInterface
      * @param integer $cmd
      * @param string  $q
      */
-    public function __construct(Connection $connection)
+    public function __construct(ConnectionInterface $connection)
     {
         $this->connection = $connection;
     }

src/Connection.php

@@ -282,6 +282,16 @@ public function handleConnectionClosed()
             $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'),
+                $command,
+                $this
+            ));
+        }
     }
 
     /**

src/Protocal/Parser.php

@@ -379,6 +379,12 @@ protected function onClose()
 
             if ($command->equals(Command::QUIT)) {
                 $command->emit('success');
+            } else {
+                $command->emit('error', array(
+                    new \RuntimeException('Connection lost'),
+                    $command,
+                    $command->getConnection()
+                ));
             }
         }
     }

tests/ConnectionTest.php

@@ -4,6 +4,7 @@
 
 use React\MySQL\Connection;
 use React\MySQL\Exception;
+use React\Socket\Server;
 
 class ConnectionTest extends BaseTestCase
 {
@@ -111,6 +112,59 @@ public function testCloseWhileConnectingWillBeQueuedAfterConnection()
         $loop->run();
     }
 
+    public function testPingAfterConnectWillEmitErrorWhenServerClosesConnection()
+    {
+        $this->expectOutputString('Connection lost');
+
+        $loop = \React\EventLoop\Factory::create();
+
+        $server = new Server(0, $loop);
+        $server->on('connection', function ($connection) use ($server) {
+            $server->close();
+            $connection->close();
+        });
+
+        $parts = parse_url($server->getAddress());
+        $options = $this->getConnectionOptions();
+        $options['host'] = $parts['host'];
+        $options['port'] = $parts['port'];
+
+        $conn = new Connection($loop, $options);
+
+        $conn->connect(function ($err) {
+            echo $err ? $err->getMessage() : 'OK';
+        });
+
+        $loop->run();
+    }
+
+    public function testConnectWillEmitErrorWhenServerClosesConnection()
+    {
+        $this->expectOutputString('Connection lost');
+
+        $loop = \React\EventLoop\Factory::create();
+
+        $server = new Server(0, $loop);
+        $server->on('connection', function ($connection) use ($server) {
+            $server->close();
+            $connection->close();
+        });
+
+        $parts = parse_url($server->getAddress());
+        $options = $this->getConnectionOptions();
+        $options['host'] = $parts['host'];
+        $options['port'] = $parts['port'];
+
+        $conn = new Connection($loop, $options);
+
+        $conn->connect(function () { });
+        $conn->ping(function ($err) {
+            echo $err ? $err->getMessage() : 'OK';
+        });
+
+        $loop->run();
+    }
+
     public function testPingAndCloseWhileConnectingWillBeQueuedAfterConnection()
     {
         $this->expectOutputString('connectedpingclosed');

tests/Protocal/ParserTest.php

@@ -0,0 +1,44 @@
+<?php
+
+namespace React\Tests\MySQL\Protocal;
+
+use React\MySQL\Commands\QueryCommand;
+use React\MySQL\Executor;
+use React\MySQL\Protocal\Parser;
+use React\Stream\ThroughStream;
+use React\Tests\MySQL\BaseTestCase;
+
+class ParserTest extends BaseTestCase
+{
+    public function testClosingStreamEmitsCloseEvent()
+    {
+        $stream = new ThroughStream();
+        $connection = $this->getMockBuilder('React\MySQL\ConnectionInterface')->disableOriginalConstructor()->getMock();
+        $executor = new Executor($connection);
+
+        $parser = new Parser($stream, $executor);
+        $parser->start();
+
+        $parser->on('close', $this->expectCallableOnce());
+
+        $stream->close();
+    }
+
+    public function testClosingStreamEmitsErrorForCurrentCommand()
+    {
+        $stream = new ThroughStream();
+        $connection = $this->getMockBuilder('React\MySQL\ConnectionInterface')->disableOriginalConstructor()->getMock();
+        $executor = new Executor($connection);
+
+        $parser = new Parser($stream, $executor);
+        $parser->start();
+
+        $command = new QueryCommand($connection);
+        $command->on('error', $this->expectCallableOnce());
+
+        // hack to inject command as current command
+        $parser->setOptions(array('currCommand' => $command));
+
+        $stream->close();
+    }
+}