30125731428

Committed 24 Jul 2026 08:52PM UTC coverage: 89.337%. First build

Build # 30125731428

Build Type

Pull #331

github

Committed by

web-flow

Commit Message

Merge 938b71859 into 14fdda615

Pull Request Pull Request #331: Roku Cloud Emulator support

Coverage Stats

999 of 1164 branches covered (85.82%)

Branch coverage included in aggregate %.

446 of 541 new or added lines in 6 files covered. (82.44%)

1439 of 1565 relevant lines covered (91.95%)

27.62 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.37

/src/RceVideoSignalingClient.ts

/* eslint-disable camelcase -- the Janus wire protocol uses snake_case fields */
import { EventEmitter } from 'events';
import * as WebSocket from 'ws';
import type { IceServer } from './RceManagementClient';

/**
 * Negotiates a Roku Cloud Emulator (RCE) device's video/audio stream from its Janus gateway, using
 * the standard Janus WebSocket JSON gateway protocol (subprotocol 'janus-protocol'):
 *
 *   1. connect, then `{janus:'create'}` for a session
 *   2. `{janus:'attach', plugin:'janus.plugin.streaming'}` for a plugin handle
 *   3. `{janus:'message', body:{request:'watch', id, pin}}`, which resolves with an `event`
 *      carrying a jsep SDP offer
 *   4. the caller answers the offer (with its own RTCPeerConnection) and calls sendAnswer(), which
 *      sends `{janus:'message', body:{request:'start'}, jsep: answer}`
 *   5. the caller trickles its local ICE candidates via sendCandidate()/sendCandidatesComplete(),
 *      and a keepalive is sent every 25s (Janus sessions time out at 60s)
 *
 * This client owns the Janus *signaling* session only: it has no WebRTC dependency and never
 * creates a peer connection of its own. Offers, answers and ICE candidates are plain data in and
 * out; the caller is responsible for its own RTCPeerConnection and for feeding this class the
 * answer/candidates it produces.
 *
 * The reason this lives here rather than in the caller: the Janus WebSocket host used by RCE
 * instances requires an `Authorization: Bearer <management api token>` header on the WebSocket
 * handshake itself, which only a Node WebSocket client can set (a browser WebSocket cannot set
 * handshake headers), so this class is meant to run somewhere with Node's `ws`, handing the
 * resulting offer/answer/candidates off to wherever the actual peer connection lives (for example
 * across a message channel to a browser or webview).
 *
 * Janus acknowledges an asynchronous plugin request immediately with `{janus:'ack'}`, then delivers
 * the actual result later as `{janus:'event', ...}` (or `{janus:'success', ...}` for core-level
 * requests like create/attach/destroy). Both are treated as the resolution of the request that
 * shares their `transaction` id; the intervening `ack` is otherwise ignored.
 */
export class RceVideoSignalingClient extends EventEmitter {
    constructor(
        private readonly config: RceVideoSignalingConfig,
        options: RceVideoSignalingClientOptions = {}
    ) {
        super();
        this.createWebSocket = options.createWebSocket ?? ((url, requestOptions) => new WebSocket(url, 'janus-protocol', requestOptions));
        this.keepaliveIntervalMs = options.keepaliveIntervalMs ?? 25000;
        this.negotiationTimeoutMs = options.negotiationTimeoutMs ?? 20000;
    }

    private readonly createWebSocket: (url: string, requestOptions: WebSocket.ClientOptions) => WebSocket;

    private readonly keepaliveIntervalMs: number;

    private readonly negotiationTimeoutMs: number;

    private webSocket: WebSocket | undefined;

    private sessionId: number | undefined;

    private handleId: number | undefined;

    private keepaliveTimerId: ReturnType<typeof setInterval> | undefined;

    private readonly pendingRequests = new Map<string, PendingJanusRequest>();

    private transactionCounter = 0;

    /**
     * Type-safe wrapper around EventEmitter#on for this class's event map.
     */
    public on<K extends keyof RceVideoSignalingClientEvents>(event: K, listener: RceVideoSignalingClientEvents[K]): this {
        super.on(event, listener as (...args: any[]) => void);
        return this;
    }

