Line data Source code
1 : // SPDX-License-Identifier: LGPL-2.1-or-later
2 : import { join_data } from './common';
3 : import { Deferred } from './deferred';
4 : import { event_mixin } from './event-mixin';
5 : import { ensure_transport, transport_globals } from './transport';
6 :
7 : /* -------------------------------------------------------------------------
8 : * Channels
9 : *
10 : * Public: https://cockpit-project.org/guide/latest/api-base1.html
11 : */
12 :
13 367 : export function Channel(options) {
14 367 : const self = this;
15 :
16 : /* We can trigger events */
17 367 : event_mixin(self, { });
18 :
19 367 : let transport;
20 367 : let ready = null;
21 367 : let closed = null;
22 367 : let waiting = null;
23 367 : let received_done = false;
24 367 : let sent_done = false;
25 367 : let id = null;
26 367 : const binary = (options.binary === true);
27 :
28 : /*
29 : * Queue while waiting for transport, items are tuples:
30 : * [is_control ? true : false, payload]
31 : */
32 367 : const queue = [];
33 :
34 : /* Handy for callers, but not used by us */
35 367 : self.valid = true;
36 367 : self.options = options;
37 367 : self.binary = binary;
38 367 : self.id = id;
39 :
40 364 : function on_message(payload) {
41 66 : if (received_done) {
42 66 : console.warn("received message after done");
43 66 : self.close("protocol-error");
44 66 : } else {
45 364 : self.dispatchEvent("message", payload);
46 364 : }
47 364 : }
48 :
49 386 : function on_close(data) {
50 386 : closed = data;
51 386 : self.valid = false;
52 386 : if (transport && id)
53 386 : transport.unregister(id);
54 185 : if (closed.message && !options.err)
55 103 : console.warn(closed.message);
56 386 : self.dispatchEvent("close", closed);
57 386 : if (waiting)
58 88 : waiting.resolve(closed);
59 386 : }
60 :
61 364 : function on_ready(data) {
62 364 : ready = data;
63 364 : self.dispatchEvent("ready", ready);
64 364 : }
65 :
66 367 : function on_control(data) {
67 366 : if (data.command == "close") {
68 366 : on_close(data);
69 366 : return;
70 363 : } else if (data.command == "ready") {
71 364 : on_ready(data);
72 364 : }
73 :
74 364 : const done = data.command === "done";
75 66 : if (done && received_done) {
76 66 : console.warn("received two done commands on channel");
77 66 : self.close("protocol-error");
78 66 : } else {
79 364 : if (done)
80 363 : received_done = true;
81 364 : self.dispatchEvent("control", data);
82 364 : }
83 367 : }
84 :
85 367 : function send_payload(payload) {
86 367 : if (!binary) {
87 367 : if (typeof payload !== "string")
88 66 : payload = String(payload);
89 367 : }
90 367 : transport.send_message(payload, id);
91 367 : }
92 :
93 367 : ensure_transport(function(trans) {
94 367 : transport = trans;
95 367 : if (closed)
96 367 : return;
97 :
98 367 : id = transport.next_channel();
99 367 : self.id = id;
100 :
101 : /* Register channel handlers */
102 367 : transport.register(id, on_control, on_message);
103 :
104 : /* Now open the channel */
105 367 : const command = { };
106 367 : for (const i in options)
107 367 : if (i !== "binary")
108 367 : command[i] = options[i];
109 : /* handle binary specially: Our JS API has always been boolean, while the wire protocol is
110 : * a string with the only valid value "raw". */
111 367 : if (binary)
112 78 : command.binary = "raw";
113 367 : command.command = "open";
114 367 : command.channel = id;
115 :
116 367 : if (!command.host) {
117 367 : if (transport_globals.default_host)
118 364 : command.host = transport_globals.default_host;
119 367 : }
120 :
121 367 : command["flow-control"] = true;
122 367 : transport.send_control(command);
123 :
124 : /* Now drain the queue */
125 367 : while (queue.length > 0) {
126 367 : const item = queue.shift();
127 66 : if (item[0]) {
128 66 : item[1].channel = id;
129 66 : transport.send_control(item[1]);
130 66 : } else {
131 367 : send_payload(item[1]);
132 367 : }
133 367 : }
134 367 : });
135 :
136 367 : self.send = function send(message) {
137 367 : if (closed)
138 66 : console.warn("sending message on closed channel");
139 367 : else if (sent_done)
140 66 : console.warn("sending message after done");
141 367 : else if (!transport)
142 364 : queue.push([false, message]);
143 : else
144 364 : send_payload(message);
145 367 : };
146 :
147 44 : self.control = function control(options) {
148 6 : options = options || { };
149 44 : if (!options.command)
150 16 : options.command = "options";
151 44 : if (options.command === "done")
152 35 : sent_done = true;
153 44 : options.channel = id;
154 44 : if (!transport)
155 6 : queue.push([true, options]);
156 : else
157 44 : transport.send_control(options);
158 44 : };
159 :
160 35 : self.wait = function wait(callback) {
161 35 : if (!waiting) {
162 35 : waiting = new Deferred();
163 4 : if (closed) {
164 4 : waiting.reject(closed);
165 4 : } else if (ready) {
166 5 : waiting.resolve(ready);
167 5 : } else {
168 34 : self.addEventListener("ready", function(event, data) {
169 34 : waiting.resolve(data);
170 34 : });
171 3 : self.addEventListener("close", function(event, data) {
172 3 : waiting.reject(data);
173 3 : });
174 35 : }
175 35 : }
176 35 : const promise = waiting.promise;
177 35 : if (callback)
178 34 : promise.then(callback, callback);
179 35 : return promise;
180 35 : };
181 :
182 280 : self.close = function close(options) {
183 280 : if (closed)
184 280 : return;
185 :
186 278 : if (!options)
187 78 : options = { };
188 251 : else if (typeof options == "string")
189 169 : options = { problem: options };
190 278 : options.command = "close";
191 278 : options.channel = id;
192 :
193 278 : if (!transport)
194 34 : queue.push([true, options]);
195 : else
196 278 : transport.send_control(options);
197 278 : on_close(options);
198 280 : };
199 :
200 360 : self.buffer = function buffer(callback) {
201 360 : const buffers = [];
202 360 : buffers.callback = callback;
203 357 : buffers.squash = function squash() {
204 357 : return join_data(buffers, binary);
205 357 : };
206 :
207 356 : function on_message(event, data) {
208 356 : buffers.push(data);
209 219 : if (buffers.callback) {
210 219 : const block = join_data(buffers, binary);
211 219 : if (block.length > 0) {
212 219 : const consumed = buffers.callback.call(self, block);
213 62 : if (typeof consumed !== "number" || consumed === block.length) {
214 219 : buffers.length = 0;
215 62 : } else if (consumed === 0) {
216 62 : buffers.length = 1;
217 62 : buffers[0] = block;
218 62 : } else if (consumed !== 0) {
219 62 : buffers.length = 1;
220 62 : if (block.subarray)
221 62 : buffers[0] = block.subarray(consumed);
222 62 : else if (block.substring)
223 62 : buffers[0] = block.substring(consumed);
224 : else
225 62 : buffers[0] = block.slice(consumed);
226 62 : }
227 219 : }
228 219 : }
229 356 : }
230 :
231 357 : function on_close() {
232 357 : self.removeEventListener("message", on_message);
233 357 : self.removeEventListener("close", on_close);
234 357 : }
235 :
236 360 : self.addEventListener("message", on_message);
237 360 : self.addEventListener("close", on_close);
238 :
239 360 : return buffers;
240 360 : };
241 :
242 0 : self.toString = function toString() {
243 0 : const host = options.host || "localhost";
244 0 : return "[Channel " + (self.valid ? id : "<invalid>") + " -> " + host + "]";
245 0 : };
246 367 : }
|
