Merge pull request #89 from clue-labs/cancellation

Improve promise cancellation and close underlying socket connection
ratchetphp · Jan 14, 2019 · 3a7d5b7 · 3a7d5b7
2 parents 0aaacb8 + 45e8ef0
commit 3a7d5b7
Show file tree

Hide file tree

Showing 2 changed files with 83 additions and 4 deletions.
diff --git a/src/Connector.php b/src/Connector.php
@@ -49,9 +49,21 @@ public function __invoke($url, array $subProtocols = [], array $headers = []) {
 
         $uriString = $scheme . '://' . $uri->getHost() . ':' . $port;
 
-        return $connector->connect($uriString)->then(function(ConnectionInterface $conn) use ($request, $subProtocols) {
-            $futureWsConn = new Deferred;
+        $connecting = $connector->connect($uriString);
 
+        $futureWsConn = new Deferred(function ($_, $reject) use ($url, $connecting) {
+            $reject(new \RuntimeException(
+                'Connection to ' . $url . ' cancelled during handshake'
+            ));
+
+            // either close active connection or cancel pending connection attempt
+            $connecting->then(function (ConnectionInterface $connection) {
+                $connection->close();
+            });
+            $connecting->cancel();
+        });
+
+        $connecting->then(function(ConnectionInterface $conn) use ($request, $subProtocols, $futureWsConn) {
             $earlyClose = function() use ($futureWsConn) {
                 $futureWsConn->reject(new \RuntimeException('Connection closed before handshake'));
             };
@@ -98,9 +110,9 @@ public function __invoke($url, array $subProtocols = [], array $headers = []) {
 
             $stream->on('data', $headerParser);
             $stream->write(gPsr\str($request));
+        }, array($futureWsConn, 'reject'));
 
-            return $futureWsConn->promise();
-        });
+        return $futureWsConn->promise();
     }
 
     /**

diff --git a/tests/unit/ConnectorTest.php b/tests/unit/ConnectorTest.php
@@ -4,6 +4,7 @@
 use Ratchet\Client\Connector;
 use React\EventLoop\Factory;
 use React\Promise\RejectedPromise;
+use React\Promise\Promise;
 
 class ConnectorTest extends TestCase
 {
@@ -36,4 +37,70 @@ public function testSecureConnectionUsesTlsScheme($uri, $expectedConnectorUri) {
 
         $pawlConnector($uri);
     }
+
+    public function testConnectorRejectsWhenUnderlyingSocketConnectorRejects()
+    {
+        $exception = new RuntimeException('Connection failed');
+
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+        $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
+        $connector->expects($this->once())->method('connect')->willReturn(\React\Promise\reject($exception));
+
+        $pawlConnector = new Connector($loop, $connector);
+
+        $promise = $pawlConnector('ws://localhost');
+
+        $actual = null;
+        $promise->then(null, function ($reason) use (&$actual) {
+            $actual = $reason;
+        });
+        $this->assertSame($exception, $actual);
+    }
+
+    public function testCancelConnectorShouldCancelUnderlyingSocketConnectorWhenSocketConnectionIsPending()
+    {
+        $promise = new Promise(function () { }, function () use (&$cancelled) {
+            ++$cancelled;
+        });
+
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+        $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
+        $connector->expects($this->once())->method('connect')->willReturn($promise);
+
+        $pawlConnector = new Connector($loop, $connector);
+
+        $promise = $pawlConnector('ws://localhost');
+
+        $this->assertNull($cancelled);
+        $promise->cancel();
+        $this->assertEquals(1, $cancelled);
+
+        $message = null;
+        $promise->then(null, function ($reason) use (&$message) {
+            $message = $reason->getMessage();
+        });
+        $this->assertEquals('Connection to ws://localhost cancelled during handshake', $message);
+    }
+
+    public function testCancelConnectorShouldCloseUnderlyingSocketConnectionWhenHandshakeIsPending()
+    {
+        $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
+        $connection->expects($this->once())->method('close');
+
+        $loop = $this->getMockBuilder('React\EventLoop\LoopInterface')->getMock();
+        $connector = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock();
+        $connector->expects($this->once())->method('connect')->willReturn(\React\Promise\resolve($connection));
+
+        $pawlConnector = new Connector($loop, $connector);
+
+        $promise = $pawlConnector('ws://localhost');
+
+        $promise->cancel();
+
+        $message = null;
+        $promise->then(null, function ($reason) use (&$message) {
+            $message = $reason->getMessage();
+        });
+        $this->assertEquals('Connection to ws://localhost cancelled during handshake', $message);
+    }
 }