From 17fd4ee2787f11748113972e629401c7db15effa Mon Sep 17 00:00:00 2001 From: Elias Nahum Date: Fri, 7 May 2021 08:20:41 -0400 Subject: [PATCH] MM-32591: Reliable Websockets: Client side changes (#5362) --- app/actions/websocket/index.ts | 36 ++++-- app/client/websocket/index.ts | 109 +++++++++++++++--- .../network_indicator/network_indicator.js | 50 ++++---- app/mm-redux/types/config.ts | 1 + 4 files changed, 146 insertions(+), 50 deletions(-) diff --git a/app/actions/websocket/index.ts b/app/actions/websocket/index.ts index d8e980661..4457e96fa 100644 --- a/app/actions/websocket/index.ts +++ b/app/actions/websocket/index.ts @@ -57,6 +57,7 @@ export function init(additionalOptions: any = {}) { connUrl += `${Client4.getUrlVersion()}/websocket`; websocketClient.setFirstConnectCallback(() => dispatch(handleFirstConnect())); websocketClient.setEventCallback((evt: WebSocketMessage) => dispatch(handleEvent(evt))); + websocketClient.setMissedEventsCallback(() => dispatch(doMissedEvents())); websocketClient.setReconnectCallback(() => dispatch(handleReconnect())); websocketClient.setCloseCallback((connectFailCount: number) => dispatch(handleClose(connectFailCount))); @@ -81,15 +82,19 @@ export function close(shouldReconnect = false): GenericAction { }; } +function wsConnected(timestamp = Date.now()) { + return { + type: GeneralTypes.WEBSOCKET_SUCCESS, + timestamp, + data: null, + }; +} + export function doFirstConnect(now: number) { return async (dispatch: DispatchFunc, getState: GetStateFunc): Promise => { const state = getState(); const {lastDisconnectAt} = state.websocket; - const actions: Array = [{ - type: GeneralTypes.WEBSOCKET_SUCCESS, - timestamp: now, - data: null, - }]; + const actions: Array = [wsConnected(now)]; if (isMinimumServerVersion(Client4.getServerVersion(), 5, 14) && lastDisconnectAt) { const currentUserId = getCurrentUserId(state); @@ -112,6 +117,14 @@ export function doFirstConnect(now: number) { }; } +export function doMissedEvents() { + return async (dispatch: DispatchFunc): Promise => { + dispatch(wsConnected()); + + return {data: true}; + }; +} + export function doReconnect(now: number) { return async (dispatch: DispatchFunc, getState: GetStateFunc): Promise => { const state = getState(); @@ -122,11 +135,7 @@ export function doReconnect(now: number) { const {lastDisconnectAt} = state.websocket; const actions: Array = []; - dispatch({ - type: GeneralTypes.WEBSOCKET_SUCCESS, - timestamp: now, - data: null, - }); + dispatch(wsConnected(now)); try { const {data: me}: any = await dispatch(loadMe(null, null, true)); @@ -273,13 +282,16 @@ export function handleUserTypingEvent(msg: WebSocketMessage) { } function handleFirstConnect() { - return (dispatch: DispatchFunc) => { + return (dispatch: DispatchFunc, getState: GetStateFunc) => { + const state = getState(); + const config = getConfig(state); const now = Date.now(); - if (reconnect) { + if (reconnect && config?.EnableReliableWebSockets !== 'true') { reconnect = false; return dispatch(doReconnect(now)); } + return dispatch(doFirstConnect(now)); }; } diff --git a/app/client/websocket/index.ts b/app/client/websocket/index.ts index b8ddb3ce1..312c24fa2 100644 --- a/app/client/websocket/index.ts +++ b/app/client/websocket/index.ts @@ -3,6 +3,10 @@ import {Platform} from 'react-native'; +import {WebsocketEvents} from '@constants'; +import {getConfig} from '@mm-redux/selectors/entities/general'; +import Store from '@store/store'; + const MAX_WEBSOCKET_FAILS = 7; const MIN_WEBSOCKET_RETRY_TIME = 3000; // 3 sec @@ -11,22 +15,34 @@ const MAX_WEBSOCKET_RETRY_TIME = 300000; // 5 mins class WebSocketClient { conn?: WebSocket; connectionUrl: string; + connectionTimeout: any; + connectionId: string | null; token: string|null; - sequence: number; + + // responseSequence is the number to track a response sent + // via the websocket. A response will always have the same sequence number + // as the request. + responseSequence: number; + + // serverSequence is the incrementing sequence number from the + // server-sent event stream. + serverSequence: number; connectFailCount: number; eventCallback?: Function; firstConnectCallback?: Function; + missedEventsCallback?: Function; reconnectCallback?: Function; errorCallback?: Function; closeCallback?: Function; connectingCallback?: Function; stop: boolean; - connectionTimeout: any; constructor() { this.connectionUrl = ''; + this.connectionId = ''; this.token = null; - this.sequence = 1; + this.responseSequence = 1; + this.serverSequence = 0; this.connectFailCount = 0; this.stop = false; } @@ -55,10 +71,6 @@ class WebSocketClient { return; } - if (this.connectFailCount === 0) { - console.log('websocket connecting to ' + connectionUrl); //eslint-disable-line no-console - } - if (this.connectingCallback) { this.connectingCallback(); } @@ -87,11 +99,30 @@ class WebSocketClient { return; } - this.conn = new WebSocket(connectionUrl, [], {headers: {origin}, ...(additionalOptions || {})}); + let url = connectionUrl; + const config = getConfig(Store.redux?.getState()); + const reliableWebSockets = config?.EnableReliableWebSockets === 'true'; + + if (reliableWebSockets) { + // Add connection id, and last_sequence_number to the query param. + // We cannot also send it as part of the auth_challenge, because the session cookie is already sent with the request. + url = `${connectionUrl}?connection_id=${this.connectionId}&sequence_number=${this.serverSequence}`; + } + + if (this.connectFailCount === 0) { + console.log('websocket connecting to ' + url); //eslint-disable-line no-console + } + + this.conn = new WebSocket(url, [], {headers: {origin}, ...(additionalOptions || {})}); this.connectionUrl = connectionUrl; this.token = token; this.conn!.onopen = () => { + // No need to reset sequence number here. + if (!reliableWebSockets) { + this.serverSequence = 0; + } + if (token) { // we check for the platform as a workaround until we fix on the server that further authentications // are ignored @@ -100,8 +131,10 @@ class WebSocketClient { if (this.connectFailCount > 0) { console.log('websocket re-established connection'); //eslint-disable-line no-console - if (this.reconnectCallback) { + if (!reliableWebSockets && this.reconnectCallback) { this.reconnectCallback(); + } else if (reliableWebSockets && this.serverSequence && this.missedEventsCallback) { + this.missedEventsCallback(); } } else if (this.firstConnectCallback) { this.firstConnectCallback(); @@ -113,7 +146,7 @@ class WebSocketClient { this.conn!.onclose = () => { this.conn = undefined; - this.sequence = 1; + this.responseSequence = 1; if (this.connectFailCount === 0) { console.log('websocket closed'); //eslint-disable-line no-console @@ -164,11 +197,55 @@ class WebSocketClient { this.conn!.onmessage = (evt: any) => { const msg = JSON.parse(evt.data); + + // 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.seq_reply) { if (msg.error) { console.warn(msg); //eslint-disable-line no-console } } else if (this.eventCallback) { + if (reliableWebSockets) { + // We check the hello packet, which is always the first packet in a stream. + if (msg.event === WebsocketEvents.HELLO && this.reconnectCallback) { + //eslint-disable-next-line no-console + console.log('got connection id ', msg.data.connection_id); + + // 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) { + //eslint-disable-next-line no-console + console.log('long timeout, or server restart, or sequence number is not found.'); + this.reconnectCallback(); + 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) { + // eslint-disable-next-line no-console + console.log('missed websocket event, act_seq=' + msg.seq + ' exp_seq=' + this.serverSequence); + + // 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; + } + } else if (msg.seq !== this.serverSequence && this.reconnectCallback) { + // eslint-disable-next-line no-console + console.log('missed websocket event, act_seq=' + msg.seq + ' exp_seq=' + this.serverSequence); + this.reconnectCallback(); + } + + this.serverSequence = msg.seq + 1; this.eventCallback(msg); } }; @@ -187,6 +264,10 @@ class WebSocketClient { this.firstConnectCallback = callback; } + setMissedEventsCallback(callback: Function) { + this.missedEventsCallback = callback; + } + setReconnectCallback(callback: Function) { this.reconnectCallback = callback; } @@ -202,19 +283,17 @@ class WebSocketClient { close(stop = false) { this.stop = stop; this.connectFailCount = 0; - this.sequence = 1; + this.responseSequence = 1; + if (this.conn && this.conn.readyState === WebSocket.OPEN) { - this.conn.onclose = () => {}; //eslint-disable-line @typescript-eslint/no-empty-function this.conn.close(); - this.conn = undefined; - console.log('websocket closed'); //eslint-disable-line no-console } } sendMessage(action: string, data: any) { const msg = { action, - seq: this.sequence++, + seq: this.responseSequence++, data, }; diff --git a/app/components/network_indicator/network_indicator.js b/app/components/network_indicator/network_indicator.js index 45367bbe5..3fc193755 100644 --- a/app/components/network_indicator/network_indicator.js +++ b/app/components/network_indicator/network_indicator.js @@ -14,17 +14,16 @@ import { import NetInfo from '@react-native-community/netinfo'; import {SafeAreaView} from 'react-native-safe-area-context'; -import {RequestStatus} from '@mm-redux/constants'; -import EventEmitter from '@mm-redux/utils/event_emitter'; - import CompassIcon from '@components/compass_icon'; import FormattedText from '@components/formatted_text'; import {ViewTypes} from '@constants'; import {INDICATOR_BAR_HEIGHT} from '@constants/view'; -import networkConnectionListener, {checkConnection} from '@utils/network'; -import {t} from '@utils/i18n'; - import PushNotifications from '@init/push_notifications'; +import {debounce} from '@mm-redux/actions/helpers'; +import {RequestStatus} from '@mm-redux/constants'; +import EventEmitter from '@mm-redux/utils/event_emitter'; +import {t} from '@utils/i18n'; +import networkConnectionListener, {checkConnection} from '@utils/network'; const MAX_WEBSOCKET_RETRIES = 3; const CONNECTION_RETRY_SECONDS = 5; @@ -226,23 +225,7 @@ export default class NetworkIndicator extends PureComponent { }; handleAppStateChange = async (appState) => { - const {actions, currentChannelId} = this.props; - const active = appState === 'active'; - if (active) { - this.connect(true); - - if (currentChannelId) { - // Clear the notifications for the current channel after one second - // this is done so we can cancel it in case the app is brought to the - // foreground by tapping a notification from another channel - this.clearNotificationTimeout = setTimeout(() => { - PushNotifications.clearChannelNotifications(currentChannelId); - actions.markChannelViewedAndReadOnReconnect(currentChannelId); - }, 1000); - } - } else { - this.handleWebSocket(false); - } + this.onStateChange(appState); }; handleConnectionChange = ({hasInternet, serverReachable}) => { @@ -302,6 +285,27 @@ export default class NetworkIndicator extends PureComponent { this.show(); }; + onStateChange = debounce((appState) => { + const {actions, currentChannelId} = this.props; + const active = appState === 'active'; + + if (active) { + this.connect(true); + + if (currentChannelId) { + // Clear the notifications for the current channel after one second + // this is done so we can cancel it in case the app is brought to the + // foreground by tapping a notification from another channel + this.clearNotificationTimeout = setTimeout(() => { + PushNotifications.clearChannelNotifications(currentChannelId); + actions.markChannelViewedAndReadOnReconnect(currentChannelId); + }, 1000); + } + } else { + this.handleWebSocket(false); + } + }, 300); + show = () => { if (!this.visible) { this.visible = true; diff --git a/app/mm-redux/types/config.ts b/app/mm-redux/types/config.ts index 8facf9808..b7c9ef2ef 100644 --- a/app/mm-redux/types/config.ts +++ b/app/mm-redux/types/config.ts @@ -75,6 +75,7 @@ export type Config = { EnablePreviewFeatures: string; EnablePreviewModeBanner: string; EnablePublicLink: string; + EnableReliableWebSockets: string; EnableSaml: string; EnableSignInWithEmail: string; EnableSignInWithUsername: string;