564 строки
20 KiB
TypeScript
564 строки
20 KiB
TypeScript
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
|
|
// See LICENSE.txt for license information.
|
|
|
|
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 type WebSocketClientConfig = {
|
|
maxWebSocketFails: number;
|
|
minWebSocketRetryTime: number;
|
|
maxWebSocketRetryTime: number;
|
|
reconnectJitterRange: number;
|
|
newWebSocketFn: (url: string) => WebSocket;
|
|
clientPingInterval: number;
|
|
}
|
|
|
|
const defaultWebSocketClientConfig: WebSocketClientConfig = {
|
|
maxWebSocketFails: 7,
|
|
minWebSocketRetryTime: 3000, // 3 seconds
|
|
maxWebSocketRetryTime: 300000, // 5 minutes
|
|
reconnectJitterRange: 2000, // 2 seconds
|
|
newWebSocketFn: (url: string) => {
|
|
return new WebSocket(url);
|
|
},
|
|
clientPingInterval: 30000, // 30 seconds
|
|
};
|
|
|
|
export default class WebSocketClient {
|
|
private config: WebSocketClientConfig;
|
|
|
|
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<MessageListener>();
|
|
private firstConnectListeners = new Set<FirstConnectListener>();
|
|
private reconnectListeners = new Set<ReconnectListener>();
|
|
private missedMessageListeners = new Set<MissedMessageListener>();
|
|
private errorListeners = new Set<ErrorListener>();
|
|
private closeListeners = new Set<CloseListener>();
|
|
|
|
private connectionId: string | null;
|
|
private serverHostname: string | null;
|
|
private postedAck: boolean;
|
|
|
|
private pingInterval: ReturnType<typeof setInterval> | null;
|
|
|
|
private reconnectTimeout: ReturnType<typeof setTimeout> | null;
|
|
|
|
constructor(config?: Partial<WebSocketClientConfig>) {
|
|
this.conn = null;
|
|
this.connectionUrl = null;
|
|
this.responseSequence = 1;
|
|
this.serverSequence = 0;
|
|
this.connectFailCount = 0;
|
|
this.responseCallbacks = {};
|
|
this.connectionId = '';
|
|
this.serverHostname = '';
|
|
this.postedAck = false;
|
|
this.reconnectTimeout = null;
|
|
this.config = {...defaultWebSocketClientConfig, ...config};
|
|
this.pingInterval = null;
|
|
}
|
|
|
|
// 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, postedAck?: boolean) {
|
|
if (this.conn) {
|
|
return;
|
|
}
|
|
|
|
// We have a timeout waiting to re-initialize the websocket.
|
|
// We should wait until that fires before initializing,
|
|
// otherwise we may not respect the configured backoff.
|
|
if (this.reconnectTimeout) {
|
|
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
|
|
}
|
|
|
|
if (typeof postedAck != 'undefined') {
|
|
this.postedAck = postedAck;
|
|
}
|
|
|
|
// 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.
|
|
const websocketUrl = `${connectionUrl}?connection_id=${this.connectionId}&sequence_number=${this.serverSequence}${this.postedAck ? '&posted_ack=true' : ''}`;
|
|
if (this.config.newWebSocketFn) {
|
|
this.conn = this.config.newWebSocketFn(websocketUrl);
|
|
} else {
|
|
this.conn = new WebSocket(websocketUrl);
|
|
}
|
|
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.stopPingInterval();
|
|
|
|
// Send a ping immediately to test the socket
|
|
var waitingForPong = true;
|
|
this.ping(() => {
|
|
waitingForPong = false;
|
|
});
|
|
|
|
// And every 30 seconds after, checking to ensure
|
|
// we're getting responses from the server
|
|
this.pingInterval = setInterval(
|
|
() => {
|
|
if (!waitingForPong) {
|
|
waitingForPong = true;
|
|
this.ping(() => {
|
|
waitingForPong = false;
|
|
});
|
|
return;
|
|
}
|
|
|
|
console.log('ping received no response within time limit: re-establishing websocket'); //eslint-disable-line no-console
|
|
|
|
// We are not calling this.close() because we need to auto-restart.
|
|
this.connectFailCount = 0;
|
|
this.responseSequence = 1;
|
|
this.stopPingInterval();
|
|
this.conn?.close(); // Will auto-reconnect after configured retry time
|
|
},
|
|
this.config.clientPingInterval);
|
|
|
|
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));
|
|
|
|
// Make sure we stop pinging if the connection is closed
|
|
this.stopPingInterval();
|
|
|
|
// If we've failed a bunch of connections then start backing off
|
|
let retryTime = this.config.minWebSocketRetryTime;
|
|
if (this.connectFailCount > this.config.maxWebSocketFails) {
|
|
retryTime = retryTime * this.connectFailCount * this.connectFailCount;
|
|
if (retryTime > this.config.maxWebSocketRetryTime) {
|
|
retryTime = this.config.maxWebSocketRetryTime;
|
|
}
|
|
}
|
|
|
|
// Applying jitter to avoid thundering herd problems.
|
|
retryTime += Math.random() * this.config.reconnectJitterRange;
|
|
|
|
// If we already have a reconnect timeout waiting,
|
|
// we should let that handle the next connection.
|
|
if (this.reconnectTimeout) {
|
|
return;
|
|
}
|
|
|
|
this.reconnectTimeout = setTimeout(
|
|
() => {
|
|
this.reconnectTimeout = null;
|
|
this.initialize(this.connectionUrl, token, this.postedAck);
|
|
},
|
|
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;
|
|
|
|
// Also update the server hostname
|
|
this.serverHostname = msg.data.server_hostname;
|
|
}
|
|
|
|
// 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.reconnectTimeout) {
|
|
clearTimeout(this.reconnectTimeout);
|
|
this.reconnectTimeout = null;
|
|
}
|
|
this.stopPingInterval();
|
|
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
|
|
}
|
|
}
|
|
|
|
stopPingInterval() {
|
|
if (this.pingInterval) {
|
|
clearInterval(this.pingInterval);
|
|
this.pingInterval = null;
|
|
}
|
|
}
|
|
|
|
ping(responseCallback?: (msg: any) => void) {
|
|
const msg = {
|
|
action: 'ping',
|
|
seq: this.responseSequence++,
|
|
};
|
|
|
|
if (responseCallback) {
|
|
this.responseCallbacks[msg.seq] = responseCallback;
|
|
}
|
|
|
|
if (this.conn && this.conn.readyState === WebSocket.OPEN) {
|
|
this.conn.send(JSON.stringify(msg));
|
|
}
|
|
}
|
|
|
|
sendMessage(action: string, data: any, responseCallback?: (msg: any) => void) {
|
|
const msg = {
|
|
action,
|
|
seq: this.responseSequence++,
|
|
data,
|
|
};
|
|
|
|
if (responseCallback) {
|
|
this.responseCallbacks[msg.seq] = responseCallback;
|
|
}
|
|
|
|
// Only try to send the message if the websocket is open.
|
|
// If the websocket is closed here, we will drop the message.
|
|
if (this.conn && this.conn.readyState === WebSocket.OPEN) {
|
|
this.conn.send(JSON.stringify(msg));
|
|
}
|
|
}
|
|
|
|
userTyping(channelId: string, parentId: string, callback?: () => void) {
|
|
const data = {
|
|
channel_id: channelId,
|
|
parent_id: parentId,
|
|
};
|
|
this.sendMessage('user_typing', data, callback);
|
|
}
|
|
|
|
updateActiveChannel(channelId: string, callback?: (msg: any) => void) {
|
|
const data = {
|
|
channel_id: channelId,
|
|
};
|
|
this.sendMessage('presence', data, callback);
|
|
}
|
|
|
|
updateActiveTeam(teamId: string, callback?: (msg: any) => void) {
|
|
const data = {
|
|
team_id: teamId,
|
|
};
|
|
this.sendMessage('presence', data, callback);
|
|
}
|
|
|
|
updateActiveThread(isThreadView: boolean, channelId: string, callback?: (msg: any) => void) {
|
|
const data = {
|
|
thread_channel_id: channelId,
|
|
is_thread_view: isThreadView,
|
|
};
|
|
this.sendMessage('presence', data, callback);
|
|
}
|
|
|
|
userUpdateActiveStatus(userIsActive: boolean, manual: boolean, callback?: () => void) {
|
|
const data = {
|
|
user_is_active: userIsActive,
|
|
manual,
|
|
};
|
|
this.sendMessage('user_update_active_status', data, callback);
|
|
}
|
|
|
|
acknowledgePostedNotification(postId: string, status: string, reason?: string, postedData?: string) {
|
|
const data = {
|
|
post_id: postId,
|
|
user_agent: window.navigator.userAgent,
|
|
status,
|
|
reason,
|
|
data: postedData,
|
|
};
|
|
|
|
this.sendMessage('posted_notify_ack', data);
|
|
}
|
|
|
|
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<string, boolean>;
|
|
user_id: string;
|
|
channel_id: string;
|
|
team_id: string;
|
|
}
|
|
|
|
export type WebSocketMessage<T = any> = {
|
|
event: string;
|
|
data: T;
|
|
broadcast: WebSocketBroadcast;
|
|
seq: number;
|
|
}
|