Update Amp usage

Fixed a few errors and used byte-stream for reading and writing.

composer.json

@@ -22,7 +22,8 @@
         "webmozart/glob": "^4.1",
         "webmozart/path-util": "^2.3",
         "symfony/console": "^3.0||^4.0",
-        "amphp/amp": "^2.1"
+        "amphp/amp": "^2.1",
+        "amphp/byte-stream": "^1.5"
     },
     "bin": ["psalm", "psalter", "psalm-language-server", "psalm-plugin"],
     "autoload": {

psalm.xml.dist

@@ -36,7 +36,7 @@
     </ignoreExceptions>
 
     <stubs>
-        <file name="src/Psalm/Internal/Stubs/SabreEvent.php"/>
+        <file name="src/Psalm/Internal/Stubs/Amp.php"/>
     </stubs>
 
     <plugins>

src/Psalm/Internal/LanguageServer/Client/TextDocument.php

@@ -7,6 +7,7 @@ use Psalm\Internal\LanguageServer\ClientHandler;
 use LanguageServerProtocol\{Diagnostic, TextDocumentItem, TextDocumentIdentifier};
 use Amp\Promise;
 use JsonMapper;
+use function Amp\call;
 
 /**
  * Provides method handlers for all textDocument/* methods
@@ -53,21 +54,15 @@ class TextDocument
      */
     public function xcontent(TextDocumentIdentifier $textDocument): Promise
     {
-        $promise = $this->handler->request(
+        return call(
+            function () use ($textDocument) {
+                $result = yield $this->handler->request(
                     'textDocument/xcontent',
                     ['textDocument' => $textDocument]
                 );
 
-        $promise->onResolve(
-            /**
-             * @param object $result
-             * @return object
-             */
-            function ($result) {
                 return $this->mapper->map($result, new TextDocumentItem);
             }
         );
-
-        return $promise;
     }
 }

src/Psalm/Internal/LanguageServer/ClientHandler.php

@@ -4,7 +4,9 @@ declare(strict_types = 1);
 namespace Psalm\Internal\LanguageServer;
 
 use AdvancedJsonRpc;
+use Amp\Deferred;
 use Amp\Promise;
+use function Amp\call;
 
 /**
  * @internal
@@ -38,24 +40,20 @@ class ClientHandler
      *
      * @param string $method The method to call
      * @param array|object $params The method parameters
-     * @return Promise Resolved with the result of the request or rejected with an error
+     * @return Promise <mixed> Resolved with the result of the request or rejected with an error
      */
     public function request(string $method, $params): Promise
     {
         $id = $this->idGenerator->generate();
 
-        $promise = $this->protocolWriter->write(
+        return call(function () use ($id, $method, $params) {
+            yield $this->protocolWriter->write(
                 new Message(
                     new AdvancedJsonRpc\Request($id, $method, (object) $params)
                 )
             );
 
-        $promise->onResolve(
+            $deferred = new Deferred();
-            /**
-             * @return Promise
-             */
-            function () use ($id) {
-                $deferred = new \Amp\Deferred();
 
             $listener =
                 /**
@@ -63,6 +61,7 @@ class ClientHandler
                  * @return void
                  */
                 function (Message $msg) use ($id, $deferred, &$listener) {
+                    error_log('request handler');
                     /**
                      * @psalm-suppress UndefinedPropertyFetch
                      * @psalm-suppress MixedArgument
@@ -82,10 +81,7 @@ class ClientHandler
                 };
             $this->protocolReader->on('message', $listener);
             return $deferred->promise();
-            }
+        });
-        );
-
-        return $promise;
     }
 
     /**

src/Psalm/Internal/LanguageServer/LanguageServer.php

@@ -26,9 +26,10 @@ use Psalm\Internal\LanguageServer\Server\TextDocument;
 use LanguageServerProtocol\{Range, Position, Diagnostic, DiagnosticSeverity};
 use AdvancedJsonRpc;
 use Amp\Promise;
-use function Amp\coroutine;
 use Throwable;
 use Webmozart\PathUtil\Path;
+use function Amp\call;
+use function Amp\asyncCoroutine;
 
 /**
  * @internal
@@ -100,10 +101,7 @@ class LanguageServer extends AdvancedJsonRpc\Dispatcher
         $this->protocolReader->on(
             'message',
             /** @return void */
-            function (Message $msg) {
+            asyncCoroutine(function (Message $msg) {
-                \Amp\call(
-                    /** @return \Generator<int, Promise, mixed, void> */
-                    function () use ($msg) {
                 if (!$msg->body) {
                     return;
                 }
@@ -147,25 +145,18 @@ class LanguageServer extends AdvancedJsonRpc\Dispatcher
                     } else {
                         $responseBody = new AdvancedJsonRpc\SuccessResponse($msg->body->id, $result);
                     }
-                            $this->protocolWriter->write(new Message($responseBody));
+                    yield $this->protocolWriter->write(new Message($responseBody));
-                        }
-                    }
-                );
                 }
+            })
         );
 
         $this->protocolReader->on(
             'readMessageGroup',
             /** @return void */
-            function () {
-                \Amp\call(
-                    /** @return null */
             function () {
                 $this->doAnalysis();
             }
         );
-            }
-        );
 
         $this->client = new LanguageClient($reader, $writer);
     }
@@ -186,7 +177,7 @@ class LanguageServer extends AdvancedJsonRpc\Dispatcher
         string $rootPath = null,
         int $processId = null
     ): Promise {
-        return \Amp\call(
+        return call(
             /** @return \Generator<int, true, mixed, InitializeResult> */
             function () use ($capabilities, $rootPath, $processId) {
                 // Eventually, this might block on something. Leave it as a generator.
@@ -442,22 +433,4 @@ class LanguageServer extends AdvancedJsonRpc\Dispatcher
         }
         return $filepath;
     }
-
-    /**
-     * Throws an exception on the next tick.
-     * Useful for letting a promise crash the process on rejection.
-     *
-     * @param Throwable $err
-     * @return void
-     * @psalm-suppress PossiblyUnusedMethod
-     */
-    public static function crash(Throwable $err)
-    {
-        \Amp\Loop::defer(
-            /** @return void */
-            function () use ($err) {
-                throw $err;
-            }
-        );
-    }
 }

src/Psalm/Internal/LanguageServer/ProtocolStreamReader.php

@@ -4,9 +4,9 @@ declare(strict_types = 1);
 namespace Psalm\Internal\LanguageServer;
 
 use AdvancedJsonRpc\Message as MessageBody;
+use Amp\ByteStream\ResourceInputStream;
 use Exception;
-use Psalm\Internal\LanguageServer\Message;
+use function Amp\asyncCall;
-use Amp\Loop;
 
 /**
  * Source: https://github.com/felixfbecker/php-language-server/tree/master/src/ProtocolStreamReader.php
@@ -18,11 +18,6 @@ class ProtocolStreamReader implements ProtocolReader
     const PARSE_HEADERS = 1;
     const PARSE_BODY = 2;
 
-    /** @var resource */
-    private $input;
-
-    /** @var string */
-    private $read_watcher;
     /**
      * This is checked by ProtocolStreamReader so that it will stop reading from streams in the forked process.
      * There could be buffered bytes in stdin/over TCP, those would be processed by TCP if it were not for this check.
@@ -45,49 +40,36 @@ class ProtocolStreamReader implements ProtocolReader
      */
     public function __construct($input)
     {
-        $this->input = $input;
+        $input = new ResourceInputStream($input);
+        asyncCall(function () use ($input): \Generator {
+            while (($chunk = yield $input->read()) !== null) {
+                /** @var string $chunk */
+                $this->readMessages($chunk);
+            }
 
-        $read_watcher = Loop::onReadable(
-            $this->input,
-            /** @return void */
-            function () {
-                if (feof($this->input)) {
-                    // If stream_select reported a status change for this stream,
-                    // but the stream is EOF, it means it was closed.
             $this->emitClose();
-                    return;
+        });
-                }
-                if (!$this->is_accepting_new_requests) {
-                    // If we fork, don't read any bytes in the input buffer from the worker process.
-                    $this->emitClose();
-                    return;
-                }
-                $emitted_messages = $this->readMessages();
-                if ($emitted_messages > 0) {
-                    $this->emit('readMessageGroup');
-                }
-            }
-        );
-
-        $this->read_watcher = $read_watcher;
 
         $this->on(
             'close',
             /** @return void */
-            function () {
+            static function () use ($input) {
-                Loop::cancel($this->read_watcher);
+                $input->close();
             }
         );
     }
 
     /**
+     * @param string $buffer
+     *
      * @return int
      */
-    private function readMessages() : int
+    private function readMessages(string $buffer) : int
     {
         $emitted_messages = 0;
-        while (($c = fgetc($this->input)) !== false && $c !== '') {
+        $i = 0;
-            $this->buffer .= $c;
+        while (($buffer[$i] ?? '') !== '') {
+            $this->buffer .= $buffer[$i++];
             switch ($this->parsing_mode) {
                 case self::PARSE_HEADERS:
                     if ($this->buffer === "\r\n") {

src/Psalm/Internal/LanguageServer/ProtocolStreamWriter.php

@@ -3,13 +3,8 @@ declare(strict_types = 1);
 
 namespace Psalm\Internal\LanguageServer;
 
-use Psalm\Internal\LanguageServer\Message;
+use Amp\ByteStream\ResourceOutputStream;
-use Amp\{
+use Amp\Promise;
-    Deferred,
-    Loop,
-    Promise
-};
-use RuntimeException;
 
 /**
  * @internal
@@ -17,26 +12,16 @@ use RuntimeException;
 class ProtocolStreamWriter implements ProtocolWriter
 {
     /**
-     * @var resource $output
+     * @var \Amp\ByteStream\ResourceOutputStream
      */
     private $output;
 
-    /**
-     * @var ?string
-     */
-    private $output_watcher;
-
-    /**
-     * @var array<int, array{message: string, deferred: Deferred}> $messages
-     */
-    private $messages = [];
-
     /**
      * @param resource $output
      */
     public function __construct($output)
     {
-        $this->output = $output;
+        $this->output = new ResourceOutputStream($output);
     }
 
     /**
@@ -44,58 +29,6 @@ class ProtocolStreamWriter implements ProtocolWriter
      */
     public function write(Message $msg): Promise
     {
-        // if the message queue is currently empty, register a write handler.
+        return $this->output->write((string)$msg);
-        if (empty($this->messages)) {
-            $this->output_watcher = Loop::onWritable(
-                $this->output,
-                /** @return void */
-                function () {
-                    $this->flush();
-                }
-            );
-        }
-
-        $deferred = new \Amp\Deferred();
-        $this->messages[] = [
-            'message' => (string)$msg,
-            'deferred' => $deferred
-        ];
-        return $deferred->promise();
-    }
-
-    /**
-     * Writes pending messages to the output stream.
-     *
-     * @return void
-     */
-    private function flush()
-    {
-        $keepWriting = true;
-        while ($keepWriting) {
-            $message = $this->messages[0]['message'];
-            $deferred = $this->messages[0]['deferred'];
-
-            $bytesWritten = @fwrite($this->output, $message);
-
-            if ($bytesWritten > 0) {
-                $message = substr($message, $bytesWritten);
-            }
-
-            // Determine if this message was completely sent
-            if (strlen($message) === 0) {
-                array_shift($this->messages);
-
-                // This was the last message in the queue, remove the write handler.
-                if (count($this->messages) === 0 && $this->output_watcher) {
-                    Loop::cancel($this->output_watcher);
-                    $keepWriting = false;
-                }
-
-                $deferred->resolve();
-            } else {
-                $this->messages[0]['message'] = $message;
-                $keepWriting = false;
-            }
-        }
     }
 }

src/Psalm/Internal/LanguageServer/Server/TextDocument.php

@@ -42,7 +42,7 @@ use Psalm\Internal\LanguageServer\Index\ReadableIndex;
 use Psalm\Internal\Analyzer\FileAnalyzer;
 use Psalm\Internal\Analyzer\ClassLikeAnalyzer;
 use Amp\Promise;
-use function Amp\coroutine;
+use Amp\Success;
 use function Psalm\Internal\LanguageServer\{waitForEvent, isVendored};
 
 /**
@@ -174,15 +174,6 @@ class TextDocument
      */
     public function definition(TextDocumentIdentifier $textDocument, Position $position): Promise
     {
-        return \Amp\call(
-            /**
-             * @return \Generator<int, true, mixed, Hover|Location>
-             */
-            function () use ($textDocument, $position) {
-                if (false) {
-                    yield true;
-                }
-
         $file_path = LanguageServer::uriToPath($textDocument->uri);
 
         try {
@@ -190,11 +181,11 @@ class TextDocument
         } catch (\Psalm\Exception\UnanalyzedFileException $e) {
             $this->codebase->file_provider->openFile($file_path);
             $this->server->queueFileAnalysis($file_path, $textDocument->uri);
-                    return new Hover([]);
+            return new Success(new Hover([]));
         }
 
         if ($reference_location === null) {
-                    return new Hover([]);
+            return new Success(new Hover([]));
         }
 
         list($reference) = $reference_location;
@@ -202,17 +193,17 @@ class TextDocument
         $code_location = $this->codebase->getSymbolLocation($file_path, $reference);
 
         if (!$code_location) {
-                    return new Hover([]);
+            return new Success(new Hover([]));
         }
 
-                return new Location(
+        return new Success(
+            new Location(
                 LanguageServer::pathToUri($code_location->file_path),
                 new Range(
                     new Position($code_location->getLineNumber() - 1, $code_location->getColumn() - 1),
                     new Position($code_location->getEndLineNumber() - 1, $code_location->getEndColumn() - 1)
                 )
-                );
+            )
-            }
         );
     }
 
@@ -226,15 +217,6 @@ class TextDocument
      */
     public function hover(TextDocumentIdentifier $textDocument, Position $position): Promise
     {
-        return \Amp\call(
-            /**
-             * @return \Generator<int, true, mixed, Hover>
-             */
-            function () use ($textDocument, $position) {
-                if (false) {
-                    yield true;
-                }
-
         $file_path = LanguageServer::uriToPath($textDocument->uri);
 
         try {
@@ -242,11 +224,11 @@ class TextDocument
         } catch (\Psalm\Exception\UnanalyzedFileException $e) {
             $this->codebase->file_provider->openFile($file_path);
             $this->server->queueFileAnalysis($file_path, $textDocument->uri);
-                    return new Hover([]);
+            return new Success(new Hover([]));
         }
 
         if ($reference_location === null) {
-                    return new Hover([]);
+            return new Success(new Hover([]));
         }
 
         list($reference, $range) = $reference_location;
@@ -257,9 +239,7 @@ class TextDocument
             $this->codebase->getSymbolInformation($file_path, $reference)
         );
 
-                return new Hover($contents, $range);
+        return new Success(new Hover($contents, $range));
-            }
-        );
     }
 
     /**
@@ -278,22 +258,13 @@ class TextDocument
      */
     public function completion(TextDocumentIdentifier $textDocument, Position $position): Promise
     {
-        return \Amp\call(
-            /**
-             * @return \Generator<int, true, mixed, array<empty, empty>|CompletionList>
-             */
-            function () use ($textDocument, $position) {
-                if (false) {
-                    yield true;
-                }
-
         $file_path = LanguageServer::uriToPath($textDocument->uri);
 
         $completion_data = $this->codebase->getCompletionDataAtPosition($file_path, $position);
 
         if (!$completion_data) {
             error_log('completion not found at ' . $position->line . ':' . $position->character);
-                    return [];
+            return new Success([]);
         }
 
         list($recent_type, $gap) = $completion_data;
@@ -352,7 +323,7 @@ class TextDocument
                 }
             } catch (\Exception $e) {
                 error_log($e->getMessage());
-                        return [];
+                return new Success([]);
             }
 
             $completion_items = $gap === '->'
@@ -362,8 +333,6 @@ class TextDocument
             error_log('Found ' . count($completion_items) . ' items');
         }
 
-                return new CompletionList($completion_items, false);
+        return new Success(new CompletionList($completion_items, false));
-            }
-        );
     }
 }

src/Psalm/Internal/Stubs/Amp.php

@@ -0,0 +1,41 @@
+<?php
+
+namespace Amp;
+
+/**
+ * @template TReturn
+ * @param callable():\Generator<mixed, mixed, mixed, TReturn> $gen
+ * @return Promise<TReturn>
+ */
+function coroutine(callable $gen) : Promise {}
+
+/**
+ * @template TReturn
+ * @param callable():(\Generator<mixed, mixed, mixed, TReturn>|null) $gen
+ * @return Promise<TReturn>
+ */
+function call(callable $gen) : Promise {}
+
+/**
+ * @template TReturn
+ */
+interface Promise {
+    /**
+     * @param callable(\Throwable|null $exception, TReturn|null $result):void
+     * @return void
+     */
+    function onResolve(callable $onResolved);
+}
+
+/**
+ * @template TReturn
+ *
+ * @template-implements Promise<TReturn>
+ */
+class Success implements Promise {
+    /**
+     * @param callable(\Throwable|null $exception, TReturn|null $result):void
+     * @return void
+     */
+    function onResolve(callable $onResolved) {}
+}

src/Psalm/Internal/Stubs/SabreEvent.php

@@ -1,27 +0,0 @@
-<?php
-
-namespace Amp;
-
-/**
- * @template TReturn
- * @param callable():\Generator<mixed, mixed, mixed, TReturn> $gen
- * @return Promise<TReturn>
- */
-function coroutine(callable $gen) : Promise {}
-
-/**
- * @template TReturn
- * @param callable():(\Generator<mixed, mixed, mixed, TReturn>|null) $gen
- * @return Promise<TReturn>
- */
-function call(callable $gen) : Promise {}
-
-/**
- * @template TReturn
- */
-class Promise {
-    /**
-     * @return TReturn
-     */
-    function wait() {}
-}