Apply timeout correctly during socket checkout and TLS handshakes

Fixes #145.

.travis.yml

@@ -22,6 +22,7 @@ install:
   - chmod +x coverage/bin/phpcov
 
   - composer show
+  - cd test/tls && ./regenerate.sh && cd -
 
 script:
   - phpdbg -qrr vendor/bin/phpunit --verbose --coverage-php coverage/cov/main.cov

lib/DefaultClient.php

@@ -6,6 +6,7 @@
 use Amp\Artax\Cookie\CookieFormatException;
 use Amp\Artax\Cookie\CookieJar;
 use Amp\Artax\Cookie\NullCookieJar;
+use Amp\Artax\Internal\CombinedCancellationToken;
 use Amp\Artax\Internal\Parser;
 use Amp\Artax\Internal\PublicSuffixList;
 use Amp\Artax\Internal\RequestCycle;
@@ -14,6 +15,7 @@
 use Amp\ByteStream\Message;
 use Amp\ByteStream\ZlibInputStream;
 use Amp\CancellationToken;
+use Amp\CancelledException;
 use Amp\Deferred;
 use Amp\Delayed;
 use Amp\Dns\ResolutionException;
@@ -26,6 +28,7 @@
 use Amp\Socket\ClientTlsContext;
 use Amp\Socket\ConnectException;
 use Amp\Success;
+use Amp\TimeoutCancellationToken;
 use Amp\Uri\InvalidUriException;
 use Amp\Uri\Uri;
 use function Amp\asyncCall;
