src/Pipe/WritablePipe.php
private function send(string $data, float $timeout = 0, bool $end = false)
    {
        if (!$this->isWritable()) {
            throw new UnwritableException('The stream is no longer writable.');
        }

        $length = strlen($data);
        $written = 0;

        if ($end) {
            $this->writable = false;
        }

        $resource = $this->getResource();

        try {
            if ($this->writeQueue->isEmpty()) {
                if (0 === $length) {
                    return $written;
                }

                // Error reporting suppressed since fwrite() emits E_WARNING if the pipe is broken or the buffer is full.
                $written = @fwrite($resource, $data, self::CHUNK_SIZE);

                if (false === $written) {
                    $message = 'Failed to write to stream.';
                    if ($error = error_get_last()) {
                        $message .= sprintf(' Errno: %d; %s', $error['type'], $error['message']);
                    }
                    throw new FailureException($message);
                }

                if ($length <= $written) {
                    return $written;
                }

                $data = substr($data, $written);
            }

            if (null === $this->await) {
                $this->await = $this->createAwait($resource, $this->writeQueue);
                $this->await->listen($timeout);
            } elseif (!$this->await->isPending()) {
                $this->await->listen($timeout);
            }

            $delayed = new Delayed($this->onCancelled);
            $this->writeQueue->push([$data, $written, $timeout, $delayed]);

            return yield $delayed;
        } catch (Throwable $exception) {
            $this->free($exception);
            throw $exception;
        } finally {
            if ($end && is_resource($resource)) {
                stream_socket_shutdown($resource, STREAM_SHUT_WR);
            }
        }
    }
private function send(string $data, float $timeout = 0, bool $end = false)
    {
        if (!$this->isWritable()) {
            throw new UnwritableException('The stream is no longer writable.');
        }

        $length = strlen($data);
        $written = 0;

        if ($end) {
            $this->writable = false;
        }

        $resource = $this->getResource();

        try {
            if ($this->writeQueue->isEmpty()) {
                if (0 === $length) {
                    return $written;
                }

                // Error reporting suppressed since fwrite() emits E_WARNING if the pipe is broken or the buffer is full.
                $written = @fwrite($resource, $data, self::CHUNK_SIZE);

                if (false === $written) {
                    $message = 'Failed to write to stream.';
                    if ($error = error_get_last()) {
                        $message .= sprintf(' Errno: %d; %s', $error['type'], $error['message']);
                    }
                    throw new FailureException($message);
                }

                if ($length <= $written) {
                    return $written;
                }

                $data = substr($data, $written);
            }

            if (null === $this->await) {
                $this->await = $this->createAwait($resource, $this->writeQueue);
                $this->await->listen($timeout);
            } elseif (!$this->await->isPending()) {
                $this->await->listen($timeout);
            }

            $delayed = new Delayed($this->onCancelled);
            $this->writeQueue->push([$data, $written, $timeout, $delayed]);

            return yield $delayed;
        } catch (Throwable $exception) {
            $this->free($exception);
            throw $exception;
        } finally {
            if ($end && is_resource($resource)) {
                stream_socket_shutdown($resource, STREAM_SHUT_WR);
            }
        }
    }
Writes the given data to the stream, immediately making the stream unwritable if $end is true.

@param string $data
@param int $timeout
@param bool $end

@return \Generator

@throws \Icicle\Stream\Exception\UnwritableException If the stream is no longer writable.
@throws \Icicle\Stream\Exception\FailureException If writing to the stream fails.