*/ final class ReadableIterableStream implements ReadableStream, \IteratorAggregate { use ReadableStreamIteratorAggregate; use ForbidCloning; use ForbidSerialization; /** @var ConcurrentIterator|null */ private ?ConcurrentIterator $iterator; private ?\Throwable $exception = null; private bool $pending = false; private readonly DeferredFuture $onClose; /** * @param iterable $iterable */ public function __construct(iterable $iterable) { $this->iterator = $iterable instanceof ConcurrentIterator ? $iterable : Pipeline::fromIterable($iterable)->getIterator(); $this->onClose = new DeferredFuture; } public function read(?Cancellation $cancellation = null): ?string { if ($this->exception) { throw $this->exception; } if ($this->pending) { throw new PendingReadError; } if ($this->iterator === null) { return null; } $this->pending = true; try { if (!$this->iterator->continue($cancellation)) { $this->iterator = null; return null; } $chunk = $this->iterator->getValue(); if (!\is_string($chunk)) { throw new StreamException(\sprintf( "Unexpected iterable value of type %s, expected string", \get_debug_type($chunk) )); } return $chunk; } catch (\Throwable $exception) { if ($exception instanceof CancelledException && $cancellation?->isRequested()) { throw $exception; // Read cancelled, stream did not fail. } if ($exception instanceof DisposedException) { $exception = new ClosedException('Stream manually closed', previous: $exception); } throw $this->exception = $exception; } finally { $this->pending = false; } } public function isReadable(): bool { return $this->iterator !== null; } public function close(): void { $this->iterator?->dispose(); $this->iterator = null; if (!$this->onClose->isComplete()) { $this->onClose->complete(); } } public function isClosed(): bool { return !$this->isReadable(); } public function onClose(\Closure $onClose): void { $this->onClose->getFuture()->finally($onClose); } }