Line data Source code
1 : /*
2 : * Copyright (C) 2024 Red Hat, Inc.
3 : * SPDX-License-Identifier: LGPL-2.1-or-later
4 : */
5 :
6 5 : import cockpit from 'cockpit';
7 :
8 : // These are the same values used by the bridge (in channel.py)
9 5 : const BLOCK_SIZE = 16 << 10; // 16kiB
10 5 : const FLOW_WINDOW = 2 << 20; // 2MiB
11 :
12 2 : function debug(...args: unknown[]) {
13 0 : if (window.debugging == 'all' || window.debugging?.includes('upload'))
14 0 : console.debug('upload', ...args);
15 2 : }
16 :
17 1 : class UploadError extends Error {
18 1 : name = 'UploadError';
19 5 : }
20 :
21 2 : class Waiter {
22 1 : unblock = () => { };
23 2 : block(): Promise<void> {
24 2 : return new Promise(resolve => { this.unblock = resolve });
25 2 : }
26 5 : }
27 :
28 2 : export async function upload(
29 2 : destination: string,
30 2 : contents: Blob,
31 2 : progress?: (bytes_sent: number) => void,
32 2 : signal?: AbortSignal,
33 2 : options?: cockpit.JsonObject
34 2 : ) {
35 2 : let close_message = null as (cockpit.JsonObject | null);
36 2 : let outstanding = 0; // for flow control
37 2 : 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 2 : const event_waiter = new Waiter();
44 :
45 2 : if (signal) {
46 2 : signal.throwIfAborted(); // early exit
47 2 : signal.addEventListener('abort', event_waiter.unblock);
48 2 : }
49 :
50 2 : const opts = {
51 2 : payload: 'fsreplace1',
52 2 : path: destination,
53 2 : binary: true,
54 2 : size: contents.size,
55 2 : 'send-acks': 'bytes',
56 2 : ...options,
57 2 : } as const;
58 2 : debug('requesting channel', opts);
59 2 : const channel = cockpit.channel(opts);
60 2 : channel.addEventListener('control', (_ev, message) => {
61 2 : debug('control', message);
62 2 : if (message.command === 'ack') {
63 2 : cockpit.assert(typeof message.bytes === 'number', 'bytes not a number');
64 2 : delivered += message.bytes;
65 2 : if (progress) {
66 2 : debug('progress', delivered);
67 2 : progress(delivered);
68 2 : }
69 2 : outstanding -= message.bytes;
70 2 : debug('outstanding -- to', outstanding);
71 2 : event_waiter.unblock();
72 2 : }
73 2 : });
74 2 : channel.addEventListener('close', (_ev, message) => {
75 2 : debug('close', message);
76 2 : close_message = message;
77 2 : event_waiter.unblock();
78 2 : });
79 :
80 2 : try {
81 2 : 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 2 : let read;
93 2 : if (typeof ReadableStreamBYOBReader === 'function') {
94 2 : const reader = contents.stream().getReader({ mode: 'byob' });
95 2 : 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 2 : let eof = false;
125 :
126 : // eslint-disable-next-line no-unmodified-loop-condition
127 2 : 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 2 : if (!eof && outstanding < FLOW_WINDOW) {
136 2 : const { done, value } = await read();
137 2 : if (done) {
138 2 : debug('sending done');
139 2 : channel.control({ command: 'done' });
140 2 : eof = true;
141 2 : } else {
142 2 : debug('sending', value.length, 'bytes');
143 2 : channel.send(value);
144 2 : outstanding += value.length;
145 2 : debug('outstanding ++ to', outstanding);
146 2 : }
147 2 : if (signal) {
148 2 : signal.throwIfAborted();
149 2 : }
150 2 : } else {
151 2 : debug('sleeping', outstanding, 'of', FLOW_WINDOW, 'eof', eof);
152 2 : await event_waiter.block();
153 2 : }
154 2 : if (signal) {
155 2 : signal.throwIfAborted();
156 2 : }
157 2 : }
158 :
159 1 : if (close_message.problem) {
160 1 : throw new UploadError(cockpit.message(close_message));
161 1 : } else {
162 2 : cockpit.assert(typeof close_message.tag === 'string', "tag missing on close message");
163 2 : return close_message.tag;
164 2 : }
165 2 : } finally {
166 2 : debug('finally');
167 2 : channel.close(); // maybe we got aborted
168 2 : }
169 2 : }
|