    /**
     * Connect and negotiate as far as the SDP offer. Resolves with the offer (and the configured
     * ice servers, echoed back for the caller's convenience) once Janus's 'watch' request returns
     * one. Rejects (and tears the session down via stop()) if negotiation has not reached that point
     * within `negotiationTimeoutMs`, so a silently unresponsive gateway (a WAF sinkhole, a dropped
     * session) fails loudly instead of leaving the caller waiting forever.
     */
    public async connect(): Promise<RceVideoSignalingOffer> {
        let negotiationSettled = false;
        let timeoutHandle: ReturnType<typeof setTimeout> | undefined;

        const negotiationPromise = this.negotiate();
        //without this, a negotiate() rejection arriving after the timeout already won the race below
        //(for example because the timeout's stop() rejected an in-flight sendRequest) would otherwise
        //be an unhandled promise rejection
        negotiationPromise.catch(() => { });

        const timeoutPromise = new Promise<never>((resolve, reject) => {
            timeoutHandle = setTimeout(() => {
                if (negotiationSettled) {
                    return;
                }
                negotiationSettled = true;
                this.stop();
                reject(new Error(`Timed out negotiating the Janus stream '${this.config.streamId}'`));
            }, this.negotiationTimeoutMs);
        });

        try {
            return await Promise.race([negotiationPromise, timeoutPromise]);
        } finally {
            negotiationSettled = true;
            clearTimeout(timeoutHandle);
        }
    }

    private async negotiate(): Promise<RceVideoSignalingOffer> {
        const webSocket = this.createWebSocket(this.config.websocketUrl, {
            headers: { Authorization: `Bearer ${this.config.apiToken}` }
        });
        this.webSocket = webSocket;

        const connected = new Promise<void>((resolve, reject) => {
            webSocket.once('open', () => resolve());
            webSocket.once('error', (error: Error) => reject(new Error(`Failed to connect to the Janus WebSocket: ${error.message}`)));
        });
        webSocket.on('message', (data: WebSocket.RawData) => this.handleMessage(data.toString()));
        webSocket.on('close', () => {
            this.stopKeepalive();
            this.emit('close');
        });

        await connected;
        //from here on, a socket error is a session-lifetime error rather than a failed connection attempt
        webSocket.removeAllListeners('error');
        webSocket.on('error', (error: Error) => {
            this.emit('error', new Error(`Janus WebSocket error: ${error.message}`));
        });

        const createResponse = await this.sendRequest({ janus: 'create' });
        this.sessionId = createResponse.data?.id;
        this.startKeepalive();

        const attachResponse = await this.sendRequest({
            janus: 'attach',
            session_id: this.sessionId,
            plugin: 'janus.plugin.streaming'
        });
        this.handleId = attachResponse.data?.id;

        const watchResponse = await this.sendRequest({
            janus: 'message',
            session_id: this.sessionId,
            handle_id: this.handleId,
            body: {
                request: 'watch',
                id: this.config.streamId,
                //a falsy pin (undefined, null, or empty string) all mean "no pin", unlike streamId
                //above, which is sent as-is since a stream id of 0 is a legitimate value
                ...(this.config.pin ? { pin: this.config.pin } : {})
            }
        });

        const offer = watchResponse.jsep;
        if (!offer?.sdp) {
            throw new Error(`Janus did not return an SDP offer for stream '${this.config.streamId}'`);
        }

        return { offer: offer, iceServers: this.config.iceServers ?? [] };
    }

    /**
     * Answer the offer returned by connect(). Sends the `start` plugin message with the answer and
     * resolves once Janus's response to it arrives.
     */
    public async sendAnswer(jsep: RceVideoJsep): Promise<void> {
        await this.sendRequest({
            janus: 'message',
            session_id: this.sessionId,
            handle_id: this.handleId,
            body: { request: 'start' },
            jsep: jsep
        });
    }

    /**
     * Trickle a single local ICE candidate to Janus.
     */
    public sendCandidate(candidate: unknown): void {
        this.sendFireAndForget({
            janus: 'trickle',
            session_id: this.sessionId,
            handle_id: this.handleId,
            candidate: candidate
        });
    }

    /**
     * Tell Janus local ICE gathering has finished.
     */
    public sendCandidatesComplete(): void {
        this.sendFireAndForget({
            janus: 'trickle',
            session_id: this.sessionId,
            handle_id: this.handleId,
            candidate: { completed: true }
        });
    }

