Add psalm and fix found issues

.travis.yml

@@ -10,7 +10,6 @@ php:
 
 matrix:
   allow_failures:
-    - php: 7.4
     - php: nightly
   fast_finish: true
 
@@ -36,10 +35,9 @@ install:
 
 script:
   - vendor/bin/phpunit --coverage-text --verbose --coverage-clover build/logs/clover.xml;
-
   - php composer-require-checker.phar check composer.json --config-file $PWD/composer-require-check.json
-
   - PHP_CS_FIXER_IGNORE_ENV=1 php vendor/bin/php-cs-fixer --diff --dry-run -v fix
+  - vendor/bin/psalm
 
 after_script:
   - travis_retry php php-coveralls.phar -v

composer.json

@@ -45,7 +45,8 @@
         "amphp/http-server": "^2-rc4",
         "kelunik/link-header-rfc5988": "^1.0",
         "clue/socks-react": "^1.0",
-        "amphp/react-adapter": "^2.1"
+        "amphp/react-adapter": "^2.1",
+        "vimeo/psalm": "^3.9@dev"
     },
     "suggest": {
         "ext-zlib": "Allows using compression for response bodies.",

examples/.helper/functions.php

@@ -11,7 +11,7 @@ function dumpRequestTrace(Request $request): void
     \printf(
         "%s %s HTTP/%s\r\n",
         $request->getMethod(),
-        $request->getUri(),
+        (string) $request->getUri(),
         \implode('+', $request->getProtocolVersions())
     );
 

examples/concurrency/1-concurrent-fetch.php

@@ -8,7 +8,7 @@
 
 require __DIR__ . '/../.helper/functions.php';
 
-Loop::run(static function () use ($argv) {
+Loop::run(static function () use ($argv): \Generator {
     $uris = [
         "https://google.com/",
         "https://github.com/",
@@ -18,7 +18,7 @@
     // Instantiate the HTTP client
     $client = HttpClientBuilder::buildDefault();
 
-    $requestHandler = static function (string $uri) use ($client) {
+    $requestHandler = static function (string $uri) use ($client): \Generator {
         /** @var Response $response */
         $response = yield $client->request(new Request($uri));
 

examples/proxy/socks4.php

@@ -32,6 +32,8 @@ public function connect(
                 ?ConnectContext $context = null,
                 ?CancellationToken $token = null
             ): Promise {
+                $context = $context ?? (new ConnectContext);
+
                 return call(static function () use ($uri, $context) {
                     $options = $context->toStreamContextArray();
 
@@ -53,12 +55,15 @@ public function connect(
 
                     /** @var Connection $connection */
                     $connection = yield $connector->connect($records[0]->getValue() . ':' . $socketAddress->getPort());
+                    /** @psalm-suppress InternalMethod */
                     $connection->pause();
 
                     if ($context->getTlsContext()) {
+                        /** @psalm-suppress InternalProperty */
                         \stream_context_set_option($connection->stream, $options);
                     }
 
+                    /** @psalm-suppress InternalProperty */
                     return ResourceSocket::fromClientSocket($connection->stream);
                 });
             }

examples/proxy/socks5.php

@@ -28,6 +28,8 @@ public function connect(
                 ?ConnectContext $context = null,
                 ?CancellationToken $token = null
             ): Promise {
+                $context = $context ?? new ConnectContext;
+
                 return call(static function () use ($uri, $context) {
                     $options = $context->toStreamContextArray();
 
@@ -41,12 +43,15 @@ public function connect(
 
                     /** @var Connection $connection */
                     $connection = yield $connector->connect($uri);
+                    /** @psalm-suppress InternalMethod */
                     $connection->pause();
 
                     if ($context->getTlsContext()) {
+                        /** @psalm-suppress InternalProperty */
                         \stream_context_set_option($connection->stream, $options);
                     }
 
+                    /** @psalm-suppress InternalProperty */
                     return ResourceSocket::fromClientSocket($connection->stream);
                 });
             }

examples/streaming/1-large-response.php

@@ -12,15 +12,15 @@
 require __DIR__ . '/../.helper/functions.php';
 
 // https://stackoverflow.com/a/2510540/2373138
-function formatBytes(int $size, int $precision = 2)
+function formatBytes(int $size, int $precision = 2): string
 {
     $base = \log($size, 1024);
     $suffixes = ['bytes', 'kB', 'MB', 'GB', 'TB'];
 
     return \round(1024 ** ($base - \floor($base)), $precision) . ' ' . $suffixes[(int) $base];
 }
 
-Loop::run(static function () {
+Loop::run(static function (): \Generator {
     try {
         $start = getCurrentTime();
 
@@ -47,7 +47,7 @@ function formatBytes(int $size, int $precision = 2)
             $response->getProtocolVersion(),
             $response->getStatus(),
             $response->getReason(),
-            $response->getRequest()->getUri()
+            (string) $response->getRequest()->getUri()
         );
 
         foreach ($response->getHeaders() as $field => $values) {

examples/streaming/2-http1-http2.php

@@ -13,7 +13,7 @@
 require __DIR__ . '/../.helper/functions.php';
 
 // https://stackoverflow.com/a/2510540/2373138
-function formatBytes(int $size, int $precision = 2)
+function formatBytes(int $size, int $precision = 2): string
 {
     $base = \log($size, 1024);
     $suffixes = ['bytes', 'kB', 'MB', 'GB', 'TB'];

psalm.xml

@@ -0,0 +1,51 @@
+<?xml version="1.0"?>
+<psalm
+        totallyTyped="false"
+        errorLevel="2"
+        phpVersion="7.2"
+        resolveFromConfigFile="true"
+        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+        xmlns="https://getpsalm.org/schema/config"
+        xsi:schemaLocation="https://getpsalm.org/schema/config vendor/vimeo/psalm/config.xsd"
+>
+    <projectFiles>
+        <directory name="examples"/>
+        <directory name="src"/>
+        <ignoreFiles>
+            <directory name="vendor"/>
+        </ignoreFiles>
+    </projectFiles>
+
+    <issueHandlers>
+        <StringIncrement>
+            <errorLevel type="suppress">
+                <directory name="examples"/>
+                <directory name="src"/>
+            </errorLevel>
+        </StringIncrement>
+
+        <DuplicateClass>
+            <errorLevel type="suppress">
+                <directory name="examples"/>
+            </errorLevel>
+        </DuplicateClass>
+
+        <RedundantConditionGivenDocblockType>
+            <errorLevel type="suppress">
+                <directory name="src"/>
+            </errorLevel>
+        </RedundantConditionGivenDocblockType>
+
+        <DocblockTypeContradiction>
+            <errorLevel type="suppress">
+                <directory name="src"/>
+            </errorLevel>
+        </DocblockTypeContradiction>
+
+        <MissingClosureParamType>
+            <errorLevel type="suppress">
+                <directory name="src"/>
+            </errorLevel>
+        </MissingClosureParamType>
+    </issueHandlers>
+</psalm>

src/Body/FileBody.php

@@ -33,10 +33,10 @@ public function createBodyStream(): InputStream
         $handlePromise = open($this->path, "r");
 
         return new class($handlePromise) implements InputStream {
-            /** @var Promise */
+            /** @var Promise<InputStream> */
             private $promise;
 
-            /** @var InputStream */
+            /** @var InputStream|null */
             private $stream;
 
             public function __construct(Promise $promise)

src/Body/FormBody.php

@@ -13,12 +13,18 @@
 
 final class FormBody implements RequestBody
 {
+    /** @var (array{0: string, 1: string, 2: string, 3: null}|array{0: string, 1: FileBody, 2: string, 3: string})[] */
     private $fields = [];
+    /** @var string */
     private $boundary;
+    /** @var bool */
     private $isMultipart = false;
 
+    /** @var string|null */
     private $cachedBody;
+    /** @var Promise<int>|null */
     private $cachedLength;
+    /** @var list<string|FileBody>|null */
     private $cachedFields;
 
     /**
@@ -123,7 +129,11 @@ private function getMultipartFieldArray(): array
         foreach ($this->fields as $fieldArr) {
             [$name, $field, $contentType, $fileName] = $fieldArr;
 
+            \assert($field instanceof FileBody ? ($fileName !== null) : ($fileName === null));
+
             $fields[] = "--{$this->boundary}\r\n";
+
+            /** @psalm-suppress PossiblyNullArgument */
             $fields[] = $field instanceof FileBody
                 ? $this->generateMultipartFileHeader($name, $fileName, $contentType)
                 : $this->generateMultipartFieldHeader($name, $contentType);
@@ -217,7 +227,8 @@ public function getBodyLength(): Promise
             return $this->cachedLength = new Success(\strlen($this->getFormEncodedBodyString()));
         }
 
-        return $this->cachedLength = call(function () {
+        /** @var Promise<int> $lengthPromise */
+        $lengthPromise = call(function (): \Generator {
             $fields = $this->getMultipartFieldArray();
             $length = 0;
 
@@ -231,5 +242,7 @@ public function getBodyLength(): Promise
 
             return $length;
         });
+
+        return $this->cachedLength = $lengthPromise;
     }
 }

src/Connection/ConnectionLimitingPool.php

@@ -23,7 +23,7 @@ final class ConnectionLimitingPool implements ConnectionPool
     /** @var ConnectionFactory */
     private $connectionFactory;
 
-    /** @var Promise[][] */
+    /** @var array<string, \ArrayObject<int, Promise<Connection>>> */
     private $connections = [];
 
     /** @var int[] */
@@ -235,6 +235,8 @@ private function getStreamFor(string $uri, Request $request, CancellationToken $
                 return;
             }
 
+            \assert($connection !== null);
+
             $this->openConnectionCount++;
 
             if ($isHttps) {
@@ -330,7 +332,7 @@ private function dropConnection(string $uri, int $connectionId): void
     {
         unset($this->connections[$uri][$connectionId], $this->activeRequestCounts[$connectionId]);
 
-        if (empty($this->connections[$uri])) {
+        if ($this->connections[$uri]->count() === 0) {
             unset($this->connections[$uri], $this->waitForPriorConnection[$uri]);
         }
     }

src/Connection/Http1Connection.php

@@ -49,7 +49,7 @@ final class Http1Connection implements Connection
     private const MAX_KEEP_ALIVE_TIMEOUT = 60;
     private const PROTOCOL_VERSIONS = ['1.0', '1.1'];
 
-    /** @var EncryptableSocket */
+    /** @var EncryptableSocket|null */
     private $socket;
 
     /** @var bool */
@@ -226,7 +226,9 @@ private function request(Request $request, CancellationToken $cancellation, Stre
                     yield $eventListener->abort($request, $e);
                 }
 
-                $this->socket->close();
+                if ($this->socket !== null) {
+                    $this->socket->close();
+                }
 
                 throw $e;
             } finally {
@@ -372,6 +374,7 @@ private function readResponse(
                             // to resolve promise with headers.
                             $chunk = null;
 
+                            /** @psalm-suppress PossiblyNullReference */
                             do {
                                 /** @noinspection CallableParameterUseCaseInTypeContextInspection */
                                 $parser->parse($chunk);
@@ -383,6 +386,11 @@ private function readResponse(
                                 if (!$backpressure instanceof Success) {
                                     yield $this->withCancellation($backpressure, $bodyCancellationToken);
                                 }
+
+                                /** @psalm-suppress TypeDoesNotContainNull */
+                                if ($this->socket === null) {
+                                    throw new SocketException('Socket closed prior to response completion');
+                                }
                             } while (null !== $chunk = yield $this->socket->read());
 
                             $originalCancellation->throwIfRequested();
@@ -439,9 +447,9 @@ private function readResponse(
 
             throw new SocketException(\sprintf(
                 "Receiving the response headers for '%s' failed, because the socket to '%s' @ '%s' closed early with %d bytes received within %d milliseconds",
-                $request->getUri()->withUserInfo(''),
-                $request->getUri()->withUserInfo('')->getAuthority(),
-                $this->socket->getRemoteAddress(),
+                (string) $request->getUri()->withUserInfo(''),
+                (string) $request->getUri()->withUserInfo('')->getAuthority(),
+                $this->socket === null ? '???' : (string) $this->socket->getRemoteAddress(),
                 \strlen($parser->getBuffer()),
                 getCurrentTime() - $start
             ));
@@ -452,6 +460,10 @@ private function readResponse(
 
     private function handleUpgradeResponse(Request $request, Response $response, string $buffer): Response
     {
+        if ($this->socket === null) {
+            throw new SocketException('Socket closed while upgrading');
+        }
+
         $socket = new UpgradedSocket($this->socket, $buffer);
         $this->free(); // Mark this connection as unusable without closing socket.
 
@@ -479,7 +491,7 @@ private function handleUpgradeResponse(Request $request, Response $response, str
      */
     private function getRemainingTime(): int
     {
-        $timestamp = $this->lastUsedAt + $this->explicitTimeout ? $this->priorTimeout * 1000 : $this->timeoutGracePeriod;
+        $timestamp = $this->lastUsedAt + ($this->explicitTimeout ? $this->priorTimeout * 1000 : $this->timeoutGracePeriod);
         return \max(0, $timestamp - getCurrentTime());
     }
 
@@ -520,8 +532,8 @@ private function determineKeepAliveTimeout(Response $response): int
     {
         $request = $response->getRequest();
 
-        $requestConnHeader = $request->getHeader('connection');
-        $responseConnHeader = $response->getHeader('connection');
+        $requestConnHeader = $request->getHeader('connection') ?? '';
+        $responseConnHeader = $response->getHeader('connection') ?? '';
 
         if (!\strcasecmp($requestConnHeader, 'close')) {
             return 0;
@@ -570,6 +582,11 @@ private function writeRequest(
     ): \Generator {
         try {
             $rawHeaders = $this->generateRawHeader($request, $protocolVersion);
+
+            if ($this->socket === null) {
+                throw new UnprocessedRequestException(new SocketException('Socket closed before request started'));
+            }
+
             yield $this->socket->write($rawHeaders);
 
             if ($request->getMethod() === 'CONNECT') {
@@ -580,6 +597,10 @@ private function writeRequest(
             $chunking = $request->getHeader("transfer-encoding") === "chunked";
             $remainingBytes = $request->getHeader("content-length");
 
+            if ($remainingBytes !== null) {
+                $remainingBytes = (int) $remainingBytes;
+            }
+
             if ($chunking && $protocolVersion === "1.0") {
                 throw new InvalidRequestException($request, "Can't send chunked bodies over HTTP/1.0");
             }

src/Connection/Http2Connection.php

@@ -24,7 +24,7 @@ final class Http2Connection implements Connection
     /** @var EncryptableSocket */
     private $socket;
 
-    /** @var Http2Processor */
+    /** @var Http2ConnectionProcessor */
     private $processor;
 
     /** @var int */