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