// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. // See LICENSE.txt for license information. const MAX_WEBSOCKET_FAILS = 7; const MIN_WEBSOCKET_RETRY_TIME = 3000; // 3 sec const MAX_WEBSOCKET_RETRY_TIME = 300000; // 5 mins const JITTER_RANGE = 2000; // 2 sec const WEBSOCKET_HELLO = 'hello'; export type MessageListener = (msg: WebSocketMessage) => void; export type FirstConnectListener = () => void; export type ReconnectListener = () => void; export type MissedMessageListener = () => void; export type ErrorListener = (event: Event) => void; export type CloseListener = (connectFailCount: number) => void; export default class WebSocketClient { private conn: WebSocket | null; private connectionUrl: string | null; // responseSequence is the number to track a response sent // via the websocket. A response will always have the same sequence number // as the request. private responseSequence: number; // serverSequence is the incrementing sequence number from the // server-sent event stream. private serverSequence: number; private connectFailCount: number; private responseCallbacks: {[x: number]: ((msg: any) => void)}; /** * @deprecated Use messageListeners instead */ private eventCallback: MessageListener | null = null; /** * @deprecated Use firstConnectListeners instead */ private firstConnectCallback: FirstConnectListener | null = null; /** * @deprecated Use reconnectListeners instead */ private reconnectCallback: ReconnectListener | null = null; /** * @deprecated Use missedMessageListeners instead */ private missedEventCallback: MissedMessageListener | null = null; /** * @deprecated Use errorListeners instead */ private errorCallback: ErrorListener | null = null; /** * @deprecated Use closeListeners instead */ private closeCallback: CloseListener | null = null; private messageListeners = new Set(); private firstConnectListeners = new Set(); private reconnectListeners = new Set(); private missedMessageListeners = new Set(); private errorListeners = new Set(); private closeListeners = new Set(); private connectionId: string | null; constructor() { this.conn = null; this.connectionUrl = null; this.responseSequence = 1; this.serverSequence = 0; this.connectFailCount = 0; this.responseCallbacks = {}; this.connectionId = ''; } // on connect, only send auth cookie and blank state. // on hello, get the connectionID and store it. // on reconnect, send cookie, connectionID, sequence number. initialize(connectionUrl = this.connectionUrl, token?: string) { if (this.conn) { return; } if (connectionUrl == null) { console.log('websocket must have connection url'); //eslint-disable-line no-console return; } if (this.connectFailCount === 0) { console.log('websocket connecting to ' + connectionUrl); //eslint-disable-line no-console } // Add connection id, and last_sequence_number to the query param. // We cannot use a cookie because it will bleed across tabs. // We cannot also send it as part of the auth_challenge, because the session cookie is already sent with the request. this.conn = new WebSocket(`${connectionUrl}?connection_id=${this.connectionId}&sequence_number=${this.serverSequence}`); this.connectionUrl = connectionUrl; this.conn.onopen = () => { if (token) { this.sendMessage('authentication_challenge', {token}); } if (this.connectFailCount > 0) { console.log('websocket re-established connection'); //eslint-disable-line no-console this.reconnectCallback?.(); this.reconnectListeners.forEach((listener) => listener()); } else if (this.firstConnectCallback || this.firstConnectListeners.size > 0) { this.firstConnectCallback?.(); this.firstConnectListeners.forEach((listener) => listener()); } this.connectFailCount = 0; }; this.conn.onclose = () => { this.conn = null; this.responseSequence = 1; if (this.connectFailCount === 0) { console.log('websocket closed'); //eslint-disable-line no-console } this.connectFailCount++; this.closeCallback?.(this.connectFailCount); this.closeListeners.forEach((listener) => listener(this.connectFailCount)); let retryTime = MIN_WEBSOCKET_RETRY_TIME; // If we've failed a bunch of connections then start backing off if (this.connectFailCount > MAX_WEBSOCKET_FAILS) { retryTime = MIN_WEBSOCKET_RETRY_TIME * this.connectFailCount * this.connectFailCount; if (retryTime > MAX_WEBSOCKET_RETRY_TIME) { retryTime = MAX_WEBSOCKET_RETRY_TIME; } } // Applying jitter to avoid thundering herd problems. retryTime += Math.random() * JITTER_RANGE; setTimeout( () => { this.initialize(connectionUrl, token); }, retryTime, ); }; this.conn.onerror = (evt) => { if (this.connectFailCount <= 1) { console.log('websocket error'); //eslint-disable-line no-console console.log(evt); //eslint-disable-line no-console } this.errorCallback?.(evt); this.errorListeners.forEach((listener) => listener(evt)); }; this.conn.onmessage = (evt) => { const msg = JSON.parse(evt.data); if (msg.seq_reply) { // This indicates a reply to a websocket request. // We ignore sequence number validation of message responses // and only focus on the purely server side event stream. if (msg.error) { console.log(msg); //eslint-disable-line no-console } if (this.responseCallbacks[msg.seq_reply]) { this.responseCallbacks[msg.seq_reply](msg); Reflect.deleteProperty(this.responseCallbacks, msg.seq_reply); } } else if (this.eventCallback || this.messageListeners.size > 0) { // We check the hello packet, which is always the first packet in a stream. if (msg.event === WEBSOCKET_HELLO && (this.missedEventCallback || this.missedMessageListeners.size > 0)) { console.log('got connection id ', msg.data.connection_id); //eslint-disable-line no-console // If we already have a connectionId present, and server sends a different one, // that means it's either a long timeout, or server restart, or sequence number is not found. // Then we do the sync calls, and reset sequence number to 0. if (this.connectionId !== '' && this.connectionId !== msg.data.connection_id) { console.log('long timeout, or server restart, or sequence number is not found.'); //eslint-disable-line no-console this.missedEventCallback?.(); for (const listener of this.missedMessageListeners) { try { listener(); } catch (e) { console.log(`missed message listener "${listener.name}" failed: ${e}`); // eslint-disable-line no-console } } this.serverSequence = 0; } // If it's a fresh connection, we have to set the connectionId regardless. // And if it's an existing connection, setting it again is harmless, and keeps the code simple. this.connectionId = msg.data.connection_id; } // Now we check for sequence number, and if it does not match, // we just disconnect and reconnect. if (msg.seq !== this.serverSequence) { console.log('missed websocket event, act_seq=' + msg.seq + ' exp_seq=' + this.serverSequence); //eslint-disable-line no-console // We are not calling this.close() because we need to auto-restart. this.connectFailCount = 0; this.responseSequence = 1; this.conn?.close(); // Will auto-reconnect after MIN_WEBSOCKET_RETRY_TIME. return; } this.serverSequence = msg.seq + 1; this.eventCallback?.(msg); this.messageListeners.forEach((listener) => listener(msg)); } }; } /** * @deprecated Use addMessageListener instead */ setEventCallback(callback: MessageListener) { this.eventCallback = callback; } addMessageListener(listener: MessageListener) { this.messageListeners.add(listener); if (this.messageListeners.size > 5) { // eslint-disable-next-line no-console console.warn(`WebSocketClient has ${this.messageListeners.size} message listeners registered`); } } removeMessageListener(listener: MessageListener) { this.messageListeners.delete(listener); } /** * @deprecated Use addFirstConnectListener instead */ setFirstConnectCallback(callback: FirstConnectListener) { this.firstConnectCallback = callback; } addFirstConnectListener(listener: FirstConnectListener) { this.firstConnectListeners.add(listener); if (this.firstConnectListeners.size > 5) { // eslint-disable-next-line no-console console.warn(`WebSocketClient has ${this.firstConnectListeners.size} first connect listeners registered`); } } removeFirstConnectListener(listener: FirstConnectListener) { this.firstConnectListeners.delete(listener); } /** * @deprecated Use addReconnectListener instead */ setReconnectCallback(callback: ReconnectListener) { this.reconnectCallback = callback; } addReconnectListener(listener: ReconnectListener) { this.reconnectListeners.add(listener); if (this.reconnectListeners.size > 5) { // eslint-disable-next-line no-console console.warn(`WebSocketClient has ${this.reconnectListeners.size} reconnect listeners registered`); } } removeReconnectListener(listener: ReconnectListener) { this.reconnectListeners.delete(listener); } /** * @deprecated Use addMissedMessageListener instead */ setMissedEventCallback(callback: MissedMessageListener) { this.missedEventCallback = callback; } addMissedMessageListener(listener: MissedMessageListener) { this.missedMessageListeners.add(listener); if (this.missedMessageListeners.size > 5) { // eslint-disable-next-line no-console console.warn(`WebSocketClient has ${this.missedMessageListeners.size} missed message listeners registered`); } } removeMissedMessageListener(listener: MissedMessageListener) { this.missedMessageListeners.delete(listener); } /** * @deprecated Use addErrorListener instead */ setErrorCallback(callback: ErrorListener) { this.errorCallback = callback; } addErrorListener(listener: ErrorListener) { this.errorListeners.add(listener); if (this.errorListeners.size > 5) { // eslint-disable-next-line no-console console.warn(`WebSocketClient has ${this.errorListeners.size} error listeners registered`); } } removeErrorListener(listener: ErrorListener) { this.errorListeners.delete(listener); } /** * @deprecated Use addCloseListener instead */ setCloseCallback(callback: CloseListener) { this.closeCallback = callback; } addCloseListener(listener: CloseListener) { this.closeListeners.add(listener); if (this.closeListeners.size > 5) { // eslint-disable-next-line no-console console.warn(`WebSocketClient has ${this.closeListeners.size} close listeners registered`); } } removeCloseListener(listener: CloseListener) { this.closeListeners.delete(listener); } close() { this.connectFailCount = 0; this.responseSequence = 1; if (this.conn && this.conn.readyState === WebSocket.OPEN) { this.conn.onclose = () => {}; this.conn.close(); this.conn = null; console.log('websocket closed'); //eslint-disable-line no-console } } sendMessage(action: string, data: any, responseCallback?: () => void) { const msg = { action, seq: this.responseSequence++, data, }; if (responseCallback) { this.responseCallbacks[msg.seq] = responseCallback; } if (this.conn && this.conn.readyState === WebSocket.OPEN) { this.conn.send(JSON.stringify(msg)); } else if (!this.conn || this.conn.readyState === WebSocket.CLOSED) { this.conn = null; this.initialize(); } } userTyping(channelId: string, parentId: string, callback?: () => void) { const data = { channel_id: channelId, parent_id: parentId, }; this.sendMessage('user_typing', data, callback); } userUpdateActiveStatus(userIsActive: boolean, manual: boolean, callback?: () => void) { const data = { user_is_active: userIsActive, manual, }; this.sendMessage('user_update_active_status', data, callback); } getStatuses(callback?: () => void) { this.sendMessage('get_statuses', null, callback); } getStatusesByIds(userIds: string[], callback?: () => void) { const data = { user_ids: userIds, }; this.sendMessage('get_statuses_by_ids', data, callback); } } export type WebSocketBroadcast = { omit_users: Record; user_id: string; channel_id: string; team_id: string; } export type WebSocketMessage = { event: string; data: T; broadcast: WebSocketBroadcast; seq: number; }