Update for changes to Closable interface

Also require PHP 8.1 and use features.

Author: trowski

.github/workflows/ci.yml

@@ -9,9 +9,6 @@ jobs:
     strategy:
       matrix:
         include:
-          - operating-system: 'ubuntu-latest'
-            php-version: '8.0'
-
           - operating-system: 'ubuntu-latest'
             php-version: '8.1'
 
@@ -47,7 +44,7 @@ jobs:
         uses: shivammathur/setup-php@v2
         with:
           php-version: ${{ matrix.php-version }}
-          extensions: eio-beta, uv-amphp/ext-uv@master, fiber-amphp/ext-fiber@master
+          extensions: eio-beta, uv-amphp/ext-uv@master
 
       - name: Get Composer cache directory
         id: composer-cache
@@ -95,4 +92,3 @@ jobs:
       - name: Run composer-require-checker
         run: php composer-require-checker.phar check composer.json --config-file $PWD/composer-require-check.json
         if: runner.os != 'Windows' && matrix.composer-require-checker-version != 'none'
-

composer.json

@@ -32,7 +32,7 @@
         }
     ],
     "require": {
-        "php": ">=8",
+        "php": ">=8.1",
         "amphp/amp": "^3",
         "amphp/byte-stream": "^2",
         "amphp/cache": "^2",

psalm.xml

@@ -1,8 +1,7 @@
 <?xml version="1.0"?>
 <psalm
-        totallyTyped="false"
         errorLevel="2"
-        phpVersion="8.0"
+        phpVersion="8.1"
         resolveFromConfigFile="true"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xmlns="https://getpsalm.org/schema/config"

src/Driver/BlockingFile.php

@@ -5,6 +5,7 @@
 use Amp\ByteStream\ClosedException;
 use Amp\ByteStream\StreamException;
 use Amp\Cancellation;
+use Amp\DeferredFuture;
 use Amp\File\File;
 
 final class BlockingFile implements File
@@ -14,6 +15,8 @@ final class BlockingFile implements File
     private string $path;
     private string $mode;
 
+    private readonly DeferredFuture $onClose;
+
     /**
      * @param resource $handle An open filesystem descriptor.
      * @param string $path File path.
@@ -24,13 +27,19 @@ public function __construct($handle, string $path, string $mode)
         $this->handle = $handle;
         $this->path = $path;
         $this->mode = $mode;
+
+        $this->onClose = new DeferredFuture;
     }
 
     public function __destruct()
     {
         if ($this->handle !== null) {
             @\fclose($this->handle);
         }
+
+        if (!$this->onClose->isComplete()) {
+            $this->onClose->complete();
+        }
     }
 
     public function read(?Cancellation $cancellation = null, int $length = self::DEFAULT_READ_LENGTH): ?string
@@ -93,6 +102,10 @@ public function close(): void
         $handle = $this->handle;
         $this->handle = null;
 
+        if (!$this->onClose->isComplete()) {
+            $this->onClose->complete();
+        }
+
         try {
             \set_error_handler(function ($type, $message) {
                 throw new StreamException("Failed closing file '{$this->path}': {$message}");
@@ -113,6 +126,11 @@ public function isClosed(): bool
         return $this->handle === null;
     }
 
+    public function onClose(\Closure $onClose): void
+    {
+        $this->onClose->getFuture()->finally($onClose);
+    }
+
     public function truncate(int $size): void
     {
         if ($this->handle === null) {

src/Driver/EioFile.php

@@ -14,7 +14,7 @@
 
 final class EioFile implements File
 {
-    private Internal\EioPoll $poll;
+    private readonly Internal\EioPoll $poll;
 
     /** @var resource eio file handle. */
     private $fh;
@@ -35,12 +35,10 @@ final class EioFile implements File
 
     private ?Future $closing = null;
 
+    private readonly DeferredFuture $onClose;
+
     /**
-     * @param Internal\EioPoll $poll
      * @param resource $fh
-     * @param string $path
-     * @param string $mode
-     * @param int $size
      */
     public function __construct(Internal\EioPoll $poll, $fh, string $path, string $mode, int $size)
     {
@@ -52,6 +50,14 @@ public function __construct(Internal\EioPoll $poll, $fh, string $path, string $m
         $this->position = ($mode[0] === "a") ? $size : 0;
 
         $this->queue = new \SplQueue;
+        $this->onClose = new DeferredFuture;
+    }
+
+    public function __destruct()
+    {
+        if (!$this->onClose->isComplete()) {
+            $this->onClose->complete();
+        }
     }
 
     public function read(?Cancellation $cancellation = null, int $length = self::DEFAULT_READ_LENGTH): ?string
@@ -63,7 +69,7 @@ public function read(?Cancellation $cancellation = null, int $length = self::DEF
         $this->isActive = true;
 
         $remaining = $this->size - $this->position;
-        $length = $length > $remaining ? $remaining : $length;
+        $length = \min($length, $remaining);
 
         $deferred = new DeferredFuture;
         $this->poll->listen();
@@ -152,14 +158,13 @@ public function close(): void
             return;
         }
 
-        $deferred = new DeferredFuture;
-        $this->closing = $deferred->getFuture();
+        $this->closing = $this->onClose->getFuture();
         $this->poll->listen();
 
         \eio_close($this->fh, \EIO_PRI_DEFAULT, static function (DeferredFuture $deferred): void {
             // Ignore errors when closing file, as the handle will become invalid anyway.
-            $deferred->complete(null);
-        }, $deferred);
+            $deferred->complete();
+        }, $this->onClose);
 
         try {
             $this->closing->await();
@@ -173,6 +178,11 @@ public function isClosed(): bool
         return $this->closing !== null;
     }
 
