From c886e3294496c223295f0efd371b222596300df4 Mon Sep 17 00:00:00 2001 From: Cees-Jan Kiewiet Date: Wed, 27 Feb 2019 22:30:04 +0100 Subject: [PATCH] Implement the Happy Eye Balls RFC's By using the happy eye balls algorithm as described in RFC6555 and RFC8305 it will connect to the quickest responding server with a preference for IPv6. --- README.md | 55 +++ src/HappyEyeBallsConnectionBuilder.php | 317 ++++++++++++++++ src/HappyEyeBallsConnector.php | 53 +++ tests/HappyEyeBallsConnectorTest.php | 488 +++++++++++++++++++++++++ tests/TestCase.php | 2 +- 5 files changed, 914 insertions(+), 1 deletion(-) create mode 100644 src/HappyEyeBallsConnectionBuilder.php create mode 100644 src/HappyEyeBallsConnector.php create mode 100644 tests/HappyEyeBallsConnectorTest.php diff --git a/README.md b/README.md index a9e4729d..e75bcc9c 100644 --- a/README.md +++ b/README.md @@ -43,6 +43,7 @@ handle multiple concurrent connections without blocking. * [Connector](#connector) * [Advanced client usage](#advanced-client-usage) * [TcpConnector](#tcpconnector) + * [HappyEyeBallsConnector](#happyeyeballsconnector) * [DnsConnector](#dnsconnector) * [SecureConnector](#secureconnector) * [TimeoutConnector](#timeoutconnector) @@ -1154,6 +1155,60 @@ be used to set up the TLS peer name. This is used by the `SecureConnector` and `DnsConnector` to verify the peer name and can also be used if you want a custom TLS peer name. +#### HappyEyeBallsConnector + +The `HappyEyeBallsConnector` class implements the +[`ConnectorInterface`](#connectorinterface) and allows you to create plaintext +TCP/IP connections to any hostname-port-combination. Internally it implements the +happy eyeballs algorythm from [`RFC6555`](https://tools.ietf.org/html/rfc6555) and +[`RFC8305`](https://tools.ietf.org/html/rfc8305) to support IPv6 and IPv4 hostnames. + +It does so by decorating a given `TcpConnector` instance so that it first +looks up the given domain name via DNS (if applicable) and then establishes the +underlying TCP/IP connection to the resolved target IP address. + +Make sure to set up your DNS resolver and underlying TCP connector like this: + +```php +$dnsResolverFactory = new React\Dns\Resolver\Factory(); +$dns = $dnsResolverFactory->createCached('8.8.8.8', $loop); + +$dnsConnector = new React\Socket\HappyEyeBallsConnector($loop, $tcpConnector, $dns); + +$dnsConnector->connect('www.google.com:80')->then(function (React\Socket\ConnectionInterface $connection) { + $connection->write('...'); + $connection->end(); +}); + +$loop->run(); +``` + +See also the [examples](examples). + +Pending connection attempts can be cancelled by cancelling its pending promise like so: + +```php +$promise = $dnsConnector->connect('www.google.com:80'); + +$promise->cancel(); +``` + +Calling `cancel()` on a pending promise will cancel the underlying DNS lookups +and/or the underlying TCP/IP connection(s) and reject the resulting promise. + + +> Advanced usage: Internally, the `HappyEyeBallsConnector` relies on a `Resolver` to +look up the IP addresses for the given hostname. +It will then replace the hostname in the destination URI with this IP's and +append a `hostname` query parameter and pass this updated URI to the underlying +connector. +The Happy Eye Balls algorythm describes looking the IPv6 and IPv4 address for +the given hostname so this connector sends out two DNS lookups for the A and +AAAA records. It then uses all IP addresses (both v6 and v4) and tries to +connect to all of them with a 50ms interval in between. Alterating between IPv6 +and IPv4 addresses. When a connection is established all the other DNS lookups +and connection attempts are cancelled. + #### DnsConnector The `DnsConnector` class implements the diff --git a/src/HappyEyeBallsConnectionBuilder.php b/src/HappyEyeBallsConnectionBuilder.php new file mode 100644 index 00000000..8cd4c11d --- /dev/null +++ b/src/HappyEyeBallsConnectionBuilder.php @@ -0,0 +1,317 @@ + false, + Message::TYPE_AAAA => false, + ); + public $resolverPromises = array(); + public $connectionPromises = array(); + public $connectQueue = array(); + public $timer; + public $parts; + public $ipsCount = 0; + public $failureCount = 0; + public $resolve; + public $reject; + + public function __construct(LoopInterface $loop, ConnectorInterface $connector, ResolverInterface $resolver, $uri, $host, $parts) + { + $this->loop = $loop; + $this->connector = $connector; + $this->resolver = $resolver; + $this->uri = $uri; + $this->host = $host; + $this->parts = $parts; + } + + public function connect() + { + $that = $this; + return new Promise\Promise(function ($resolve, $reject) use ($that) { + $lookupResolve = function ($type) use ($that, $resolve, $reject) { + return function (array $ips) use ($that, $type, $resolve, $reject) { + unset($that->resolverPromises[$type]); + $that->resolved[$type] = true; + + $that->mixIpsIntoConnectQueue($ips); + + if ($that->timer instanceof TimerInterface) { + return; + } + + $that->check($resolve, $reject); + }; + }; + + $ipv4Deferred = null; + $that->resolverPromises[Message::TYPE_AAAA] = $that->resolve(Message::TYPE_AAAA, $reject)->then($lookupResolve(Message::TYPE_AAAA))->then(function () use (&$ipv4Deferred) { + if ($ipv4Deferred instanceof Promise\Deferred) { + $ipv4Deferred->resolve(); + } + }); + $that->resolverPromises[Message::TYPE_A] = $that->resolve(Message::TYPE_A, $reject)->then(function ($ips) use ($that, &$ipv4Deferred) { + if ($that->resolved[Message::TYPE_AAAA] === true) { + return Promise\resolve($ips); + } + + /** + * Delay A lookup by 50ms sending out connection to IPv4 addresses when IPv6 records haven't + * resolved yet as per RFC. + * + * @link https://tools.ietf.org/html/rfc8305#section-3 + */ + $ipv4Deferred = new Promise\Deferred(); + $deferred = new Promise\Deferred(); + + $timer = $that->loop->addTimer($that::RESOLVE_WAIT, function () use ($deferred, $ips) { + $ipv4Deferred = null; + $deferred->resolve($ips); + }); + + $ipv4Deferred->promise()->then(function () use ($that, $timer, $deferred, $ips) { + $that->loop->cancelTimer($timer); + $deferred->resolve($ips); + }); + + return $deferred->promise(); + })->then($lookupResolve(Message::TYPE_A)); + }, function ($_, $reject) use ($that) { + $that->cleanUp(); + + $reject(new \RuntimeException('Connection to ' . $that->uri . ' cancelled during DNS lookup')); + + $_ = $reject = null; + }); + } + + /** + * @internal + */ + public function resolve($type, $reject) + { + $that = $this; + return $that->resolver->resolveAll($that->host, $type)->then(null, function () use ($type, $reject, $that) { + unset($that->resolverPromises[$type]); + $that->resolved[$type] = true; + + if ($that->hasBeenResolved() === false) { + return; + } + + if ($that->ipsCount === 0) { + $that->resolved = null; + $that->resolverPromises = null; + $reject(new \RuntimeException('Connection to ' . $that->uri . ' failed during DNS lookup: DNS error')); + } + }); + } + + /** + * @internal + */ + public function check($resolve, $reject) + { + if (\count($this->connectQueue) === 0 && ($this->resolved[Message::TYPE_A] === false || $this->resolved[Message::TYPE_AAAA] === false) && $this->timer instanceof TimerInterface) { + return; + } + + $ip = \array_shift($this->connectQueue); + + if (\count($this->connectQueue) === 0 && $this->resolved[Message::TYPE_A] === true && $this->resolved[Message::TYPE_AAAA] === true && $this->timer instanceof TimerInterface) { + $this->loop->cancelTimer($this->timer); + $this->timer = null; + } + + $that = $this; + $that->connectionPromises[$ip] = $this->attemptConnection($ip)->then(function ($connection) use ($that, $ip, $resolve) { + unset($that->connectionPromises[$ip]); + + $that->cleanUp(); + + $resolve($connection); + }, function () use ($that, $ip, $resolve, $reject) { + unset($that->connectionPromises[$ip]); + + $that->failureCount++; + + if ($that->hasBeenResolved() === false) { + return; + } + + if ($that->ipsCount === $that->failureCount) { + $that->cleanUp(); + + $reject(new \RuntimeException('All attempts to connect to "' . $that->host . '" have failed')); + } + }); + + /** + * As long as we haven't connected yet keep popping an IP address of the connect queue until one of them + * succeeds or they all fail. We will wait 100ms between connection attempts as per RFC. + * + * @link https://tools.ietf.org/html/rfc8305#section-5 + */ + if ((\count($this->connectQueue) > 0 || ($this->resolved[Message::TYPE_A] === false || $this->resolved[Message::TYPE_AAAA] === false)) && $this->timer === null) { + $this->timer = $this->loop->addPeriodicTimer(self::CONNECT_INTERVAL, function () use ($that, $resolve, $reject) { + $that->check($resolve, $reject); + }); + } + } + + /** + * @internal + */ + public function attemptConnection($ip) + { + $promise = null; + $that = $this; + + return new Promise\Promise( + function ($resolve, $reject) use (&$promise, $that, $ip) { + $uri = ''; + + // prepend original scheme if known + if (isset($that->parts['scheme'])) { + $uri .= $that->parts['scheme'] . '://'; + } + + if (\strpos($ip, ':') !== false) { + // enclose IPv6 addresses in square brackets before appending port + $uri .= '[' . $ip . ']'; + } else { + $uri .= $ip; + } + + // append original port if known + if (isset($that->parts['port'])) { + $uri .= ':' . $that->parts['port']; + } + + // append orignal path if known + if (isset($that->parts['path'])) { + $uri .= $that->parts['path']; + } + + // append original query if known + if (isset($that->parts['query'])) { + $uri .= '?' . $that->parts['query']; + } + + // append original hostname as query if resolved via DNS and if + // destination URI does not contain "hostname" query param already + $args = array(); + \parse_str(isset($that->parts['query']) ? $that->parts['query'] : '', $args); + if ($that->host !== $ip && !isset($args['hostname'])) { + $uri .= (isset($that->parts['query']) ? '&' : '?') . 'hostname=' . \rawurlencode($that->host); + } + + // append original fragment if known + if (isset($that->parts['fragment'])) { + $uri .= '#' . $that->parts['fragment']; + } + + $promise = $that->connector->connect($uri); + $promise->then($resolve, $reject); + }, + function ($_, $reject) use (&$promise, $that) { + // cancellation should reject connection attempt + // (try to) cancel pending connection attempt + $reject(new \RuntimeException('Connection to ' . $that->uri . ' cancelled during connection attempt')); + + if ($promise instanceof CancellablePromiseInterface) { + // overwrite callback arguments for PHP7+ only, so they do not show + // up in the Exception trace and do not cause a possible cyclic reference. + $_ = $reject = null; + + $promise->cancel(); + $promise = null; + } + } + ); + } + + /** + * @internal + */ + public function cleanUp() + { + /** @var CancellablePromiseInterface $promise */ + foreach ($this->connectionPromises as $index => $connectionPromise) { + if ($connectionPromise instanceof CancellablePromiseInterface) { + $connectionPromise->cancel(); + } + } + + /** @var CancellablePromiseInterface $promise */ + foreach ($this->resolverPromises as $index => $resolverPromise) { + if ($resolverPromise instanceof CancellablePromiseInterface) { + $resolverPromise->cancel(); + } + } + + if ($this->timer instanceof TimerInterface) { + $this->loop->cancelTimer($this->timer); + $this->timer = null; + } + } + + /** + * @internal + */ + public function hasBeenResolved() + { + foreach ($this->resolved as $typeHasBeenResolved) { + if ($typeHasBeenResolved === false) { + return false; + } + } + + return true; + } + + /** + * Mixes an array of IP addresses into the connect queue in such a way they alternate when attempting to connect. + * The goal behind it is first attempt to connect to IPv6, then to IPv4, then to IPv6 again until one of those + * attempts succeeds. + * + * @link https://tools.ietf.org/html/rfc8305#section-4 + * + * @internal + */ + public function mixIpsIntoConnectQueue(array $ips) + { + $this->ipsCount += \count($ips); + $connectQueueStash = $this->connectQueue; + $this->connectQueue = array(); + while (\count($connectQueueStash) > 0 || \count($ips) > 0) { + if (\count($ips) > 0) { + $this->connectQueue[] = \array_shift($ips); + } + if (\count($connectQueueStash) > 0) { + $this->connectQueue[] = \array_shift($connectQueueStash); + } + } + } +} \ No newline at end of file diff --git a/src/HappyEyeBallsConnector.php b/src/HappyEyeBallsConnector.php new file mode 100644 index 00000000..1d4a21df --- /dev/null +++ b/src/HappyEyeBallsConnector.php @@ -0,0 +1,53 @@ +loop = $loop; + $this->connector = $connector; + $this->resolver = $resolver; + } + + public function connect($uri) + { + + if (\strpos($uri, '://') === false) { + $parts = \parse_url('tcp://' . $uri); + unset($parts['scheme']); + } else { + $parts = \parse_url($uri); + } + + if (!$parts || !isset($parts['host'])) { + return Promise\reject(new \InvalidArgumentException('Given URI "' . $uri . '" is invalid')); + } + + $host = \trim($parts['host'], '[]'); + + // skip DNS lookup / URI manipulation if this URI already contains an IP + if (false !== \filter_var($host, \FILTER_VALIDATE_IP)) { + return $this->connector->connect($uri); + } + + $builder = new HappyEyeBallsConnectionBuilder( + $this->loop, + $this->connector, + $this->resolver, + $uri, + $host, + $parts + ); + return $builder->connect(); + } +} diff --git a/tests/HappyEyeBallsConnectorTest.php b/tests/HappyEyeBallsConnectorTest.php new file mode 100644 index 00000000..0ce1ed5f --- /dev/null +++ b/tests/HappyEyeBallsConnectorTest.php @@ -0,0 +1,488 @@ +loop = new StreamSelectLoop(); + $this->tcp = $this->getMockBuilder('React\Socket\ConnectorInterface')->getMock(); + $this->resolver = $this->getMockBuilder('React\Dns\Resolver\ResolverInterface')->disableOriginalConstructor()->getMock(); + + $this->connector = new HappyEyeBallsConnector($this->loop, $this->tcp, $this->resolver); + } + + public function testHappyFlow() + { + $first = new Deferred(); + $this->resolver->expects($this->exactly(2))->method('resolveAll')->with($this->equalTo('example.com'), $this->anything())->willReturn($first->promise()); + $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock(); + $this->tcp->expects($this->exactly(1))->method('connect')->with($this->equalTo('1.2.3.4:80?hostname=example.com'))->willReturn(Promise\resolve($connection)); + + $promise = $this->connector->connect('example.com:80'); + $first->resolve(array('1.2.3.4')); + + $resolvedConnection = Block\await($promise, $this->loop); + + self::assertSame($connection, $resolvedConnection); + } + + public function testThatAnyOtherPendingConnectionAttemptsWillBeCanceled() + { + $connection = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock(); + $lookupAttempts = array( + Promise\reject(new \Exception('error')), + Promise\resolve(array('1.2.3.4', '5.6.7.8', '9.10.11.12')), + ); + $connectionAttempts = array( + new Promise\Promise(function () {}, $this->expectCallableOnce()), + Promise\resolve($connection), + new Promise\Promise(function () {}, $this->expectCallableNever()), + ); + $this->resolver->expects($this->exactly(2))->method('resolveAll')->with($this->equalTo('example.com'), $this->anything())->will($this->returnCallback(function () use (&$lookupAttempts) { + return array_shift($lookupAttempts); + })); + $this->tcp->expects($this->exactly(2))->method('connect')->with($this->isType('string'))->will($this->returnCallback(function () use (&$connectionAttempts) { + return array_shift($connectionAttempts); + })); + + $promise = $this->connector->connect('example.com:80'); + + $resolvedConnection = Block\await($promise, $this->loop); + + self::assertSame($connection, $resolvedConnection); + } + + public function testPassByResolverIfGivenIp() + { + $this->resolver->expects($this->never())->method('resolveAll'); + $this->tcp->expects($this->once())->method('connect')->with($this->equalTo('127.0.0.1:80'))->will($this->returnValue(Promise\resolve())); + + $this->connector->connect('127.0.0.1:80'); + + $this->loop->run(); + } + + public function testPassByResolverIfGivenIpv6() + { + $this->resolver->expects($this->never())->method('resolveAll'); + $this->tcp->expects($this->once())->method('connect')->with($this->equalTo('[::1]:80'))->will($this->returnValue(Promise\reject())); + + $this->connector->connect('[::1]:80'); + + $this->loop->run(); + } + + public function testPassThroughResolverIfGivenHost() + { + $this->resolver->expects($this->exactly(2))->method('resolveAll')->with($this->equalTo('google.com'), $this->anything())->will($this->returnValue(Promise\resolve(array('1.2.3.4')))); + $this->tcp->expects($this->exactly(2))->method('connect')->with($this->equalTo('1.2.3.4:80?hostname=google.com'))->will($this->returnValue(Promise\reject())); + + $this->connector->connect('google.com:80'); + + $this->loop->run(); + } + + public function testPassThroughResolverIfGivenHostWhichResolvesToIpv6() + { + $this->resolver->expects($this->exactly(2))->method('resolveAll')->with($this->equalTo('google.com'), $this->anything())->will($this->returnValue(Promise\resolve(array('::1')))); + $this->tcp->expects($this->exactly(2))->method('connect')->with($this->equalTo('[::1]:80?hostname=google.com'))->will($this->returnValue(Promise\reject())); + + $this->connector->connect('google.com:80'); + + $this->loop->run(); + } + + public function testPassByResolverIfGivenCompleteUri() + { + $this->resolver->expects($this->never())->method('resolveAll'); + $this->tcp->expects($this->once())->method('connect')->with($this->equalTo('scheme://127.0.0.1:80/path?query#fragment'))->will($this->returnValue(Promise\reject())); + + $this->connector->connect('scheme://127.0.0.1:80/path?query#fragment'); + + $this->loop->run(); + } + + public function testPassThroughResolverIfGivenCompleteUri() + { + $this->resolver->expects($this->exactly(2))->method('resolveAll')->with($this->equalTo('google.com'), $this->anything())->will($this->returnValue(Promise\resolve(array('1.2.3.4')))); + $this->tcp->expects($this->exactly(2))->method('connect')->with($this->equalTo('scheme://1.2.3.4:80/path?query&hostname=google.com#fragment'))->will($this->returnValue(Promise\reject())); + + $this->connector->connect('scheme://google.com:80/path?query#fragment'); + + $this->loop->run(); + } + + public function testPassThroughResolverIfGivenExplicitHost() + { + $this->resolver->expects($this->exactly(2))->method('resolveAll')->with($this->equalTo('google.com'), $this->anything())->will($this->returnValue(Promise\resolve(array('1.2.3.4')))); + $this->tcp->expects($this->exactly(2))->method('connect')->with($this->equalTo('scheme://1.2.3.4:80/?hostname=google.de'))->will($this->returnValue(Promise\reject())); + + $this->connector->connect('scheme://google.com:80/?hostname=google.de'); + + $this->loop->run(); + } + + /** + * @dataProvider provideIpvAddresses + */ + public function testIpv4ResolvesFirstSoButIPv6IsTheFirstToConnect(array $ipv6, array $ipv4) + { + $this->resolver->expects($this->at(0))->method('resolveAll')->with('google.com', Message::TYPE_AAAA)->will($this->returnValue(Promise\Timer\resolve(0.001, $this->loop)->then(function () use ($ipv6) { + return Promise\resolve($ipv6); + }))); + $this->resolver->expects($this->at(1))->method('resolveAll')->with('google.com', Message::TYPE_A)->will($this->returnValue(Promise\resolve($ipv4))); + $i = 0; + while (count($ipv6) > 0 && count($ipv4) > 0) { + $this->tcp->expects($this->at($i++))->method('connect')->with($this->equalTo('scheme://[' . array_shift($ipv6) . ']:80/?hostname=google.com'))->will($this->returnValue(Promise\resolve())); + $this->tcp->expects($this->at($i++))->method('connect')->with($this->equalTo('scheme://' . array_shift($ipv4) . ':80/?hostname=google.com'))->will($this->returnValue(Promise\resolve())); + } + + $this->connector->connect('scheme://google.com:80/?hostname=google.com'); + + $this->loop->run(); + } + + /** + * @dataProvider provideIpvAddresses + */ + public function testIpv6ResolvesFirstSoIsTheFirstToConnect(array $ipv6, array $ipv4) + { + $deferred = new Deferred(); + + $this->resolver->expects($this->at(0))->method('resolveAll')->with('google.com', Message::TYPE_AAAA)->will($this->returnValue(Promise\resolve($ipv6))); + $this->resolver->expects($this->at(1))->method('resolveAll')->with('google.com', Message::TYPE_A)->will($this->returnValue($deferred->promise())); + $this->tcp->expects($this->any())->method('connect')->with($this->stringContains(']:80/?hostname=google.com'))->will($this->returnValue(Promise\reject())); + + $this->connector->connect('scheme://google.com:80/?hostname=google.com'); + + $this->loop->addTimer(0.8, function () use ($deferred) { + $deferred->reject(); + }); + + $this->loop->run(); + } + + /** + * @dataProvider provideIpvAddresses + */ + public function testIpv6DoesntResolvesWhileIpv4DoesFirstSoIpv4Connects(array $ipv6, array $ipv4) + { + $deferred = new Deferred(); + + $this->resolver->expects($this->at(0))->method('resolveAll')->with('google.com', Message::TYPE_AAAA)->will($this->returnValue($deferred->promise())); + $this->resolver->expects($this->at(1))->method('resolveAll')->with('google.com', Message::TYPE_A)->will($this->returnValue(Promise\resolve($ipv4))); + $this->tcp->expects($this->any())->method('connect')->with($this->stringContains(':80/?hostname=google.com'))->will($this->returnValue(Promise\reject())); + + $this->connector->connect('scheme://google.com:80/?hostname=google.com'); + + $this->loop->addTimer(0.8, function () use ($deferred) { + $deferred->reject(); + }); + + $this->loop->run(); + } + + /** + * @dataProvider provideIpvAddresses + */ + public function testAttemptsToConnectBothIpv6AndIpv4Addresses(array $ipv6, array $ipv4) + { + $this->resolver->expects($this->at(0))->method('resolveAll')->with('google.com', Message::TYPE_AAAA)->will($this->returnValue(Promise\resolve($ipv6))); + $this->resolver->expects($this->at(1))->method('resolveAll')->with('google.com', Message::TYPE_A)->will($this->returnValue(Promise\resolve($ipv4))); + + $i = 0; + while (count($ipv6) > 0 && count($ipv4) > 0) { + $this->tcp->expects($this->at($i++))->method('connect')->with($this->equalTo('scheme://[' . array_shift($ipv6) . ']:80/?hostname=google.com'))->will($this->returnValue(Promise\resolve())); + $this->tcp->expects($this->at($i++))->method('connect')->with($this->equalTo('scheme://' . array_shift($ipv4) . ':80/?hostname=google.com'))->will($this->returnValue(Promise\resolve())); + } + + + $this->connector->connect('scheme://google.com:80/?hostname=google.com'); + + $this->loop->run(); + } + + /** + * @dataProvider provideIpvAddresses + */ + public function testThatTheIpv4ConnectionWillWait100MilisecondsWhenIpv6AndIpv4ResolveSimultaniously(array $ipv6, array $ipv4) + { + $timings = array(); + $this->resolver->expects($this->at(0))->method('resolveAll')->with('google.com', Message::TYPE_AAAA)->will($this->returnValue(Promise\resolve(array(array_shift($ipv6))))); + $this->resolver->expects($this->at(1))->method('resolveAll')->with('google.com', Message::TYPE_A)->will($this->returnValue(Promise\resolve(array(array_shift($ipv4))))); + $this->tcp->expects($this->at(0))->method('connect')->with($this->equalTo('scheme://[1:2:3:4]:80/?hostname=google.com'))->will($this->returnCallback(function () use (&$timings) { + $timings[Message::TYPE_AAAA] = microtime(true); + + return new Promise\Promise(function () {}); + })); + $this->tcp->expects($this->at(1))->method('connect')->with($this->equalTo('scheme://1.2.3.4:80/?hostname=google.com'))->will($this->returnCallback(function () use (&$timings) { + $timings[Message::TYPE_A] = microtime(true); + + return new Promise\Promise(function () {}); + })); + + $this->connector->connect('scheme://google.com:80/?hostname=google.com'); + + $this->loop->run(); + + self::assertGreaterThan(0.1, $timings[Message::TYPE_A] - $timings[Message::TYPE_AAAA]); + } + + /** + * @dataProvider provideIpvAddresses + */ + public function testAssert100MilisecondsBetweenConnectionAttempts(array $ipv6, array $ipv4) + { + $timings = array(); + $this->resolver->expects($this->at(0))->method('resolveAll')->with('google.com', Message::TYPE_AAAA)->will($this->returnValue(Promise\resolve($ipv6))); + $this->resolver->expects($this->at(1))->method('resolveAll')->with('google.com', Message::TYPE_A)->will($this->returnValue(Promise\resolve($ipv4))); + $this->tcp->expects($this->any())->method('connect')->with($this->stringContains(':80/?hostname=google.com'))->will($this->returnCallback(function () use (&$timings) { + $timings[] = microtime(true); + + return new Promise\Promise(function () {}); + })); + + $this->connector->connect('scheme://google.com:80/?hostname=google.com'); + + $this->loop->run(); + + for ($i = 0; $i < (count($timings) - 1); $i++) { + self::assertGreaterThan(0.1, $timings[$i + 1] - $timings[$i]); + } + } + + public function testAttemptsToConnectBothIpv6AndIpv4AddressesAlternatingIpv6AndIpv4AddressesWhenMoreThenOneIsResolvedPerFamily() + { + $this->resolver->expects($this->at(0))->method('resolveAll')->with('google.com', Message::TYPE_AAAA)->will($this->returnValue( + Promise\Timer\resolve(0.1, $this->loop)->then(function () { + return Promise\resolve(array('1:2:3:4', '5:6:7:8', '9:10:11:12')); + }) + )); + $this->resolver->expects($this->at(1))->method('resolveAll')->with('google.com', Message::TYPE_A)->will($this->returnValue( + Promise\Timer\resolve(0.1, $this->loop)->then(function () { + return Promise\resolve(array('1.2.3.4', '5.6.7.8', '9.10.11.12')); + }) + )); + $this->tcp->expects($this->at(0))->method('connect')->with($this->equalTo('scheme://[1:2:3:4]:80/?hostname=google.com'))->will($this->returnValue(Promise\reject())); + $this->tcp->expects($this->at(1))->method('connect')->with($this->equalTo('scheme://1.2.3.4:80/?hostname=google.com'))->will($this->returnValue(Promise\reject())); + $this->tcp->expects($this->at(2))->method('connect')->with($this->equalTo('scheme://[5:6:7:8]:80/?hostname=google.com'))->will($this->returnValue(Promise\reject())); + $this->tcp->expects($this->at(3))->method('connect')->with($this->equalTo('scheme://5.6.7.8:80/?hostname=google.com'))->will($this->returnValue(Promise\reject())); + $this->tcp->expects($this->at(4))->method('connect')->with($this->equalTo('scheme://[9:10:11:12]:80/?hostname=google.com'))->will($this->returnValue(Promise\reject())); + $this->tcp->expects($this->at(5))->method('connect')->with($this->equalTo('scheme://9.10.11.12:80/?hostname=google.com'))->will($this->returnValue(Promise\reject())); + + $this->connector->connect('scheme://google.com:80/?hostname=google.com'); + + $this->loop->run(); + } + + public function testRejectsImmediatelyIfUriIsInvalid() + { + $this->resolver->expects($this->never())->method('resolveAll'); + $this->tcp->expects($this->never())->method('connect'); + + $promise = $this->connector->connect('////'); + + $promise->then($this->expectCallableNever(), $this->expectCallableOnce()); + + $this->loop->run(); + } + + /** + * @expectedException RuntimeException + * @expectedExceptionMessage Connection failed + */ + public function testRejectsWithTcpConnectorRejectionIfGivenIp() + { + $that = $this; + $promise = Promise\reject(new \RuntimeException('Connection failed')); + $this->resolver->expects($this->never())->method('resolveAll'); + $this->tcp->expects($this->once())->method('connect')->with($this->equalTo('1.2.3.4:80'))->willReturn($promise); + + $promise = $this->connector->connect('1.2.3.4:80'); + $this->loop->addTimer(0.5, function () use ($that, $promise) { + $promise->cancel(); + + $that->throwRejection($promise); + }); + + $this->loop->run(); + } + + /** + * @expectedException RuntimeException + * @expectedExceptionMessage All attempts to connect to "example.com" have failed + * @dataProvider provideIpvAddresses + */ + public function testRejectsWithTcpConnectorRejectionAfterDnsIsResolved(array $ipv6, array $ipv4) + { + $that = $this; + $promise = Promise\reject(new \RuntimeException('Connection failed')); + $this->resolver->expects($this->at(0))->method('resolveAll')->with($this->equalTo('example.com'), $this->anything())->willReturn(Promise\resolve($ipv6)); + $this->resolver->expects($this->at(1))->method('resolveAll')->with($this->equalTo('example.com'), $this->anything())->willReturn(Promise\resolve($ipv4)); + $this->tcp->expects($this->any())->method('connect')->with($this->stringContains(':80?hostname=example.com'))->willReturn($promise); + + $promise = $this->connector->connect('example.com:80'); + $this->loop->addTimer(0.8, function () use ($that, $promise) { + $promise->cancel(); + + $that->throwRejection($promise); + }); + + $this->loop->run(); + } + + /** + * @expectedException RuntimeException + * @expectedExceptionMessage Connection to example.invalid:80 failed during DNS lookup: DNS error + */ + public function testSkipConnectionIfDnsFails() + { + $that = $this; + $this->resolver->expects($this->exactly(2))->method('resolveAll')->with($this->equalTo('example.invalid'), $this->anything())->willReturn(Promise\reject(new \RuntimeException('DNS error'))); + $this->tcp->expects($this->never())->method('connect'); + + $promise = $this->connector->connect('example.invalid:80'); + + $this->loop->addTimer(0.5, function () use ($that, $promise) { + $that->throwRejection($promise); + }); + + $this->loop->run(); + } + + /** + * @expectedException RuntimeException + * @expectedExceptionMessage Connection to example.com:80 cancelled during DNS lookup + */ + public function testCancelDuringDnsCancelsDnsAndDoesNotStartTcpConnection() + { + $that = $this; + $this->resolver->expects($this->exactly(2))->method('resolveAll')->with('example.com', $this->anything())->will($this->returnCallback(function () use ($that) { + return new Promise\Promise(function () { }, $that->expectCallableExactly(1)); + })); + $this->tcp->expects($this->never())->method('connect'); + + $promise = $this->connector->connect('example.com:80'); + $this->loop->addTimer(0.05, function () use ($that, $promise) { + $promise->cancel(); + + $that->throwRejection($promise); + }); + + $this->loop->run(); + } + + public function testCancelDuringTcpConnectionCancelsTcpConnectionIfGivenIp() + { + $pending = new Promise\Promise(function () { }, $this->expectCallableOnce()); + $this->resolver->expects($this->never())->method('resolveAll'); + $this->tcp->expects($this->once())->method('connect')->with($this->equalTo('1.2.3.4:80'))->willReturn($pending); + + $promise = $this->connector->connect('1.2.3.4:80'); + $this->loop->addTimer(0.5, function () use ($promise) { + $promise->cancel(); + }); + + $this->loop->run(); + } + + /** + * @dataProvider provideIpvAddresses + */ + public function testCancelDuringTcpConnectionCancelsTcpConnectionAfterDnsIsResolved(array $ipv6, array $ipv4) + { + $pending = new Promise\Promise(function () { }, $this->expectCallableOnce()); + $this->resolver->expects($this->at(0))->method('resolveAll')->with($this->equalTo('example.com'), $this->anything())->willReturn(Promise\resolve($ipv6)); + $this->resolver->expects($this->at(1))->method('resolveAll')->with($this->equalTo('example.com'), $this->anything())->willReturn(Promise\resolve($ipv4)); + $this->tcp->expects($this->any())->method('connect')->with($this->stringContains(':80?hostname=example.com'))->willReturn($pending); + + $promise = $this->connector->connect('example.com:80'); + $this->loop->addTimer(0.8, function () use ($promise) { + $promise->cancel(); + }); + + $this->loop->run(); + } + + /** + * @expectedException RuntimeException + * @expectedExceptionMessage All attempts to connect to "example.com" have failed + */ + public function testCancelDuringTcpConnectionCancelsTcpConnectionWithTcpRejectionAfterDnsIsResolved() + { + $first = new Deferred(); + $this->resolver->expects($this->exactly(2))->method('resolveAll')->with($this->equalTo('example.com'), $this->anything())->willReturn($first->promise()); + $pending = new Promise\Promise(function () { }, function () { + throw new \RuntimeException('Connection cancelled'); + }); + $this->tcp->expects($this->exactly(2))->method('connect')->with($this->equalTo('1.2.3.4:80?hostname=example.com'))->willReturn($pending); + + $promise = $this->connector->connect('example.com:80'); + $first->resolve(array('1.2.3.4')); + + $that = $this; + $this->loop->addTimer(0.5, function () use ($promise, $that) { + $promise->cancel(); + + $that->throwRejection($promise); + }); + + $this->loop->run(); + } + + /** + * @internal + */ + public function throwRejection($promise) + { + $ex = null; + $promise->then(null, function ($e) use (&$ex) { + $ex = $e; + }); + + throw $ex; + } + + public function provideIpvAddresses() + { + $ipv6 = array( + array('1:2:3:4'), + array('1:2:3:4', '5:6:7:8'), + array('1:2:3:4', '5:6:7:8', '9:10:11:12'), + array('1:2:3:4', '5:6:7:8', '9:10:11:12', '13:14:15:16'), + ); + $ipv4 = array( + array('1.2.3.4'), + array('1.2.3.4', '5.6.7.8'), + array('1.2.3.4', '5.6.7.8', '9.10.11.12'), + array('1.2.3.4', '5.6.7.8', '9.10.11.12', '13.14.15.16'), + ); + + $ips = array(); + + foreach ($ipv6 as $v6) { + foreach ($ipv4 as $v4) { + $ips[] = array( + $v6, + $v4 + ); + } + } + + return $ips; + } +} diff --git a/tests/TestCase.php b/tests/TestCase.php index b9b383af..4c4adbd9 100644 --- a/tests/TestCase.php +++ b/tests/TestCase.php @@ -10,7 +10,7 @@ class TestCase extends BaseTestCase { - protected function expectCallableExactly($amount) + public function expectCallableExactly($amount) { $mock = $this->createCallableMock(); $mock