Add TimeoutConnector decorator

README.md

@@ -67,6 +67,20 @@ $secureConnector->create('www.google.com', 443)->then(function (React\Stream\Str
 $loop->run();
 ```
 
+### Connection timeouts
+
+The `TimeoutConnector` class decorates any given `Connector` instance.
+It provides the same `create()` method, but will automatically reject the
+underlying connection attempt if it takes too long.
+
+```php
+$timeoutConnector = new React\SocketClient\TimeoutConnector($connector, 3.0, $loop);
+
+$timeoutConnector->create('google.com', 80)->then(function (React\Stream\Stream $stream) {
+    // connection succeeded within 3.0 seconds
+});
+```
+
 ### Unix domain sockets
 
 Similarly, the `UnixConnector` class can be used to connect to Unix domain socket (UDS)

composer.json

@@ -8,7 +8,8 @@
         "react/dns": "0.4.*",
         "react/event-loop": "0.4.*",
         "react/stream": "0.4.*",
-        "react/promise": "~2.0"
+        "react/promise": "~2.0",
+        "react/promise-timer": "~1.0"
     },
     "autoload": {
         "psr-4": {

src/TimeoutConnector.php

@@ -0,0 +1,26 @@
+<?php
+
+namespace React\SocketClient;
+
+use React\SocketClient\ConnectorInterface;
+use React\EventLoop\LoopInterface;
+use React\Promise\Timer;
+
+class TimeoutConnector implements ConnectorInterface
+{
+    private $connector;
+    private $timeout;
+    private $loop;
+
+    public function __construct(ConnectorInterface $connector, $timeout, LoopInterface $loop)
+    {
+        $this->connector = $connector;
+        $this->timeout = $timeout;
+        $this->loop = $loop;
+    }
+
+    public function create($host, $port)
+    {
+        return Timer\timeout($this->connector->create($host, $port), $this->timeout, $this->loop);
+    }
+}

tests/TimeoutConnectorTest.php

@@ -0,0 +1,86 @@
+<?php
+
+namespace React\Tests\SocketClient;
+
+use React\SocketClient\TimeoutConnector;
+use React\Promise;
+use React\EventLoop\Factory;
+
+class TimeoutConnectorTest extends TestCase
+{
+    public function testRejectsOnTimeout()
+    {
+        $promise = new Promise\Promise(function () { });
+
+        $connector = $this->getMock('React\SocketClient\ConnectorInterface');
+        $connector->expects($this->once())->method('create')->with('google.com', 80)->will($this->returnValue($promise));
+
+        $loop = Factory::create();
+
+        $timeout = new TimeoutConnector($connector, 0.01, $loop);
+
+        $timeout->create('google.com', 80)->then(
+            $this->expectCallableNever(),
+            $this->expectCallableOnce()
+        );
+
+        $loop->run();
+    }
+
+    public function testRejectsWhenConnectorRejects()
+    {
+        $promise = Promise\reject();
+
+        $connector = $this->getMock('React\SocketClient\ConnectorInterface');
+        $connector->expects($this->once())->method('create')->with('google.com', 80)->will($this->returnValue($promise));
+
+        $loop = Factory::create();
+
+        $timeout = new TimeoutConnector($connector, 5.0, $loop);
+
+        $timeout->create('google.com', 80)->then(
+            $this->expectCallableNever(),
+            $this->expectCallableOnce()
+        );
+
+        $loop->run();
+    }
+
+    public function testResolvesWhenConnectorResolves()
+    {
+        $promise = Promise\resolve();
+
+        $connector = $this->getMock('React\SocketClient\ConnectorInterface');
+        $connector->expects($this->once())->method('create')->with('google.com', 80)->will($this->returnValue($promise));
+
+        $loop = Factory::create();
+
+        $timeout = new TimeoutConnector($connector, 5.0, $loop);
+
+        $timeout->create('google.com', 80)->then(
+            $this->expectCallableOnce(),
+            $this->expectCallableNever()
+        );
+
+        $loop->run();
+    }
+
+    public function testRejectsAndCancelsPendingPromiseOnTimeout()
+    {
+        $promise = new Promise\Promise(function () { }, $this->expectCallableOnce());
+
+        $connector = $this->getMock('React\SocketClient\ConnectorInterface');
+        $connector->expects($this->once())->method('create')->with('google.com', 80)->will($this->returnValue($promise));
+
+        $loop = Factory::create();
+
+        $timeout = new TimeoutConnector($connector, 0.01, $loop);
+
+        $timeout->create('google.com', 80)->then(
+            $this->expectCallableNever(),
+            $this->expectCallableOnce()
+        );
+
+        $loop->run();
+    }
+}