Line data Source code
1 758 : /*
2 : * Copyright (C) 2024 Red Hat, Inc.
3 : * SPDX-License-Identifier: GPL-3.0-or-later
4 : */
5 :
6 : import type { JsonObject } from './_internal/common';
7 : import { Transport, ensure_transport, transport_globals } from './_internal/transport';
8 : import { EventEmitter } from './event';
9 :
10 : export type ChannelPayload = string | Uint8Array;
11 :
12 : export interface BaseChannelOptions extends JsonObject {
13 : command?: never;
14 : channel?: never;
15 : binary?: boolean;
16 : host?: string;
17 : payload?: string;
18 : superuser?: "try" | "require";
19 : }
20 :
21 : export interface BinaryChannelOptions extends BaseChannelOptions {
22 : binary: true;
23 : }
24 :
25 : export interface TextChannelOptions extends BaseChannelOptions {
26 : binary?: false;
27 : }
28 :
29 : export type ChannelOptions<P extends ChannelPayload> =
30 : P extends Uint8Array ?
31 : BinaryChannelOptions
32 : : P extends string ?
33 : TextChannelOptions | undefined | void
34 : :
35 : BaseChannelOptions
36 : ;
37 :
38 : type ChannelOpenOptions<P extends ChannelPayload> = ChannelOptions<P> & {
39 : payload: string;
40 : };
41 :
42 : export interface ChannelControlMessage extends JsonObject {
43 : command: string;
44 : }
45 :
46 : interface ChannelEvents<P extends ChannelPayload = string> {
47 : control(options: ChannelControlMessage): void;
48 : done(options: ChannelControlMessage): void;
49 : ready(options: ChannelControlMessage): void;
50 : close(options: ChannelControlMessage): void;
51 : data(data: P): void;
52 : }
53 :
54 501 : export class Channel<out P extends ChannelPayload = string> extends EventEmitter<ChannelEvents<P>> {
55 501 : id: string | null = null; // can be unassigned during transport startup
56 : readonly options: ChannelOpenOptions<P>;
57 : readonly binary: boolean;
58 :
59 501 : #transport: Transport | null = null;
60 501 : #received: Partial<Record<"close" | "ready" | "done", ChannelControlMessage>> = {};
61 501 : #queue: ([true, ChannelControlMessage] | [false, P])[] = [];
62 501 : #sent_done: boolean = false;
63 :
64 501 : #on_control(control: ChannelControlMessage): void {
65 501 : const command = control.command;
66 :
67 111 : if (command === 'ready' || command === 'close' || command === 'done') {
68 92 : if (this.#received[command]) {
69 92 : console.error('received duplicate control message', this.id, this.options, control);
70 92 : return;
71 92 : }
72 :
73 501 : this.#received[command] = control;
74 501 : this.emit(command, control);
75 92 : } else {
76 92 : this.emit('control', control);
77 92 : }
78 :
79 149 : if (command === 'close') {
80 149 : if (this.#transport && this.id)
81 149 : this.#transport.unregister(this.id);
82 93 : if (control.message && !this.options?.err)
83 93 : console.warn('channel error', control.message, this.id, this.options);
84 149 : }
85 501 : }
86 :
87 : /**
88 : * Open a new channel to the bridge.
89 : *
90 : * @options: The options for the channel. A payload type must be specified.
91 : */
92 501 : constructor(options: ChannelOpenOptions<P>) {
93 501 : super();
94 :
95 501 : this.options = { ...options };
96 501 : this.binary = (options?.binary === true);
97 :
98 501 : ensure_transport(transport => {
99 501 : if (this.#received.close)
100 501 : return;
101 :
102 501 : this.#transport = transport;
103 501 : this.id = transport.next_channel();
104 501 : transport.register(
105 501 : this.id,
106 501 : control => {
107 92 : if (typeof control.command !== 'string') {
108 92 : console.error('Received control message without command', this.id, this.options, control);
109 92 : } else {
110 501 : this.#on_control(control as ChannelControlMessage);
111 501 : }
112 501 : },
113 158 : data => {
114 29 : if (this.binary && typeof data === 'string') {
115 29 : console.error('Text message received on binary channel', this.id, this.options, data);
116 29 : } else if (!this.binary && typeof data !== 'string') {
117 29 : console.error('Binary message received on text channel', this.id, this.options, data);
118 29 : } else {
119 158 : this.emit('data', data as P);
120 158 : }
121 158 : }
122 501 : );
123 :
124 : // We need to delay sending the open message until after we have
125 : // the transport because we need to set the host field.
126 :
127 : // Make a copy so we can modify some fields.
128 501 : const command: JsonObject = { ...this.options };
129 :
130 498 : if (!command.host && transport_globals.default_host) {
131 498 : command.host = transport_globals.default_host;
132 498 : }
133 :
134 92 : if (this.binary) {
135 92 : command.binary = "raw";
136 92 : } else {
137 501 : delete command.binary;
138 501 : }
139 :
140 : // Go direct: we need this to go before the rest of the queue
141 501 : transport.send_control({ ...command, command: 'open', channel: this.id, 'flow-control': true });
142 :
143 : // Now send everything else from the queue
144 92 : for (const [is_control, message] of this.#queue) {
145 92 : if (is_control) {
146 92 : transport.send_control({ ...message, channel: this.id });
147 92 : } else {
148 92 : transport.send_message(message, this.id);
149 92 : }
150 92 : }
151 501 : this.#queue = [];
152 501 : });
153 501 : }
154 :
155 : /**
156 : * Sends a payload frame.
157 : *
158 : * You may not call this after you've sent a 'done' control message or
159 : * after the channel has been closed. This implies that you need to
160 : * register a 'close' event handler, and stop sending data after it's
161 : * called.
162 : *
163 : * @message the payload to send, either a string or a Uint8Array.
164 : */
165 2 : send_data(message: P): void {
166 0 : if (this.#sent_done) {
167 0 : console.error('sending data after .done()', this.id, this.options, message);
168 0 : } else if (this.#received.close) {
169 0 : console.error('sending data after close', this.id, this.options, message);
170 0 : } else if (this.#transport && this.id) {
171 2 : this.#transport.send_message(message, this.id);
172 0 : } else {
173 0 : this.#queue.push([false, message]);
174 0 : }
175 2 : }
176 :
177 : /**
178 : * Sends a control message on the channel.
179 : *
180 : * You may not call this after the channel is closed. This implies that
181 : * you need to register a 'close' event handler, and stop sending data
182 : * after it's called.
183 : *
184 : * @options: the message to send. A command must be specified.
185 : */
186 27 : send_control(options: ChannelControlMessage): void {
187 4 : if (this.#received.close) {
188 4 : console.error('sending control after close', this.id, this.options, options);
189 4 : return;
190 4 : }
191 :
192 : // A sent close message gets handled as if the exact same close message
193 : // was received. This allows signalling your own code for cancellation, etc.
194 25 : if (options.command === 'close') {
195 25 : this.#on_control(options);
196 4 : } else if (options.command === 'done') {
197 6 : this.#sent_done = true;
198 6 : }
199 :
200 27 : if (this.#transport && this.id) {
201 27 : this.#transport.send_control({ ...options, channel: this.id });
202 4 : } else {
203 4 : this.#queue.push([true, options]);
204 4 : }
205 27 : }
206 :
207 : /**
208 : * Sends a done control message on the channel. This is something like
209 : * EOF: it means that you won't send any more data using `.send_data()`.
210 : *
211 : * @options: optional extra arguments for the message.
212 : */
213 0 : done(options?: JsonObject): void {
214 0 : this.send_control({ ...options, command: 'done' });
215 0 : }
216 :
217 : /**
218 : * Closes the channel.
219 : *
220 : * This means that you're completely finished with the channel. Any
221 : * underlying resources will be freed as soon as possible. When you call
222 : * this you'll receive a 'close' signal (synchronously) and then nothing
223 : * else.
224 : *
225 : * @problem: a problem code. If this is unset it implies something like a
226 : * "successful" close. Otherwise, it indicates an error.
227 : * @options: the bridge will ignore this, but it will be thrown as the
228 : * result of any pending wait() operations and passed to the 'close' signal
229 : * handler, so you can use it to communicate with your own code.
230 : */
231 27 : close(problem?: string, options?: JsonObject): void {
232 25 : if (!this.#received.close) {
233 4 : this.send_control({ ...options, ...problem && { problem }, command: 'close' });
234 25 : }
235 27 : }
236 :
237 : /**
238 : * Waits for the result of the channel open request.
239 : *
240 : * @return: the content of the ready message, on success
241 : * @throws: the content of the close message, on fail
242 : */
243 2 : wait(): Promise<JsonObject> {
244 2 : return new Promise((resolve, reject) => {
245 : // If we got ready and closed then it's not an error.
246 : // Resolve with the ready message.
247 0 : if (this.#received.ready) {
248 0 : resolve(this.#received.ready);
249 0 : } else if (this.#received.close) {
250 0 : reject(this.#received.close);
251 0 : } else {
252 2 : this.on('ready', resolve);
253 2 : this.on('close', reject);
254 2 : }
255 2 : });
256 2 : }
257 :
258 : /**
259 : * Provides a text description of the channel.
260 : */
261 0 : toString(): string {
262 0 : const state =
263 0 : (!this.id && 'waiting for transport') ||
264 0 : (this.#received.close?.problem && `${this.id} error ${this.#received.close.problem}`) ||
265 0 : (this.#received.close && `${this.id} closed`) ||
266 0 : (this.#received.ready && `${this.id} opened`) ||
267 0 : `${this.id} waiting for open`;
268 :
269 0 : const host = this.options?.host || "localhost";
270 :
271 0 : return `[Channel ${state} -> ${this.options?.payload}@${host}]`;
272 0 : }
273 758 : }
|
