Multi Server Executor

Author: WyriHaximus

src/Query/MultiServerExecutor.php

@@ -0,0 +1,102 @@
+<?php
+
+namespace React\Dns\Query;
+
+use React\EventLoop\LoopInterface;
+use React\EventLoop\TimerInterface;
+use React\Promise\CancellablePromiseInterface;
+use React\Promise\Promise;
+
+final class MultiServerExecutor implements ExecutorInterface
+{
+    private $loop;
+
+    /**
+     * @var ExecutorInterface[]
+     */
+    public $executors = array();
+
+    private $executorsCount = 0;
+
+    public function __construct($nameservers, LoopInterface $loop)
+    {
+        $this->loop = $loop;
+        foreach ($nameservers as $nameserver) {
+            $this->executors[] = new UdpTransportExecutor($nameserver, $loop);
+            $this->executorsCount++;
+        }
+    }
+
+    public function query(Query $query)
+    {
+        $that = $this;
+        $executorsIndexes = range(0, $this->executorsCount - 1);
+        shuffle($executorsIndexes);
+        $loop = $this->loop;
+        $promises = array();
+        $timer = null;
+        return new Promise(function ($resolve, $reject) use (&$promises, &$timer, &$executorsIndexes, $that, $loop, $query) {
+            $resolveWrap = function ($index) use (&$promises, &$timer, $resolve, $loop) {
+                return function ($result) use ($index, &$promises, &$timer, $resolve, $loop) {
+                    $resolve($result);
+
+                    unset($promises[$index]);
+
+                    if ($timer instanceof TimerInterface) {
+                        $loop->cancelTimer($timer);
+                    }
+
+                    foreach ($promises as $promise) {
+                        if ($promise instanceof CancellablePromiseInterface) {
+                            $promise->cancel();
+                        }
+                    }
+                };
+            };
+            $rejectWrap = function ($index) use (&$promises, &$timer, $reject, $loop) {
+                return function ($error) use ($index, &$promises, &$timer, $reject, $loop) {
+                    $reject($error);
+
+                    unset($promises[$index]);
+
+                    if ($timer instanceof TimerInterface) {
+                        $loop->cancelTimer($timer);
+                    }
+
+                    foreach ($promises as $promise) {
+                        if ($promise instanceof CancellablePromiseInterface) {
+                            $promise->cancel();
+                        }
+                    }
+                };
+            };
+            $timer = $loop->addPeriodicTimer(0.05, function () use (&$promises, &$timer, &$executorsIndexes, $that, $loop, $query, $resolveWrap, $rejectWrap) {
+                if (count($executorsIndexes) <= 0) {
+                    $loop->cancelTimer($timer);
+
+                    return;
+                }
+
+                $index = array_pop($executorsIndexes);
+                $promise = $that->executors[$index]->query($query);
+                $promise->then($resolveWrap($index), $rejectWrap);
+                $promises[$index] = $promise;
+            });
+
+            $index = array_pop($executorsIndexes);
+            $promise = $that->executors[$index]->query($query);
+            $promise->then($resolveWrap($index), $rejectWrap);
+            $promises[$index] = $promise;
+        }, function () use (&$promises, &$timer, $loop) {
+            if ($timer instanceof TimerInterface) {
+                $loop->cancelTimer($timer);
+            }
+
+            foreach ($promises as $promise) {
+                if ($promise instanceof CancellablePromiseInterface) {
+                    $promise->cancel();
+                }
+            }
+        });
+    }
+}