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