Add request maximum timeout

README.md

@@ -15,6 +15,7 @@ Event-driven, streaming HTTP client for [ReactPHP](https://reactphp.org).
 
 * [Basic usage](#basic-usage)
   * [Client](#client)
+  * [Request Timeouts](#request-timeouts)
   * [Example](#example)
 * [Advanced usage](#advanced-usage)
   * [Unix domain sockets](#unix-domain-sockets)
@@ -112,6 +113,19 @@ Interesting events emitted by Response:
   preceeded by an `error` event.
   This will also be forwarded to the request and emit a `close` event there.
 
+### Request Timeouts
+
+Each request can have their own separate maximum timeout
+(connection timeout is implemented in the [`react/socket`](https://github.com/reactphp/socket) component).
+The maximum timeout is measured from the start of the request until the end of the response (completely received).
+
+The maximum time for a request to take can be set for each `Request` using their `setTimeout` method.
+The method takes a single argument indicating the timeout in seconds (as integer or float).
+
+The timeout has to be set before the request starts.
+Once the timeout is reached, a `React\HttpClient\TimeoutException` will be emitted as `error` event.
+The stream will be forcefully closed.
+
 ### Example
 
 ```php
@@ -129,9 +143,13 @@ $request->on('response', function ($response) {
         echo 'DONE';
     });
 });
-$request->on('error', function (\Exception $e) {
+$request->on('error', function (Exception $e) {
     echo $e;
 });
+
+// allow the request to take 30 seconds
+$request->setTimeout(30.0); 
+
 $request->end();
 $loop->run();
 ```

src/Client.php

@@ -8,6 +8,7 @@
 
 class Client
 {
+    private $loop;
     private $connector;
 
     public function __construct(LoopInterface $loop, ConnectorInterface $connector = null)
@@ -16,13 +17,17 @@ public function __construct(LoopInterface $loop, ConnectorInterface $connector =
             $connector = new Connector($loop);
         }
 
+        $this->loop = $loop;
         $this->connector = $connector;
     }
 
     public function request($method, $url, array $headers = array(), $protocolVersion = '1.0')
     {
         $requestData = new RequestData($method, $url, $headers, $protocolVersion);
 
-        return new Request($this->connector, $requestData);
+        $request = new Request($this->connector, $requestData);
+        $request->setLoop($this->loop);
+
+        return $request;
     }
 }

src/Request.php

@@ -3,6 +3,7 @@
 namespace React\HttpClient;
 
 use Evenement\EventEmitter;
+use React\EventLoop\LoopInterface;
 use React\Promise;
 use React\Socket\ConnectionInterface;
 use React\Socket\ConnectorInterface;
@@ -22,6 +23,7 @@ class Request extends EventEmitter implements WritableStreamInterface
     const STATE_HEAD_WRITTEN = 2;
     const STATE_END = 3;
 
+    private $loop;
     private $connector;
     private $requestData;
 
@@ -32,13 +34,33 @@ class Request extends EventEmitter implements WritableStreamInterface
     private $ended = false;
 
     private $pendingWrites = '';
+    private $timeout = 0.0;
+    private $timeoutTimer;
 
     public function __construct(ConnectorInterface $connector, RequestData $requestData)
     {
         $this->connector = $connector;
         $this->requestData = $requestData;
     }
 
+    // non-BC way to add the loop to the request
+
+    /** @internal */
+    public function setLoop(LoopInterface $loop)
+    {
+        $this->loop = $loop;
+    }
+
+    // non-BC way to add a timeout
+
+    public function setTimeout($timeout)
+    {
+        $this->timeout = $timeout;
+    }
+
+    // we may want to merge these two methods into the constructor
+    // in the next major version
+
     public function isWritable()
     {
         return self::STATE_END > $this->state && !$this->ended;
@@ -47,12 +69,19 @@ public function isWritable()
     private function writeHead()
     {
         $this->state = self::STATE_WRITING_HEAD;
+        $that = $this;
+        $timeout = $this->timeout;
+
+        if ($timeout > 0.0) {
+            $this->timeoutTimer = $this->loop->addTimer($this->timeout, function () use ($that, $timeout) {
+                $that->closeError(new TimeoutException('The request took longer than expected ('.$timeout.' seconds)'));
+            });
+        }
 
         $requestData = $this->requestData;
         $streamRef = &$this->stream;
         $stateRef = &$this->state;
         $pendingWrites = &$this->pendingWrites;
-        $that = $this;
 
         $promise = $this->connect();
         $promise->then(
@@ -210,6 +239,11 @@ public function close()
             return;
         }
 
+        if ($this->timeoutTimer !== null) {
+            $this->loop->cancelTimer($this->timeoutTimer);
+            $this->timeoutTimer = null;
+        }
+
         $this->state = self::STATE_END;
         $this->pendingWrites = '';
 

src/TimeoutException.php

@@ -0,0 +1,10 @@
+<?php
+
+
+namespace React\HttpClient;
+
+use RuntimeException;
+
+class TimeoutException extends RuntimeException {
+
+}

tests/RequestTest.php

@@ -2,8 +2,11 @@
 
 namespace React\Tests\HttpClient;
 
+use Clue\React\Block;
+use React\EventLoop\Factory;
 use React\HttpClient\Request;
 use React\HttpClient\RequestData;
+use React\HttpClient\TimeoutException;
 use React\Stream\Stream;
 use React\Stream\DuplexResourceStream;
 use React\Promise\RejectedPromise;
@@ -708,4 +711,85 @@ public function chunkedStreamDecoder()
         $request->handleData("\n3\t\nody\r\n0\t\n\r\n");
 
     }
+
+    /** @test */
+    public function timeoutRequest()
+    {
+        $requestData = new RequestData('GET', 'http://www.example.com');
+        $request = new Request($this->connector, $requestData);
+
+        $loop = Factory::create();
+        $request->setLoop($loop);
+        $request->setTimeout(2.0);
+
+        $deferred = new Deferred();
+
+        $request->on('end', function () use ($deferred) {
+            $deferred->resolve();
+        });
+
+        $request->on('error', function ($e) use ($deferred) {
+            $deferred->reject($e);
+        });
+
+        $this->successfulConnectionMock();
+
+        $request->end();
+
+        $this->stream->expects($this->once())
+            ->method('emit')
+            ->with('data', array("1\r\nb\r"));
+
+        $loop->addTimer(1.0, function () use ($request) {
+            $request->handleData("HTTP/1.0 200 OK\r\n");
+            $request->handleData("Transfer-Encoding: chunked\r\n");
+            $request->handleData("\r\n1\r\nb\r");
+            $request->handleData("\n3\t\nody\r\n0\t\n\r\n");
+            $request->emit('end');
+            $request->emit('close');
+        });
+
+        Block\await($deferred->promise(), $loop, 5.0);
+    }
+
+    /**
+     * @test
+     * @expectedException \React\HttpClient\TimeoutException
+     */
+    public function timeoutRequestMissingEndTimingOut()
+    {
+        $requestData = new RequestData('GET', 'http://www.example.com');
+        $request = new Request($this->connector, $requestData);
+
+        $loop = Factory::create();
+        $request->setLoop($loop);
+        $request->setTimeout(2.0);
+
+        $deferred = new Deferred();
+
+        $request->on('end', function () use ($deferred) {
+            $deferred->resolve();
+        });
+
+        $request->on('error', function ($e) use ($deferred) {
+            $deferred->reject($e);
+        });
+
+        $this->successfulConnectionMock();
+
+        $request->end();
+
+        $this->stream->expects($this->once())
+            ->method('emit')
+            ->with('data', array("1\r\nb\r"));
+
+        $loop->addTimer(1.0, function () use ($request) {
+            $request->handleData("HTTP/1.0 200 OK\r\n");
+            $request->handleData("Transfer-Encoding: chunked\r\n");
+            $request->handleData("\r\n1\r\nb\r");
+            $request->handleData("\n3\t\nody\r\n0\t\n\r\n");
+        });
+
+        Block\await($deferred->promise(), $loop, 5.0);
+    }
 }