+    public function onClose(\Closure $onClose): void
+    {
+        $this->onClose->getFuture()->finally($onClose);
+    }
+
     public function truncate(int $size): void
     {
         if ($this->isActive && $this->queue->isEmpty()) {

src/Driver/ParallelFile.php

@@ -5,6 +5,7 @@
 use Amp\ByteStream\ClosedException;
 use Amp\ByteStream\StreamException;
 use Amp\Cancellation;
+use Amp\DeferredFuture;
 use Amp\File\File;
 use Amp\File\Internal;
 use Amp\File\PendingOperationError;
@@ -16,7 +17,7 @@
 
 final class ParallelFile implements File
 {
-    private Internal\FileWorker $worker;
+    private readonly Internal\FileWorker $worker;
 
     private ?int $id;
 
@@ -38,13 +39,8 @@ final class ParallelFile implements File
 
     private ?Future $closing = null;
 
-    /**
-     * @param Internal\FileWorker $worker
-     * @param int $id
-     * @param string $path
-     * @param int $size
-     * @param string $mode
-     */
+    private readonly DeferredFuture $onClose;
+
     public function __construct(Internal\FileWorker $worker, int $id, string $path, int $size, string $mode)
     {
         $this->worker = $worker;
@@ -53,6 +49,8 @@ public function __construct(Internal\FileWorker $worker, int $id, string $path,
         $this->size = $size;
         $this->mode = $mode;
         $this->position = $this->mode[0] === 'a' ? $this->size : 0;
+
+        $this->onClose = new DeferredFuture;
     }
 
     public function __destruct()
@@ -62,6 +60,10 @@ public function __destruct()
             $worker = $this->worker;
             EventLoop::queue(static fn () => $worker->execute(new Internal\FileTask('fclose', [], $id)));
         }
+
+        if (!$this->onClose->isComplete()) {
+            $this->onClose->complete();
+        }
     }
 
     public function close(): void
@@ -83,14 +85,23 @@ public function close(): void
             $this->worker->execute(new Internal\FileTask('fclose', [], $id));
         });
 
-        $this->closing->await();
+        try {
+            $this->closing->await();
+        } finally {
+            $this->onClose->complete();
+        }
     }
 
     public function isClosed(): bool
     {
         return $this->closing !== null;
     }
 
