Support Promise cancellation

composer.json

@@ -7,7 +7,7 @@
         "php": ">=5.3.0",
         "react/cache": "~0.4.0|~0.3.0",
         "react/socket": "~0.4.0|~0.3.0",
-        "react/promise": "~2.0|~1.1"
+        "react/promise": "~2.1|~1.2"
     },
     "autoload": {
         "psr-4": { "React\\Dns\\": "src" }

src/Query/CancellationException.php

@@ -0,0 +1,7 @@
+<?php
+
+namespace React\Dns\Query;
+
+class CancellationException extends \RuntimeException
+{
+}

src/Query/Executor.php

@@ -53,7 +53,12 @@ public function doQuery($nameserver, $transport, $queryData, $name)
         $loop = $this->loop;
 
         $response = new Message();
-        $deferred = new Deferred();
+        $deferred = new Deferred(function ($resolve, $reject) use (&$timer, &$conn, $name) {
+            $reject(new CancellationException(sprintf('DNS query for %s has been cancelled', $name)));
+
+            $timer->cancel();
+            $conn->close();
+        });
 
         $retryWithTcp = function () use ($that, $nameserver, $queryData, $name) {
             return $that->doQuery($nameserver, 'tcp', $queryData, $name);

src/Query/RetryExecutor.php

@@ -17,35 +17,28 @@ public function __construct(ExecutorInterface $executor, $retries = 2)
 
     public function query($nameserver, Query $query)
     {
-        $deferred = new Deferred();
-
-        $this->tryQuery($nameserver, $query, $this->retries, $deferred);
-
-        return $deferred->promise();
+        return $this->tryQuery($nameserver, $query, $this->retries);
     }
 
-    public function tryQuery($nameserver, Query $query, $retries, $deferred)
+    public function tryQuery($nameserver, Query $query, $retries)
     {
         $that = $this;
-        $errorback = function ($error) use ($nameserver, $query, $retries, $deferred, $that) {
+        $errorback = function ($error) use ($nameserver, $query, $retries, $that) {
             if (!$error instanceof TimeoutException) {
-                $deferred->reject($error);
-                return;
+                throw $error;
             }
             if (0 >= $retries) {
-                $error = new \RuntimeException(
+                throw new \RuntimeException(
                     sprintf("DNS query for %s failed: too many retries", $query->name),
                     0,
                     $error
                 );
-                $deferred->reject($error);
-                return;
             }
-            $that->tryQuery($nameserver, $query, $retries-1, $deferred);
+            return $that->tryQuery($nameserver, $query, $retries-1);
         };
 
-        $this->executor
+        return $this->executor
             ->query($nameserver, $query)
-            ->then(array($deferred, 'resolve'), $errorback);
+            ->then(null, $errorback);
     }
 }

tests/FunctionalResolverTest.php

@@ -32,4 +32,17 @@ public function testResolveInvalidRejects()
 
         $this->loop->run();
     }
+
+    public function testResolveCancelledRejectsImmediately()
+    {
+        $promise = $this->resolver->resolve('google.com');
+        $promise->then($this->expectCallableNever(), $this->expectCallableOnce());
+        $promise->cancel();
+
+        $time = microtime(true);
+        $this->loop->run();
+        $time = microtime(true) - $time;
+
+        $this->assertLessThan(0.1, $time);
+    }
 }

tests/Query/ExecutorTest.php

@@ -61,6 +61,43 @@ public function resolveShouldCreateTcpRequestIfRequestIsLargerThan512Bytes()
         $this->executor->query('8.8.8.8:53', $query, function () {}, function () {});
     }
 
+    /** @test */
+    public function resolveShouldCloseConnectionWhenCancelled()
+    {
+        $conn = $this->createConnectionMock();
+        $conn->expects($this->once())->method('close');
+
+        $timer = $this->getMock('React\EventLoop\Timer\TimerInterface');
+        $this->loop
+            ->expects($this->any())
+            ->method('addTimer')
+            ->will($this->returnValue($timer));
+
+        $this->executor = $this->createExecutorMock();
+        $this->executor
+            ->expects($this->once())
+            ->method('createConnection')
+            ->with('8.8.8.8:53', 'udp')
+            ->will($this->returnValue($conn));
+
+        $query = new Query('igor.io', Message::TYPE_A, Message::CLASS_IN, 1345656451);
+        $promise = $this->executor->query('8.8.8.8:53', $query);
+
+        $promise->cancel();
+
+        $errorback = $this->createCallableMock();
+        $errorback
+            ->expects($this->once())
+            ->method('__invoke')
+            ->with($this->logicalAnd(
+                $this->isInstanceOf('React\Dns\Query\CancellationException'),
+                $this->attribute($this->equalTo('DNS query for igor.io has been cancelled'), 'message')
+            )
+        );
+
+        $promise->then($this->expectCallableNever(), $errorback);
+    }
+
     /** @test */
     public function resolveShouldRetryWithTcpIfResponseIsTruncated()
     {

tests/Query/RetryExecutorTest.php

@@ -9,6 +9,8 @@
 use React\Dns\Query\TimeoutException;
 use React\Dns\Model\Record;
 use React\Promise;
+use React\Promise\Deferred;
+use React\Dns\Query\CancellationException;
 
 class RetryExecutorTest extends TestCase
 {
@@ -125,6 +127,41 @@ public function queryShouldForwardNonTimeoutErrors()
         $retryExecutor->query('8.8.8.8', $query)->then($callback, $errorback);
     }
 
+    /**
+     * @covers React\Dns\Query\RetryExecutor
+     * @test
+     */
+    public function queryShouldCancelQueryOnCancel()
+    {
+        $cancelled = 0;
+
+        $executor = $this->createExecutorMock();
+        $executor
+            ->expects($this->once())
+            ->method('query')
+            ->with('8.8.8.8', $this->isInstanceOf('React\Dns\Query\Query'))
+            ->will($this->returnCallback(function ($domain, $query) use (&$cancelled) {
+                $deferred = new Deferred(function ($resolve, $reject) use (&$cancelled) {
+                    ++$cancelled;
+                    $reject(new CancellationException('Cancelled'));
+                });
+
+                return $deferred->promise();
+            })
+        );
+
+        $retryExecutor = new RetryExecutor($executor, 2);
+
+        $query = new Query('igor.io', Message::TYPE_A, Message::CLASS_IN, 1345656451);
+        $promise = $retryExecutor->query('8.8.8.8', $query);
+
+        $promise->then($this->expectCallableNever(), $this->expectCallableOnce());
+
+        $this->assertEquals(0, $cancelled);
+        $promise->cancel();
+        $this->assertEquals(1, $cancelled);
+    }
+
     protected function expectCallableOnce()
     {
         $mock = $this->createCallableMock();