Merge pull request #16 from clue/react04

Support React v0.4 (while preserving BC)

Author: clue

composer.json

@@ -15,12 +15,12 @@
     },
     "require": {
         "php": ">=5.3",
-        "react/event-loop": "0.3.*",
-        "react/socket-client": "0.3.*",
-        "react/socket": "0.3.*",
-        "react/dns": "0.3.*",
-        "react/stream": "0.3.*",
-        "react/promise": "~1.0",
-        "evenement/evenement": "~1.0"
+        "react/event-loop": "0.3.*|0.4.*",
+        "react/socket-client": "0.3.*|0.4.*",
+        "react/socket": "0.3.*|0.4.*",
+        "react/dns": "0.3.*|0.4.*",
+        "react/stream": "0.3.*|0.4.*",
+        "react/promise": "~1.0|~2.0",
+        "evenement/evenement": "~1.0|~2.0"
     }
 }

composer.lock

@@ -2,7 +2,6 @@
 
 namespace Clue\React\Socks;
 
-use React\Promise\When;
 use React\Promise\Deferred;
 use React\Dns\Resolver\Factory as DnsFactory;
 use React\Dns\Resolver\Resolver;
@@ -133,7 +132,9 @@ public function createConnector()
     public function getConnection($host, $port)
     {
         if (strlen($host) > 255 || $port > 65535 || $port < 0) {
-            return When::reject(new InvalidArgumentException('Invalid target specified'));
+            $deferred = new Deferred();
+            $deferred->reject(new InvalidArgumentException('Invalid target specified'));
+            return $deferred->promise();
         }
         $deferred = new Deferred();
 
@@ -155,40 +156,43 @@ public function getConnection($host, $port)
 
         $loop = $this->loop;
         $that = $this;
-        When::all(
-            array(
-                $this->connector->create($this->socksHost, $this->socksPort)->then(
-                    null,
-                    function ($error) {
-                        throw new Exception('Unable to connect to socks server', 0, $error);
-                    }
-                ),
-                $this->resolve($host)->then(
-                    null,
+
+        // simultaneously start TCP connection and DNS resolution
+        $connecting = $this->connector->create($this->socksHost, $this->socksPort);
+        $resolving = $this->resolve($host);
+
+        $deferred->resolve($connecting->then(
+            function (Stream $stream) use ($that, $resolving, $loop, $timerTimeout, $protocolVersion, $auth, $timestampTimeout, $port) {
+                // connection established, wait for DNS resolver
+                return $resolving->then(
+                    function ($host) use ($stream, $port, $timestampTimeout, $that, $loop, $timerTimeout, $protocolVersion, $auth) {
+                        // DNS resolver completed => cancel timeout
+                        $loop->cancelTimer($timerTimeout);
+
+                        $timeout = max($timestampTimeout - microtime(true), 0.1);
+                        return $that->handleConnectedSocks($stream, $host, $port, $timeout, $protocolVersion, $auth);
+                    },
                     function ($error) {
                         throw new Exception('Unable to resolve remote hostname', 0, $error);
                     }
-                )
-            ),
-            function ($fulfilled) use ($deferred, $port, $timestampTimeout, $that, $loop, $timerTimeout, $protocolVersion, $auth) {
-                $loop->cancelTimer($timerTimeout);
-
-                $timeout = max($timestampTimeout - microtime(true), 0.1);
-                $deferred->resolve($that->handleConnectedSocks($fulfilled[0], $fulfilled[1], $port, $timeout, $protocolVersion, $auth));
+                );
             },
-            function ($error) use ($deferred, $loop, $timerTimeout) {
+            function ($error) use ($loop, $timerTimeout) {
                 $loop->cancelTimer($timerTimeout);
-                $deferred->reject(new Exception('Unable to connect to socks server', 0, $error));
+                throw new Exception('Unable to connect to socks server', 0, $error);
             }
-        );
+        ));
+
         return $deferred->promise();
     }
 
     private function resolve($host)
     {
         // return if it's already an IP or we want to resolve remotely (socks 4 only supports resolving locally)
         if (false !== filter_var($host, FILTER_VALIDATE_IP) || ($this->protocolVersion !== '4' && !$this->resolveLocal)) {
-            return When::resolve($host);
+            $deferred = new Deferred();
+            $deferred->resolve($host);
+            return $deferred->promise();
         }
 
         return $this->resolver->resolve($host);
@@ -197,10 +201,9 @@ private function resolve($host)
     public function handleConnectedSocks(Stream $stream, $host, $port, $timeout, $protocolVersion, $auth=null)
     {
         $deferred = new Deferred();
-        $resolver = $deferred->resolver();
 
-        $timerTimeout = $this->loop->addTimer($timeout, function () use ($resolver) {
-            $resolver->reject(new Exception('Timeout while establishing socks session'));
+        $timerTimeout = $this->loop->addTimer($timeout, function () use ($deferred) {
+            $deferred->reject(new Exception('Timeout while establishing socks session'));
         });
 
         $reader = new StreamReader($stream);
@@ -211,14 +214,14 @@ public function handleConnectedSocks(Stream $stream, $host, $port, $timeout, $pr
         } else {
             $promise = $this->handleSocks4($stream, $host, $port, $reader);
         }
-        $promise->then(function () use ($resolver, $stream) {
-            $resolver->resolve($stream);
-        }, function($error) use ($resolver) {
-            $resolver->reject(new Exception('Unable to communicate...', 0, $error));
+        $promise->then(function () use ($deferred, $stream) {
+            $deferred->resolve($stream);
+        }, function($error) use ($deferred) {
+            $deferred->reject(new Exception('Unable to communicate...', 0, $error));
         });
 
         $loop = $this->loop;
-        $deferred->then(
+        $deferred->promise()->then(
             function (Stream $stream) use ($timerTimeout, $loop, $reader) {
                 $loop->cancelTimer($timerTimeout);
                 $stream->removeAllListeners('end');
@@ -237,8 +240,8 @@ function ($error) use ($stream, $timerTimeout, $loop, $reader) {
             }
         );
 
-        $stream->on('end', function (Stream $stream) use ($resolver) {
-            $resolver->reject(new Exception('Premature end while establishing socks session'));
+        $stream->on('end', function (Stream $stream) use ($deferred) {
+            $deferred->reject(new Exception('Premature end while establishing socks session'));
         });
 
         return $deferred->promise();

src/Client.php

@@ -4,7 +4,6 @@
 
 use Evenement\EventEmitter;
 use React\Socket\ServerInterface;
-use React\Promise\When;
 use React\Promise\PromiseInterface;
 use React\Stream\Stream;
 use React\Dns\Resolver\Factory as DnsFactory;
@@ -15,6 +14,7 @@
 use \UnexpectedValueException;
 use \InvalidArgumentException;
 use \Exception;
+use React\Promise\Deferred;
 
 class Server extends EventEmitter
 {
@@ -73,7 +73,9 @@ public function setAuth($auth)
             if ($ret instanceof PromiseInterface) {
                 return $ret;
             }
-            return $ret ? When::resolve() : When::reject();
+            $deferred = new Deferred();
+            $ret ? $deferred->resolve() : $deferred->reject();
+            return $deferred->promise();
         };
     }