Support Promise cancellation for SecureConnector

Author: clue

README.md

@@ -135,6 +135,17 @@ $secureConnector->create('www.google.com', 443)->then(function (React\Stream\Str
 $loop->run();
 ```
 
+Pending connection attempts can be cancelled by cancelling its pending promise like so:
+
+```php
+$promise = $secureConnector->create($host, $port);
+
+$promise->cancel();
+```
+
+Calling `cancel()` on a pending promise will cancel the underlying TCP/IP
+connection and/or the SSL/TLS negonation and reject the resulting promise.
+
 You can optionally pass additional
 [SSL context options](http://php.net/manual/en/context.ssl.php)
 to the constructor like this:

src/SecureConnector.php

@@ -5,6 +5,7 @@
 use React\EventLoop\LoopInterface;
 use React\Stream\Stream;
 use React\Promise;
+use React\Promise\CancellablePromiseInterface;
 
 class SecureConnector implements ConnectorInterface
 {
@@ -39,7 +40,7 @@ public function create($host, $port)
         }
 
         $encryption = $this->streamEncryption;
-        return $this->connector->create($host, $port)->then(function (Stream $stream) use ($context, $encryption) {
+        return $this->connect($host, $port)->then(function (Stream $stream) use ($context, $encryption) {
             // (unencrypted) TCP/IP connection succeeded
 
             // set required SSL/TLS context options
@@ -55,4 +56,30 @@ public function create($host, $port)
             });
         });
     }
+
+    private function connect($host, $port)
+    {
+        $promise = $this->connector->create($host, $port);
+
+        return new Promise\Promise(
+            function ($resolve, $reject) use ($promise) {
+                // resolve/reject with result of TCP/IP connection
+                $promise->then($resolve, $reject);
+            },
+            function ($_, $reject) use ($promise) {
+                // cancellation should reject connection attempt
+                $reject(new \RuntimeException('Connection attempt cancelled during TCP/IP connection'));
+
+                // forefully close TCP/IP connection if it completes despite cancellation
+                $promise->then(function (Stream $stream) {
+                    $stream->close();
+                });
+
+                // (try to) cancel pending TCP/IP connection
+                if ($promise instanceof CancellablePromiseInterface) {
+                    $promise->cancel();
+                }
+            }
+        );
+    }
 }

src/StreamEncryption.php

@@ -66,7 +66,10 @@ public function toggle(Stream $stream, $toggle)
 
         // TODO: add write() event to make sure we're not sending any excessive data
 
-        $deferred = new Deferred();
+        $deferred = new Deferred(function ($_, $reject) use ($toggle) {
+            // cancelling this leaves this stream in an inconsistent state…
+            $reject(new \RuntimeException('Cancelled toggling encryption ' . $toggle ? 'on' : 'off'));
+        });
 
         // get actual stream socket from stream instance
         $socket = $stream->stream;
@@ -82,15 +85,18 @@ public function toggle(Stream $stream, $toggle)
         $wrap = $this->wrapSecure && $toggle;
         $loop = $this->loop;
 
-        return $deferred->promise()->then(function () use ($stream, $wrap, $loop) {
+        return $deferred->promise()->then(function () use ($stream, $socket, $wrap, $loop) {
+            $this->loop->removeReadStream($socket);
+
             if ($wrap) {
                 return new SecureStream($stream, $loop);
             }
 
             $stream->resume();
 
             return $stream;
-        }, function($error) use ($stream) {
+        }, function($error) use ($stream, $socket) {
+            $this->loop->removeReadStream($socket);
             $stream->resume();
             throw $error;
         });
@@ -103,12 +109,8 @@ public function toggleCrypto($socket, Deferred $deferred, $toggle)
         restore_error_handler();
 
         if (true === $result) {
-            $this->loop->removeReadStream($socket);
-
             $deferred->resolve();
         } else if (false === $result) {
-            $this->loop->removeReadStream($socket);
-
             $deferred->reject(new UnexpectedValueException(
                 sprintf("Unable to complete SSL/TLS handshake: %s", $this->errstr),
                 $this->errno

tests/SecureConnectorTest.php

@@ -0,0 +1,58 @@
+<?php
+
+namespace React\Tests\SocketClient;
+
+use React\Promise;
+use React\SocketClient\SecureConnector;
+
+class SecureConnectorTest extends TestCase
+{
+    private $loop;
+    private $tcp;
+    private $connector;
+
+    public function setUp()
+    {
+        $this->loop = $this->getMock('React\EventLoop\LoopInterface');
+        $this->tcp = $this->getMock('React\SocketClient\ConnectorInterface');
+        $this->connector = new SecureConnector($this->tcp, $this->loop);
+    }
+
+    public function testConnectionWillWaitForTcpConnection()
+    {
+        $pending = new Promise\Promise(function () { });
+        $this->tcp->expects($this->once())->method('create')->with($this->equalTo('example.com'), $this->equalTo(80))->will($this->returnValue($pending));
+
+        $promise = $this->connector->create('example.com', 80);
+
+        $this->assertInstanceOf('React\Promise\PromiseInterface', $promise);
+    }
+
+    public function testCancelDuringTcpConnectionCancelsTcpConnection()
+    {
+        $pending = new Promise\Promise(function () { }, $this->expectCallableOnce());
+        $this->tcp->expects($this->once())->method('create')->with($this->equalTo('example.com'), $this->equalTo(80))->will($this->returnValue($pending));
+
+        $promise = $this->connector->create('example.com', 80);
+        $promise->cancel();
+
+        $promise->then($this->expectCallableNever(), $this->expectCallableOnce());
+    }
+
+    public function testCancelClosesStreamIfTcpResolvesDespiteCancellation()
+    {
+        $stream = $this->getMockBuilder('React\Stream\Stream')->disableOriginalConstructor()->setMethods(array('close'))->getMock();
+        $stream->expects($this->once())->method('close');
+
+        $pending = new Promise\Promise(function () { }, function ($resolve) use ($stream) {
+            $resolve($stream);
+        });
+
+        $this->tcp->expects($this->once())->method('create')->with($this->equalTo('example.com'), $this->equalTo(80))->will($this->returnValue($pending));
+
+        $promise = $this->connector->create('example.com', 80);
+        $promise->cancel();
+
+        $promise->then($this->expectCallableNever(), $this->expectCallableOnce());
+    }
+}