import { actions } from '../types/protocolmessagecommon'; import ProtocolMessage, { stringify as stringifyProtocolMessage, fromValues as protocolMessageFromValues, } from 'common/lib/types/protocolmessage'; import * as Utils from 'common/lib/util/utils'; import Protocol, { PendingMessage } from './protocol'; import Defaults, { getAgentString } from 'common/lib/util/defaults'; import Platform, { TransportImplementations } from 'common/platform'; import EventEmitter from '../util/eventemitter'; import MessageQueue from './messagequeue'; import Logger from '../util/logger'; import ConnectionStateChange from 'common/lib/client/connectionstatechange'; import ConnectionErrors, { isRetriable } from './connectionerrors'; import ErrorInfo, { IPartialErrorInfo, PartialErrorInfo } from 'common/lib/types/errorinfo'; import Auth from 'common/lib/client/auth'; import Message, { getMessagesSize } from 'common/lib/types/message'; import Multicaster, { MulticasterInstance } from 'common/lib/util/multicaster'; import Transport, { TransportCtor } from './transport'; import * as API from '../../../../ably'; import { ErrCallback } from 'common/types/utils'; import HttpStatusCodes from 'common/constants/HttpStatusCodes'; import BaseRealtime from '../client/baserealtime'; import { NormalisedClientOptions } from 'common/types/ClientOptions'; import TransportName, { TransportNames } from 'common/constants/TransportName'; let globalObject = typeof global !== 'undefined' ? global : typeof window !== 'undefined' ? window : self; const haveWebStorage = () => typeof Platform.WebStorage !== 'undefined' && Platform.WebStorage?.localSupported; const haveSessionStorage = () => typeof Platform.WebStorage !== 'undefined' && Platform.WebStorage?.sessionSupported; const noop = function () {}; const transportPreferenceName = 'ably-transport-preference'; function bundleWith(dest: ProtocolMessage, src: ProtocolMessage, maxSize: number) { let action; if (dest.channel !== src.channel) { /* RTL6d3 */ return false; } if ((action = dest.action) !== actions.PRESENCE && action !== actions.MESSAGE) { /* RTL6d - can only bundle messages or presence */ return false; } if (action !== src.action) { /* RTL6d4 */ return false; } const kind = action === actions.PRESENCE ? 'presence' : 'messages', proposed = (dest as Record)[kind].concat((src as Record)[kind]), size = getMessagesSize(proposed); if (size > maxSize) { /* RTL6d1 */ return false; } if (!Utils.allSame(proposed, 'clientId')) { /* RTL6d2 */ return false; } if ( !proposed.every(function (msg: Message) { return !msg.id; }) ) { /* RTL6d7 */ return false; } /* we're good to go! */ (dest as Record)[kind] = proposed; return true; } type RecoveryContext = { connectionKey: string; msgSerial: number; channelSerials: { [name: string]: string }; }; function decodeRecoveryKey(recoveryKey: NormalisedClientOptions['recover']): RecoveryContext | null { try { return JSON.parse(recoveryKey as string); } catch (e) { return null; } } export class TransportParams { options: NormalisedClientOptions; host: string | null; mode: string; format?: Utils.Format; connectionKey?: string; stream?: any; heartbeats?: boolean; constructor(options: NormalisedClientOptions, host: string | null, mode: string, connectionKey?: string) { this.options = options; this.host = host; this.mode = mode; this.connectionKey = connectionKey; this.format = options.useBinaryProtocol ? Utils.Format.msgpack : Utils.Format.json; } getConnectParams(authParams: Record): Record { const params = authParams ? Utils.copy(authParams) : {}; const options = this.options; switch (this.mode) { case 'resume': params.resume = this.connectionKey as string; break; case 'recover': { const recoveryContext = decodeRecoveryKey(options.recover); if (recoveryContext) { params.recover = recoveryContext.connectionKey; } break; } default: } if (options.clientId !== undefined) { params.clientId = options.clientId; } if (options.echoMessages === false) { params.echo = 'false'; } if (this.format !== undefined) { params.format = this.format; } if (this.stream !== undefined) { params.stream = this.stream; } if (this.heartbeats !== undefined) { params.heartbeats = this.heartbeats; } params.v = Defaults.protocolVersion; params.agent = getAgentString(this.options); if (options.transportParams !== undefined) { Utils.mixin(params, options.transportParams); } return params as Record; } toString(): string { let result = '[mode=' + this.mode; if (this.host) { result += ',host=' + this.host; } if (this.connectionKey) { result += ',connectionKey=' + this.connectionKey; } if (this.format) { result += ',format=' + this.format; } result += ']'; return result; } } type ConnectionState = { state: string; terminal?: boolean; queueEvents?: boolean; sendEvents?: boolean; failState?: string; retryDelay?: number; retryImmediately?: boolean; error?: IPartialErrorInfo; }; class ConnectionManager extends EventEmitter { supportedTransports: Partial> = {}; realtime: BaseRealtime; options: NormalisedClientOptions; states: Record; state: ConnectionState; errorReason: IPartialErrorInfo | null; queuedMessages: MessageQueue; msgSerial: number; connectionDetails?: Record; connectionId?: string; connectionKey?: string; connectionStateTtl: number; maxIdleInterval: number | null; transports: TransportName[]; baseTransport?: TransportName; webSocketTransportAvailable?: true; transportPreference: string | null; domains: string[]; activeProtocol: null | Protocol; pendingTransport?: Transport; proposedTransport?: Transport; host: string | null; lastAutoReconnectAttempt: number | null; lastActivity: number | null; forceFallbackHost: boolean; transitionTimer?: number | NodeJS.Timeout | null; suspendTimer?: number | NodeJS.Timeout | null; retryTimer?: number | NodeJS.Timeout | null; disconnectedRetryCount: number = 0; pendingChannelMessagesState: { // Whether a message is currently being processed isProcessing: boolean; // The messages remaining to be processed (excluding any message currently being processed) queue: { message: ProtocolMessage; transport: Transport }[]; } = { isProcessing: false, queue: [] }; webSocketSlowTimer: NodeJS.Timeout | null; wsCheckResult: boolean | null; webSocketGiveUpTimer: NodeJS.Timeout | null; abandonedWebSocket: boolean; connectCounter: number; constructor(realtime: BaseRealtime, options: NormalisedClientOptions) { super(realtime.logger); this.realtime = realtime; this.initTransports(); this.options = options; const timeouts = options.timeouts; /* connectingTimeout: leave webSocketConnectTimeout (~6s) to try the * websocket transport, then realtimeRequestTimeout (~10s) to establish * the base transport in case that fails */ const connectingTimeout = timeouts.webSocketConnectTimeout + timeouts.realtimeRequestTimeout; this.states = { initialized: { state: 'initialized', terminal: false, queueEvents: true, sendEvents: false, failState: 'disconnected', }, connecting: { state: 'connecting', terminal: false, queueEvents: true, sendEvents: false, retryDelay: connectingTimeout, failState: 'disconnected', }, connected: { state: 'connected', terminal: false, queueEvents: false, sendEvents: true, failState: 'disconnected', }, disconnected: { state: 'disconnected', terminal: false, queueEvents: true, sendEvents: false, retryDelay: timeouts.disconnectedRetryTimeout, failState: 'disconnected', }, suspended: { state: 'suspended', terminal: false, queueEvents: false, sendEvents: false, retryDelay: timeouts.suspendedRetryTimeout, failState: 'suspended', }, closing: { state: 'closing', terminal: false, queueEvents: false, sendEvents: false, retryDelay: timeouts.realtimeRequestTimeout, failState: 'closed', }, closed: { state: 'closed', terminal: true, queueEvents: false, sendEvents: false, failState: 'closed' }, failed: { state: 'failed', terminal: true, queueEvents: false, sendEvents: false, failState: 'failed' }, }; this.state = this.states.initialized; this.errorReason = null; this.queuedMessages = new MessageQueue(this.logger); this.msgSerial = 0; this.connectionDetails = undefined; this.connectionId = undefined; this.connectionKey = undefined; this.connectionStateTtl = timeouts.connectionStateTtl; this.maxIdleInterval = null; this.transports = Utils.intersect(options.transports || Defaults.defaultTransports, this.supportedTransports); this.transportPreference = null; if (this.transports.includes(TransportNames.WebSocket)) { this.webSocketTransportAvailable = true; } if (this.transports.includes(TransportNames.XhrPolling)) { this.baseTransport = TransportNames.XhrPolling; } else if (this.transports.includes(TransportNames.Comet)) { this.baseTransport = TransportNames.Comet; } this.domains = Defaults.getHosts(options); this.activeProtocol = null; this.host = null; this.lastAutoReconnectAttempt = null; this.lastActivity = null; this.forceFallbackHost = false; this.connectCounter = 0; this.wsCheckResult = null; this.webSocketSlowTimer = null; this.webSocketGiveUpTimer = null; this.abandonedWebSocket = false; Logger.logAction(this.logger, Logger.LOG_MINOR, 'Realtime.ConnectionManager()', 'started'); Logger.logAction( this.logger, Logger.LOG_MICRO, 'Realtime.ConnectionManager()', 'requested transports = [' + (options.transports || Defaults.defaultTransports) + ']', ); Logger.logAction( this.logger, Logger.LOG_MICRO, 'Realtime.ConnectionManager()', 'available transports = [' + this.transports + ']', ); Logger.logAction( this.logger, Logger.LOG_MICRO, 'Realtime.ConnectionManager()', 'http domains = [' + this.domains + ']', ); if (!this.transports.length) { const msg = 'no requested transports available'; Logger.logAction(this.logger, Logger.LOG_ERROR, 'realtime.ConnectionManager()', msg); throw new Error(msg); } const addEventListener = Platform.Config.addEventListener; if (addEventListener) { /* intercept close event in browser to persist connection id if requested */ if (haveSessionStorage() && typeof options.recover === 'function') { addEventListener('beforeunload', this.persistConnection.bind(this)); } if (options.closeOnUnload === true) { addEventListener('beforeunload', () => { Logger.logAction( this.logger, Logger.LOG_MAJOR, 'Realtime.ConnectionManager()', 'beforeunload event has triggered the connection to close as closeOnUnload is true', ); this.requestState({ state: 'closing' }); }); } /* Listen for online and offline events */ addEventListener('online', () => { if (this.state == this.states.disconnected || this.state == this.states.suspended) { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager caught browser ‘online’ event', 'reattempting connection', ); this.requestState({ state: 'connecting' }); } else if (this.state == this.states.connecting) { // RTN20c: if 'online' event recieved while CONNECTING, abandon connection attempt and retry this.pendingTransport?.off(); this.disconnectAllTransports(); this.startConnect(); } }); addEventListener('offline', () => { if (this.state == this.states.connected) { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager caught browser ‘offline’ event', 'disconnecting active transport', ); // Not sufficient to just go to the 'disconnected' state, want to // force all transports to reattempt the connection. Will immediately // retry. this.disconnectAllTransports(); } }); } } /********************* * transport management *********************/ // Used by tests static supportedTransports(additionalImplementations: TransportImplementations) { const storage: TransportStorage = { supportedTransports: {} }; this.initTransports(additionalImplementations, storage); return storage.supportedTransports; } private static initTransports(additionalImplementations: TransportImplementations, storage: TransportStorage) { const implementations = { ...Platform.Transports.bundledImplementations, ...additionalImplementations }; [TransportNames.WebSocket, ...Platform.Transports.order].forEach((transportName) => { const transport = implementations[transportName]; if (transport && transport.isAvailable()) { storage.supportedTransports[transportName] = transport; } }); } initTransports() { ConnectionManager.initTransports(this.realtime._additionalTransportImplementations, this); } createTransportParams(host: string | null, mode: string): TransportParams { return new TransportParams(this.options, host, mode, this.connectionKey); } getTransportParams(callback: Function): void { const decideMode = (modeCb: Function) => { if (this.connectionKey) { modeCb('resume'); return; } if (typeof this.options.recover === 'string') { modeCb('recover'); return; } const recoverFn = this.options.recover, lastSessionData = this.getSessionRecoverData(), sessionRecoveryName = this.sessionRecoveryName(); if (lastSessionData && typeof recoverFn === 'function') { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.getTransportParams()', 'Calling clientOptions-provided recover function with last session data (recovery scope: ' + sessionRecoveryName + ')', ); recoverFn(lastSessionData, (shouldRecover?: boolean) => { if (shouldRecover) { this.options.recover = lastSessionData.recoveryKey; modeCb('recover'); } else { modeCb('clean'); } }); return; } modeCb('clean'); }; decideMode((mode: string) => { const transportParams = this.createTransportParams(null, mode); if (mode === 'recover') { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.getTransportParams()', 'Transport recovery mode = recover; recoveryKey = ' + this.options.recover, ); const recoveryContext = decodeRecoveryKey(this.options.recover); if (recoveryContext) { this.msgSerial = recoveryContext.msgSerial; } } else { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.getTransportParams()', 'Transport params = ' + transportParams.toString(), ); } callback(transportParams); }); } /** * Attempt to connect using a given transport * @param transportParams * @param candidate, the transport to try * @param callback */ tryATransport(transportParams: TransportParams, candidate: TransportName, callback: Function): void { Logger.logAction(this.logger, Logger.LOG_MICRO, 'ConnectionManager.tryATransport()', 'trying ' + candidate); this.proposedTransport = Transport.tryConnect( this.supportedTransports[candidate]!, this, this.realtime.auth, transportParams, (wrappedErr: { error: ErrorInfo; event: string } | null, transport?: Transport) => { const state = this.state; if (state == this.states.closing || state == this.states.closed || state == this.states.failed) { if (transport) { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.tryATransport()', 'connection ' + state.state + ' while we were attempting the transport; closing ' + transport, ); transport.close(); } callback(true); return; } if (wrappedErr) { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.tryATransport()', 'transport ' + candidate + ' ' + wrappedErr.event + ', err: ' + wrappedErr.error.toString(), ); /* Comet transport onconnect token errors can be dealt with here. * Websocket ones only happen after the transport claims to be viable, * so are dealt with as non-onconnect token errors */ if ( Auth.isTokenErr(wrappedErr.error) && !(this.errorReason && Auth.isTokenErr(this.errorReason as ErrorInfo)) ) { this.errorReason = wrappedErr.error; /* re-get a token and try again */ Utils.whenPromiseSettles(this.realtime.auth._forceNewToken(null, null), (err: ErrorInfo | null) => { if (err) { this.actOnErrorFromAuthorize(err); return; } this.tryATransport(transportParams, candidate, callback); }); } else if (wrappedErr.event === 'failed') { /* Error that's fatal to the connection */ this.notifyState({ state: 'failed', error: wrappedErr.error }); callback(true); } else if (wrappedErr.event === 'disconnected') { if (!isRetriable(wrappedErr.error)) { /* Error received from the server that does not call for trying a fallback host, eg a rate limit */ this.notifyState({ state: this.states.connecting.failState as string, error: wrappedErr.error }); callback(true); } else { /* Error with that transport only; continue trying other fallback hosts */ callback(false); } } return; } Logger.logAction( this.logger, Logger.LOG_MICRO, 'ConnectionManager.tryATransport()', 'viable transport ' + candidate + '; setting pending', ); this.setTransportPending(transport as Transport, transportParams); callback(null, transport); }, ); } /** * Called when a transport is indicated to be viable, and the ConnectionManager * expects to activate this transport as soon as it is connected. * @param transport * @param transportParams */ setTransportPending(transport: Transport, transportParams: TransportParams): void { const mode = transportParams.mode; Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.setTransportPending()', 'transport = ' + transport + '; mode = ' + mode, ); this.pendingTransport = transport; this.cancelWebSocketSlowTimer(); this.cancelWebSocketGiveUpTimer(); transport.once('connected', (error: ErrorInfo, connectionId: string, connectionDetails: Record) => { this.activateTransport(error, transport, connectionId, connectionDetails); if (mode === 'recover' && this.options.recover) { /* After a successful recovery, we unpersist, as a recovery key cannot * be used more than once */ delete this.options.recover; this.unpersistConnection(); } }); const self = this; transport.on(['disconnected', 'closed', 'failed'], function (this: { event: string }, error: ErrorInfo) { self.deactivateTransport(transport, this.event, error); }); this.emit('transport.pending', transport); } /** * Called when a transport is connected, and the connectionmanager decides that * it will now be the active transport. Returns whether or not it activated * the transport (if the connection is closing/closed it will choose not to). * @param transport the transport instance * @param connectionId the id of the new active connection * @param connectionDetails the details of the new active connection */ activateTransport( error: ErrorInfo, transport: Transport, connectionId: string, connectionDetails: Record, ): boolean { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.activateTransport()', 'transport = ' + transport, ); if (error) { Logger.logAction(this.logger, Logger.LOG_ERROR, 'ConnectionManager.activateTransport()', 'error = ' + error); } if (connectionId) { Logger.logAction( this.logger, Logger.LOG_MICRO, 'ConnectionManager.activateTransport()', 'connectionId = ' + connectionId, ); } if (connectionDetails) { Logger.logAction( this.logger, Logger.LOG_MICRO, 'ConnectionManager.activateTransport()', 'connectionDetails = ' + JSON.stringify(connectionDetails), ); } this.persistTransportPreference(transport); /* if the connectionmanager moved to the closing/closed state before this * connection event, then we won't activate this transport */ const existingState = this.state, connectedState = this.states.connected.state; Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.activateTransport()', 'current state = ' + existingState.state, ); if ( existingState.state == this.states.closing.state || existingState.state == this.states.closed.state || existingState.state == this.states.failed.state ) { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.activateTransport()', 'Disconnecting transport and abandoning', ); transport.disconnect(); return false; } delete this.pendingTransport; /* if the transport is not connected then don't activate it */ if (!transport.isConnected) { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.activateTransport()', 'Declining to activate transport ' + transport + ' since it appears to no longer be connected', ); return false; } /* the given transport is connected; this will immediately * take over as the active transport */ const existingActiveProtocol = this.activeProtocol; this.activeProtocol = new Protocol(transport); this.host = transport.params.host; const connectionKey = connectionDetails.connectionKey; if (connectionKey && this.connectionKey != connectionKey) { this.setConnection(connectionId, connectionDetails, !!error); } /* Rebroadcast any new connectionDetails from the active transport, which * can come at any time (eg following a reauth), and emit an RTN24 UPDATE * event. (Listener added on nextTick because we're in a transport.on('connected') * callback at the moment; if we add it now we'll be adding it to the end * of the listeners array and it'll be called immediately) */ this.onConnectionDetailsUpdate(connectionDetails, transport); Platform.Config.nextTick(() => { transport.on( 'connected', (connectedErr: ErrorInfo, _connectionId: string, connectionDetails: Record) => { this.onConnectionDetailsUpdate(connectionDetails, transport); this.emit('update', new ConnectionStateChange(connectedState, connectedState, null, connectedErr)); }, ); }); /* If previously not connected, notify the state change (including any * error). */ if (existingState.state === this.states.connected.state) { if (error) { this.errorReason = this.realtime.connection.errorReason = error; this.emit('update', new ConnectionStateChange(connectedState, connectedState, null, error)); } } else { this.notifyState({ state: 'connected', error: error }); this.errorReason = this.realtime.connection.errorReason = error || null; } /* Send after the connection state update, as Channels hooks into this to * resend attaches on a new transport if necessary */ this.emit('transport.active', transport); /* Gracefully terminate existing protocol */ if (existingActiveProtocol) { if (existingActiveProtocol.messageQueue.count() > 0) { /* We could just requeue pending messages on the new transport, but * actually this should never happen: transports should only take over * from other active transports when upgrading, and upgrading waits for * the old transport to be idle. So log an error. */ Logger.logAction( this.logger, Logger.LOG_ERROR, 'ConnectionManager.activateTransport()', 'Previous active protocol (for transport ' + existingActiveProtocol.transport.shortName + ', new one is ' + transport.shortName + ') finishing with ' + existingActiveProtocol.messageQueue.count() + ' messages still pending', ); } if (existingActiveProtocol.transport === transport) { const msg = 'Assumption violated: activating a transport that was also the transport for the previous active protocol; transport = ' + transport.shortName + '; stack = ' + new Error().stack; Logger.logAction(this.logger, Logger.LOG_ERROR, 'ConnectionManager.activateTransport()', msg); } else { existingActiveProtocol.finish(); } } return true; } /** * Called when a transport is no longer the active transport. This can occur * in any transport connection state. * @param transport */ deactivateTransport(transport: Transport, state: string, error: ErrorInfo): void { const currentProtocol = this.activeProtocol, wasActive = currentProtocol && currentProtocol.getTransport() === transport, wasPending = transport === this.pendingTransport, noTransportsScheduledForActivation = this.noTransportsScheduledForActivation(); Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.deactivateTransport()', 'transport = ' + transport, ); Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.deactivateTransport()', 'state = ' + state + (wasActive ? '; was active' : wasPending ? '; was pending' : '') + (noTransportsScheduledForActivation ? '' : '; another transport is scheduled for activation'), ); if (error && error.message) Logger.logAction( this.logger, Logger.LOG_MICRO, 'ConnectionManager.deactivateTransport()', 'reason = ' + error.message, ); if (wasActive) { Logger.logAction( this.logger, Logger.LOG_MICRO, 'ConnectionManager.deactivateTransport()', 'Getting, clearing, and requeuing ' + (this.activeProtocol as Protocol).messageQueue.count() + ' pending messages', ); this.queuePendingMessages((currentProtocol as Protocol).getPendingMessages()); /* Clear any messages we requeue to allow the protocol to become idle.*/ (currentProtocol as Protocol).clearPendingMessages(); this.activeProtocol = this.host = null; } this.emit('transport.inactive', transport); /* this transport state change is a state change for the connectionmanager if * - the transport was the active transport and there are no transports * which are connected and scheduled for activation, just waiting for the * active transport to finish what its doing; or * - the transport was the active transport and the error was fatal (so * unhealable by another transport); or * - there is no active transport, and this is the last remaining * pending transport (so we were in the connecting state) */ if ( (wasActive && noTransportsScheduledForActivation) || (wasActive && state === 'failed') || state === 'closed' || (currentProtocol === null && wasPending) ) { /* If we're disconnected with a 5xx we need to try fallback hosts * (RTN14d), but (a) due to how the upgrade sequence works, the * host/transport selection sequence only cares about getting to * `preconnect` (eg establishing a websocket) getting a `disconnected` * protocol message afterwards is too late; and (b) host retry only * applies to connectBase unless the stored preference transport doesn't * work. We solve this by unpersisting the transport preference and * setting an instance variable to force fallback hosts to be used (if * any) here. Bit of a kludge, but no real better alternatives without * rewriting the entire thing */ if (state === 'disconnected' && error && (error.statusCode as number) > 500 && this.domains.length > 1) { this.unpersistTransportPreference(); this.forceFallbackHost = true; /* and try to connect again to try a fallback host without waiting for the usual 15s disconnectedRetryTimeout */ this.notifyState({ state: state, error: error, retryImmediately: true }); return; } /* TODO remove below line once realtime sends token errors as DISCONNECTEDs */ const newConnectionState = state === 'failed' && Auth.isTokenErr(error) ? 'disconnected' : state; this.notifyState({ state: newConnectionState, error: error }); return; } } /* Helper that returns true if there are no transports which are pending, * have been connected, and are just waiting for onceNoPending to fire before * being activated */ noTransportsScheduledForActivation(): boolean { return !this.pendingTransport || !this.pendingTransport.isConnected; } setConnection(connectionId: string, connectionDetails: Record, hasConnectionError?: boolean): void { /* if connectionKey changes but connectionId stays the same, then just a * transport change on the same connection. If connectionId changes, we're * on a new connection, with implications for msgSerial and channel state */ /* If no previous connectionId, don't reset the msgSerial as it may have * been set by recover data (unless the recover failed) */ const prevConnId = this.connectionId, connIdChanged = prevConnId && prevConnId !== connectionId, recoverFailure = !prevConnId && hasConnectionError; if (connIdChanged || recoverFailure) { Logger.logAction(this.logger, Logger.LOG_MINOR, 'ConnectionManager.setConnection()', 'Resetting msgSerial'); this.msgSerial = 0; // RTN19a2: In the event of a new connectionId, previous msgSerials are // meaningless. this.queuedMessages.resetSendAttempted(); } if (this.connectionId !== connectionId) { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.setConnection()', 'New connectionId; reattaching any attached channels', ); } this.realtime.connection.id = this.connectionId = connectionId; this.realtime.connection.key = this.connectionKey = connectionDetails.connectionKey; } clearConnection(): void { this.realtime.connection.id = this.connectionId = undefined; this.realtime.connection.key = this.connectionKey = undefined; this.msgSerial = 0; this.unpersistConnection(); } createRecoveryKey(): string | null { // RTN16g2. if (!this.connectionKey) { return null; } return JSON.stringify({ connectionKey: this.connectionKey, msgSerial: this.msgSerial, channelSerials: this.realtime.channels.channelSerials(), }); } checkConnectionStateFreshness(): void { if (!this.lastActivity || !this.connectionId) { return; } const sinceLast = Date.now() - this.lastActivity; if (sinceLast > this.connectionStateTtl + (this.maxIdleInterval as number)) { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.checkConnectionStateFreshness()', 'Last known activity from realtime was ' + sinceLast + 'ms ago; discarding connection state', ); this.clearConnection(); this.states.connecting.failState = 'suspended'; } } /** * Called when the connectionmanager wants to persist transport * state for later recovery. Only applicable in the browser context. */ persistConnection(): void { if (haveSessionStorage()) { const recoveryKey = this.createRecoveryKey(); if (recoveryKey) { this.setSessionRecoverData({ recoveryKey: recoveryKey, disconnectedAt: Date.now(), location: globalObject.location, clientId: this.realtime.auth.clientId, }); } } } /** * Called when the connectionmanager wants to persist transport * state for later recovery. Only applicable in the browser context. */ unpersistConnection(): void { this.clearSessionRecoverData(); } getActiveTransportFormat(): Utils.Format | undefined { return this.activeProtocol?.getTransport().format; } /********************* * state management *********************/ getError(): IPartialErrorInfo | string { if (this.errorReason) { // create new PartialErrorInfo so it has the correct stack trace // which points to the place which caused us to return this error. const newError = PartialErrorInfo.fromValues(this.errorReason); newError.cause = this.errorReason; return newError; } return this.getStateError(); } getStateError(): ErrorInfo { return (ConnectionErrors as Record ErrorInfo>)[this.state.state]?.(); } activeState(): boolean | void { return this.state.queueEvents || this.state.sendEvents; } enactStateChange(stateChange: ConnectionStateChange): void { const action = 'Connection state'; const message = stateChange.current + (stateChange.reason ? '; reason: ' + stateChange.reason : ''); if (stateChange.current === 'failed') { Logger.logAction(this.logger, Logger.LOG_ERROR, action, message); } else { Logger.logAction(this.logger, Logger.LOG_MAJOR, action, message); } Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.enactStateChange', 'setting new state: ' + stateChange.current + '; reason = ' + (stateChange.reason && (stateChange.reason as ErrorInfo).message), ); const newState = (this.state = this.states[stateChange.current as string]); if (stateChange.reason) { this.errorReason = stateChange.reason; // TODO remove this type assertion after fixing https://github.com/ably/ably-js/issues/1405 this.realtime.connection.errorReason = stateChange.reason as ErrorInfo; } if (newState.terminal || newState.state === 'suspended') { /* suspended is nonterminal, but once in the suspended state, realtime * will have discarded our connection state, so futher connection * attempts should start from scratch */ this.clearConnection(); } this.emit('connectionstate', stateChange); } /**************************************** * ConnectionManager connection lifecycle ****************************************/ startTransitionTimer(transitionState: ConnectionState): void { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.startTransitionTimer()', 'transitionState: ' + transitionState.state, ); if (this.transitionTimer) { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.startTransitionTimer()', 'clearing already-running timer', ); clearTimeout(this.transitionTimer as number); } this.transitionTimer = setTimeout(() => { if (this.transitionTimer) { this.transitionTimer = null; Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager ' + transitionState.state + ' timer expired', 'requesting new state: ' + transitionState.failState, ); this.notifyState({ state: transitionState.failState as string }); } }, transitionState.retryDelay); } cancelTransitionTimer(): void { Logger.logAction(this.logger, Logger.LOG_MINOR, 'ConnectionManager.cancelTransitionTimer()', ''); if (this.transitionTimer) { clearTimeout(this.transitionTimer as number); this.transitionTimer = null; } } startSuspendTimer(): void { if (this.suspendTimer) return; this.suspendTimer = setTimeout(() => { if (this.suspendTimer) { this.suspendTimer = null; Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager suspend timer expired', 'requesting new state: suspended', ); this.states.connecting.failState = 'suspended'; this.notifyState({ state: 'suspended' }); } }, this.connectionStateTtl); } checkSuspendTimer(state: string): void { if (state !== 'disconnected' && state !== 'suspended' && state !== 'connecting') this.cancelSuspendTimer(); } cancelSuspendTimer(): void { this.states.connecting.failState = 'disconnected'; if (this.suspendTimer) { clearTimeout(this.suspendTimer as number); this.suspendTimer = null; } } startRetryTimer(interval: number): void { this.retryTimer = setTimeout(() => { Logger.logAction(this.logger, Logger.LOG_MINOR, 'ConnectionManager retry timer expired', 'retrying'); this.retryTimer = null; this.requestState({ state: 'connecting' }); }, interval); } cancelRetryTimer(): void { if (this.retryTimer) { clearTimeout(this.retryTimer as NodeJS.Timeout); this.retryTimer = null; } } startWebSocketSlowTimer() { this.webSocketSlowTimer = setTimeout(() => { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager WebSocket slow timer', 'checking connectivity', ); this.checkWsConnectivity() .then(() => { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager WebSocket slow timer', 'ws connectivity check succeeded', ); this.wsCheckResult = true; }) .catch(() => { Logger.logAction( this.logger, Logger.LOG_MAJOR, 'ConnectionManager WebSocket slow timer', 'ws connectivity check failed', ); this.wsCheckResult = false; }); if (this.realtime.http.checkConnectivity) { Utils.whenPromiseSettles(this.realtime.http.checkConnectivity(), (err, connectivity) => { if (err || !connectivity) { Logger.logAction( this.logger, Logger.LOG_MAJOR, 'ConnectionManager WebSocket slow timer', 'http connectivity check failed', ); this.cancelWebSocketGiveUpTimer(); this.notifyState({ state: 'disconnected', error: new ErrorInfo('Unable to connect (network unreachable)', 80003, 404), }); } else { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager WebSocket slow timer', 'http connectivity check succeeded', ); } }); } }, this.options.timeouts.webSocketSlowTimeout); } cancelWebSocketSlowTimer() { if (this.webSocketSlowTimer) { clearTimeout(this.webSocketSlowTimer); this.webSocketSlowTimer = null; } } startWebSocketGiveUpTimer(transportParams: TransportParams) { this.webSocketGiveUpTimer = setTimeout(() => { if (!this.wsCheckResult) { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager WebSocket give up timer', 'websocket connection took more than 10s; ' + (this.baseTransport ? 'trying base transport' : ''), ); if (this.baseTransport) { this.abandonedWebSocket = true; this.proposedTransport?.dispose(); this.pendingTransport?.dispose(); this.connectBase(transportParams, ++this.connectCounter); } else { // if we don't have a base transport to fallback to, just let the websocket connection attempt time out Logger.logAction( this.logger, Logger.LOG_MAJOR, 'ConnectionManager WebSocket give up timer', 'websocket connectivity appears to be unavailable but no other transports to try', ); } } }, this.options.timeouts.webSocketConnectTimeout); } cancelWebSocketGiveUpTimer() { if (this.webSocketGiveUpTimer) { clearTimeout(this.webSocketGiveUpTimer); this.webSocketGiveUpTimer = null; } } notifyState(indicated: ConnectionState): void { const state = indicated.state; /* We retry immediately if: * - something disconnects us while we're connected, or * - a viable (but not yet active) transport fails due to a token error (so * this.errorReason will be set, and startConnect will do a forced * authorize). If this.errorReason is already set (to a token error), * then there has been at least one previous attempt to connect that also * failed for a token error, so by RTN14b we go to DISCONNECTED and wait * before trying again */ const retryImmediately = state === 'disconnected' && (this.state === this.states.connected || indicated.retryImmediately || (this.state === this.states.connecting && indicated.error && Auth.isTokenErr(indicated.error) && !(this.errorReason && Auth.isTokenErr(this.errorReason as ErrorInfo)))); Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.notifyState()', 'new state: ' + state + (retryImmediately ? '; will retry connection immediately' : ''), ); /* do nothing if we're already in the indicated state */ if (state == this.state.state) return; /* kill timers (possibly excepting suspend timer depending on the notified * state), as these are superseded by this notification */ this.cancelTransitionTimer(); this.cancelRetryTimer(); this.cancelWebSocketSlowTimer(); this.cancelWebSocketGiveUpTimer(); this.checkSuspendTimer(indicated.state); if (state === 'suspended' || state === 'connected') { this.disconnectedRetryCount = 0; } /* do nothing if we're unable to move from the current state */ if (this.state.terminal) return; /* process new state */ const newState = this.states[indicated.state]; let retryDelay = newState.retryDelay; if (newState.state === 'disconnected') { this.disconnectedRetryCount++; retryDelay = Utils.getRetryTime(newState.retryDelay as number, this.disconnectedRetryCount); } const change = new ConnectionStateChange( this.state.state, newState.state, retryDelay, indicated.error || (ConnectionErrors as Partial ErrorInfo>>)[newState.state]?.(), ); if (retryImmediately) { const autoReconnect = () => { if (this.state === this.states.disconnected) { this.lastAutoReconnectAttempt = Date.now(); this.requestState({ state: 'connecting' }); } }; const sinceLast = this.lastAutoReconnectAttempt && Date.now() - this.lastAutoReconnectAttempt + 1; if (sinceLast && sinceLast < 1000) { Logger.logAction( this.logger, Logger.LOG_MICRO, 'ConnectionManager.notifyState()', 'Last reconnect attempt was only ' + sinceLast + 'ms ago, waiting another ' + (1000 - sinceLast) + 'ms before trying again', ); setTimeout(autoReconnect, 1000 - sinceLast); } else { Platform.Config.nextTick(autoReconnect); } } else if (state === 'disconnected' || state === 'suspended') { this.startRetryTimer(retryDelay as number); } /* If going into disconnect/suspended (and not retrying immediately), or a * terminal state, ensure there are no orphaned transports hanging around. */ if ((state === 'disconnected' && !retryImmediately) || state === 'suspended' || newState.terminal) { /* Wait till the next tick so the connection state change is enacted, * so aborting transports doesn't trigger redundant state changes */ Platform.Config.nextTick(() => { this.disconnectAllTransports(); }); } if (state == 'connected' && !this.activeProtocol) { Logger.logAction( this.logger, Logger.LOG_ERROR, 'ConnectionManager.notifyState()', 'Broken invariant: attempted to go into connected state, but there is no active protocol', ); } /* implement the change and notify */ this.enactStateChange(change); if (this.state.sendEvents) { this.sendQueuedMessages(); } else if (!this.state.queueEvents) { this.realtime.channels.propogateConnectionInterruption(state, change.reason); this.failQueuedMessages(change.reason as ErrorInfo); // RTN7c } } requestState(request: any): void { const state = request.state; Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.requestState()', 'requested state: ' + state + '; current state: ' + this.state.state, ); if (state == this.state.state) return; /* silently do nothing */ /* kill running timers, as this request supersedes them */ this.cancelWebSocketSlowTimer(); this.cancelWebSocketGiveUpTimer(); this.cancelTransitionTimer(); this.cancelRetryTimer(); /* for suspend timer check rather than cancel -- eg requesting a connecting * state should not reset the suspend timer */ this.checkSuspendTimer(state); if (state == 'connecting' && this.state.state == 'connected') return; if (state == 'closing' && this.state.state == 'closed') return; const newState = this.states[state], change = new ConnectionStateChange( this.state.state, newState.state, null, request.error || (ConnectionErrors as Partial ErrorInfo>>)[newState.state]?.(), ); this.enactStateChange(change); if (state == 'connecting') { Platform.Config.nextTick(() => { this.startConnect(); }); } if (state == 'closing') { this.closeImpl(); } } startConnect(): void { if (this.state !== this.states.connecting) { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.startConnect()', 'Must be in connecting state to connect, but was ' + this.state.state, ); return; } const auth = this.realtime.auth; /* The point of the connectCounter mechanism is to ensure that the * connection procedure can be cancelled. We want disconnectAllTransports * to be able to stop any in-progress connection, even before it gets to * the stage of having a pending (or even a proposed) transport that it can * dispose() of. So we check that it's still current after any async stage, * up until the stage that is synchronous with instantiating a transport */ const connectCount = ++this.connectCounter; const connect = () => { this.checkConnectionStateFreshness(); this.getTransportParams((transportParams: TransportParams) => { if (transportParams.mode === 'recover' && transportParams.options.recover) { const recoveryContext = decodeRecoveryKey(transportParams.options.recover); if (recoveryContext) { this.realtime.channels.recoverChannels(recoveryContext.channelSerials); } } if (connectCount !== this.connectCounter) { return; } this.connectImpl(transportParams, connectCount); }); }; Logger.logAction(this.logger, Logger.LOG_MINOR, 'ConnectionManager.startConnect()', 'starting connection'); this.startSuspendTimer(); this.startTransitionTimer(this.states.connecting); if (auth.method === 'basic') { connect(); } else { const authCb = (err: ErrorInfo | null) => { if (connectCount !== this.connectCounter) { return; } if (err) { this.actOnErrorFromAuthorize(err); } else { connect(); } }; if (this.errorReason && Auth.isTokenErr(this.errorReason as ErrorInfo)) { /* Force a refetch of a new token */ Utils.whenPromiseSettles(auth._forceNewToken(null, null), authCb); } else { Utils.whenPromiseSettles(auth._ensureValidAuthCredentials(false), authCb); } } } /* * there are, at most, two transports available with which a connection may * be attempted: web_socket and/or a base transport (xhr_polling in browsers, * comet in nodejs). web_socket is always preferred, and the base transport is * only used in case web_socket connectivity appears to be unavailable. * * connectImpl begins the transport selection process by checking which transports * are available, and if there is a cached preference. It then defers to the * transport-specific connect methods: connectWs and connectBase. * * It is also responsible for invalidating the cache in the case that a base * transport preference is stored but web socket connectivity is now available. * * handling of the case where we need to failover from web_socket to the base * transport is implemented in the connectWs method. */ connectImpl(transportParams: TransportParams, connectCount: number): void { const state = this.state.state; if (state !== this.states.connecting.state) { /* Only keep trying as long as in the 'connecting' state (or 'connected' * for upgrading). Any operation can put us into 'disconnected' to cancel * connection attempts and wait before retrying, or 'failed' to fail. */ Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.connectImpl()', 'Must be in connecting state to connect, but was ' + state, ); return; } const transportPreference = this.getTransportPreference(); // If transport preference is for a non-ws transport but websocket is now available, unpersist the preference for next time if (transportPreference && transportPreference === this.baseTransport && this.webSocketTransportAvailable) { this.checkWsConnectivity() .then(() => { this.unpersistTransportPreference(); if (this.state === this.states.connecting) { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.connectImpl():', 'web socket connectivity available, cancelling connection attempt with ' + this.baseTransport, ); this.disconnectAllTransports(); this.connectWs(transportParams, ++this.connectCounter); } }) .catch(noop); } if ( (transportPreference && transportPreference === this.baseTransport) || (this.baseTransport && !this.webSocketTransportAvailable) ) { this.connectBase(transportParams, connectCount); } else { this.connectWs(transportParams, connectCount); } } /* * connectWs starts two timers to monitor the success of a web_socket connection attempt: * - webSocketSlowTimer: if this timer fires before the connection succeeds, * cm will simultaneously check websocket and http/xhr connectivity. if the http * connectivity check fails, we give up the connection sequence entirely and * transition to disconnected. if the websocket connectivity check fails then * we assume no ws connectivity and failover to base transport. in the case that * the checks succeed, we continue with websocket and wait for it to try fallback hosts * and, if unsuccessful, ultimately transition to disconnected. * - webSocketGiveUpTimer: if this timer fires, and the preceding websocket * connectivity check is still pending then we assume that there is an issue * with the transport and fallback to base transport. */ connectWs(transportParams: TransportParams, connectCount: number) { Logger.logAction(this.logger, Logger.LOG_MICRO, 'ConnectionManager.connectWs()'); this.wsCheckResult = null; this.abandonedWebSocket = false; this.startWebSocketSlowTimer(); this.startWebSocketGiveUpTimer(transportParams); this.tryTransportWithFallbacks('web_socket', transportParams, true, connectCount, () => { return this.wsCheckResult !== false && !this.abandonedWebSocket; }); } connectBase(transportParams: TransportParams, connectCount: number) { Logger.logAction(this.logger, Logger.LOG_MICRO, 'ConnectionManager.connectBase()'); if (this.baseTransport) { this.tryTransportWithFallbacks(this.baseTransport, transportParams, false, connectCount, () => true); } else { this.notifyState({ state: 'disconnected', error: new ErrorInfo('No transports left to try', 80000, 404), }); } } tryTransportWithFallbacks( transportName: TransportName, transportParams: TransportParams, ws: boolean, connectCount: number, shouldContinue: () => boolean, ): void { Logger.logAction( this.logger, Logger.LOG_MICRO, 'ConnectionManager.tryTransportWithFallbacks()', transportName, ); const giveUp = (err: IPartialErrorInfo) => { this.notifyState({ state: this.states.connecting.failState as string, error: err }); }; const candidateHosts = this.domains.slice(); const hostAttemptCb = (fatal: boolean, transport: Transport) => { if (connectCount !== this.connectCounter) { return; } if (!shouldContinue()) { if (transport) { transport.dispose(); } return; } if (!transport && !fatal) { tryFallbackHosts(); } }; /* first try to establish a connection with the priority host with http transport */ const host = candidateHosts.shift(); if (!host) { giveUp(new ErrorInfo('Unable to connect (no available host)', 80003, 404)); return; } transportParams.host = host; /* this is what we'll be doing if the attempt for the main host fails */ const tryFallbackHosts = () => { /* if there aren't any fallback hosts, fail */ if (!candidateHosts.length) { giveUp(new ErrorInfo('Unable to connect (and no more fallback hosts to try)', 80003, 404)); return; } /* before trying any fallback (or any remaining fallback) we decide if * there is a problem with the ably host, or there is a general connectivity * problem */ if (!this.realtime.http.checkConnectivity) { giveUp(new PartialErrorInfo('Internal error: Http.checkConnectivity not set', null, 500)); return; } Utils.whenPromiseSettles( this.realtime.http.checkConnectivity(), (err?: ErrorInfo | null, connectivity?: boolean) => { if (connectCount !== this.connectCounter) { return; } if (!shouldContinue()) { return; } /* we know err won't happen but handle it here anyway */ if (err) { giveUp(err); return; } if (!connectivity) { /* the internet isn't reachable, so don't try the fallback hosts */ giveUp(new ErrorInfo('Unable to connect (network unreachable)', 80003, 404)); return; } /* the network is there, so there's a problem with the main host, or * its dns. Try the fallback hosts. We could try them simultaneously but * that would potentially cause a huge spike in load on the load balancer */ transportParams.host = Utils.arrPopRandomElement(candidateHosts); this.tryATransport(transportParams, transportName, hostAttemptCb); }, ); }; if (this.forceFallbackHost && candidateHosts.length) { this.forceFallbackHost = false; tryFallbackHosts(); return; } this.tryATransport(transportParams, transportName, hostAttemptCb); } closeImpl(): void { Logger.logAction(this.logger, Logger.LOG_MINOR, 'ConnectionManager.closeImpl()', 'closing connection'); this.cancelSuspendTimer(); this.startTransitionTimer(this.states.closing); if (this.pendingTransport) { Logger.logAction( this.logger, Logger.LOG_MICRO, 'ConnectionManager.closeImpl()', 'Closing pending transport: ' + this.pendingTransport, ); this.pendingTransport.close(); } if (this.activeProtocol) { Logger.logAction( this.logger, Logger.LOG_MICRO, 'ConnectionManager.closeImpl()', 'Closing active transport: ' + this.activeProtocol.getTransport(), ); this.activeProtocol.getTransport().close(); } /* If there was an active transport, this will probably be * preempted by the notifyState call in deactivateTransport */ this.notifyState({ state: 'closed' }); } onAuthUpdated(tokenDetails: API.TokenDetails, callback: Function): void { switch (this.state.state) { case 'connected': { Logger.logAction( this.logger, Logger.LOG_MICRO, 'ConnectionManager.onAuthUpdated()', 'Sending AUTH message on active transport', ); /* Do any transport-specific new-token action */ const activeTransport = this.activeProtocol?.getTransport(); if (activeTransport && activeTransport.onAuthUpdated) { activeTransport.onAuthUpdated(tokenDetails); } const authMsg = protocolMessageFromValues({ action: actions.AUTH, auth: { accessToken: tokenDetails.token, }, }); this.send(authMsg); /* The answer will come back as either a connectiondetails event * (realtime sends a CONNECTED to acknowledge the reauth) or a * statechange to failed */ const successListener = () => { this.off(failureListener); callback(null, tokenDetails); }; const failureListener = (stateChange: ConnectionStateChange) => { if (stateChange.current === 'failed') { this.off(successListener); this.off(failureListener); callback(stateChange.reason || this.getStateError()); } }; this.once('connectiondetails', successListener); this.on('connectionstate', failureListener); break; } case 'connecting': Logger.logAction( this.logger, Logger.LOG_MICRO, 'ConnectionManager.onAuthUpdated()', 'Aborting current connection attempts in order to start again with the new auth details', ); this.disconnectAllTransports(); /* fallthrough to add statechange listener */ default: { Logger.logAction( this.logger, Logger.LOG_MICRO, 'ConnectionManager.onAuthUpdated()', 'Connection state is ' + this.state.state + '; waiting until either connected or failed', ); const listener = (stateChange: ConnectionStateChange) => { switch (stateChange.current) { case 'connected': this.off(listener); callback(null, tokenDetails); break; case 'failed': case 'closed': case 'suspended': this.off(listener); callback(stateChange.reason || this.getStateError()); break; default: /* ignore till we get either connected or failed */ break; } }; this.on('connectionstate', listener); if (this.state.state === 'connecting') { /* can happen if in the connecting state but no transport was pending * yet, so disconnectAllTransports did not trigger a disconnected state */ this.startConnect(); } else { this.requestState({ state: 'connecting' }); } } } } disconnectAllTransports(): void { Logger.logAction( this.logger, Logger.LOG_MINOR, 'ConnectionManager.disconnectAllTransports()', 'Disconnecting all transports', ); /* This will prevent any connection procedure in an async part of one of its early stages from continuing */ this.connectCounter++; if (this.pendingTransport) { Logger.logAction( this.logger, Logger.LOG_MICRO, 'ConnectionManager.disconnectAllTransports()', 'Disconnecting pending transport: ' + this.pendingTransport, ); this.pendingTransport.disconnect(); } delete this.pendingTransport; if (this.proposedTransport) { Logger.logAction( this.logger, Logger.LOG_MICRO, 'ConnectionManager.disconnectAllTransports()', 'Disconnecting proposed transport: ' + this.pendingTransport, ); this.proposedTransport.disconnect(); } delete this.pendingTransport; if (this.activeProtocol) { Logger.logAction( this.logger, Logger.LOG_MICRO, 'ConnectionManager.disconnectAllTransports()', 'Disconnecting active transport: ' + this.activeProtocol.getTransport(), ); this.activeProtocol.getTransport().disconnect(); } /* No need to notify state disconnected; disconnecting the active transport * will have that effect */ } /****************** * event queueing ******************/ send(msg: ProtocolMessage, queueEvent?: boolean, callback?: ErrCallback): void { callback = callback || noop; const state = this.state; if (state.sendEvents) { Logger.logAction(this.logger, Logger.LOG_MICRO, 'ConnectionManager.send()', 'sending event'); this.sendImpl(new PendingMessage(msg, callback)); return; } const shouldQueue = queueEvent && state.queueEvents; if (!shouldQueue) { const err = 'rejecting event, queueEvent was ' + queueEvent + ', state was ' + state.state; Logger.logAction(this.logger, Logger.LOG_MICRO, 'ConnectionManager.send()', err); callback(this.errorReason || new ErrorInfo(err, 90000, 400)); return; } if (this.logger.shouldLog(Logger.LOG_MICRO)) { Logger.logAction( this.logger, Logger.LOG_MICRO, 'ConnectionManager.send()', 'queueing msg; ' + stringifyProtocolMessage( msg, this.realtime._RealtimePresence, this.realtime._Annotations, this.realtime._liveObjectsPlugin, ), ); } this.queue(msg, callback); } sendImpl(pendingMessage: PendingMessage): void { const msg = pendingMessage.message; /* If have already attempted to send this, resend with the same msgSerial, * so Ably can dedup if the previous send succeeded */ if (pendingMessage.ackRequired && !pendingMessage.sendAttempted) { msg.msgSerial = this.msgSerial++; } try { (this.activeProtocol as Protocol).send(pendingMessage); } catch (e) { Logger.logAction( this.logger, Logger.LOG_ERROR, 'ConnectionManager.sendImpl()', 'Unexpected exception in transport.send(): ' + (e as Error).stack, ); } } queue(msg: ProtocolMessage, callback: ErrCallback): void { Logger.logAction(this.logger, Logger.LOG_MICRO, 'ConnectionManager.queue()', 'queueing event'); const lastQueued = this.queuedMessages.last(); const maxSize = this.options.maxMessageSize; /* If have already attempted to send a message, don't merge more messages * into it, as if the previous send actually succeeded and realtime ignores * the dup, they'll be lost */ if (lastQueued && !lastQueued.sendAttempted && bundleWith(lastQueued.message, msg, maxSize)) { if (!lastQueued.merged) { lastQueued.callback = Multicaster.create(this.logger, [lastQueued.callback]); lastQueued.merged = true; } (lastQueued.callback as MulticasterInstance).push(callback); } else { this.queuedMessages.push(new PendingMessage(msg, callback)); } } sendQueuedMessages(): void { Logger.logAction( this.logger, Logger.LOG_MICRO, 'ConnectionManager.sendQueuedMessages()', 'sending ' + this.queuedMessages.count() + ' queued messages', ); let pendingMessage; while ((pendingMessage = this.queuedMessages.shift())) this.sendImpl(pendingMessage); } queuePendingMessages(pendingMessages: Array): void { if (pendingMessages && pendingMessages.length) { Logger.logAction( this.logger, Logger.LOG_MICRO, 'ConnectionManager.queuePendingMessages()', 'queueing ' + pendingMessages.length + ' pending messages', ); this.queuedMessages.prepend(pendingMessages); } } failQueuedMessages(err: ErrorInfo): void { const numQueued = this.queuedMessages.count(); if (numQueued > 0) { Logger.logAction( this.logger, Logger.LOG_ERROR, 'ConnectionManager.failQueuedMessages()', 'failing ' + numQueued + ' queued messages, err = ' + Utils.inspectError(err), ); this.queuedMessages.completeAllMessages(err); } } onChannelMessage(message: ProtocolMessage, transport: Transport): void { this.pendingChannelMessagesState.queue.push({ message, transport }); if (!this.pendingChannelMessagesState.isProcessing) { this.processNextPendingChannelMessage(); } } private processNextPendingChannelMessage() { if (this.pendingChannelMessagesState.queue.length > 0) { this.pendingChannelMessagesState.isProcessing = true; const pendingChannelMessage = this.pendingChannelMessagesState.queue.shift()!; this.processChannelMessage(pendingChannelMessage.message) .catch((err) => { Logger.logAction( this.logger, Logger.LOG_ERROR, 'ConnectionManager.processNextPendingChannelMessage() received error ', err, ); }) .finally(() => { this.pendingChannelMessagesState.isProcessing = false; this.processNextPendingChannelMessage(); }); } } private async processChannelMessage(message: ProtocolMessage) { await this.realtime.channels.processChannelMessage(message); } async ping(): Promise { if (this.state.state !== 'connected') { throw new ErrorInfo('Unable to ping service; not connected', 40000, 400); } const transport = this.activeProtocol?.getTransport(); if (!transport) { throw this.getStateError(); } Logger.logAction(this.logger, Logger.LOG_MINOR, 'ConnectionManager.ping()', 'transport = ' + transport); const pingStart = Date.now(); const id = Utils.cheapRandStr(); return Utils.withTimeoutAsync( new Promise((resolve) => { const onHeartbeat = (responseId: string) => { if (responseId === id) { transport.off('heartbeat', onHeartbeat); resolve(Date.now() - pingStart); } }; transport.on('heartbeat', onHeartbeat); transport.ping(id); }), this.options.timeouts.realtimeRequestTimeout, 'Timeout waiting for heartbeat response', ); } abort(error: ErrorInfo): void { (this.activeProtocol as Protocol).getTransport().fail(error); } getTransportPreference(): TransportName { return this.transportPreference || (haveWebStorage() && Platform.WebStorage?.get?.(transportPreferenceName)); } persistTransportPreference(transport: Transport): void { this.transportPreference = transport.shortName; if (haveWebStorage()) { Platform.WebStorage?.set?.(transportPreferenceName, transport.shortName); } } unpersistTransportPreference(): void { this.transportPreference = null; if (haveWebStorage()) { Platform.WebStorage?.remove?.(transportPreferenceName); } } /* This method is only used during connection attempts, so implements RSA4c1, RSA4c2, * and RSA4d. It is generally not invoked for serverside-triggered reauths or manual * reauths, so RSA4c3 does not apply, except (per per RSA4d1) in the case that the auth * server returns 403. */ actOnErrorFromAuthorize(err: ErrorInfo): void { if (err.code === 40171) { /* No way to reauth */ this.notifyState({ state: 'failed', error: err }); } else if (err.code === 40102) { this.notifyState({ state: 'failed', error: err }); } else if (err.statusCode === HttpStatusCodes.Forbidden) { const msg = 'Client configured authentication provider returned 403; failing the connection'; Logger.logAction(this.logger, Logger.LOG_ERROR, 'ConnectionManager.actOnErrorFromAuthorize()', msg); this.notifyState({ state: 'failed', error: new ErrorInfo(msg, 80019, 403, err) }); } else { const msg = 'Client configured authentication provider request failed'; Logger.logAction(this.logger, Logger.LOG_MINOR, 'ConnectionManager.actOnErrorFromAuthorize', msg); this.notifyState({ state: this.state.failState as string, error: new ErrorInfo(msg, 80019, 401, err) }); } } onConnectionDetailsUpdate(connectionDetails: Record, transport: Transport): void { if (!connectionDetails) { return; } this.connectionDetails = connectionDetails; if (connectionDetails.maxMessageSize) { this.options.maxMessageSize = connectionDetails.maxMessageSize; } const clientId = connectionDetails.clientId; if (clientId) { const err = this.realtime.auth._uncheckedSetClientId(clientId); if (err) { Logger.logAction(this.logger, Logger.LOG_ERROR, 'ConnectionManager.onConnectionDetailsUpdate()', err.message); /* Errors setting the clientId are fatal to the connection */ transport.fail(err); return; } } const connectionStateTtl = connectionDetails.connectionStateTtl; if (connectionStateTtl) { this.connectionStateTtl = connectionStateTtl; } this.maxIdleInterval = connectionDetails.maxIdleInterval; this.emit('connectiondetails', connectionDetails); } checkWsConnectivity() { const wsConnectivityCheckUrl = this.options.wsConnectivityCheckUrl || Defaults.wsConnectivityCheckUrl; const ws = new Platform.Config.WebSocket(wsConnectivityCheckUrl); return new Promise((resolve, reject) => { let finished = false; ws.onopen = () => { if (!finished) { finished = true; resolve(); ws.close(); } }; ws.onclose = ws.onerror = () => { if (!finished) { finished = true; reject(); } }; }); } sessionRecoveryName() { return this.options.recoveryKeyStorageName || 'ably-connection-recovery'; } getSessionRecoverData() { return haveSessionStorage() && Platform.WebStorage?.getSession?.(this.sessionRecoveryName()); } setSessionRecoverData(value: any) { return haveSessionStorage() && Platform.WebStorage?.setSession?.(this.sessionRecoveryName(), value); } clearSessionRecoverData() { return haveSessionStorage() && Platform.WebStorage?.removeSession?.(this.sessionRecoveryName()); } } export default ConnectionManager; export interface TransportStorage { supportedTransports: Partial>; }