Add SelectiveTransportExecutor to retry with TCP if UDP is truncated and automatically select transport protocol when no explicit scheme is given in Factory

README.md

@@ -19,6 +19,7 @@ easily be used to create a DNS server.
 * [Advanced usage](#advanced-usage)
   * [UdpTransportExecutor](#udptransportexecutor)
   * [TcpTransportExecutor](#tcptransportexecutor)
+  * [SelectiveTransportExecutor](#selectivetransportexecutor)
   * [HostsFileExecutor](#hostsfileexecutor)
 * [Install](#install)
 * [Tests](#tests)
@@ -350,6 +351,54 @@ $executor = new CoopExecutor(
   packages. Higher-level components should take advantage of the Socket
   component instead of reimplementing this socket logic from scratch.
 
+### SelectiveTransportExecutor
+
+The `SelectiveTransportExecutor` class can be used to
+Send DNS queries over a UDP or TCP/IP stream transport.
+
+This class will automatically choose the correct transport protocol to send
+a DNS query to your DNS server. It will always try to send it over the more
+efficient UDP transport first. If this query yields a size related issue
+(truncated messages), it will retry over a streaming TCP/IP transport.
+
+For more advanced usages one can utilize this class directly.
+The following example looks up the `IPv6` address for `reactphp.org`.
+
+```php
+$executor = new SelectiveTransportExecutor($udpExecutor, $tcpExecutor);
+
+$executor->query(
+    new Query($name, Message::TYPE_AAAA, Message::CLASS_IN)
+)->then(function (Message $message) {
+    foreach ($message->answers as $answer) {
+        echo 'IPv6: ' . $answer->data . PHP_EOL;
+    }
+}, 'printf');
+```
+
+Note that this executor only implements the logic to select the correct
+transport for the given DNS query. Implementing the correct transport logic,
+implementing timeouts and any retry logic is left up to the given executors,
+see also [`UdpTransportExecutor`](#udptransportexecutor) and
+[`TcpTransportExecutor`](#tcptransportexecutor) for more details.
+
+Note that this executor is entirely async and as such allows you to execute
+any number of queries concurrently. You should probably limit the number of
+concurrent queries in your application or you're very likely going to face
+rate limitations and bans on the resolver end. For many common applications,
+you may want to avoid sending the same query multiple times when the first
+one is still pending, so you will likely want to use this in combination with
+a `CoopExecutor` like this:
+
+```php
+$executor = new CoopExecutor(
+    new SelectiveTransportExecutor(
+        $datagramExecutor,
+        $streamExecutor
+    )
+);
+```
+
 ### HostsFileExecutor
 
 Note that the above `UdpTransportExecutor` class always performs an actual DNS query.

composer.json

@@ -7,7 +7,7 @@
         "php": ">=5.3.0",
         "react/cache": "^1.0 || ^0.6 || ^0.5",
         "react/event-loop": "^1.0 || ^0.5",
-        "react/promise": "^2.1 || ^1.2.1",
+        "react/promise": "^2.7 || ^1.2.1",
         "react/promise-timer": "^1.2"
     },
     "require-dev": {

src/Query/SelectiveTransportExecutor.php

@@ -0,0 +1,85 @@
+<?php
+
+namespace React\Dns\Query;
+
+use React\Promise\Promise;
+
+/**
+ * Send DNS queries over a UDP or TCP/IP stream transport.
+ *
+ * This class will automatically choose the correct transport protocol to send
+ * a DNS query to your DNS server. It will always try to send it over the more
+ * efficient UDP transport first. If this query yields a size related issue
+ * (truncated messages), it will retry over a streaming TCP/IP transport.
+ *
+ * For more advanced usages one can utilize this class directly.
+ * The following example looks up the `IPv6` address for `reactphp.org`.
+ *
+ * ```php
+ * $executor = new SelectiveTransportExecutor($udpExecutor, $tcpExecutor);
+ *
+ * $executor->query(
+ *     new Query($name, Message::TYPE_AAAA, Message::CLASS_IN)
+ * )->then(function (Message $message) {
+ *     foreach ($message->answers as $answer) {
+ *         echo 'IPv6: ' . $answer->data . PHP_EOL;
+ *     }
+ * }, 'printf');
+ * ```
+ *
+ * Note that this executor only implements the logic to select the correct
+ * transport for the given DNS query. Implementing the correct transport logic,
+ * implementing timeouts and any retry logic is left up to the given executors,
+ * see also [`UdpTransportExecutor`](#udptransportexecutor) and
+ * [`TcpTransportExecutor`](#tcptransportexecutor) for more details.
+ *
+ * Note that this executor is entirely async and as such allows you to execute
+ * any number of queries concurrently. You should probably limit the number of
+ * concurrent queries in your application or you're very likely going to face
+ * rate limitations and bans on the resolver end. For many common applications,
+ * you may want to avoid sending the same query multiple times when the first
+ * one is still pending, so you will likely want to use this in combination with
+ * a `CoopExecutor` like this:
+ *
+ * ```php
+ * $executor = new CoopExecutor(
+ *     new SelectiveTransportExecutor(
+ *         $datagramExecutor,
+ *         $streamExecutor
+ *     )
+ * );
+ * ```
+ */
+class SelectiveTransportExecutor implements ExecutorInterface
+{
+    private $datagramExecutor;
+    private $streamExecutor;
+
+    public function __construct(ExecutorInterface $datagramExecutor, ExecutorInterface $streamExecutor)
+    {
+        $this->datagramExecutor = $datagramExecutor;
+        $this->streamExecutor = $streamExecutor;
+    }
+
+    public function query(Query $query)
+    {
+        $stream = $this->streamExecutor;
+        $pending = $this->datagramExecutor->query($query);
+
+        return new Promise(function ($resolve, $reject) use (&$pending, $stream, $query) {
+            $pending->then(
+                $resolve,
+                function ($e) use (&$pending, $stream, $query, $resolve, $reject) {
+                    if ($e->getCode() === (\defined('SOCKET_EMSGSIZE') ? \SOCKET_EMSGSIZE : 90)) {
+                        $pending = $stream->query($query)->then($resolve, $reject);
+                    } else {
+                        $reject($e);
+                    }
+                }
+            );
+        }, function () use (&$pending) {
+            $pending->cancel();
+            $pending = null;
+        });
+    }
+}

src/Query/UdpTransportExecutor.php

@@ -121,7 +121,8 @@ public function query(Query $query)
         $queryData = $this->dumper->toBinary($request);
         if (isset($queryData[512])) {
             return \React\Promise\reject(new \RuntimeException(
-                'DNS query for ' . $query->name . ' failed: Query too large for UDP transport'
+                'DNS query for ' . $query->name . ' failed: Query too large for UDP transport',
+                \defined('SOCKET_EMSGSIZE') ? \SOCKET_EMSGSIZE : 90
             ));
         }
 
@@ -172,7 +173,10 @@ public function query(Query $query)
             \fclose($socket);
 
             if ($response->tc) {
-                $deferred->reject(new \RuntimeException('DNS query for ' . $query->name . ' failed: The server returned a truncated result for a UDP query, but retrying via TCP is currently not supported'));
+                $deferred->reject(new \RuntimeException(
+                    'DNS query for ' . $query->name . ' failed: The server returned a truncated result for a UDP query',
+                    \defined('SOCKET_EMSGSIZE') ? \SOCKET_EMSGSIZE : 90
+                ));
                 return;
             }
 

src/Resolver/Factory.php

@@ -10,6 +10,7 @@
 use React\Dns\Query\ExecutorInterface;
 use React\Dns\Query\HostsFileExecutor;
 use React\Dns\Query\RetryExecutor;
+use React\Dns\Query\SelectiveTransportExecutor;
 use React\Dns\Query\TcpTransportExecutor;
 use React\Dns\Query\TimeoutExecutor;
 use React\Dns\Query\UdpTransportExecutor;
@@ -84,24 +85,39 @@ private function createExecutor($nameserver, LoopInterface $loop)
         $parts = \parse_url($nameserver);
 
         if (isset($parts['scheme']) && $parts['scheme'] === 'tcp') {
-            $executor = new TimeoutExecutor(
-                new TcpTransportExecutor($nameserver, $loop),
-                5.0,
-                $loop
-            );
+            $executor = $this->createTcpExecutor($nameserver, $loop);
+        } elseif (isset($parts['scheme']) && $parts['scheme'] === 'udp') {
+            $executor = $this->createUdpExecutor($nameserver, $loop);
         } else {
-            $executor = new RetryExecutor(
-                new TimeoutExecutor(
-                    new UdpTransportExecutor(
-                        $nameserver,
-                        $loop
-                    ),
-                    5.0,
-                    $loop
-                )
+            $executor = new SelectiveTransportExecutor(
+                $this->createUdpExecutor($nameserver, $loop),
+                $this->createTcpExecutor($nameserver, $loop)
             );
         }
 
         return new CoopExecutor($executor);
     }
+
+    private function createTcpExecutor($nameserver, LoopInterface $loop)
+    {
+        return new TimeoutExecutor(
+            new TcpTransportExecutor($nameserver, $loop),
+            5.0,
+            $loop
+        );
+    }
+
+    private function createUdpExecutor($nameserver, LoopInterface $loop)
+    {
+        return new RetryExecutor(
+            new TimeoutExecutor(
+                new UdpTransportExecutor(
+                    $nameserver,
+                    $loop
+                ),
+                5.0,
+                $loop
+            )
+        );
+    }
 }