<?phpnamespace App\Xapi;use JumPHP\Async\Task;use JumPHP\Kernel\Platform;/** * One big HTTP transfer on a stream of its own (a VM exported as an XVA, an XVA imported): XAPI's HTTP handlers GET /export and * PUT /import. Never the whole file in memory, never a blocking wait: * - download(): the body goes to the file chunk by chunk as it arrives (Content-Length, chunked, or until the server closes); * - upload(): the file is read and written to the stream a chunk at a time, at most $budget bytes a frame, and not while the * stream says it still holds more than HIGH_WATER bytes to send (backpressure: TlsStream::buffered()). * progress() = bytes / Content-Length (or the size expected). The Task settles with the HttpReply of the server (its body: the * small text of an answer, never the file); a failure of the network fails it. * * $t = HttpStreamTransfer::download($connect, $host, '/export?session_id=…&ref=…&use_compression=true', '/home/me/web-01.xva'); * $t->task()->then(fn (HttpReply $r) => ..., fn (\Throwable $e) => ...); // and $t->poll($dt) each frame */final class HttpStreamTransfer{ public const CHUNK = 65536; public const HIGH_WATER = 1048576; /** seconds without a byte in either direction: the transfer failed */ public const IDLE = 120.0; public string $state = 'connecting'; /** bytes of the body written to the file (download) or sent (upload) */ public float $bytes = 0.0; /** the size of the body: Content-Length, the file's size, or what the caller expects (0: unknown) */ public float $total = 0.0; /** bytes written to the stream per frame at most (upload) */ public float $budget; public int $status = 0; /** @var (\Closure(self): void)|null called each frame while it runs (the progress of an Events line) */ public ?\Closure $onProgress = null; private Task $task; private ?object $stream = null; private bool $open = false; private string $head = ''; private bool $headDone = false; /** @var array<string, string> */ private array $headers = []; private string $replyBody = ''; /** @var resource|null */ private $file = null; private float $idle = 0.0; // chunked decoding, as the bytes come private bool $chunked = false; private float $chunkLeft = -1; private string $chunkLine = ''; private bool $saving = false; private bool $finished = false; /** @param \Closure(): object $connect */ private function __construct(private \Closure $connect, private string $hostHeader, private string $method, private string $path, private string $local) { $this->task = new Task(); $this->budget = Platform::isWasm() ? 262144.0 : 1048576.0; } /** GET $path, its body into $file (created when the server answers 200; removed if the transfer fails). */ public static function download(\Closure $connect, string $hostHeader, string $path, string $file, float $expected = 0.0): self { $t = new self($connect, $hostHeader, 'GET', $path, $file); $t->total = $expected; $t->start(); return $t; } /** PUT $path with $file as the body (Content-Length: its size). */ public static function upload(\Closure $connect, string $hostHeader, string $path, string $file, string $method = 'PUT'): self { $t = new self($connect, $hostHeader, $method, $path, $file); $size = @filesize($file); $handle = $size === false ? false : @fopen($file, 'rb'); if ($handle === false) { $t->fail("Cannot read the file $file."); return $t; } $t->file = $handle; $t->total = (float)$size; $t->start(); return $t; } public function task(): Task { return $this->task; } public function isDone(): bool { return $this->finished; } public function progress(): float { return $this->total > 0 ? max(0.0, min(1.0, $this->bytes / $this->total)) : 0.0; } private function start(): void { try { $this->stream = ($this->connect)(); } catch (\Throwable $e) { $this->fail($e->getMessage()); return; } $this->stream->on('open', function (): void { if ($this->finished) { return; } $this->open = true; $this->state = $this->method === 'GET' ? 'receiving' : 'sending'; $lines = "{$this->method} {$this->path} HTTP/1.1\r\nHost: {$this->hostHeader}\r\nUser-Agent: XenCenter-JumPHP\r\nConnection: close\r\n"; if ($this->method !== 'GET') { $lines .= 'Content-Length: ' . sprintf('%.0F', $this->total) . "\r\nContent-Type: application/octet-stream\r\n"; } $this->stream?->write($lines . "\r\n"); }); $this->stream->on('data', fn (string $bytes) => $this->received($bytes)); $this->stream->on('error', fn (string $message) => $this->fail($message)); $this->stream->on('close', function (?string $why = null): void { if ($this->finished) { return; } if ($this->headDone && !$this->chunked && !isset($this->headers['content-length'])) { $this->complete(); // a body read until the server closes return; } $this->fail($why ?? ($this->headDone ? 'The connection closed before the end of the transfer.' : 'The server closed the connection without answering.')); }); } /** Each frame: the next chunks of an upload, the idle timeout. */ public function poll(float $dt): void { if ($this->finished) { return; } $this->idle += $dt; if ($this->onProgress !== null) { ($this->onProgress)($this); } if ($this->idle > self::IDLE) { $this->fail('No data for ' . (int)self::IDLE . ' s: the transfer stopped.'); return; } if ($this->state !== 'sending' || !$this->open || $this->file === null || $this->stream === null) { return; } $sent = 0.0; while ($sent < $this->budget && $this->bytes < $this->total) { if (method_exists($this->stream, 'buffered') && $this->stream->buffered() > self::HIGH_WATER) { return; // the stream still holds a lot to send: wait for it } $chunk = fread($this->file, (int)min(self::CHUNK, $this->total - $this->bytes)); if ($chunk === false || $chunk === '') { $this->fail('The file ended before its size (' . sprintf('%.0F', $this->total) . ' bytes): it changed while it was sent.'); return; } if ($this->stream->write($chunk) === false) { $this->fail('The connection closed while the file was sent.'); return; } $this->bytes += strlen($chunk); $sent += strlen($chunk); $this->idle = 0; } if ($this->bytes >= $this->total) { fclose($this->file); $this->file = null; $this->state = 'waiting'; // the whole body is out: the server's answer comes next } } public function cancel(string $why = 'Cancelled.'): void { $this->fail($why); } private function received(string $bytes): void { if ($this->finished) { return; } $this->idle = 0; if (!$this->headDone) { $this->head .= $bytes; $end = strpos($this->head, "\r\n\r\n"); if ($end === false) { if (strlen($this->head) > 65536) { $this->fail('The answer of the server is not HTTP.'); } return; } $rest = (string)substr($this->head, $end + 4); $this->parseHead(substr($this->head, 0, $end)); $this->head = ''; $this->headDone = true; if ($this->method !== 'GET' && $this->state === 'sending' && $this->status >= 300) { // the server refused the upload before its end (no session, no space...): stop sending $this->closeFile(); $this->state = 'waiting'; } $this->saving = $this->method === 'GET' && $this->status === 200; if ($this->saving) { $handle = @fopen($this->local, 'wb'); if ($handle === false) { $this->fail("Cannot write the file {$this->local}."); return; } $this->file = $handle; if (isset($this->headers['content-length'])) { $this->total = (float)$this->headers['content-length']; } } if ($rest === '' && isset($this->headers['content-length']) && (float)$this->headers['content-length'] == 0) { $this->complete(); return; } $bytes = $rest; if ($bytes === '') { return; } } $this->chunked ? $this->dechunk($bytes) : $this->body($bytes); if (!$this->finished && !$this->chunked && isset($this->headers['content-length']) && $this->bodyBytes() >= (float)$this->headers['content-length']) { $this->complete(); } } private function bodyBytes(): float { return $this->saving ? $this->bytes : (float)strlen($this->replyBody); } /** Bytes of the body: to the file (a download that answered 200), else kept as the answer's text (small). */ private function body(string $bytes): void { if ($this->saving && $this->file !== null) { if (@fwrite($this->file, $bytes) !== strlen($bytes)) { $this->fail("Cannot write the file {$this->local} (disk full?)."); return; } $this->bytes += strlen($bytes); } elseif (strlen($this->replyBody) < 65536) { $this->replyBody .= $bytes; } } /** Transfer-Encoding: chunked, decoded as it comes: "<hex size>\r\n<bytes>\r\n" ... "0\r\n\r\n". */ private function dechunk(string $data): void { while ($data !== '' && !$this->finished) { if ($this->chunkLeft > 0) { $part = (string)substr($data, 0, (int)min($this->chunkLeft, strlen($data))); $this->body($part); $this->chunkLeft -= strlen($part); $data = (string)substr($data, strlen($part)); if ($this->chunkLeft == 0) { $this->chunkLeft = -2; // its CRLF comes next } continue; } $this->chunkLine .= $data; $data = ''; if ($this->chunkLeft == -2) { if (strlen($this->chunkLine) < 2) { return; } $this->chunkLine = (string)substr($this->chunkLine, 2); $this->chunkLeft = -1; } $eol = strpos($this->chunkLine, "\r\n"); if ($eol === false) { return; } $size = (float)hexdec(trim(explode(';', substr($this->chunkLine, 0, $eol))[0])); $data = (string)substr($this->chunkLine, $eol + 2); $this->chunkLine = ''; if ($size == 0) { $this->complete(); // the trailers do not matter return; } $this->chunkLeft = $size; } } private function parseHead(string $head): void { $lines = explode("\r\n", $head); $this->status = preg_match('#^HTTP/\d(?:\.\d)?\s+(\d{3})#', (string)array_shift($lines), $m) ? (int)$m[1] : 0; foreach ($lines as $line) { $colon = strpos($line, ':'); if ($colon !== false) { $this->headers[strtolower(trim(substr($line, 0, $colon)))] = trim(substr($line, $colon + 1)); } } $this->chunked = str_contains(strtolower($this->headers['transfer-encoding'] ?? ''), 'chunked'); } private function complete(): void { if ($this->finished) { return; } $this->finished = true; $this->state = 'done'; $this->closeFile(); $stream = $this->stream; $this->stream = null; $stream?->close(); $this->task->settle(new HttpReply($this->status, $this->headers, $this->replyBody)); } private function fail(string $message): void { if ($this->finished) { return; } $this->finished = true; $this->state = 'failed'; $this->closeFile(); if ($this->saving) { @unlink($this->local); // a part of an XVA is no XVA } $stream = $this->stream; $this->stream = null; $stream?->close(); $this->task->settle(null, XapiError::transport($message)); } private function closeFile(): void { if ($this->file !== null) { fclose($this->file); $this->file = null; } }}