    /**
     * Tear the session down: best-effort session destroy, then close the socket and clear the
     * keepalive timer. Safe to call more than once, or before connect() has finished.
     */
    public stop(): void {
        this.stopKeepalive();

        if (this.sessionId !== undefined) {
            this.sendFireAndForget({ janus: 'destroy', session_id: this.sessionId });
        }

        if (this.webSocket) {
            const webSocket = this.webSocket;
            webSocket.removeAllListeners();
            //closing (or terminating) a socket that is still CONNECTING makes ws abort the handshake
            //and emit an 'error' ("WebSocket was closed before the connection was established"). With
            //no listener, Node's EventEmitter throws that error rather than swallowing it, so retain a
            //no-op listener here before closing rather than leaving the socket listener-less
            webSocket.on('error', () => { });
            try {
                //close() waits on the closing handshake, which never completes for a socket that never
                //finished opening; terminate() tears the connection down immediately instead
                if (webSocket.readyState === WebSocket.CONNECTING) {
                    webSocket.terminate();
                } else {
                    webSocket.close();
                }
            } catch {
                //ws can also throw synchronously here; either way the socket is being discarded
            }
            this.webSocket = undefined;
        }

        for (const pendingRequest of this.pendingRequests.values()) {
            pendingRequest.reject(new Error(`Janus signaling session for stream '${this.config.streamId}' was stopped`));
        }
        this.pendingRequests.clear();

        this.sessionId = undefined;
        this.handleId = undefined;
    }

    private handleMessage(rawData: string): void {
        let message: JanusIncomingMessage;
        try {
            message = JSON.parse(rawData);
        } catch {
            return;
        }

        if (message.janus === 'ack') {
            //acknowledges receipt of an asynchronous request; the real response arrives later as a
            //'success' or 'event' carrying the same transaction
            return;
        }
        if (message.janus === 'success' || message.janus === 'event') {
            this.settlePendingRequest(message.transaction, message, undefined);
            return;
        }
        if (message.janus === 'error') {
            const errorMessage = this.describeJanusError(message);
            const wasPending = this.settlePendingRequest(message.transaction, undefined, errorMessage);
            if (!wasPending) {
                this.emit('error', new Error(errorMessage));
            }
            return;
        }
        if (message.janus === 'hangup') {
            this.emit('error', new Error(`Janus hung up on stream '${this.config.streamId}'${message.reason ? `: ${message.reason}` : ''}`));
        }
        //keepalive acks, webrtcup/media/slowlink notifications, and other informational events are
        //not currently surfaced
    }

    /**
     * Resolves or rejects the pending request matching `transaction`, if there is one. A 'success' or
     * 'event' message that carries a plugin-level error (wrong pin, unknown stream id, and so on) is
     * still a Janus-protocol success, but is treated as a rejection here so the real reason surfaces
     * instead of, for example, connect() later failing with a generic "no SDP offer" message.
     * @returns whether a pending request was found (and settled)
     */
    private settlePendingRequest(transaction: string | undefined, message: JanusIncomingMessage | undefined, errorMessage: string | undefined): boolean {
        if (transaction === undefined) {
            return false;
        }
        const pendingRequest = this.pendingRequests.get(transaction);
        if (!pendingRequest) {
            return false;
        }
        this.pendingRequests.delete(transaction);

        const pluginErrorMessage = message ? this.describePluginError(message) : undefined;
        if (errorMessage !== undefined) {
            pendingRequest.reject(new Error(errorMessage));
        } else if (pluginErrorMessage !== undefined) {
            pendingRequest.reject(new Error(pluginErrorMessage));
        } else {
            pendingRequest.resolve(message);
        }
        return true;
    }

    private describeJanusError(message: JanusIncomingMessage): string {
        const reason = message.error?.reason ?? 'unknown error';
        const code = message.error?.code;
        return `Janus error for stream '${this.config.streamId}'${code !== undefined ? ` (code ${code})` : ''}: ${reason}`;
    }

    /**
     * Describes a streaming-plugin-level error (for example a wrong pin or unknown stream id),
     * which arrives as a normal 'event' with no jsep rather than a top-level `{janus:'error'}`.
     */
    private describePluginError(message: JanusIncomingMessage): string | undefined {
        const pluginErrorText = message.plugindata?.data?.error;
        if (pluginErrorText === undefined) {
            return undefined;
        }
        const errorCode = message.plugindata?.data?.error_code;
        return `Janus plugin error for stream '${this.config.streamId}'${errorCode !== undefined ? ` (code ${errorCode})` : ''}: ${pluginErrorText}`;
    }

    private sendRequest(request: Record<string, unknown>): Promise<JanusIncomingMessage> {
        const transaction = this.nextTransactionId();
        return new Promise<JanusIncomingMessage>((resolve, reject) => {
            this.pendingRequests.set(transaction, { resolve: resolve, reject: reject });
            this.webSocket.send(JSON.stringify(this.withTransactionAndSecret(request, transaction)));
        });
    }

    private sendFireAndForget(request: Record<string, unknown>): void {
        this.webSocket?.send(JSON.stringify(this.withTransactionAndSecret(request, this.nextTransactionId())));
    }

