* * @internal */ final class ConcurrentIteratorChannel implements Channel { use ForbidCloning; use ForbidSerialization; private readonly DeferredFuture $onClose; /** * @param ConcurrentIterator $receive * @param Queue $send */ public function __construct( private readonly ConcurrentIterator $receive, private readonly Queue $send, ) { $this->onClose = new DeferredFuture(); } public function __destruct() { $this->close(); } public function isClosed(): bool { return $this->send->isComplete(); } public function close(): void { if (!$this->send->isComplete()) { $this->send->complete(); } $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"); } return $this->receive->getValue(); } public function send(mixed $data): void { if ($data === null) { throw new ChannelException("Cannot send null on a channel"); } if ($this->send->isComplete()) { throw new ChannelException("Cannot send on a closed channel"); } $this->send->push($data); } }