Line data Source code
1 : /*
2 : * Copyright (C) 2024 Red Hat, Inc.
3 : * SPDX-License-Identifier: LGPL-2.1-or-later
4 : */
5 :
6 2 : import cockpit from 'cockpit';
7 :
8 : // These are the same values used by the bridge (in channel.py)
9 2 : const BLOCK_SIZE = 16 << 10; // 16kiB
10 2 : const FLOW_WINDOW = 2 << 20; // 2MiB
11 :
12 0 : function debug(...args: unknown[]) {
13 0 : if (window.debugging == 'all' || window.debugging?.includes('upload'))
14 0 : console.debug('upload', ...args);
15 0 : }
16 :
17 0 : class UploadError extends Error {
18 0 : name = 'UploadError';
19 2 : }
20 :
21 0 : class Waiter {
22 0 : unblock = () => { };
23 0 : block(): Promise<void> {
24 0 : return new Promise(resolve => { this.unblock = resolve });
25 0 : }
26 2 : }
27 :
28 0 : export async function upload(
29 0 : destination: string,
30 0 : contents: Blob,
31 0 : progress?: (bytes_sent: number) => void,
32 0 : signal?: AbortSignal,
33 0 : options?: cockpit.JsonObject
34 0 : ) {
35 0 : let close_message = null as (cockpit.JsonObject | null);
36 0 : let outstanding = 0; // for flow control
37 0 : let delivered = 0; // for progress reporting
38 :
39 : // This variable is the most important thing in this function. The main
40 : // upload loop will do work for as long as it can, and then it .block()s on
41 : // the waiter until something changes (ack, close, abort, etc). All of
42 : // those things call .unblock() to resume the loop.
43 0 : const event_waiter = new Waiter();
44 :
45 0 : if (signal) {
46 0 : signal.throwIfAborted(); // early exit
47 0 : signal.addEventListener('abort', event_waiter.unblock);
48 0 : }
49 :
50 0 : const opts = {
51 0 : payload: 'fsreplace1',
52 0 : path: destination,
53 0 : binary: true,
54 0 : size: contents.size,
55 0 : 'send-acks': 'bytes',
56 0 : ...options,
57 0 : } as const;
58 0 : debug('requesting channel', opts);
59 0 : const channel = cockpit.channel(opts);
60 0 : channel.addEventListener('control', (_ev, message) => {
61 0 : debug('control', message);
62 0 : if (message.command === 'ack') {
63 0 : cockpit.assert(typeof message.bytes === 'number', 'bytes not a number');
64 0 : delivered += message.bytes;
65 0 : if (progress) {
66 0 : debug('progress', delivered);
67 0 : progress(delivered);
68 0 : }
69 0 : outstanding -= message.bytes;
70 0 : debug('outstanding -- to', outstanding);
71 0 : event_waiter.unblock();
72 0 : }
73 0 : });
74 0 : channel.addEventListener('close', (_ev, message) => {
75 0 : debug('close', message);
76 0 : close_message = message;
77 0 : event_waiter.unblock();
78 0 : });
79 :
80 0 : try {
81 0 : debug('starting file send', contents);
82 :
83 : /* We want to use the "bring your own buffer" (byob) API so that we can
84 : * decide the size of the blocks to read from the file: this is needed
85 : * for flow control reasons and also to respect internal limitations in
86 : * cockpit-ws. "byob" is not available on WebKit, though:
87 : *
88 : * https://caniuse.com/mdn-api_readablestreambyobreader
89 : *
90 : * Check if the API is available, and fake it if not.
91 : */
92 0 : let read;
93 0 : if (typeof ReadableStreamBYOBReader === 'function') {
94 0 : const reader = contents.stream().getReader({ mode: 'byob' });
95 0 : read = () => reader.read(new Uint8Array(BLOCK_SIZE));
96 0 : } else {
97 : // fallback code (no 'byob' available)
98 0 : const reader = contents.stream().getReader();
99 0 : let buffer: Uint8Array | null = null;
100 0 : read = async () => {
101 : // No buffered data? Try a read.
102 0 : if (!buffer) {
103 0 : const { done, value } = await reader.read();
104 0 : if (done) {
105 0 : return { done, value };
106 0 : } else {
107 0 : buffer = value;
108 0 : }
109 0 : }
110 :
111 : // Return the buffered data: if length < size, return it all, else return a slice
112 0 : if (buffer.length < BLOCK_SIZE) {
113 0 : const value = buffer;
114 0 : buffer = null;
115 0 : return { done: false, value };
116 0 : } else {
117 0 : const value = buffer.slice(0, BLOCK_SIZE);
118 0 : buffer = buffer.slice(BLOCK_SIZE);
119 0 : return { done: false, value };
120 0 : }
121 0 : };
122 0 : }
123 :
124 0 : let eof = false;
125 :
126 : // eslint-disable-next-line no-unmodified-loop-condition
127 0 : while (!close_message) {
128 : /* We do the following steps for as long as the channel is open:
129 : * - if there is room to write more data, do that
130 : * - otherwise, block on the waiter until something changes
131 : * - in any case, check for cancellation, repeat
132 : * The idea here is that each loop iteration will `await` one
133 : * thing, and once it returns, we need to re-evaluate our state.
134 : */
135 0 : if (!eof && outstanding < FLOW_WINDOW) {
136 0 : const { done, value } = await read();
137 0 : if (done) {
138 0 : debug('sending done');
139 0 : channel.control({ command: 'done' });
140 0 : eof = true;
141 0 : } else {
142 0 : debug('sending', value.length, 'bytes');
143 0 : channel.send(value);
144 0 : outstanding += value.length;
145 0 : debug('outstanding ++ to', outstanding);
146 0 : }
147 0 : if (signal) {
148 0 : signal.throwIfAborted();
149 0 : }
150 0 : } else {
151 0 : debug('sleeping', outstanding, 'of', FLOW_WINDOW, 'eof', eof);
152 0 : await event_waiter.block();
153 0 : }
154 0 : if (signal) {
155 0 : signal.throwIfAborted();
156 0 : }
157 0 : }
158 :
159 0 : if (close_message.problem) {
160 0 : throw new UploadError(cockpit.message(close_message));
161 0 : } else {
162 0 : cockpit.assert(typeof close_message.tag === 'string', "tag missing on close message");
163 0 : return close_message.tag;
164 0 : }
165 0 : } finally {
166 0 : debug('finally');
167 0 : channel.close(); // maybe we got aborted
168 0 : }
169 0 : }
|
