import { Agent } from 'http' ; import { EventEmitter } from 'eventemitter3' ; import WebSocket from 'ws' ; import Finity , { StateMachine } from 'finity' ; import PQueue from 'p-queue' ; import PCancelable from 'p-cancelable' ; import { WebClient , WebAPICallResult , WebAPICallError , ErrorCode as APICallErrorCode , RetryOptions , TLSOptions , RTMConnectArguments , RTMStartArguments , addAppMetadata , } from '@slack/web-api' ; import { KeepAlive } from './KeepAlive' ; import { LogLevel , Logger , getLogger } from './logger' ; import { websocketErrorWithOriginal , platformErrorFromEvent , noReplyReceivedError , sendWhileDisconnectedError , sendWhileNotReadyError , } from './errors' ; const packageJson = require ( '../package.json' ) ; enum UnrecoverableRTMStartError { NotAuthed = 'not_authed' , InvalidAuth = 'invalid_auth' , AccountInactive = 'account_inactive' , UserRemovedFromTeam = 'user_removed_from_team' , TeamDisabled = 'team_disabled' } /**
 * This interface is the minimum that the RTMClient depends on from the intersection of `rtm.start` and `rtm.connect`
 * responses.
 */ interface RTMStartResult extends WebAPICallResult { self : { id : string ; } ; team : { id : string ; } ; } /**
 * An RTMClient allows programs to communicate with the  https .
 * This object uses the EventEmitter pattern to dispatch incoming events and has several methods for sending outgoing
 * messages.
  export class RTMClient extends EventEmitter { /**
     * Whether or not the client is currently connected to the RTM API
     */ public connected : boolean = false ; /**
     * Whether or not the client has authenticated to the RTM API. This occurs when the connect method
     * completes, and a WebSocket URL is available for the client's connection.
     */ public authenticated : boolean = false ; /**
     * The user ID for the connected client.
     */ public activeUserId ? : string ; /**
     * The team ID for the workspace the client is connected to.
     */ public activeTeamId ? : string ; /**
     * Internal use web client
     */ private webClient : WebClient ; /**
     * An agent used to manage TCP connections for requests. Most commonly used to implement proxy support. See
     * npm packages `tunnel` and `https-proxy-agent` for information on how to construct a proxy agent.
     */ private agentConfig ? : Agent ; /**
     * Whether this client will automatically reconnect when (not manually) disconnected
     */ private autoReconnect : boolean ; /**
     * Use the `rtm.connect` method to connect when true, or the `rtm.start` method when false
     */ private useRtmConnect : boolean ; /**
     * The number of milliseconds to wait upon connection for reply messages from the previous connection. The default
     * value is 2 seconds.
     */ private replyAckOnReconnectTimeout : number = 2000 ; /**
     * State machine that backs the transition and action behavior
     */ private stateMachine : StateMachine < string , string > ; /**
     * Configuration for the state machine
     */ private stateMachineConfig = Finity . configure ( ) . initialState ( 'disconnected' ) . on ( 'start' ) . transitionTo ( 'connecting' ) . onEnter ( ( ) => { this . logger . debug ( 'pausing outgoing event queue' ) ; this . outgoingEventQueue . pause ( ) ; this . awaitingReplyList . forEach ( ( p ) => { p . cancel ( ) ; } ) ; } ) . state ( 'connecting' ) . submachine ( Finity . configure ( ) . initialState ( 'authenticating' ) . do ( ( ) => { const connectMethod = this . useRtmConnect ? 'rtm.connect' : 'rtm.start' ; return this . webClient . apiCall ( connectMethod , this . startOpts !== undefined ? this . startOpts : { } ) . then ( ( result ) => { const startData = result as RTMStartResult ; this . activeUserId = startData . self . id ; this . activeTeamId = startData . team . id ; return result ; } ) ; } ) . onSuccess ( ) . transitionTo ( 'authenticated' ) . onFailure ( ) . transitionTo ( 'reconnecting' ) . withCondition ( ( context ) => { const error = context . error as WebAPICallError ; this . logger . info ( `unable to RTM start: ${ error . message }` ) ; this . emit ( 'unable_to_rtm_start' , error ) ; let isRecoverable = true ; if ( error . code === APICallErrorCode . PlatformError && ( Object . values ( UnrecoverableRTMStartError ) as string [ ] ) . includes ( error . data . error ) ) { isRecoverable = false ; } else if ( error . code === APICallErrorCode . RequestError ) { isRecoverable = false ; } else if ( error . code === APICallErrorCode . HTTPError ) { isRecoverable = false ; } return this . autoReconnect && isRecoverable ; } ) . transitionTo ( 'failed' ) . state ( 'authenticated' ) . onEnter ( ( _state , context ) => { this . authenticated = true ; this . setupWebsocket ( context . result . url ) ; setImmediate ( ( ) => { this . emit ( 'authenticated' , context . result ) ; } ) ; } ) . on ( 'websocket open' ) . transitionTo ( 'handshaking' ) . state ( 'handshaking' ) . state ( 'failed' ) . onEnter ( ( _state , context ) => { this . stateMachine . handle ( 'failure' , context . error ) ; } ) . global ( ) . onStateEnter ( ( state ) => { this . logger . debug ( `transitioning to state: connecting:${ state }` ) ; } ) . getConfig ( ) ) . on ( 'server hello' ) . transitionTo ( 'connected' ) . on ( 'websocket close' ) . transitionTo ( 'reconnecting' ) . withCondition ( ( ) => this . autoReconnect ) . transitionTo ( 'disconnected' ) . withAction ( ( ) => { this . teardownWebsocket ( ) ; } ) . on ( 'failure' ) . transitionTo ( 'disconnected' ) . on ( 'explicit disconnect' ) . transitionTo ( 'disconnecting' ) . state ( 'connected' ) . onEnter ( ( ) => { this . connected = true ; } ) . submachine ( Finity . configure ( ) . initialState ( 'resuming' ) . on ( 'replay finished' ) . transitionTo ( 'ready' ) . onTimeout ( this . replyAckOnReconnectTimeout ) . transitionTo ( 'ready' ) . onExit ( ( ) => { this . awaitingReplyList . forEach ( ( p ) => { p . cancel ( ) ; } ) ; } ) . state ( 'ready' ) . onEnter ( ( ) => { this . keepAlive . start ( this ) ; setImmediate ( ( ) => { this . logger . debug ( 'resuming outgoing event queue' ) ; this . outgoingEventQueue . start ( ) ; this . emit ( 'ready' ) ; } ) ; } ) . global ( ) . onStateEnter ( ( state ) => { this . logger . debug ( `transitioning to state: connected:${ state }` ) ; } ) . getConfig ( ) ) . on ( 'websocket close' ) . transitionTo ( 'reconnecting' ) . withCondition ( ( ) => this . autoReconnect ) . transitionTo ( 'disconnected' ) . withAction ( ( ) => { this . teardownWebsocket ( ) ; } ) . on ( 'explicit disconnect' ) . transitionTo ( 'disconnecting' ) . onExit ( ( ) => { this . connected = false ; this . authenticated = false ; this . activeUserId = undefined ; this . activeTeamId = undefined ; this . keepAlive . stop ( ) ; this . outgoingEventQueue . pause ( ) ; } ) . state ( 'disconnecting' ) . onEnter ( ( ) => { if ( this . websocket !== undefined ) { this . websocket . close ( ) ; } } ) . on ( 'websocket close' ) . transitionTo ( 'disconnected' ) . onExit ( ( ) => this . teardownWebsocket ( ) ) . state ( 'reconnecting' ) . do ( ( ) => { this . keepAlive . stop ( ) ; return Promise . resolve ( true ) ; } ) . onSuccess ( ) . transitionTo ( 'connecting' ) . onExit ( ( ) => this . teardownWebsocket ( ) ) . global ( ) . onStateEnter ( ( state , context ) => { this . logger . debug ( `transitioning to state: ${ state }` ) ; if ( state === 'disconnected' ) { this . emit ( state , context . eventPayload ) ; } else { this . emit ( state ) ; } } ) . getConfig ( ) ; /**
     * The client's websocket
     */ private websocket ? : WebSocket ; /**
     * The last message ID used for an outgoing message
     */ private messageId = 1 ; /**
     * A cache of the options used to start the connection, so that it can be reused during reconnections.
     */ private startOpts ? : RTMStartOptions ; /**
     * The instance of KeepAlive used to monitor this client's connection.
     */ private keepAlive : KeepAlive ; /**
     * A queue of tasks used to serialize outgoing messages and to allow the client to buffer outgoing messages when
     * its not in the 'ready' state. This queue is paused and resumed as the state machine transitions.
     */ private outgoingEventQueue = new PQueue ( { concurrency : 1 } ) ; /**
     * A list of cancelable Promises that each represent a caller waiting on the server to acknowledge an outgoing
     * message with a response (an incoming message containing a "reply_to" property with the outgoing message's ID).
     * This list is emptied by canceling all the promises when the client no longer expects to receive any replies from
     * the server (when its disconnected or when its reconnected and doesn't expect replies for past outgoing messages).
     * The list is a sparse array, where the indexes are message IDs for the sent messages.
     */ private awaitingReplyList : PCancelable < RTMCallResult > [ ] = [ ] ; /**
     * Configuration for custom TLS handling
     */ private tlsConfig : TLSOptions ; /**
     * The name used to prefix all logging generated from this object
     */ private static loggerName = 'RTMClient' ; /**
     * This object's logger instance
     */ private logger : Logger ; public constructor ( token , { slackApiUrl = 'https://slack.com/api/' , logger = undefined , logLevel = LogLevel . INFO , retryConfig , agent = undefined , autoReconnect = true , useRtmConnect = true , clientPingTimeout , serverPongTimeout , replyAckOnReconnectTimeout = 2000 , tls = undefined , } = { } ) { super ( ) ; this . webClient = new WebClient ( token , { slackApiUrl , logger , logLevel , retryConfig , agent , tls , maxRequestConcurrency : 1 , } ) ; this . agentConfig = agent ; this . autoReconnect = autoReconnect ; this . useRtmConnect = useRtmConnect ; this . replyAckOnReconnectTimeout = replyAckOnReconnectTimeout ; this . tlsConfig = tls !== undefined ? tls : { } ; this . keepAlive = new KeepAlive ( { clientPingTimeout , serverPongTimeout , logger , logLevel , } ) ; this . keepAlive . on ( 'recommend_reconnect' , ( ) => { if ( this . websocket !== undefined ) { this . websocket . close ( ) ; this . keepAlive . stop ( ) ; } } , this ) ; this . logger = getLogger ( RTMClient . loggerName , logLevel , logger ) ; this . stateMachine = Finity . start ( this . stateMachineConfig ) ; this . logger . debug ( 'initialized' ) ; } /**
     * Begin an RTM session using the provided options. This method must be called before any messages can
     * be sent or received.
     */ public start ( options ? ) { this . logger . debug ( 'start()' ) ; this . startOpts = options ; this . stateMachine . handle ( 'start' ) ; return new Promise ( ( resolve , reject ) => { this . once ( 'authenticated' , ( result ) => { this . removeListener ( 'disconnected' , reject ) ; resolve ( result ) ; } ) ; this . once ( 'disconnected' , ( err ) => { this . removeListener ( 'authenticated' , resolve ) ; reject ( err ) ; } ) ; } ) ; } /**
     * End an RTM session. After this method is called no messages will be sent or received unless you call
     * start() again later.
     */ public disconnect ( ) { return new Promise ( ( resolve , reject ) => { this . logger . debug ( 'manual disconnect' ) ; this . once ( 'disconnected' , ( err ) => { if ( err instanceof Error ) { reject ( err ) ; } else { resolve ( ) ; } } ) ; this . stateMachine . handle ( 'explicit disconnect' ) ; } ) ; } param text param conversationId public async sendMessage ( text , conversationId ) { return this . addOutgoingEvent ( true , 'message' , { text , channel : conversationId } ) ; } param conversationId public sendTyping ( conversationId ) { return this . addOutgoingEvent ( false , 'typing' , { channel : conversationId } ) ; } param userIds public subscribePresence ( userIds ) { return this . addOutgoingEvent ( false , 'presence_sub' , { ids : userIds } ) ; } param awaitReply param type param body public addOutgoingEvent ( awaitReply , type , body ? ) ; public addOutgoingEvent ( awaitReply , type , body ? ) ; public addOutgoingEvent ( awaitReply , type , body ? ) { const awaitReplyTask = ( messageId ) => { const replyPromise = new PCancelable < RTMCallResult > ( ( resolve , reject , onCancel ) => { const eventHandler = ( _type , event ) => { if ( event . reply_to === messageId ) { this . off ( 'slack_event' , eventHandler ) ; if ( event . error !== undefined ) { const error = platformErrorFromEvent ( event as RTMCallResult & { error : { msg : string ; } ; } ) ; return reject ( error ) ; } return resolve ( event ) ; } } ; onCancel ( ( ) => { this . off ( 'slack_event' , eventHandler ) ; reject ( noReplyReceivedError ( ) ) ; } ) ; this . on ( 'slack_event' , eventHandler ) ; } ) ; this . awaitingReplyList [ messageId ] = replyPromise ; return replyPromise ; } ; const sendTask = ( ) : Promise < RTMCallResult | void > => { const sendPromise = this . send ( type , body ) ; if ( awaitReply ) { return sendPromise . then ( awaitReplyTask ) ; } return sendPromise . then ( ( ) => Promise . resolve ( ) ) ; } ; return this . outgoingEventQueue . add ( sendTask ) ; } param type param body public send ( type , body = { } ) { const message = { ... body , type , id : this . nextMessageId ( ) , } ; return new Promise ( ( resolve , reject ) => { this . logger . debug ( `send() in state: ${ this . stateMachine . getStateHierarchy ( ) }` ) ; if ( this . websocket === undefined ) { this . logger . error ( 'cannot send message when client is not connected' ) ; reject ( sendWhileDisconnectedError ( ) ) ; } else if ( ! ( this . stateMachine . getCurrentState ( ) === 'connected' && this . stateMachine . getStateHierarchy ( ) [ 1 ] === 'ready' ) ) { this . logger . error ( 'cannot send message when client is not ready' ) ; reject ( sendWhileNotReadyError ( ) ) ; } else { this . emit ( 'outgoing_message' , message ) ; const flatMessage = JSON . stringify ( message ) ; this . logger . debug ( `sending message on websocket: ${ flatMessage }` ) ; this . websocket . send ( flatMessage , ( error ) => { if ( error !== undefined ) { this . logger . error ( `failed to send message on websocket: ${ error . message }` ) ; return reject ( websocketErrorWithOriginal ( error ) ) ; } return resolve ( message . id ) ; } ) ; } } ) ; } /**
     * Atomically increments and returns a message ID for the next message.
     */ private nextMessageId ( ) { this . messageId += 1 ; return this . messageId ; } /**
     * Set up method for the client's websocket instance. This method will attach event listeners.
     */ private setupWebsocket ( url ) { const options = { perMessageDeflate : false , ... this . tlsConfig , } ; if ( this . agentConfig !== undefined ) { options . agent = this . agentConfig ; } this . websocket = new WebSocket ( url , options ) ; this . websocket . addEventListener ( 'open' , ( event ) => this . stateMachine . handle ( 'websocket open' , event ) ) ; this . websocket . addEventListener ( 'close' , ( event ) => this . stateMachine . handle ( 'websocket close' , event ) ) ; this . websocket . addEventListener ( 'error' , ( event ) => { this . logger . error ( `A websocket error occurred: ${ event . message }` ) ; this . emit ( 'error' , websocketErrorWithOriginal ( event . error ) ) ; } ) ; this . websocket . addEventListener ( 'message' , this . onWebsocketMessage . bind ( this ) ) ; } /**
     * Tear down method for the client's websocket instance. This method undoes the work in setupWebsocket(url).
     */ private teardownWebsocket ( ) { if ( this . websocket !== undefined ) { this . websocket . removeAllListeners ( 'open' ) ; this . websocket . removeAllListeners ( 'close' ) ; this . websocket . removeAllListeners ( 'error' ) ; this . websocket . removeAllListeners ( 'message' ) ; this . websocket = undefined ; } } /**
     * `onmessage` handler for the client's websocket. This will parse the payload and dispatch the relevant events for
     * each incoming message.
     */ private onWebsocketMessage ( { data } ) { this . logger . debug ( `received message on websocket: ${ data }` ) ; let event ; try { event = JSON . parse ( data ) ; } catch ( parseError ) { this . logger . error ( `unable to parse incoming websocket message: ${ parseError . message }\n    message contents: "${ data }"` ) ; return ; } if ( event . type === 'hello' ) { this . stateMachine . handle ( 'server hello' ) ; } if ( event . type === 'team_migration_started' ) { if ( this . websocket !== undefined ) { this . websocket . close ( ) ; } } if ( this . stateMachine . getCurrentState ( ) === 'connected' && this . stateMachine . getStateHierarchy ( ) [ 1 ] === 'resuming' && event . reply_to !== undefined && event . reply_to === this . messageId ) { this . stateMachine . handle ( 'replay finished' ) ; } this . emit ( 'slack_event' , event . type , event ) ; this . emit ( event . type , event ) ; if ( event . subtype !== undefined ) { this . emit ( `${ event . type }::${ event . subtype }` , event ) ; } } } addAppMetadata ( { name : packageJson . name , version : packageJson . version } ) ; export default RTMClient ; export interface RTMClientOptions { slackApiUrl ? : string ; logger ? : Logger ; logLevel ? : LogLevel ; retryConfig ? : RetryOptions ; agent ? : Agent ; autoReconnect ? : boolean ; useRtmConnect ? : boolean ; clientPingTimeout ? : number ; serverPongTimeout ? : number ; replyAckOnReconnectTimeout ? : number ; tls ? : TLSOptions ; } export interface RTMCallResult { ts : string ; reply_to ? : number ; error ? : { code : number ; msg : string ; } ; } export type RTMStartOptions = RTMConnectArguments | RTMStartArguments ;