    private withTransactionAndSecret(request: Record<string, unknown>, transaction: string): Record<string, unknown> {
        return {
            ...request,
            transaction: transaction,
            //the RCE Janus gateway uses API-secret auth: the janus_token value must be sent as
            //apisecret on every request, not as token (a stored-token auth field Janus also supports,
            //but this gateway does not accept - it 403s "Unauthorized request" on create with token)
            ...(this.config.janusToken !== undefined ? { apisecret: this.config.janusToken } : {})
        };
    }

    private nextTransactionId(): string {
        this.transactionCounter += 1;
        return `rce-video-${this.transactionCounter}`;
    }

    private startKeepalive(): void {
        this.keepaliveTimerId = setInterval(() => {
            if (this.sessionId !== undefined) {
                this.sendFireAndForget({ janus: 'keepalive', session_id: this.sessionId });
            }
        }, this.keepaliveIntervalMs);
    }

    private stopKeepalive(): void {
        if (this.keepaliveTimerId) {
            clearInterval(this.keepaliveTimerId);
            this.keepaliveTimerId = undefined;
        }
    }
}

/**
 * Everything needed to negotiate a stream from a running RCE device's Janus gateway (built from the
 * device's `running_device` Janus fields, plus the management api token used for the WebSocket
 * handshake).
 */
export interface RceVideoSignalingConfig {
    websocketUrl: string;
    streamId: number;
    pin?: string;
    /**
     * The management api's `janus_token` device field. This gateway uses Janus API-secret auth, so
     * it is sent as the `apisecret` field on every Janus request (not `token`, a stored-token auth
     * field Janus also supports but this gateway rejects with a 403 "Unauthorized request" on
     * create). Distinct from `apiToken`, which authenticates the WebSocket handshake itself.
     */
    janusToken?: string;
    /**
     * RCE management api bearer token, sent as `Authorization: Bearer <apiToken>` on the WebSocket
     * handshake. The Janus WebSocket host requires this; it is not the same credential as
     * `janusToken`.
     */
    apiToken: string;
    iceServers?: IceServer[];
}

export interface RceVideoSignalingClientOptions {
    createWebSocket?: (url: string, requestOptions: WebSocket.ClientOptions) => WebSocket;
    /**
     * How often to send a Janus keepalive. Defaults to 25000ms (Janus sessions time out at 60s).
     */
    keepaliveIntervalMs?: number;
    /**
     * How long connect() waits (from connecting through the 'watch' request's response) before
     * giving up and rejecting. Defaults to 20000ms.
     */
    negotiationTimeoutMs?: number;
}

export interface RceVideoSignalingOffer {
    offer: RceVideoJsep;
    iceServers: IceServer[];
}

export interface RceVideoJsep {
    type: string;
    sdp: string;
}

export interface RceVideoSignalingClientEvents {
    error: (error: Error) => void;
    close: () => void;
}

interface PendingJanusRequest {
    resolve: (message: JanusIncomingMessage) => void;
    reject: (error: Error) => void;
}

interface JanusIncomingMessage {
    janus: string;
    transaction?: string;
    data?: { id?: number };
    jsep?: RceVideoJsep;
    error?: { code?: number; reason?: string };
    reason?: string;
    /**
     * Carries a streaming-plugin-level error (as opposed to a Janus-core-level `error` message)
     * when a plugin request such as 'watch' fails, for example a wrong pin or unknown stream id
     */
    plugindata?: {
        data?: {
            error?: string;
            error_code?: number;
        };
    };
}

