<?phpnamespace App\Feature\Xva;use App\Gui\Units;use App\Model\Connection;use App\Model\LogEntry;use App\Model\Session;use App\Xapi\HttpReply;use App\Xapi\HttpStreamTransfer;use App\Xapi\XapiError;use JumPHP\Async\Task;/** * VM > Export and File > Import, as XenCenter does them: a task made with task.create (its progress and its end show in Events, * Cancel cancels it), then XAPI's HTTP handler on a connection of its own (the same road as the API: TCP, the JumPHP runtime, TLS * or the SSH tunnel), the file streamed to / from the disk (HttpStreamTransfer). * export: GET /export?session_id=…&task_id=…&ref=<vm>&use_compression=true|false -> the .xva file * import: PUT /import?session_id=…&task_id=…&sr_id=<sr> (body: the .xva) -> the new VM's ref in the task's result */final class XvaTransfers{ /** Task of the file's size in bytes. */ public static function export(Session $session, Connection $c, string $vm, string $file, bool $compress): Task { $name = $c->cache->name($vm); $entry = $session->log("Exporting VM '$name' to " . basename($file), $c->name(), $vm, LogEntry::RUNNING); // the size to expect (no Content-Length when compressed): what its disks hold $expected = 0.0; foreach ($c->cache->disksOf($vm) as $d) { if ($d['vdi'] !== null && ($d['vbd']['type'] ?? '') !== 'CD') { $expected += (float)($d['vdi']['physical_utilisation'] ?? 0); } } return self::start($session, $c, $entry, "Export of VM $name", function (string $task, array $connector) use ($c, $vm, $file, $compress, $expected): HttpStreamTransfer { $query = http_build_query(['session_id' => (string)$c->client?->session, 'task_id' => $task, 'ref' => $vm, 'use_compression' => $compress ? 'true' : 'false']); return HttpStreamTransfer::download($connector[0], $connector[1], '/export?' . $query, $file, $expected); }, fn (HttpStreamTransfer $t, string $task) => self::done($t->bytes)); } /** Task of the new VM's ref. $host: its home server (VM.set_affinity after), or null. */ public static function import(Session $session, Connection $c, string $file, string $sr, ?string $host, string $label): Task { $entry = $session->log("Importing '$label' to '" . $c->cache->name($sr) . "'", $c->name(), $sr, LogEntry::RUNNING); return self::start($session, $c, $entry, "Import of $label", function (string $task, array $connector) use ($c, $file, $sr): HttpStreamTransfer { $query = http_build_query(['session_id' => (string)$c->client?->session, 'task_id' => $task, 'sr_id' => $sr]); return HttpStreamTransfer::upload($connector[0], $connector[1], '/import?' . $query, $file); }, function (HttpStreamTransfer $t, string $task) use ($c, $entry, $host): Task { // the upload is in: XAPI finishes the import, the task's result names the VM $entry->status = LogEntry::RUNNING; $out = new Task(); $c->follow($task, $entry)->then(function (mixed $vm) use ($c, $host, $out): void { $vm = (string)$vm; if ($host !== null && $host !== '' && $vm !== '') { $c->call('VM.set_affinity', $vm, $host); } $out->settle($vm); }, fn (\Throwable $e) => $out->settle(null, $e)); return $out; }); } /** * task.create, the transfer, its progress in the Events line, its end. * * @param \Closure(string, array{0: \Closure(): object, 1: string}): HttpStreamTransfer $open * @param \Closure(HttpStreamTransfer, string): Task $then what follows a 200 */ private static function start(Session $session, Connection $c, LogEntry $entry, string $taskName, \Closure $open, \Closure $then): Task { $done = new Task(); $transport = $c->client?->transport; $connector = $transport !== null && method_exists($transport, 'connector') ? $transport->connector() : null; if ($connector === null || !$c->isConnected()) { self::fail($session, $entry, $done, new XapiError(['TRANSPORT', 'Not connected'])); return $done; } $c->call('task.create', $taskName, 'XenCenter (JumPHP)')->then(function (mixed $task) use ($session, $c, $entry, $open, $then, $connector, $done): void { $task = (string)$task; $entry->taskRef = $task; $transfer = $open($task, $connector); $transfer->onProgress = function (HttpStreamTransfer $t) use ($entry): void { $entry->progress = max($entry->progress, min(0.99, $t->progress())); $entry->detail = Units::bytes($t->bytes) . ($t->total > 0 ? ' of ' . Units::bytes($t->total) : ''); }; $c->transfers[] = $transfer; $transfer->task()->then(function (HttpReply $reply) use ($session, $c, $entry, $task, $transfer, $then, $done): void { if ($reply->status !== 200) { // the handler refused (no session, a halted VM needed, not an XVA...): the reason is the task's error, or the answer $c->call('task.get_error_info', $task)->then(function (mixed $info) use ($session, $entry, $done, $reply, $c, $task): void { $failure = array_map('strval', (array)$info) ?: ['HTTP_' . $reply->status, trim($reply->body)]; self::fail($session, $entry, $done, new XapiError($failure)); $c->call('task.destroy', $task); }, fn () => self::fail($session, $entry, $done, XapiError::transport("HTTP {$reply->status} " . trim($reply->body)))); return; } $entry->detail = Units::bytes($transfer->bytes); $then($transfer, $task)->then(function (mixed $result) use ($session, $c, $entry, $task, $done): void { if ($entry->status === LogEntry::RUNNING) { $entry->status = LogEntry::DONE; $entry->progress = 1; $entry->finished = $session->clock; $c->call('task.destroy', $task); } $done->settle($result); }, fn (\Throwable $e) => self::fail($session, $entry, $done, $e)); }, function (\Throwable $e) use ($session, $c, $entry, $task, $done): void { // the connection ended early: a Cancel of the task, or the network $c->call('task.get_status', $task)->then(function (mixed $status) use ($session, $entry, $done, $e, $c, $task): void { if ($status === 'cancelled' || $status === 'cancelling') { $entry->status = LogEntry::CANCELLED; $entry->detail = 'Cancelled'; $entry->finished = $session->clock; $done->settle(null, new XapiError(['CANCELLED'])); } else { self::fail($session, $entry, $done, $e); } $c->call('task.destroy', $task); }, fn () => self::fail($session, $entry, $done, $e)); }); }, fn (\Throwable $e) => self::fail($session, $entry, $done, $e)); return $done; } private static function fail(Session $session, LogEntry $entry, Task $done, \Throwable $e): void { if ($done->isDone()) { return; } $entry->status = LogEntry::FAILED; $entry->detail = $e->getMessage(); $entry->finished = $session->clock; $entry->progress = 1; $done->settle(null, $e); } private static function done(mixed $value): Task { $task = new Task(); $task->settle($value); return $task; }}