<?phpnamespace App\Xapi;use JumPHP\Async\Task;/** * HTTP/1.1 over one stream (a TcpConnection, an SshConnection, a TlsStream: anything with on() / write() / close()), the * requests one after the other on the same connection (keep-alive). The stream is opened when the first request comes, and * again after the server closed it. Answers with Content-Length, chunked, or "read until close". */final class HttpChannel{ /** @var list<array{0: string, 1: Task}> requests waiting (raw request, task) */ private array $queue = []; /** @var array{0: string, 1: Task}|null */ private ?array $current = null; private ?object $stream = null; private bool $open = false; private string $buffer = ''; private float $waited = 0; /** @var int a counter: events of an old stream are ignored */ private int $generation = 0; /** answers given on the current stream: a request written on a reused stream that dies before any byte is sent again, once */ private int $served = 0; private bool $onReused = false; private bool $retried = false; /** @var array{0: int, 1: array<string, string>}|null the current answer's status and headers, once its head arrived */ private ?array $head = null; private int $scanFrom = 0; private int $bodyAt = 0; private int $chunkAt = 0; private string $decoded = ''; /** @param \Closure(): object $connect opens a new stream */ public function __construct(private \Closure $connect, private string $hostHeader, private float $timeout = 40.0) { } /** @param array<string, string> $headers */ public function request(string $method, string $path, string $body = '', array $headers = []): Task { $task = new Task(); $lines = "$method $path HTTP/1.1\r\nHost: {$this->hostHeader}\r\nUser-Agent: XenCenter-JumPHP\r\nConnection: keep-alive\r\n"; foreach ($headers as $name => $value) { $lines .= "$name: $value\r\n"; } if ($body !== '' || $method === 'POST') { $lines .= 'Content-Length: ' . strlen($body) . "\r\n"; } $this->queue[] = [$lines . "\r\n" . $body, $task]; $this->next(); return $task; } public function pending(): int { return count($this->queue) + ($this->current !== null ? 1 : 0); } public function poll(float $dt): void { if ($this->current === null) { return; } $this->waited += $dt; if ($this->waited > $this->timeout) { $this->failCurrent('The server did not answer in ' . (int)$this->timeout . ' s.'); $this->drop(); $this->next(); } } public function close(): void { foreach ($this->queue as [, $task]) { $task->settle(null, XapiError::transport('Disconnected.')); } $this->queue = []; $this->failCurrent('Disconnected.'); $this->drop(); } private function next(): void { if ($this->current !== null || $this->queue === []) { return; } $this->current = array_shift($this->queue); $this->waited = 0; $this->buffer = ''; $this->resetAnswer(); $this->onReused = $this->stream !== null && $this->served > 0; if ($this->stream === null) { $this->openStream(); } if ($this->open && $this->current !== null) { $this->stream->write($this->current[0]); } } private function openStream(): void { $generation = ++$this->generation; $this->open = false; $this->served = 0; try { $stream = ($this->connect)(); } catch (\Throwable $e) { $this->failCurrent($e->getMessage()); $this->next(); // the others queued behind it try in turn (each fails the same way, nothing waits forever) return; } $this->stream = $stream; $stream->on('open', function () use ($generation): void { if ($generation !== $this->generation) { return; } $this->open = true; if ($this->current !== null) { $this->stream?->write($this->current[0]); } }); $stream->on('data', function (string $bytes) use ($generation): void { if ($generation === $this->generation) { $this->buffer .= $bytes; $this->waited = 0; // the timeout counts silence, not the length of a big answer still arriving $this->parse(false); } }); $stream->on('error', function (string $message) use ($generation): void { if ($generation === $this->generation) { $this->failCurrent($message); $this->drop(); $this->retryLater(); } }); $stream->on('close', function (?string $why = null) use ($generation): void { if ($generation !== $this->generation) { return; } // an answer read until the end of the stream (parse() then goes on with the next request, on a new stream) if ($this->current !== null && $this->buffer !== '') { $this->parse(true); if ($generation !== $this->generation) { return; } } if ($this->current !== null) { $this->failCurrent($why ?? 'The connection closed.'); } $this->drop(); $this->retryLater(); }); } private function retryLater(): void { // the next queued request opens a new stream (keep-alive ended, a failure): never in a loop, one at a time $this->next(); } private function drop(): void { $this->generation++; $stream = $this->stream; $this->stream = null; $this->open = false; $stream?->close(); } private function failCurrent(string $message): void { if ($this->current === null) { return; } if ($this->onReused && $this->buffer === '' && !$this->retried) { // a kept-alive connection the server (a NAT, the bridge) had closed while idle: the request never arrived, send it again array_unshift($this->queue, $this->current); $this->current = null; $this->onReused = false; $this->retried = true; return; } $this->retried = false; [, $task] = $this->current; $this->current = null; $task->settle(null, XapiError::transport($message)); } /** * One answer in the buffer? $atEnd: the stream ended (an answer without length is complete). Incremental: the head is parsed * once, a body is only measured until it is whole, a chunked body is decoded as its chunks come. Each byte is copied once, not * once per piece received: the first event.from of a pool is tens of MB, in the browser in thousands of WebSocket messages * (re-reading the whole buffer each time took minutes there: stuck on "Synchronizing…"). */ private function parse(bool $atEnd): void { if ($this->current === null) { return; } if ($this->head === null) { $end = strpos($this->buffer, "\r\n\r\n", $this->scanFrom); if ($end === false) { $this->scanFrom = max(0, strlen($this->buffer) - 3); return; } $lines = explode("\r\n", substr($this->buffer, 0, $end)); $status = preg_match('#^HTTP/\d(?:\.\d)?\s+(\d{3})#', (string)array_shift($lines), $m) ? (int)$m[1] : 0; $headers = []; foreach ($lines as $line) { $colon = strpos($line, ':'); if ($colon !== false) { $headers[strtolower(trim(substr($line, 0, $colon)))] = trim(substr($line, $colon + 1)); } } $this->head = [$status, $headers]; $this->bodyAt = $end + 4; $this->chunkAt = $this->bodyAt; $this->decoded = ''; } [$status, $headers] = $this->head; $closeAfter = strtolower($headers['connection'] ?? '') === 'close'; if (isset($headers['content-length'])) { $length = (int)$headers['content-length']; if (strlen($this->buffer) - $this->bodyAt < $length) { return; } $body = substr($this->buffer, $this->bodyAt, $length); $this->buffer = substr($this->buffer, $this->bodyAt + $length); } elseif (str_contains(strtolower($headers['transfer-encoding'] ?? ''), 'chunked')) { $rest = $this->dechunkMore(); if ($rest === null) { return; } $body = $this->decoded; $this->buffer = $rest; } elseif ($status === 204 || $status === 304) { $body = ''; $this->buffer = substr($this->buffer, $this->bodyAt); } else { if (!$atEnd) { return; // read until the server closes } $body = substr($this->buffer, $this->bodyAt); $this->buffer = ''; $closeAfter = true; } $this->resetAnswer(); [, $task] = $this->current; $this->current = null; if ($closeAfter) { $this->drop(); } $this->served++; $this->retried = false; $task->settle(new HttpReply($status, $headers, $body)); $this->next(); } /** The chunks complete so far go into $decoded; @return string|null what follows the last chunk, null while incomplete */ private function dechunkMore(): ?string { $data = $this->buffer; $at = $this->chunkAt; while (true) { $eol = strpos($data, "\r\n", $at); if ($eol === false) { return null; } $size = (int)hexdec(trim(explode(';', substr($data, $at, $eol - $at))[0])); if ($size === 0) { $end = strpos($data, "\r\n", $eol + 2); // trailers end with an empty line return $end === false ? null : (string)substr($data, $end + 2); } if (strlen($data) < $eol + 2 + $size + 2) { return null; } $this->decoded .= substr($data, $eol + 2, $size); $at = $eol + 2 + $size + 2; $this->chunkAt = $at; } } private function resetAnswer(): void { $this->head = null; $this->scanFrom = 0; $this->bodyAt = 0; $this->chunkAt = 0; $this->decoded = ''; } /** @return array{0: string, 1: string}|null the body and what follows it, null while incomplete */ public static function dechunk(string $data): ?array { $body = ''; $at = 0; while (true) { $eol = strpos($data, "\r\n", $at); if ($eol === false) { return null; } $size = hexdec(trim(explode(';', substr($data, $at, $eol - $at))[0])); $at = $eol + 2; if ($size == 0) { $end = strpos($data, "\r\n", $at); // trailers end with an empty line return $end === false ? null : [$body, (string)substr($data, $end + 2)]; } if (strlen($data) < $at + $size + 2) { return null; } $body .= substr($data, $at, (int)$size); $at += (int)$size + 2; } }}