+    public function onClose(\Closure $onClose): void
+    {
+        $this->onClose->getFuture()->finally($onClose);
+    }
+
     public function truncate(int $size): void
     {
         if ($this->id === null) {

src/Driver/ParallelFilesystemDriver.php

@@ -30,7 +30,6 @@ final class ParallelFilesystemDriver implements FilesystemDriver
     private Future $pendingWorker;
 
     /**
-     * @param WorkerPool|null $pool
      * @param int       $workerLimit Maximum number of workers to use from the pool for open files.
      */
     public function __construct(WorkerPool $pool = null, int $workerLimit = self::DEFAULT_WORKER_LIMIT)

src/Driver/StatusCachingFile.php

@@ -7,18 +7,17 @@
 
 final class StatusCachingFile implements File
 {
-    private File $file;
+    private readonly File $file;
 
-    /** @var callable */
-    private $invalidateCallback;
+    private readonly \Closure $invalidateCallback;
 
     /**
      * @param File $file Decorated instance.
-     * @param callable $invalidateCallback Invalidation callback.
+     * @param \Closure $invalidateCallback Invalidation callback.
      *
      * @internal
      */
-    public function __construct(File $file, callable $invalidateCallback)
+    public function __construct(File $file, \Closure $invalidateCallback)
     {
         $this->file = $file;
         $this->invalidateCallback = $invalidateCallback;
@@ -57,6 +56,11 @@ public function isClosed(): bool
         return $this->file->isClosed();
     }
 
+    public function onClose(\Closure $onClose): void
+    {
+        $this->file->onClose($onClose);
+    }
+
     public function seek(int $position, int $whence = self::SEEK_SET): int
     {
         return $this->file->seek($position, $whence);

src/Driver/StatusCachingFilesystemDriver.php

@@ -8,10 +8,8 @@
 
 final class StatusCachingFilesystemDriver implements FilesystemDriver
 {
-    /** @var FilesystemDriver */
     private FilesystemDriver $driver;
 
-    /** @var Cache */
     private Cache $statusCache;
 
     public function __construct(FilesystemDriver $driver)

src/Driver/UvFile.php

@@ -15,7 +15,7 @@
 
 final class UvFile implements File
 {
-    private Internal\UvPoll $poll;
+    private readonly Internal\UvPoll $poll;
 
     /** @var \UVLoop|resource */
     private $eventLoopHandle;
@@ -42,13 +42,11 @@ final class UvFile implements File
     /** @var bool True if ext-uv version is < 0.3.0. */
     private bool $priorVersion;
 
+    private readonly DeferredFuture $onClose;
+
     /**
-     * @param UvLoopDriver $driver
      * @param Internal\UvPoll $poll Poll for keeping the loop active.
      * @param resource $fh File handle.
-     * @param string $path
-     * @param string $mode
-     * @param int $size
      */
     public function __construct(
         UvLoopDriver $driver,
@@ -69,10 +67,18 @@ public function __construct(
         $this->position = ($mode[0] === "a") ? $size : 0;
 
         $this->queue = new \SplQueue;
+        $this->onClose = new DeferredFuture;
 
         $this->priorVersion = \version_compare(\phpversion('uv'), '0.3.0', '<');
     }
 
+    public function __destruct()
+    {
+        if (!$this->onClose->isComplete()) {
+            $this->onClose->complete();
+        }
+    }
+
     public function read(?Cancellation $cancellation = null, int $length = self::DEFAULT_READ_LENGTH): ?string
     {
         if ($this->isActive) {
@@ -242,16 +248,17 @@ public function close(): void
             return;
         }
 
-        $deferred = new DeferredFuture;
+        $deferred = $this->onClose;
+        $this->closing = $deferred->getFuture();
         $this->poll->listen();
 
         \uv_fs_close($this->eventLoopHandle, $this->fh, static function () use ($deferred): void {
             // Ignore errors when closing file, as the handle will become invalid anyway.
-            $deferred->complete(null);
+            $deferred->complete();
         });
 
         try {
-            $deferred->getFuture()->await();
+            $this->closing->await();
         } finally {
             $this->poll->done();
         }
@@ -262,6 +269,11 @@ public function isClosed(): bool
         return $this->closing !== null;
     }
 
+    public function onClose(\Closure $onClose): void
+    {
+        $this->onClose->getFuture()->finally($onClose);
+    }
+
     private function push(string $data): Future
     {
         $length = \strlen($data);