Skip to content

Commit 4662be9

Browse files
WyriHaximusWyriHaximus
committed
Implement the Happy Eye Balls RFC's
By using the happy eye balls algorythm as descripted in RFC6555 and RFC8305 it will connect to the quickest responding server with a preference for IPv6.
1 parent 2cf8dfa commit 4662be9

File tree

2 files changed

+711
-0
lines changed

2 files changed

+711
-0
lines changed

‎src/HappyEyeBallsConnector.php

Lines changed: 284 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,284 @@
1+
<?php
2+
3+
namespace React\Socket;
4+
5+
use React\Dns\Model\Message;
6+
use React\Dns\Resolver\Resolver;
7+
use React\EventLoop\LoopInterface;
8+
use React\EventLoop\TimerInterface;
9+
use React\Promise;
10+
use React\Promise\CancellablePromiseInterface;
11+
use InvalidArgumentException;
12+
use RuntimeException;
13+
14+
final class HappyEyeBallsConnector implements ConnectorInterface
15+
{
16+
private $loop;
17+
private $connector;
18+
private $resolver;
19+
20+
public function __construct(LoopInterface $loop, ConnectorInterface $connector, Resolver $resolver)
21+
{
22+
$this->loop = $loop;
23+
$this->connector = $connector;
24+
$this->resolver = $resolver;
25+
}
26+
27+
public function connect($uri)
28+
{
29+
if (\strpos($uri, '://') === false) {
30+
$parts = \parse_url('tcp://' . $uri);
31+
unset($parts['scheme']);
32+
} else {
33+
$parts = \parse_url($uri);
34+
}
35+
36+
if (!$parts || !isset($parts['host'])) {
37+
return Promise\reject(new \InvalidArgumentException('Given URI "' . $uri . '" is invalid'));
38+
}
39+
40+
$host = \trim($parts['host'], '[]');
41+
$connector = $this->connector;
42+
43+
// skip DNS lookup / URI manipulation if this URI already contains an IP
44+
if (false !== \filter_var($host, \FILTER_VALIDATE_IP)) {
45+
return $connector->connect($uri);
46+
}
47+
48+
$that = $this;
49+
$loop = $this->loop;
50+
$resolver = $this->resolver;
51+
$resolved = array();
52+
$resolverPromises = array();
53+
$connectionPromises = array();
54+
$timer = null;
55+
56+
return new Promise\Promise(function ($resolve, $reject) use (&$timer, &$resolverPromises, &$connectionPromises, &$resolved, $loop, $resolver, $uri, $parts, $host, $that) {
57+
$connectQueue = array();
58+
$ipsCount = 0;
59+
$failureCount = 0;
60+
$check = function () use (&$timer, &$resolved, &$connectQueue, &$failureCount, &$ipsCount, &$connectionPromises, &$resolverPromises, $loop, $uri, $parts, $host, $that, $resolve, $reject, &$check) {
61+
$hasBeenResolved = true;
62+
foreach ($resolved as $typeHasBeenResolved) {
63+
if ($typeHasBeenResolved === false) {
64+
$hasBeenResolved = false;
65+
66+
break;
67+
}
68+
}
69+
70+
if ($hasBeenResolved === true && \count($connectQueue) === 0 && $timer instanceof TimerInterface) {
71+
$loop->cancelTimer($timer);
72+
$timer = null;
73+
return;
74+
}
75+
76+
if (\count($connectQueue) === 0 && $timer instanceof TimerInterface) {
77+
$loop->cancelTimer($timer);
78+
$timer = null;
79+
return;
80+
}
81+
82+
$ip = \array_shift($connectQueue);
83+
84+
if (\count($connectQueue) === 0 && $timer instanceof TimerInterface) {
85+
$loop->cancelTimer($timer);
86+
$timer = null;
87+
}
88+
89+
$connectionPromises[$ip] = $that->attemptConnection($ip, $uri, $parts, $host)->then(function ($connection) use ($resolve, &$connectionPromises, &$resolverPromises, $ip, &$timer, $loop) {
90+
unset($connectionPromises[$ip]);
91+
92+
/** @var CancellablePromiseInterface $promise */
93+
foreach ($connectionPromises as $connectionPromise) {
94+
if (!($connectionPromise instanceof CancellablePromiseInterface)) {
95+
continue;
96+
}
97+
98+
$connectionPromise->cancel();
99+
$connectionPromise = null;
100+
}
101+
102+
/** @var CancellablePromiseInterface $promise */
103+
foreach ($resolverPromises as $resolverPromise) {
104+
if (!($resolverPromise instanceof CancellablePromiseInterface)) {
105+
continue;
106+
}
107+
108+
$resolverPromise->cancel();
109+
$resolverPromise = null;
110+
}
111+
112+
if ($timer instanceof TimerInterface) {
113+
$loop->cancelTimer($timer);
114+
$timer = null;
115+
}
116+
117+
$resolve($connection);
118+
}, function () use (&$failureCount, &$ipsCount, &$resolved, $reject, $loop, &$timer, $host, &$connectionPromises, $ip) {
119+
unset($connectionPromises[$ip]);
120+
121+
$failureCount++;
122+
123+
foreach ($resolved as $typeHasBeenResolved) {
124+
if ($typeHasBeenResolved === false) {
125+
return;
126+
}
127+
}
128+
129+
if ($ipsCount === $failureCount) {
130+
if ($timer instanceof TimerInterface) {
131+
$loop->cancelTimer($timer);
132+
$timer = null;
133+
}
134+
$reject(new \RuntimeException('All attempts to connect to "' . $host . '" have failed'));
135+
}
136+
});
137+
138+
if (\count($connectQueue) > 0 && $timer === null) {
139+
$timer = $loop->addPeriodicTimer(0.05, $check);
140+
}
141+
};
142+
143+
foreach (array(Message::TYPE_AAAA, Message::TYPE_A) as $type) {
144+
$resolved[$type] = false;
145+
$resolverPromises[$type] = $resolver->resolveAll($host, $type)->then(function (array $ips) use (&$connectQueue, &$resolved, $type, $check, &$ipsCount, &$resolverPromises, &$timer) {
146+
unset($resolverPromises[$type]);
147+
$resolved[$type] = true;
148+
$ipsCount += \count($ips);
149+
$connectQueueStash = $connectQueue;
150+
$connectQueue = array();
151+
while (\count($connectQueueStash) > 0 || \count($ips) > 0) {
152+
if (\count($ips) > 0) {
153+
$connectQueue[] = \array_shift($ips);
154+
}
155+
if (\count($connectQueueStash) > 0) {
156+
$connectQueue[] = \array_shift($connectQueueStash);
157+
}
158+
}
159+
$check();
160+
}, function () use (&$resolved, $type, &$resolverPromises, &$ipsCount, $reject, $uri) {
161+
unset($resolverPromises[$type]);
162+
$resolved[$type] = true;
163+
164+
foreach ($resolved as $typeHasBeenResolved) {
165+
if ($typeHasBeenResolved === false) {
166+
return;
167+
}
168+
}
169+
170+
if ($ipsCount === 0) {
171+
unset($resolverPromises, $resolved);
172+
$reject(new \RuntimeException('Connection to ' . $uri . ' failed during DNS lookup: DNS error'));
173+
$reject = null;
174+
}
175+
});
176+
}
177+
}, function ($_, $reject) use (&$resolverPromises, &$connectionPromises, &$timer, &$resolved, $loop, $uri) {
178+
if ($timer instanceof TimerInterface) {
179+
$loop->cancelTimer($timer);
180+
$timer = null;
181+
}
182+
foreach ($resolverPromises as $resolverPromise) {
183+
if ($resolverPromise instanceof CancellablePromiseInterface) {
184+
$resolverPromise->cancel();
185+
$resolverPromise = null;
186+
}
187+
}
188+
$resolverPromises = null;
189+
foreach ($connectionPromises as $connectionPromise) {
190+
if ($connectionPromise instanceof CancellablePromiseInterface) {
191+
$connectionPromise->cancel();
192+
$connectionPromise = null;
193+
}
194+
}
195+
$connectionPromises = null;
196+
197+
foreach ($resolved as $typeHasBeenResolved) {
198+
if ($typeHasBeenResolved === true) {
199+
return;
200+
}
201+
}
202+
203+
$reject(new \RuntimeException('Connection to ' . $uri . ' cancelled during DNS lookup'));
204+
205+
$_ = $reject = null;
206+
});
207+
}
208+
209+
/**
210+
* @internal
211+
*/
212+
public function attemptConnection($ip, $uri, $parts, $host)
213+
{
214+
$connector = $this->connector;
215+
$resolved = null;
216+
$promise = null;
217+
218+
return new Promise\Promise(
219+
function ($resolve, $reject) use (&$promise, &$resolved, $ip, $uri, $connector, $host, $parts) {
220+
$resolved = $ip;
221+
$uri = '';
222+
223+
// prepend original scheme if known
224+
if (isset($parts['scheme'])) {
225+
$uri .= $parts['scheme'] . '://';
226+
}
227+
228+
if (\strpos($ip, ':') !== false) {
229+
// enclose IPv6 addresses in square brackets before appending port
230+
$uri .= '[' . $ip . ']';
231+
} else {
232+
$uri .= $ip;
233+
}
234+
235+
// append original port if known
236+
if (isset($parts['port'])) {
237+
$uri .= ':' . $parts['port'];
238+
}
239+
240+
// append orignal path if known
241+
if (isset($parts['path'])) {
242+
$uri .= $parts['path'];
243+
}
244+
245+
// append original query if known
246+
if (isset($parts['query'])) {
247+
$uri .= '?' . $parts['query'];
248+
}
249+
250+
// append original hostname as query if resolved via DNS and if
251+
// destination URI does not contain "hostname" query param already
252+
$args = array();
253+
\parse_str(isset($parts['query']) ? $parts['query'] : '', $args);
254+
if ($host !== $ip && !isset($args['hostname'])) {
255+
$uri .= (isset($parts['query']) ? '&' : '?') . 'hostname=' . \rawurlencode($host);
256+
}
257+
258+
// append original fragment if known
259+
if (isset($parts['fragment'])) {
260+
$uri .= '#' . $parts['fragment'];
261+
}
262+
263+
$promise = $connector->connect($uri);
264+
$promise->then($resolve, $reject);
265+
},
266+
function ($_, $reject) use (&$promise, &$resolved, $uri) {
267+
// cancellation should reject connection attempt
268+
// (try to) cancel pending connection attempt
269+
if ($resolved === null) {
270+
$reject(new \RuntimeException('Connection to ' . $uri . ' cancelled during connection attempt'));
271+
}
272+
273+
if ($promise instanceof CancellablePromiseInterface) {
274+
// overwrite callback arguments for PHP7+ only, so they do not show
275+
// up in the Exception trace and do not cause a possible cyclic reference.
276+
$_ = $reject = null;
277+
278+
$promise->cancel();
279+
$promise = null;
280+
}
281+
}
282+
);
283+
}
284+
}

0 commit comments

Comments
 (0)