src/Pipe/WritablePipe.php
private function createAwait($resource, \SplQueue $writeQueue): Io
    {
        return Loop\await($resource, static function ($resource, bool $expired, Io $await) use ($writeQueue) {
            /** @var \Icicle\Awaitable\Delayed $delayed */
            list($data, $previous, $timeout, $delayed) = $writeQueue->shift();

            if ($expired) {
                $delayed->reject(new TimeoutException('Writing to the socket timed out.'));
                return;
            }

            $length = strlen($data);

            if (0 === $length) {
                $delayed->resolve($previous);
            } else {
                // 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 || 0 === $written) {
                    $message = 'Failed to write to stream.';
                    if ($error = error_get_last()) {
                        $message .= sprintf(' Errno: %d; %s', $error['type'], $error['message']);
                    }
                    $delayed->reject(new FailureException($message));
                    return;
                }

                if ($length <= $written) {
                    $delayed->resolve($written + $previous);
                } else {
                    $data = substr($data, $written);
                    $written += $previous;
                    $writeQueue->unshift([$data, $written, $timeout, $delayed]);
                }
            }

            if (!$writeQueue->isEmpty()) {
                list( , , $timeout) = $writeQueue->bottom();
                $await->listen($timeout);
            }
        });
    }
private function createAwait($resource, \SplQueue $writeQueue): Io
    {
        return Loop\await($resource, static function ($resource, bool $expired, Io $await) use ($writeQueue) {
            /** @var \Icicle\Awaitable\Delayed $delayed */
            list($data, $previous, $timeout, $delayed) = $writeQueue->shift();

            if ($expired) {
                $delayed->reject(new TimeoutException('Writing to the socket timed out.'));
                return;
            }

            $length = strlen($data);

            if (0 === $length) {
                $delayed->resolve($previous);
            } else {
                // 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 || 0 === $written) {
                    $message = 'Failed to write to stream.';
                    if ($error = error_get_last()) {
                        $message .= sprintf(' Errno: %d; %s', $error['type'], $error['message']);
                    }
                    $delayed->reject(new FailureException($message));
                    return;
                }

                if ($length <= $written) {
                    $delayed->resolve($written + $previous);
                } else {
                    $data = substr($data, $written);
                    $written += $previous;
                    $writeQueue->unshift([$data, $written, $timeout, $delayed]);
                }
            }

            if (!$writeQueue->isEmpty()) {
                list( , , $timeout) = $writeQueue->bottom();
                $await->listen($timeout);
            }
        });
    }
@param resource $resource
@param \SplQueue $writeQueue

@return \Icicle\Loop\Watcher\Io