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 368 : export function Channel(options) {
14 368 : const self = this;
15 :
16 : /* We can trigger events */
17 368 : event_mixin(self, { });
18 :
19 368 : let transport;
20 368 : let ready = null;
21 368 : let closed = null;
22 368 : let waiting = null;
23 368 : let received_done = false;
24 368 : let sent_done = false;
25 368 : let id = null;
26 368 : const binary = (options.binary === true);
27 :
28 : /*
29 : * Queue while waiting for transport, items are tuples:
30 : * [is_control ? true : false, payload]
31 : */
32 368 : const queue = [];
33 :
34 : /* Handy for callers, but not used by us */
35 368 : self.valid = true;
36 368 : self.options = options;
37 368 : self.binary = binary;
38 368 : self.id = id;
39 :
40 365 : function on_message(payload) {
41 67 : if (received_done) {
42 67 : console.warn("received message after done");
43 67 : self.close("protocol-error");
44 67 : } else {
45 365 : self.dispatchEvent("message", payload);
46 365 : }
47 365 : }
48 :
49 387 : function on_close(data) {
50 387 : closed = data;
51 387 : self.valid = false;
52 387 : if (transport && id)
53 387 : transport.unregister(id);
54 182 : if (closed.message && !options.err)
55 104 : console.warn(closed.message);
56 387 : self.dispatchEvent("close", closed);
57 387 : if (waiting)
58 89 : waiting.resolve(closed);
59 387 : }
60 :
61 365 : function on_ready(data) {
62 365 : ready = data;
63 365 : self.dispatchEvent("ready", ready);
64 365 : }
65 :
66 368 : function on_control(data) {
67 367 : if (data.command == "close") {
68 367 : on_close(data);
69 367 : return;
70 364 : } else if (data.command == "ready") {
71 365 : on_ready(data);
72 365 : }
73 :
74 365 : const done = data.command === "done";
75 67 : if (done && received_done) {
76 67 : console.warn("received two done commands on channel");
77 67 : self.close("protocol-error");
78 67 : } else {
79 365 : if (done)
80 364 : received_done = true;
81 365 : self.dispatchEvent("control", data);
82 365 : }
83 368 : }
84 :
85 368 : function send_payload(payload) {
86 368 : if (!binary) {
87 368 : if (typeof payload !== "string")
88 67 : payload = String(payload);
89 368 : }
90 368 : transport.send_message(payload, id);
91 368 : }
92 :
93 368 : ensure_transport(function(trans) {
94 368 : transport = trans;
95 368 : if (closed)
96 368 : return;
97 :
98 368 : id = transport.next_channel();
99 368 : self.id = id;
100 :
101 : /* Register channel handlers */
102 368 : transport.register(id, on_control, on_message);
103 :
104 : /* Now open the channel */
105 368 : const command = { };
106 368 : for (const i in options)
107 368 : if (i !== "binary")
108 368 : 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 368 : if (binary)
112 79 : command.binary = "raw";
113 368 : command.command = "open";
114 368 : command.channel = id;
115 :
116 368 : if (!command.host) {
117 368 : if (transport_globals.default_host)
118 365 : command.host = transport_globals.default_host;
119 368 : }
120 :
121 368 : command["flow-control"] = true;
122 368 : transport.send_control(command);
123 :
124 : /* Now drain the queue */
125 368 : while (queue.length > 0) {
126 368 : const item = queue.shift();
127 67 : if (item[0]) {
128 67 : item[1].channel = id;
129 67 : transport.send_control(item[1]);
130 67 : } else {
131 368 : send_payload(item[1]);
132 368 : }
133 368 : }
134 368 : });
135 :
136 368 : self.send = function send(message) {
137 368 : if (closed)
138 67 : console.warn("sending message on closed channel");
139 368 : else if (sent_done)
140 67 : console.warn("sending message after done");
141 368 : else if (!transport)
142 365 : queue.push([false, message]);
143 : else
144 365 : send_payload(message);
145 368 : };
146 :
147 42 : self.control = function control(options) {
148 4 : options = options || { };
149 42 : if (!options.command)
150 14 : options.command = "options";
151 42 : if (options.command === "done")
152 33 : sent_done = true;
153 42 : options.channel = id;
154 42 : if (!transport)
155 4 : queue.push([true, options]);
156 : else
157 42 : transport.send_control(options);
158 42 : };
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 279 : if (!options)
187 77 : options = { };
188 248 : else if (typeof options == "string")
189 167 : options = { problem: options };
190 279 : options.command = "close";
191 279 : options.channel = id;
192 :
193 279 : if (!transport)
194 32 : queue.push([true, options]);
195 : else
196 279 : transport.send_control(options);
197 279 : on_close(options);
198 280 : };
199 :
200 360 : self.buffer = function buffer(callback) {
201 360 : const buffers = [];
202 360 : buffers.callback = callback;
203 355 : buffers.squash = function squash() {
204 355 : return join_data(buffers, binary);
205 355 : };
206 :
207 353 : function on_message(event, data) {
208 353 : buffers.push(data);
209 221 : if (buffers.callback) {
210 221 : const block = join_data(buffers, binary);
211 221 : if (block.length > 0) {
212 221 : const consumed = buffers.callback.call(self, block);
213 62 : if (typeof consumed !== "number" || consumed === block.length) {
214 221 : 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 221 : }
228 221 : }
229 353 : }
230 :
231 355 : function on_close() {
232 355 : self.removeEventListener("message", on_message);
233 355 : self.removeEventListener("close", on_close);
234 355 : }
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 368 : }
|
