Extend Closable in Channel

Adds an onClose method to Channel.

.github/workflows/ci.yml

@@ -66,6 +66,9 @@ jobs:
  - name: Run tests
  run: vendor/bin/phpunit ${{ matrix.phpunit-flags }}
 
+ - name: Run static analysis
+ run: vendor/bin/psalm.phar
+
  - name: Run style fixer
  env:
  PHP_CS_FIXER_IGNORE_ENV: 1

src/Channel.php

@@ -3,6 +3,7 @@
 namespace Amp\Sync;
 
 use Amp\Cancellation;
+use Amp\Closable;
 use Amp\Serialization\SerializationException;
 
 /**
@@ -11,7 +12,7 @@
  * @template TReceive
  * @template TSend
  */
-interface Channel
+interface Channel extends Closable
 {
  /**
  * @param Cancellation|null $cancellation Cancels waiting for the next value. Note the next value is not discarded
@@ -31,16 +32,4 @@ public function receive(?Cancellation $cancellation = null): mixed;
  * @throws SerializationException If the underlying transport mechanism uses serialization and fails.
  */
  public function send(mixed $data): void;
-
- /**
- * Closes the channel, preventing further sends or receives.
- */
- public function close(): void;
-
- /**
- * Returns whether the channel has been closed.
- *
- * @return bool {@code true} if closed, otherwise {@code false}
- */
- public function isClosed(): bool;
 }

src/Internal/ConcurrentIteratorChannel.php

@@ -3,6 +3,7 @@
 namespace Amp\Sync\Internal;
 
 use Amp\Cancellation;
+use Amp\DeferredFuture;
 use Amp\Pipeline\ConcurrentIterator;
 use Amp\Pipeline\Queue;
 use Amp\Sync\Channel;
@@ -22,6 +23,8 @@
  */
 final class ConcurrentIteratorChannel implements Channel
 {
+ private readonly DeferredFuture $onClose;
+
  /**
  * @param ConcurrentIterator<TReceive> $receive
  * @param Queue<TSend> $send
@@ -30,6 +33,7 @@ public function __construct(
  private readonly ConcurrentIterator $receive,
  private readonly Queue $send,
  ) {
+ $this->onClose = new DeferredFuture();
  }
 
  public function __destruct()
@@ -49,11 +53,21 @@ public function close(): void
  }
 
  $this->receive->dispose();
+
+ if ($this->onClose->isComplete()) {
+ $this->onClose->complete();
+ }
+ }
+
+ public function onClose(\Closure $onClose): void
+ {
+ $this->onClose->getFuture()->finally($onClose);
  }
 
  public function receive(?Cancellation $cancellation = null): mixed
  {
  if (!$this->receive->continue($cancellation)) {
+ $this->close();
  throw new ChannelException("The channel closed while waiting to receive the next value");
  }