1	/* eslint-disable camelcase -- the Janus wire protocol uses snake_case fields */
2	import { EventEmitter } from 'events';	1✔
3	import * as WebSocket from 'ws';	1✔
4	import type { IceServer } from './RceManagementClient';
5
6	/**
7	* Negotiates a Roku Cloud Emulator (RCE) device's video/audio stream from its Janus gateway, using
8	* the standard Janus WebSocket JSON gateway protocol (subprotocol 'janus-protocol'):
9	*
10	* 1. connect, then `{janus:'create'}` for a session
11	* 2. `{janus:'attach', plugin:'janus.plugin.streaming'}` for a plugin handle
12	* 3. `{janus:'message', body:{request:'watch', id, pin}}`, which resolves with an `event`
13	* carrying a jsep SDP offer
14	* 4. the caller answers the offer (with its own RTCPeerConnection) and calls sendAnswer(), which
15	* sends `{janus:'message', body:{request:'start'}, jsep: answer}`
16	* 5. the caller trickles its local ICE candidates via sendCandidate()/sendCandidatesComplete(),
17	* and a keepalive is sent every 25s (Janus sessions time out at 60s)
18	*
19	* This client owns the Janus signaling session only: it has no WebRTC dependency and never
20	* creates a peer connection of its own. Offers, answers and ICE candidates are plain data in and
21	* out; the caller is responsible for its own RTCPeerConnection and for feeding this class the
22	* answer/candidates it produces.
23	*
24	* The reason this lives here rather than in the caller: the Janus WebSocket host used by RCE
25	* instances requires an `Authorization: Bearer <management api token>` header on the WebSocket
26	* handshake itself, which only a Node WebSocket client can set (a browser WebSocket cannot set
27	* handshake headers), so this class is meant to run somewhere with Node's `ws`, handing the
28	* resulting offer/answer/candidates off to wherever the actual peer connection lives (for example
29	* across a message channel to a browser or webview).
30	*
31	* Janus acknowledges an asynchronous plugin request immediately with `{janus:'ack'}`, then delivers
32	* the actual result later as `{janus:'event', ...}` (or `{janus:'success', ...}` for core-level
33	* requests like create/attach/destroy). Both are treated as the resolution of the request that
34	* shares their `transaction` id; the intervening `ack` is otherwise ignored.
35	*/
36	export class RceVideoSignalingClient extends EventEmitter {	1✔
37	constructor(
38	private readonly config: RceVideoSignalingConfig,	22✔
39	options: RceVideoSignalingClientOptions = {}	×
40	) {
41	super();	22✔
42	this.createWebSocket = options.createWebSocket ?? ((url, requestOptions) => new WebSocket(url, 'janus-protocol', requestOptions));	22!
43	this.keepaliveIntervalMs = options.keepaliveIntervalMs ?? 25000;	22✔
44	this.negotiationTimeoutMs = options.negotiationTimeoutMs ?? 20000;	22✔
45	}
46
47	private readonly createWebSocket: (url: string, requestOptions: WebSocket.ClientOptions) => WebSocket;
48
49	private readonly keepaliveIntervalMs: number;
50
51	private readonly negotiationTimeoutMs: number;
52
53	private webSocket: WebSocket \| undefined;
54
55	private sessionId: number \| undefined;
56
57	private handleId: number \| undefined;
58
59	private keepaliveTimerId: ReturnType<typeof setInterval> \| undefined;
60
61	private readonly pendingRequests = new Map<string, PendingJanusRequest>();	22✔
62
63	private transactionCounter = 0;	22✔
64
65	/**
66	* Type-safe wrapper around EventEmitter#on for this class's event map.
67	*/
68	public on<K extends keyof RceVideoSignalingClientEvents>(event: K, listener: RceVideoSignalingClientEvents[K]): this {
69	super.on(event, listener as (...args: any[]) => void);	5✔
70	return this;	5✔
71	}
72
73	/**
74	* Connect and negotiate as far as the SDP offer. Resolves with the offer (and the configured
75	* ice servers, echoed back for the caller's convenience) once Janus's 'watch' request returns
76	* one. Rejects (and tears the session down via stop()) if negotiation has not reached that point
77	* within `negotiationTimeoutMs`, so a silently unresponsive gateway (a WAF sinkhole, a dropped
78	* session) fails loudly instead of leaving the caller waiting forever.
79	*/
80	public async connect(): Promise<RceVideoSignalingOffer> {
81	let negotiationSettled = false;	22✔
82	let timeoutHandle: ReturnType<typeof setTimeout> \| undefined;
83
84	const negotiationPromise = this.negotiate();	22✔
85	//without this, a negotiate() rejection arriving after the timeout already won the race below
86	//(for example because the timeout's stop() rejected an in-flight sendRequest) would otherwise
87	//be an unhandled promise rejection
88	negotiationPromise.catch(() => { });	22✔
89
90	const timeoutPromise = new Promise<never>((resolve, reject) => {	22✔
91	timeoutHandle = setTimeout(() => {	22✔
92	if (negotiationSettled) {	4!
NEW 93	return;	×
94	}
95	negotiationSettled = true;	4✔
96	this.stop();	4✔
97	reject(new Error(`Timed out negotiating the Janus stream '${this.config.streamId}'`));	4✔
98	}, this.negotiationTimeoutMs);
99	});
100
101	try {	22✔
102	return await Promise.race([negotiationPromise, timeoutPromise]);	22✔
103	} finally {
104	negotiationSettled = true;	22✔
105	clearTimeout(timeoutHandle);	22✔
106	}
107	}
108
109	private async negotiate(): Promise<RceVideoSignalingOffer> {
110	const webSocket = this.createWebSocket(this.config.websocketUrl, {	22✔
111	headers: { Authorization: `Bearer ${this.config.apiToken}` }
112	});
113	this.webSocket = webSocket;	22✔
114
115	const connected = new Promise<void>((resolve, reject) => {	22✔
116	webSocket.once('open', () => resolve());	22✔
117	webSocket.once('error', (error: Error) => reject(new Error(`Failed to connect to the Janus WebSocket: ${error.message}`)));	22✔
118	});
119	webSocket.on('message', (data: WebSocket.RawData) => this.handleMessage(data.toString()));	55✔
120	webSocket.on('close', () => {	22✔
121	this.stopKeepalive();	1✔
122	this.emit('close');	1✔
123	});
124
125	await connected;	22✔
126	//from here on, a socket error is a session-lifetime error rather than a failed connection attempt
127	webSocket.removeAllListeners('error');	19✔
128	webSocket.on('error', (error: Error) => {	19✔
NEW 129	this.emit('error', new Error(`Janus WebSocket error: ${error.message}`));	×
130	});
131
132	const createResponse = await this.sendRequest({ janus: 'create' });	19✔
133	this.sessionId = createResponse.data?.id;	17!
134	this.startKeepalive();	17✔
135
136	const attachResponse = await this.sendRequest({	17✔
137	janus: 'attach',
138	session_id: this.sessionId,
139	plugin: 'janus.plugin.streaming'
140	});
141	this.handleId = attachResponse.data?.id;	17!
142
143	const watchResponse = await this.sendRequest({	17✔
144	janus: 'message',
145	session_id: this.sessionId,
146	handle_id: this.handleId,
147	body: {
148	request: 'watch',
149	id: this.config.streamId,
150	//a falsy pin (undefined, null, or empty string) all mean "no pin", unlike streamId
151	//above, which is sent as-is since a stream id of 0 is a legitimate value
152	...(this.config.pin ? { pin: this.config.pin } : {})	17✔
153	}
154	});
155
156	const offer = watchResponse.jsep;	16✔
157	if (!offer?.sdp) {	16!
NEW 158	throw new Error(`Janus did not return an SDP offer for stream '${this.config.streamId}'`);	×
159	}
160
161	return { offer: offer, iceServers: this.config.iceServers ?? [] };	16!
162	}
163
164	/**
165	* Answer the offer returned by connect(). Sends the `start` plugin message with the answer and
166	* resolves once Janus's response to it arrives.
167	*/
168	public async sendAnswer(jsep: RceVideoJsep): Promise<void> {
169	await this.sendRequest({	1✔
170	janus: 'message',
171	session_id: this.sessionId,
172	handle_id: this.handleId,
173	body: { request: 'start' },
174	jsep: jsep
175	});
176	}
177
178	/**
179	* Trickle a single local ICE candidate to Janus.
180	*/
181	public sendCandidate(candidate: unknown): void {
182	this.sendFireAndForget({	1✔
183	janus: 'trickle',
184	session_id: this.sessionId,
185	handle_id: this.handleId,
186	candidate: candidate
187	});
188	}
189
190	/**
191	* Tell Janus local ICE gathering has finished.
192	*/
193	public sendCandidatesComplete(): void {
194	this.sendFireAndForget({	1✔
195	janus: 'trickle',
196	session_id: this.sessionId,
197	handle_id: this.handleId,
198	candidate: { completed: true }
199	});
200	}
201
202	/**
203	* Tear the session down: best-effort session destroy, then close the socket and clear the
204	* keepalive timer. Safe to call more than once, or before connect() has finished.
205	*/
206	public stop(): void {
207	this.stopKeepalive();	32✔
208
209	if (this.sessionId !== undefined) {	32✔
210	this.sendFireAndForget({ janus: 'destroy', session_id: this.sessionId });	17✔
211	}
212
213	if (this.webSocket) {	32✔
214	const webSocket = this.webSocket;	22✔
215	webSocket.removeAllListeners();	22✔
216	//closing (or terminating) a socket that is still CONNECTING makes ws abort the handshake
217	//and emit an 'error' ("WebSocket was closed before the connection was established"). With
218	//no listener, Node's EventEmitter throws that error rather than swallowing it, so retain a
219	//no-op listener here before closing rather than leaving the socket listener-less
220	webSocket.on('error', () => { });	22✔
221	try {	22✔
222	//close() waits on the closing handshake, which never completes for a socket that never
223	//finished opening; terminate() tears the connection down immediately instead
224	if (webSocket.readyState === WebSocket.CONNECTING) {	22✔
225	webSocket.terminate();	3✔
226	} else {
227	webSocket.close();	19✔
228	}
229	} catch {
230	//ws can also throw synchronously here; either way the socket is being discarded
231	}
232	this.webSocket = undefined;	22✔
233	}
234
235	for (const pendingRequest of this.pendingRequests.values()) {	32✔
236	pendingRequest.reject(new Error(`Janus signaling session for stream '${this.config.streamId}' was stopped`));	1✔
237	}
238	this.pendingRequests.clear();	32✔
239
240	this.sessionId = undefined;	32✔
241	this.handleId = undefined;	32✔
242	}
243
244	private handleMessage(rawData: string): void {
245	let message: JanusIncomingMessage;
246	try {	55✔
247	message = JSON.parse(rawData);	55✔
248	} catch {
NEW 249	return;	×
250	}
251
252	if (message.janus === 'ack') {	55!
253	//acknowledges receipt of an asynchronous request; the real response arrives later as a
254	//'success' or 'event' carrying the same transaction
NEW 255	return;	×
256	}
257	if (message.janus === 'success' \|\| message.janus === 'event') {	55✔
258	this.settlePendingRequest(message.transaction, message, undefined);	52✔
259	return;	52✔
260	}
261	if (message.janus === 'error') {	3✔
262	const errorMessage = this.describeJanusError(message);	2✔
263	const wasPending = this.settlePendingRequest(message.transaction, undefined, errorMessage);	2✔
264	if (!wasPending) {	2✔
265	this.emit('error', new Error(errorMessage));	1✔
266	}
267	return;	2✔
268	}
269	if (message.janus === 'hangup') {	1✔
270	this.emit('error', new Error(`Janus hung up on stream '${this.config.streamId}'${message.reason ? `: ${message.reason}` : ''}`));	1!
271	}
272	//keepalive acks, webrtcup/media/slowlink notifications, and other informational events are
273	//not currently surfaced
274	}
275
276	/**
277	* Resolves or rejects the pending request matching `transaction`, if there is one. A 'success' or
278	* 'event' message that carries a plugin-level error (wrong pin, unknown stream id, and so on) is
279	* still a Janus-protocol success, but is treated as a rejection here so the real reason surfaces
280	* instead of, for example, connect() later failing with a generic "no SDP offer" message.
281	* @returns whether a pending request was found (and settled)
282	*/
283	private settlePendingRequest(transaction: string \| undefined, message: JanusIncomingMessage \| undefined, errorMessage: string \| undefined): boolean {
284	if (transaction === undefined) {	54✔
285	return false;	1✔
286	}
287	const pendingRequest = this.pendingRequests.get(transaction);	53✔
288	if (!pendingRequest) {	53!
NEW 289	return false;	×
290	}
291	this.pendingRequests.delete(transaction);	53✔
292
293	const pluginErrorMessage = message ? this.describePluginError(message) : undefined;	53✔
294	if (errorMessage !== undefined) {	53✔
295	pendingRequest.reject(new Error(errorMessage));	1✔
296	} else if (pluginErrorMessage !== undefined) {	52✔
297	pendingRequest.reject(new Error(pluginErrorMessage));	1✔
298	} else {
299	pendingRequest.resolve(message);	51✔
300	}
301	return true;	53✔
302	}
303
304	private describeJanusError(message: JanusIncomingMessage): string {
305	const reason = message.error?.reason ?? 'unknown error';	2!
306	const code = message.error?.code;	2!
307	return `Janus error for stream '${this.config.streamId}'${code !== undefined ? ` (code ${code})` : ''}: ${reason}`;	2!
308	}
309
310	/**
311	* Describes a streaming-plugin-level error (for example a wrong pin or unknown stream id),
312	* which arrives as a normal 'event' with no jsep rather than a top-level `{janus:'error'}`.
313	*/
314	private describePluginError(message: JanusIncomingMessage): string \| undefined {
315	const pluginErrorText = message.plugindata?.data?.error;	52✔
316	if (pluginErrorText === undefined) {	52✔
317	return undefined;	51✔
318	}
319	const errorCode = message.plugindata?.data?.error_code;	1!
320	return `Janus plugin error for stream '${this.config.streamId}'${errorCode !== undefined ? ` (code ${errorCode})` : ''}: ${pluginErrorText}`;	1!
321	}
322
323	private sendRequest(request: Record<string, unknown>): Promise<JanusIncomingMessage> {
324	const transaction = this.nextTransactionId();	54✔
325	return new Promise<JanusIncomingMessage>((resolve, reject) => {	54✔
326	this.pendingRequests.set(transaction, { resolve: resolve, reject: reject });	54✔
327	this.webSocket.send(JSON.stringify(this.withTransactionAndSecret(request, transaction)));	54✔
328	});
329	}
330
331	private sendFireAndForget(request: Record<string, unknown>): void {
332	this.webSocket?.send(JSON.stringify(this.withTransactionAndSecret(request, this.nextTransactionId())));	22!
333	}
334
335	private withTransactionAndSecret(request: Record<string, unknown>, transaction: string): Record<string, unknown> {
336	return {	76✔
337	...request,
338	transaction: transaction,
339	//the RCE Janus gateway uses API-secret auth: the janus_token value must be sent as
340	//apisecret on every request, not as token (a stored-token auth field Janus also supports,
341	//but this gateway does not accept - it 403s "Unauthorized request" on create with token)
342	...(this.config.janusToken !== undefined ? { apisecret: this.config.janusToken } : {})	76✔
343	};
344	}
345
346	private nextTransactionId(): string {
347	this.transactionCounter += 1;	76✔
348	return `rce-video-${this.transactionCounter}`;	76✔
349	}
350
351	private startKeepalive(): void {
352	this.keepaliveTimerId = setInterval(() => {	17✔
353	if (this.sessionId !== undefined) {	3✔
354	this.sendFireAndForget({ janus: 'keepalive', session_id: this.sessionId });	3✔
355	}
356	}, this.keepaliveIntervalMs);
357	}
358
359	private stopKeepalive(): void {
360	if (this.keepaliveTimerId) {	33✔
361	clearInterval(this.keepaliveTimerId);	17✔
362	this.keepaliveTimerId = undefined;	17✔
363	}
364	}
365	}
366
367	/**
368	* Everything needed to negotiate a stream from a running RCE device's Janus gateway (built from the
369	* device's `running_device` Janus fields, plus the management api token used for the WebSocket
370	* handshake).
371	*/
372	export interface RceVideoSignalingConfig {
373	websocketUrl: string;
374	streamId: number;
375	pin?: string;
376	/**
377	* The management api's `janus_token` device field. This gateway uses Janus API-secret auth, so
378	* it is sent as the `apisecret` field on every Janus request (not `token`, a stored-token auth
379	* field Janus also supports but this gateway rejects with a 403 "Unauthorized request" on
380	* create). Distinct from `apiToken`, which authenticates the WebSocket handshake itself.
381	*/
382	janusToken?: string;
383	/**
384	* RCE management api bearer token, sent as `Authorization: Bearer <apiToken>` on the WebSocket
385	* handshake. The Janus WebSocket host requires this; it is not the same credential as
386	* `janusToken`.
387	*/
388	apiToken: string;
389	iceServers?: IceServer[];
390	}
391
392	export interface RceVideoSignalingClientOptions {
393	createWebSocket?: (url: string, requestOptions: WebSocket.ClientOptions) => WebSocket;
394	/**
395	* How often to send a Janus keepalive. Defaults to 25000ms (Janus sessions time out at 60s).
396	*/
397	keepaliveIntervalMs?: number;
398	/**
399	* How long connect() waits (from connecting through the 'watch' request's response) before
400	* giving up and rejecting. Defaults to 20000ms.
401	*/
402	negotiationTimeoutMs?: number;
403	}
404
405	export interface RceVideoSignalingOffer {
406	offer: RceVideoJsep;
407	iceServers: IceServer[];
408	}
409
410	export interface RceVideoJsep {
411	type: string;
412	sdp: string;
413	}
414
415	export interface RceVideoSignalingClientEvents {
416	error: (error: Error) => void;
417	close: () => void;
418	}
419
420	interface PendingJanusRequest {
421	resolve: (message: JanusIncomingMessage) => void;
422	reject: (error: Error) => void;
423	}
424
425	interface JanusIncomingMessage {
426	janus: string;
427	transaction?: string;
428	data?: { id?: number };
429	jsep?: RceVideoJsep;
430	error?: { code?: number; reason?: string };
431	reason?: string;
432	/**
433	* Carries a streaming-plugin-level error (as opposed to a Janus-core-level `error` message)
434	* when a plugin request such as 'watch' fails, for example a wrong pin or unknown stream id
435	*/
436	plugindata?: {
437	data?: {
438	error?: string;
439	error_code?: number;
440	};
441	};
442	}

rokucommunity / roku-deploy / 30125731428

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous