[Dns] Handle timeouts in Executor

Author: arnaud-lb

Query/CachedExecutor.php

@@ -16,7 +16,7 @@ public function __construct(ExecutorInterface $executor, RecordCache $cache)
         $this->cache = $cache;
     }
 
-    public function query($nameserver, Query $query, $callback)
+    public function query($nameserver, Query $query, $callback, $errorback)
     {
         $cachedRecords = $this->cache->lookup($query);
         if (count($cachedRecords)) {
@@ -29,7 +29,7 @@ public function query($nameserver, Query $query, $callback)
         $this->executor->query($nameserver, $query, function ($response) use ($cache, $query, $callback) {
             $callback($response);
             $cache->storeResponseMessage($query->currentTime, $response);
-        });
+        }, $errorback);
     }
 
     private function buildResponse(Query $query, array $cachedRecords)

Query/Executor.php

@@ -15,22 +15,24 @@ class Executor implements ExecutorInterface
     private $loop;
     private $parser;
     private $dumper;
+    private $timeout;
 
-    public function __construct(LoopInterface $loop, Parser $parser, BinaryDumper $dumper)
+    public function __construct(LoopInterface $loop, Parser $parser, BinaryDumper $dumper, $timeout = 5)
     {
         $this->loop = $loop;
         $this->parser = $parser;
         $this->dumper = $dumper;
+        $this->timeout = $timeout;
     }
 
-    public function query($nameserver, Query $query, $callback)
+    public function query($nameserver, Query $query, $callback, $errorback)
     {
         $request = $this->prepareRequest($query);
 
         $queryData = $this->dumper->toBinary($request);
         $transport = strlen($queryData) > 512 ? 'tcp' : 'udp';
 
-        $this->doQuery($nameserver, $transport, $queryData, $callback);
+        $this->doQuery($nameserver, $transport, $queryData, $callback, $errorback);
     }
 
     public function prepareRequest(Query $query)
@@ -44,25 +46,33 @@ public function prepareRequest(Query $query)
         return $request;
     }
 
-    public function doQuery($nameserver, $transport, $queryData, $callback)
+    public function doQuery($nameserver, $transport, $queryData, $callback, $errorback)
     {
         $that = $this;
         $parser = $this->parser;
+        $loop = $this->loop;
 
         $response = new Message();
 
-        $retryWithTcp = function () use ($that, $nameserver, $queryData, $callback) {
-            $that->doQuery($nameserver, 'tcp', $queryData, $callback);
+        $retryWithTcp = function () use ($that, $nameserver, $queryData, $callback, $errorback) {
+            $that->doQuery($nameserver, 'tcp', $queryData, $callback, $errorback);
         };
 
+        $timer = $this->loop->addTimer($this->timeout, function () use (&$conn, $errorback) {
+            $conn->close();
+            $errorback(new TimeoutException("query timed out"));
+        });
+
         $conn = $this->createConnection($nameserver, $transport);
-        $conn->on('data', function ($data) use ($that, $retryWithTcp, $conn, $parser, $response, $transport, $callback) {
+        $conn->on('data', function ($data) use ($that, $retryWithTcp, $conn, $parser, $response, $transport, $callback, $loop, $timer) {
             $responseReady = $parser->parseChunk($data, $response);
 
             if (!$responseReady) {
                 return;
             }
 
+            $loop->cancelTimer($timer);
+
             if ($response->header->isTruncated()) {
                 if ('tcp' === $transport) {
                     throw new BadServerException('The server set the truncated bit although we issued a TCP request');

Query/ExecutorInterface.php

@@ -4,5 +4,5 @@
 
 interface ExecutorInterface
 {
-    public function query($nameserver, Query $query, $callback);
+    public function query($nameserver, Query $query, $callback, $errorback);
 }

Query/TimeoutException.php

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

Resolver/Resolver.php

@@ -27,19 +27,21 @@ public function resolve($domain, $callback, $errback = null)
     {
         $that = $this;
 
+        if (!$errback) {
+            $errback = function ($error) {
+                throw $error;
+            };
+        }
+
         $query = new Query($domain, Message::TYPE_A, Message::CLASS_IN, time());
 
         $this->executor->query($this->nameserver, $query, function (Message $response) use ($that, $callback, $errback) {
             try {
                 $that->extractAddress($response, Message::TYPE_A, $callback);
             } catch (RecordNotFoundException $e) {
-                if (!$errback) {
-                    throw $e;
-                }
-
                 $errback($e);
             }
-        });
+        }, $errback);
     }
 
     public function extractAddress(Message $response, $type, $callback)