Fix read/write timeouts with PhpiredisStreamConnection.

This is a regression of commit 4db0017, using stream_socket_recvfrom()
bypasses stream wrappers thus read/write timeouts handled by PHP. The
only way to make them work again is to fetch the raw socket from the
stream resource and use socket_import_stream() which is unfortunately
a PHP 5.4+ function.

CHANGELOG.md

@@ -22,6 +22,11 @@ v0.8.7 (2014-xx-xx)
   aside from the different hashing function used to calculate distribution, is
   in how empty hash tags like {} are treated by redis-cluster.
 
+- __FIX__: the patch applied to fix #180 introduced a regression affecting read/
+  write timeouts in `Predis\Connection\PhpiredisStreamConnection`. Unfortunately
+  the only possible solution requires PHP 5.4+. On PHP 5.3, read/write timeouts
+  will be ignored from now on.
+
 
 v0.8.6 (2014-07-15)
 ================================================================================

lib/Predis/Connection/PhpiredisStreamConnection.php

@@ -96,6 +96,52 @@ class PhpiredisStreamConnection extends StreamConnection
         return parent::checkParameters($parameters);
     }
 
+    /**
+     * {@inheritdoc}
+     */
+    protected function tcpStreamInitializer(ConnectionParametersInterface $parameters)
+    {
+        $uri = "tcp://{$parameters->host}:{$parameters->port}";
+        $flags = STREAM_CLIENT_CONNECT;
+        $socket = null;
+
+        if (isset($parameters->async_connect) && $parameters->async_connect) {
+            $flags |= STREAM_CLIENT_ASYNC_CONNECT;
+        }
+
+        if (isset($parameters->persistent) && $parameters->persistent) {
+            $flags |= STREAM_CLIENT_PERSISTENT;
+            $uri .= strpos($path = $parameters->path, '/') === 0 ? $path : "/$path";
+        }
+
+        $resource = @stream_socket_client($uri, $errno, $errstr, $parameters->timeout, $flags);
+
+        if (!$resource) {
+            $this->onConnectionError(trim($errstr), $errno);
+        }
+
+        if (isset($parameters->read_write_timeout) && function_exists('socket_import_stream')) {
+            $rwtimeout = (float) $parameters->read_write_timeout;
+            $rwtimeout = $rwtimeout > 0 ? $rwtimeout : -1;
+
+            $timeout = array(
+                'sec'  => $timeoutSeconds = floor($rwtimeout),
+                'usec' => ($rwtimeout - $timeoutSeconds) * 1000000,
+            );
+
+            $socket = $socket ?: socket_import_stream($resource);
+            @socket_set_option($socket, SOL_SOCKET, SO_SNDTIMEO, $timeout);
+            @socket_set_option($socket, SOL_SOCKET, SO_RCVTIMEO, $timeout);
+        }
+
+        if (isset($parameters->tcp_nodelay) && function_exists('socket_import_stream')) {
+            $socket = $socket ?: socket_import_stream($resource);
+            socket_set_option($socket, SOL_TCP, TCP_NODELAY, (int) $parameters->tcp_nodelay);
+        }
+
+        return $resource;
+    }
+
     /**
      * Initializes the protocol reader resource.
      */

lib/Predis/Connection/StreamConnection.php

@@ -76,7 +76,7 @@ class StreamConnection extends AbstractConnection
      * @param  ConnectionParametersInterface $parameters Parameters used to initialize the connection.
      * @return resource
      */
-    private function tcpStreamInitializer(ConnectionParametersInterface $parameters)
+    protected function tcpStreamInitializer(ConnectionParametersInterface $parameters)
     {
         $uri = "tcp://{$parameters->host}:{$parameters->port}";
         $flags = STREAM_CLIENT_CONNECT;