From a7ee80702cfd57f64ee57b9ee036d3b4f561fc58 Mon Sep 17 00:00:00 2001 From: Daniele Alessandri Date: Wed, 29 Jul 2015 18:00:38 +0200 Subject: [PATCH] Implement full support for IPv6. Using IPv6 with Predis was basically impossible due to various inconsistencies and bugs through the library, now it is supported by all the connection classes. Following the standard for IPv6 literal addresses in URI strings, the IP literal must be enclosed within square brackets when passing the parameters as a string: $parameters = 'tcp://[2001:db8:0:f101::1]:6379'; See https://tools.ietf.org/html/rfc3986#section-3.2.2 for further details. This commit also fixes #239 making redis-cluster usable with nodes using IPv6. --- src/Connection/AbstractConnection.php | 31 +++++- src/Connection/Aggregate/RedisCluster.php | 6 +- src/Connection/PhpiredisSocketConnection.php | 104 +++++++++--------- src/Connection/StreamConnection.php | 2 +- src/Connection/WebdisConnection.php | 6 +- tests/PHPUnit/PredisConnectionTestCase.php | 35 +++++- .../Connection/Aggregate/RedisClusterTest.php | 34 ++++++ tests/Predis/Connection/ParametersTest.php | 12 +- 8 files changed, 159 insertions(+), 71 deletions(-) diff --git a/src/Connection/AbstractConnection.php b/src/Connection/AbstractConnection.php index 7c2ff09d..fb865132 100644 --- a/src/Connection/AbstractConnection.php +++ b/src/Connection/AbstractConnection.php @@ -120,6 +120,29 @@ abstract class AbstractConnection implements NodeConnectionInterface return $this->read(); } + /** + * Helper method that returns an exception message augmented with useful + * details from the connection parameters. + * + * @param string $message Error message. + * + * @return string + */ + private function createExceptionMessage($message) + { + $parameters = $this->parameters; + + if ($parameters->scheme === 'unix') { + return "$message [$parameters->scheme:$parameters->path]"; + } + + if (filter_var($parameters->host, FILTER_VALIDATE_IP, FILTER_FLAG_IPV6)) { + return "$message [$parameters->scheme://[$parameters->host]:$parameters->port]"; + } + + return "$message [$parameters->scheme://$parameters->host:$parameters->port]"; + } + /** * Helper method to handle connection errors. * @@ -129,9 +152,7 @@ abstract class AbstractConnection implements NodeConnectionInterface protected function onConnectionError($message, $code = null) { CommunicationException::handle( - new ConnectionException( - $this, "$message [{$this->parameters->scheme}://{$this->getIdentifier()}]", $code - ) + new ConnectionException($this, static::createExceptionMessage($message), $code) ); } @@ -143,9 +164,7 @@ abstract class AbstractConnection implements NodeConnectionInterface protected function onProtocolError($message) { CommunicationException::handle( - new ProtocolException( - $this, "$message [{$this->parameters->scheme}://{$this->getIdentifier()}]" - ) + new ProtocolException($this, static::createExceptionMessage($message)) ); } diff --git a/src/Connection/Aggregate/RedisCluster.php b/src/Connection/Aggregate/RedisCluster.php index d790f751..ed27f0c9 100644 --- a/src/Connection/Aggregate/RedisCluster.php +++ b/src/Connection/Aggregate/RedisCluster.php @@ -275,11 +275,11 @@ class RedisCluster implements ClusterInterface, \IteratorAggregate, \Countable */ protected function createConnection($connectionID) { - $host = explode(':', $connectionID, 2); + $separator = strrpos($connectionID, ':'); return $this->connections->create(array( - 'host' => $host[0], - 'port' => $host[1], + 'host' => substr($connectionID, 0, $separator), + 'port' => substr($connectionID, $separator + 1), )); } diff --git a/src/Connection/PhpiredisSocketConnection.php b/src/Connection/PhpiredisSocketConnection.php index d5cc94fa..f93af8a5 100644 --- a/src/Connection/PhpiredisSocketConnection.php +++ b/src/Connection/PhpiredisSocketConnection.php @@ -174,22 +174,54 @@ class PhpiredisSocketConnection extends AbstractConnection $this->onConnectionError(trim($errstr), $errno); } + /** + * Gets the address of an host from connection parameters. + * + * @param ParametersInterface $parameters Parameters used to initialize the connection. + * + * @return string + */ + protected static function getAddress(ParametersInterface $parameters) + { + if (filter_var($host = $parameters->host, FILTER_VALIDATE_IP)) { + return $host; + } + + if ($host === $address = gethostbyname($host)) { + return false; + } + + return $address; + } + /** * {@inheritdoc} */ protected function createResource() { - $isUnix = $this->parameters->scheme === 'unix'; - $domain = $isUnix ? AF_UNIX : AF_INET; - $protocol = $isUnix ? 0 : SOL_TCP; + $parameters = $this->parameters; - $socket = @call_user_func('socket_create', $domain, SOCK_STREAM, $protocol); + if ($parameters->scheme === 'unix') { + $address = $parameters->path; + $domain = AF_UNIX; + $protocol = 0; + } else { + if (false === $address = self::getAddress($parameters)) { + $this->onConnectionError("Cannot resolve the address of '$parameters->host'."); + } + + $domain = filter_var($address, FILTER_VALIDATE_IP, FILTER_FLAG_IPV6) ? AF_INET6 : AF_INET; + $protocol = SOL_TCP; + } + + $socket = @socket_create($domain, SOCK_STREAM, $protocol); if (!is_resource($socket)) { $this->emitSocketError(); } - $this->setSocketOptions($socket, $this->parameters); + $this->setSocketOptions($socket, $parameters); + $this->connectWithTimeout($socket, $address, $parameters); return $socket; } @@ -202,12 +234,14 @@ class PhpiredisSocketConnection extends AbstractConnection */ private function setSocketOptions($socket, ParametersInterface $parameters) { - if (!socket_set_option($socket, SOL_TCP, TCP_NODELAY, 1)) { - $this->emitSocketError(); - } + if ($parameters->scheme !== 'unix') { + if (!socket_set_option($socket, SOL_TCP, TCP_NODELAY, 1)) { + $this->emitSocketError(); + } - if (!socket_set_option($socket, SOL_SOCKET, SO_REUSEADDR, 1)) { - $this->emitSocketError(); + if (!socket_set_option($socket, SOL_SOCKET, SO_REUSEADDR, 1)) { + $this->emitSocketError(); + } } if (isset($parameters->read_write_timeout)) { @@ -230,50 +264,20 @@ class PhpiredisSocketConnection extends AbstractConnection } } - /** - * Gets the address from the connection parameters. - * - * @param ParametersInterface $parameters Parameters used to initialize the connection. - * - * @return string - */ - protected static function getAddress(ParametersInterface $parameters) - { - if ($parameters->scheme === 'unix') { - return $parameters->path; - } - - $host = $parameters->host; - - if (ip2long($host) === false) { - if (false === $addresses = gethostbynamel($host)) { - return false; - } - - return $addresses[array_rand($addresses)]; - } - - return $host; - } - /** * Opens the actual connection to the server with a timeout. * + * @param resource $socket Socket resource. + * @param string $address IP address (DNS-resolved from hostname) * @param ParametersInterface $parameters Parameters used to initialize the connection. * * @return string */ - private function connectWithTimeout(ParametersInterface $parameters) + private function connectWithTimeout($socket, $address, ParametersInterface $parameters) { - if (false === $host = self::getAddress($parameters)) { - $this->onConnectionError("Cannot resolve the address of '$parameters->host'."); - } - - $socket = $this->getResource(); - socket_set_nonblock($socket); - if (@socket_connect($socket, $host, (int) $parameters->port) === false) { + if (@socket_connect($socket, $address, (int) $parameters->port) === false) { $error = socket_last_error(); if ($error != SOCKET_EINPROGRESS && $error != SOCKET_EALREADY) { @@ -295,9 +299,11 @@ class PhpiredisSocketConnection extends AbstractConnection if ($selected === 2) { $this->onConnectionError('Connection refused.', SOCKET_ECONNREFUSED); } + if ($selected === 0) { $this->onConnectionError('Connection timed out.', SOCKET_ETIMEDOUT); } + if ($selected === false) { $this->emitSocketError(); } @@ -308,13 +314,9 @@ class PhpiredisSocketConnection extends AbstractConnection */ public function connect() { - if (parent::connect()) { - $this->connectWithTimeout($this->parameters); - - if ($this->initCommands) { - foreach ($this->initCommands as $command) { - $this->executeCommand($command); - } + if (parent::connect() && $this->initCommands) { + foreach ($this->initCommands as $command) { + $this->executeCommand($command); } } } diff --git a/src/Connection/StreamConnection.php b/src/Connection/StreamConnection.php index 61ad4698..5bac88bb 100644 --- a/src/Connection/StreamConnection.php +++ b/src/Connection/StreamConnection.php @@ -152,7 +152,7 @@ class StreamConnection extends AbstractConnection */ protected function tcpStreamInitializer(ParametersInterface $parameters) { - $address = "tcp://{$parameters->host}:{$parameters->port}"; + $address = "tcp://[$parameters->host]:$parameters->port"; $flags = STREAM_CLIENT_CONNECT; if (isset($parameters->async_connect) && $parameters->async_connect) { diff --git a/src/Connection/WebdisConnection.php b/src/Connection/WebdisConnection.php index ea557c93..9ae7a4b8 100644 --- a/src/Connection/WebdisConnection.php +++ b/src/Connection/WebdisConnection.php @@ -119,10 +119,14 @@ class WebdisConnection implements NodeConnectionInterface $parameters = $this->getParameters(); $timeout = (isset($parameters->timeout) ? (float) $parameters->timeout : 5.0) * 1000; + if (filter_var($host = $parameters->host, FILTER_VALIDATE_IP)) { + $host = "[$host]"; + } + $options = array( CURLOPT_FAILONERROR => true, CURLOPT_CONNECTTIMEOUT_MS => $timeout, - CURLOPT_URL => "{$parameters->scheme}://{$parameters->host}:{$parameters->port}", + CURLOPT_URL => "$parameters->scheme://$host:$parameters->port", CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1, CURLOPT_POST => true, CURLOPT_WRITEFUNCTION => array($this, 'feedReader'), diff --git a/tests/PHPUnit/PredisConnectionTestCase.php b/tests/PHPUnit/PredisConnectionTestCase.php index 7b13cf10..1120b58e 100644 --- a/tests/PHPUnit/PredisConnectionTestCase.php +++ b/tests/PHPUnit/PredisConnectionTestCase.php @@ -431,12 +431,45 @@ abstract class PredisConnectionTestCase extends PredisTestCase * @group connected * @group slow * @expectedException \Predis\Connection\ConnectionException + * @expectedExceptionMessageRegExp /.* \[tcp:\/\/169.254.10.10:6379\]/ */ public function testThrowsExceptionOnConnectionTimeout() { $connection = $this->createConnectionWithParams(array( 'host' => '169.254.10.10', - 'timeout' => 0.5, + 'timeout' => 0.1, + ), false); + + $connection->connect(); + } + + /** + * @group connected + * @group slow + * @expectedException \Predis\Connection\ConnectionException + * @expectedExceptionMessageRegExp /.* \[tcp:\/\/\[0:0:0:0:0:ffff:a9fe:a0a\]:6379\]/ + */ + public function testThrowsExceptionOnConnectionTimeoutIPv6() + { + $connection = $this->createConnectionWithParams(array( + 'host' => '0:0:0:0:0:ffff:a9fe:a0a', + 'timeout' => 0.1, + ), false); + + $connection->connect(); + } + + /** + * @group connected + * @group slow + * @expectedException \Predis\Connection\ConnectionException + * @expectedExceptionMessageRegExp /.* \[unix:\/tmp\/nonexistent\/redis\.sock]/ + */ + public function testThrowsExceptionOnUnixDomainSocketNotFound() + { + $connection = $this->createConnectionWithParams(array( + 'scheme' => 'unix', + 'path' => '/tmp/nonexistent/redis.sock', ), false); $connection->connect(); diff --git a/tests/Predis/Connection/Aggregate/RedisClusterTest.php b/tests/Predis/Connection/Aggregate/RedisClusterTest.php index b041760e..2c12a78c 100644 --- a/tests/Predis/Connection/Aggregate/RedisClusterTest.php +++ b/tests/Predis/Connection/Aggregate/RedisClusterTest.php @@ -632,6 +632,40 @@ class RedisClusterTest extends PredisTestCase $this->assertSame(3, count($cluster)); } + /** + * @group disconnected + */ + public function testParseIPv6AddresseAndPortPairInRedirectionPayload() + { + $movedResponse = new Response\Error('MOVED 1970 2001:db8:0:f101::2:6379'); + + $command = Profile\Factory::getDefault()->createCommand('get', array('node:1001')); + + $connection1 = $this->getMockConnection('tcp://[2001:db8:0:f101::1]:6379'); + $connection1->expects($this->once()) + ->method('executeCommand') + ->with($command) + ->will($this->returnValue($movedResponse)); + + $connection2 = $this->getMockConnection('tcp://[2001:db8:0:f101::2]:6379'); + $connection2->expects($this->once()) + ->method('executeCommand') + ->with($command) + ->will($this->returnValue('foobar')); + + $factory = $this->getMock('Predis\Connection\Factory'); + $factory->expects($this->once()) + ->method('create') + ->with(array('host' => '2001:db8:0:f101::2', 'port' => '6379')) + ->will($this->returnValue($connection2)); + + $cluster = new RedisCluster($factory); + $cluster->useClusterSlots(false); + $cluster->add($connection1); + + $cluster->executeCommand($command); + } + /** * @group disconnected */ diff --git a/tests/Predis/Connection/ParametersTest.php b/tests/Predis/Connection/ParametersTest.php index 0b19ddb1..a42f6823 100644 --- a/tests/Predis/Connection/ParametersTest.php +++ b/tests/Predis/Connection/ParametersTest.php @@ -280,15 +280,11 @@ class ParametersTest extends PredisTestCase */ public function testParsingURIWithEmbeddedIPV6AddressShouldStripBracketsFromHost() { - $uri = 'tcp://[::1]:7000'; + $expected = array('scheme' => 'tcp', 'host' => '::1', 'port' => 7000); + $this->assertSame($expected, Parameters::parse('tcp://[::1]:7000')); - $expected = array( - 'scheme' => 'tcp', - 'host' => '::1', - 'port' => 7000, - ); - - $this->assertSame($expected, Parameters::parse($uri)); + $expected = array('scheme' => 'tcp', 'host' => '2001:db8:0:f101::1', 'port' => 7000); + $this->assertSame($expected, Parameters::parse('tcp://[2001:db8:0:f101::1]:7000')); } /**