@@ -390,17 +393,41 @@ private function withCancellation(Promise $promise, CancellationToken $cancellat
     }
 
     private function doWrite(RequestCycle $requestCycle) {
+        $timeout = $requestCycle->options[self::OP_TRANSFER_TIMEOUT];
+        $timeoutToken = new NullCancellationToken;
+
+        if ($timeout > 0) {
+            $transferTimeoutWatcher = Loop::delay($timeout, function () use ($requestCycle, $timeout) {
+                $this->fail($requestCycle, new TimeoutException(
+                    sprintf('Allowed transfer timeout exceeded: %d ms', $timeout)
+                ));
+            });
+
+            $requestCycle->bodyDeferred->promise()->onResolve(static function () use ($transferTimeoutWatcher) {
+                Loop::cancel($transferTimeoutWatcher);
+            });
+
+            $timeoutToken = new TimeoutCancellationToken($timeout);
+        }
+
         $authority = $this->generateAuthorityFromUri($requestCycle->uri);
         $socketCheckoutUri = $requestCycle->uri->getScheme() . "://{$authority}";
+        $connectTimeoutToken = new CombinedCancellationToken($requestCycle->cancellation, $timeoutToken);
 
         try {
             /** @var ClientSocket $socket */
-            $socket = yield $this->socketPool->checkout($socketCheckoutUri, $requestCycle->cancellation);
+            $socket = yield $this->socketPool->checkout($socketCheckoutUri, $connectTimeoutToken);
             $requestCycle->socket = $socket;
         } catch (ResolutionException $dnsException) {
             throw new DnsException(\sprintf("Resolving the specified domain failed: '%s'", $requestCycle->uri->getHost()), 0, $dnsException);
         } catch (ConnectException $e) {
             throw new SocketException(\sprintf("Connection to '%s' failed", $authority), 0, $e);
+        } catch (CancelledException $e) {
+            // In case of a user cancellation request, throw the expected exception
+            $requestCycle->cancellation->throwIfRequested();
+
+            // Otherwise we ran into a timeout of our TimeoutCancellationToken
+            throw new SocketException(\sprintf("Connection to '%s' timed out", $authority), 0, $e);
         }
 
         $cancellation = $requestCycle->cancellation->subscribe(function ($error) use ($requestCycle) {
@@ -419,20 +446,6 @@ private function doWrite(RequestCycle $requestCycle) {
             // Collect this here, because it fails in case the remote closes the connection directly.
             $connectionInfo = $this->collectConnectionInfo($socket);
 
-            $timeout = $requestCycle->options[self::OP_TRANSFER_TIMEOUT];
-
-            if ($timeout > 0) {
-                $transferTimeoutWatcher = Loop::delay($timeout, function () use ($requestCycle, $timeout) {
-                    $this->fail($requestCycle, new TimeoutException(
-                        sprintf('Allowed transfer timeout exceeded: %d ms', $timeout)
-                    ));
-                });
-
-                $requestCycle->bodyDeferred->promise()->onResolve(static function () use ($transferTimeoutWatcher) {
-                    Loop::cancel($transferTimeoutWatcher);
-                });
-            }
-
             $rawHeaders = $this->generateRawRequestHeaders($requestCycle->request, $requestCycle->protocolVersion);
             yield $socket->write($rawHeaders);
 

lib/Internal/CombinedCancellationToken.php

@@ -0,0 +1,52 @@
+<?php
+
+namespace Amp\Artax\Internal;
+
+use Amp\CancellationToken;
+use Amp\CancellationTokenSource;
+
+/** @internal */
+class CombinedCancellationToken implements CancellationToken {
+    private $token;
+    private $tokens = [];
+
+    public function __construct(CancellationToken ...$tokens) {
+        $tokenSource = new CancellationTokenSource;
+        $this->token = $tokenSource->getToken();
+
+        foreach ($tokens as $token) {
+            $id = $token->subscribe(function ($error) use ($tokenSource) {
+                $tokenSource->cancel($error);
+            });
+
+            $this->tokens[] = [$token, $id];
+        }
+    }
+
+    public function __destruct() {
+        foreach ($this->tokens as list($token, $id)) {
+            /** @var CancellationToken $token */
+            $token->unsubscribe($id);
+        }
+    }
+
+    /** @inheritdoc */
+    public function subscribe(callable $callback): string {
+        return $this->token->subscribe($callback);
+    }
+
+    /** @inheritdoc */
+    public function unsubscribe(string $id) {
+        $this->token->unsubscribe($id);
+    }
+
+    /** @inheritdoc */
+    public function isRequested(): bool {
+        return $this->token->isRequested();
+    }
+
+    /** @inheritdoc */
+    public function throwIfRequested() {
+        $this->token->throwIfRequested();
+    }
+}

test/ClientHttpBinIntegrationTest.php

@@ -11,14 +11,12 @@
 use Amp\Artax\RequestBody;
 use Amp\Artax\Response;
 use Amp\Artax\SocketException;
-use Amp\Artax\TimeoutException;
 use Amp\Artax\TooManyRedirectsException;
 use Amp\ByteStream\InMemoryStream;
 use Amp\ByteStream\InputStream;
 use Amp\ByteStream\IteratorStream;
 use Amp\CancellationTokenSource;
 use Amp\CancelledException;
-use Amp\Loop;
 use Amp\Promise;
 use Amp\Socket;
 use Amp\Success;
@@ -142,40 +140,6 @@ public function testIncompleteHttpResponseWithoutChunkedEncodingAndWithoutConten
         }
     }
 
-    public function testTimeout() {
-        $client = new DefaultClient;
-        $client->setOption(Client::OP_TRANSFER_TIMEOUT, 500);
-
-        $server = Socket\listen("tcp://127.0.0.1:0");
-
-        asyncCall(function () use ($server) {
-            /** @var Socket\ClientSocket $client */
-            while ($client = yield $server->accept()) {
-                yield $client->write("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n.");
-
-                Loop::delay(3000, function () use ($client) {
-                    $client->close();
-                });
-            }
-        });
-
-        try {
-            $uri = "http://" . $server->getAddress() . "/";
-
-            $start = \microtime(true);
-            $promise = $client->request($uri);
-
-            /** @var Response $response */
-            $response = wait($promise);
-            $this->expectException(TimeoutException::class);
-            $this->expectExceptionMessage("Allowed transfer timeout exceeded: 500 ms");
-            wait($response->getBody());
-        } finally {
-            $this->assertLessThan(0.6, \microtime(true) - $start);
-            $server->close();
-        }
-    }
-
     public function testDefaultUserAgentSent() {
         $uri = 'http://httpbin.org/user-agent';
         $client = new DefaultClient;

test/TimeoutTest.php

@@ -0,0 +1,121 @@
+<?php
+
+namespace Amp\Test\Artax;
+
+use Amp\Artax\Client;
+use Amp\Artax\DefaultClient;
+use Amp\Artax\HttpSocketPool;
+use Amp\Artax\Response;
+use Amp\Artax\TimeoutException;
+use Amp\CancellationToken;
+use Amp\Deferred;
+use Amp\Loop;
+use Amp\PHPUnit\TestCase;
+use Amp\Promise;
+use Amp\Socket;
+use Amp\Socket\ClientSocket;
+use function Amp\asyncCall;
+use function Amp\Promise\wait;
+
+class TimeoutTest extends TestCase {
+    /** @var DefaultClient */
+    private $client;
+
+    public function setUp() {
+        $this->client = new DefaultClient;
+    }
+
+    public function testTimeoutDuringBody() {
+        $server = Socket\listen("tcp://127.0.0.1:0");
+
+        asyncCall(function () use ($server) {
+            /** @var Socket\ClientSocket $client */
+            while ($client = yield $server->accept()) {
+                yield $client->write("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n.");
+
+                Loop::delay(3000, function () use ($client) {
+                    $client->close();
+                });
+            }
+        });
+
+        try {
+            $uri = "http://" . $server->getAddress() . "/";
+
+            $start = \microtime(true);
+            $promise = $this->client->request($uri, [Client::OP_TRANSFER_TIMEOUT => 100]);
+
+            /** @var Response $response */
+            $response = wait($promise);
+            $this->expectException(TimeoutException::class);
+            $this->expectExceptionMessage("Allowed transfer timeout exceeded: 100 ms");
+            wait($response->getBody());
+        } finally {
+            $this->assertLessThan(0.6, \microtime(true) - $start);
+            $server->close();
+        }
+    }
+
+    public function testTimeoutDuringConnect() {
+        Loop::repeat(1000, function () { /* dummy watcher, because socket pool doesn't do anything */ });
+
+        $this->client = new DefaultClient(null, new HttpSocketPool(new class implements Socket\SocketPool {
+            public function checkout(string $uri, CancellationToken $token = null): Promise {
+                $deferred = new Deferred;
+
+                if ($token) {
+                    $token->subscribe(function ($error) use ($deferred) {
+                        $deferred->fail($error);
+                    });
+                }
+
+                return $deferred->promise(); // never resolve
+            }
+
+            public function checkin(ClientSocket $socket) {
+                // ignore
+            }
+
+            public function clear(ClientSocket $socket) {
+                // ignore
+            }
+        }));
+
+        $this->expectException(TimeoutException::class);
+        $this->expectExceptionMessage("Allowed transfer timeout exceeded: 100 ms");
+
+        $this->assertRunTimeLessThan(function () {
+            wait($this->client->request("http://localhost:1337/", [Client::OP_TRANSFER_TIMEOUT => 100]));
+        }, 600);
+    }
+
+    public function testTimeoutDuringTlsEnable() {
+        $tlsContext = (new Socket\ServerTlsContext)
+            ->withDefaultCertificate(new Socket\Certificate(__DIR__ . "/tls/amphp.org.pem"));
+
+        $server = Socket\listen("tcp://127.0.0.1:0", null, $tlsContext);
+
+        asyncCall(function () use ($server) {
+            /** @var Socket\ClientSocket $client */
+            while ($client = yield $server->accept()) {
+                Loop::delay(3000, function () use ($client) {
+                    $client->close();
+                });
+            }
+        });
+
+        try {
+            $uri = "http://" . $server->getAddress() . "/";
+
+            $start = \microtime(true);
+            $promise = $this->client->request($uri, [Client::OP_TRANSFER_TIMEOUT => 100]);
+
+            $this->expectException(TimeoutException::class);
+            $this->expectExceptionMessage("Allowed transfer timeout exceeded: 100 ms");
+            wait($promise);
+        } finally {
+            $this->assertLessThan(0.6, \microtime(true) - $start);
+            $server->close();
+        }
+    }
+}

test/tls/amphp.org.crt

@@ -0,0 +1,19 @@
+-----BEGIN CERTIFICATE-----
+MIIDAjCCAeoCCQD9ibkUCk3KajANBgkqhkiG9w0BAQsFADBDMQswCQYDVQQGEwJE
+RTEQMA4GA1UECAwHR2VybWFueTEOMAwGA1UECgwFYW1waHAxEjAQBgNVBAMMCWFt
+cGhwLm9yZzAeFw0xNzEwMjMxMjI2NDFaFw0xODEwMjMxMjI2NDFaMEMxCzAJBgNV
+BAYTAkRFMRAwDgYDVQQIDAdHZXJtYW55MQ4wDAYDVQQKDAVhbXBocDESMBAGA1UE
+AwwJYW1waHAub3JnMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAvMq/
+2ilgvZokr49wMYVpN/0hpd/992vMtVJ6sMr8LXHtP2iYCcS3oP1EfOUSKphEUBfS
+NbfkmQVkOysEo4+cadahqnBd7Ml++kBvwhHyP14uVp87jyZtCR/WL5BTyByweN07
+sfdVyXSY9HsLBSr5vBtcgg9QHDiZfhouJDaWUwN80J6nK22IRT35i8bjORuqXm+4
+XNEL1zO5mAUn3nJyaKs5gnwwwgJvGc46okJtoH/uWjFrI60LwchC9+yRgaTszznJ
+P7QcUtVczWdIW5dik0KesV/Wv18tCN6S4abnZVxc1tuShV+rVXedXVfDy291ws1w
+dKMNidp3uaIKnQof3QIDAQABMA0GCSqGSIb3DQEBCwUAA4IBAQBC9XCwPLOITsv/
+M9LgcY0DZBMaoyJNhum7pxRXon7DG9k6awntyDh2h5b6oEGVFwBNMJYa0VYj6DZ+
+TMT5Rqf7y2lgx/jNQUxQCzIRCerQK1bFWqYCQ8ZQbFpJTMmcJFrv++ST8R9Lz+11
+eDinc304QFYo44fRZJcxrbCWIYV2kCR+2ivXxhXr2WL7egy8/XwVQawU0FyOB3ym
+mrIF5e8wkbst2FE7GGx87aUKqF4TubjiLg1RzuQwjEzG5C/hCXeSGhq3Zx9QfOXg
+xbFZwWxj9IxpuvM9gueIcFheFS7aquIjWe69jCNe46JgzjpO5pN38bC3a0t4ntVK
+OTwPPXAj
+-----END CERTIFICATE-----

test/tls/amphp.org.key

@@ -0,0 +1,27 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIEpAIBAAKCAQEAvMq/2ilgvZokr49wMYVpN/0hpd/992vMtVJ6sMr8LXHtP2iY
+CcS3oP1EfOUSKphEUBfSNbfkmQVkOysEo4+cadahqnBd7Ml++kBvwhHyP14uVp87
+jyZtCR/WL5BTyByweN07sfdVyXSY9HsLBSr5vBtcgg9QHDiZfhouJDaWUwN80J6n
+K22IRT35i8bjORuqXm+4XNEL1zO5mAUn3nJyaKs5gnwwwgJvGc46okJtoH/uWjFr
+I60LwchC9+yRgaTszznJP7QcUtVczWdIW5dik0KesV/Wv18tCN6S4abnZVxc1tuS
+hV+rVXedXVfDy291ws1wdKMNidp3uaIKnQof3QIDAQABAoIBAC2n9pei1BNmOKMA
+VEiVk+mHXODJd5ijSEE9bhBdNnkjCRYBjGsoWKQlO6/ckfUdF8Aq0ppNG4pqBGBO
+ufN4IoJx9mzKedxuqjeI574OKwBqHVizb2riBxJi0aB1Dd3iGkdQcURLpUJv4SGW
+tAaO6xAzqb6GR4Bbq7sUR75YOShfz6ycP4Ri3xChkwtQ9mAjMjK2bNQweO6/5hGP
+c/GI889uSWGTA4lLrJZ/SIOUFFYSUzvvzEQ03pm3h4luqPWcCmPI6nUT+ytSdntK
+U4NtE/69Vgi0DH4f/x0gv6gGSwQ2f4ncA8/08rbfO0Z7gm/B/qB53Ek/G5EiAdCB
+MVT2mEECgYEA8waSN7WOdcEUE8pWH3tnLy9lO1i2Tpx/R53smJ6xgjsMyz3oxoMI
+UkQSL9bWy0jOVk3EDUaKxvNuME80WIcVZDfcqyEq+3vaQgfc2Lz2sQ8DGJN6pWml
+u2xZNrQa6fIvKbOzpLhINgCW6yFc8Yt/CCRxuv0TqzdXJ3dRmcGaXbECgYEAxt73
+nOMSXXzeW5blFVNMtmZq6N9BgMiGp+9QV4oa4rNskjm3C6b+zHlHsisdMwR7WV9z
+Qk7DmjoSA0QzJwdvJd7RM4maZCiCF/szyAajzdWzFQRK1jt4AaQElIVzAGXLRYNV
+tdE24VBCTZOw5JQWHyy77spRnsLHfmoY+dwRU+0CgYBdLxLXq5TxT9RL9lRpg96/
+t/Of+tcEc3tWUZaLuqWjotK4B4f+vfVt3c3/a+g2UDdV3kGjOX4y5NpGSchvfXRV
+VhoVwAUEIqxAj1U+Ac6xe2kbfkkTrduvbs8Sa6K6O3OcENx793Ewy68Sf1ts/qj5
+zUeShEGLcA/KATVXuGVhMQKBgQCT/3K+qq4MLNU9y/oH+MnqJCYDz8HrmU+8wnhD
+7V88vtJZr+HtJgYRHWCh0zwTr44sYMBh5EPlDrpA/AwlS38H6948QHdJemv2dNeA
+UJCuqg5v4ZXgALm6XtmZvgZrkVXJEYhDmsMAwr3VBchrE1mzMZNaje5X/BSCm2qi
+54dtsQKBgQDQfQmh4FiGsjbXTd9ECUzdmB1Ts/1KubP4lvF4jxiSLXsQrgKRF2Vn
+nTWySTQhVjbpWZp5iDkXH9i6V+85wPyQPPNJS6abPFqLjFL1hbkEDQTGfikz4iuI
+ttDxbTSkle8ujWv1uU1cs7gMf9Ytk2K7PpWzKdiXowNkyATLje0TGQ==
+-----END RSA PRIVATE KEY-----