src/Pipe/ReadablePipe.php
public function read(int $length = 0, string $byte = null, float $timeout = 0): \Generator
    {
        while (!$this->queue->isEmpty()) {
            /** @var \Icicle\Awaitable\Delayed $delayed */
            $delayed = $this->queue->bottom();
            yield $delayed; // Wait for previous read to complete.
        }

        if (!$this->isReadable()) {
            throw new UnreadableException('The stream is no longer readable.');
        }

        if (0 > $length) {
            throw new InvalidArgumentError('The length must be a non-negative integer.');
        }

        if (0 === $length) {
            $length = self::CHUNK_SIZE;
        }

        $byte = strlen($byte) ? $byte[0] : null;

        $resource = $this->getResource();

        while ('' === ($data = $this->fetch($resource, $length, $byte))) {
            if ($this->eof($resource)) { // Close only if no data was read and at EOF.
                $this->free();
                break; // Resolve with empty string on EOF.
            }

            $this->queue->push($delayed = new Delayed($this->onCancelled));
            $this->poll->listen($timeout);

            yield $delayed;
        }

        return $data;
    }
public function read(int $length = 0, string $byte = null, float $timeout = 0): \Generator
    {
        while (!$this->queue->isEmpty()) {
            /** @var \Icicle\Awaitable\Delayed $delayed */
            $delayed = $this->queue->bottom();
            yield $delayed; // Wait for previous read to complete.
        }

        if (!$this->isReadable()) {
            throw new UnreadableException('The stream is no longer readable.');
        }

        if (0 > $length) {
            throw new InvalidArgumentError('The length must be a non-negative integer.');
        }

        if (0 === $length) {
            $length = self::CHUNK_SIZE;
        }

        $byte = strlen($byte) ? $byte[0] : null;

        $resource = $this->getResource();

        while ('' === ($data = $this->fetch($resource, $length, $byte))) {
            if ($this->eof($resource)) { // Close only if no data was read and at EOF.
                $this->free();
                break; // Resolve with empty string on EOF.
            }

            $this->queue->push($delayed = new Delayed($this->onCancelled));
            $this->poll->listen($timeout);

            yield $delayed;
        }

        return $data;
    }
{@inheritdoc}