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