/**
 * Connection that wraps a socket and provides an interface to interact with
 * the Home Assistant websocket API.
 */ import * as messages from "./messages.js" ; import { ERR_INVALID_AUTH , ERR_CONNECTION_LOST } from "./errors.js" ; import { HassEvent , MessageBase } from "./types.js" ; import { HaWebSocket } from "./socket.js" ; import type { Auth } from "./auth.js" ; const DEBUG = false ; export type ConnectionOptions = { setupRetry : number ; auth ? : Auth ; createSocket : ( options : ConnectionOptions ) => Promise < HaWebSocket > ; } ; export type ConnectionEventListener = ( conn : Connection , eventData ? : any ) => void ; type Events = "ready" | "disconnected" | "reconnect-error" ; type WebSocketPongResponse = { id : number ; type : "pong" ; } ; type WebSocketEventResponse = { id : number ; type : "event" ; event : HassEvent ; } ; type WebSocketResultResponse = { id : number ; type : "result" ; success : true ; result : any ; } ; type WebSocketResultErrorResponse = { id : number ; type : "result" ; success : false ; error : { code : string ; message : string ; } ; } ; type WebSocketResponse = WebSocketPongResponse | WebSocketEventResponse | WebSocketResultResponse | WebSocketResultErrorResponse ; type SubscriptionUnsubscribe = ( ) => Promise < void > ; interface SubscribeEventCommmandInFlight < T > { resolve : ( result ? : any ) => void ; reject : ( err : any ) => void ; callback : ( ev : T ) => void ; subscribe : ( ( ) => Promise < SubscriptionUnsubscribe > ) | undefined ; unsubscribe : SubscriptionUnsubscribe ; } type CommandWithAnswerInFlight = { resolve : ( result ? : any ) => void ; reject : ( err : any ) => void ; } ; type CommandInFlight = SubscribeEventCommmandInFlight < any > | CommandWithAnswerInFlight ; export class Connection { options : ConnectionOptions ; commandId : number ; commands : Map < number , CommandInFlight > ; eventListeners : Map < string , ConnectionEventListener [ ] > ; closeRequested : boolean ; suspendReconnectPromise ? : Promise < void > ; oldSubscriptions ? : Map < number , CommandInFlight > ; _queuedMessages ? : Array < { resolve : ( value ? : unknown ) => unknown ; reject ? : ( err : typeof ERR_CONNECTION_LOST ) => unknown ; } > ; socket ? : HaWebSocket ; /**
     * Version string of the Home Assistant instance. Set to version of last connection while reconnecting.
     */ haVersion : string ; constructor ( socket , options ) { this . options = options ; this . commandId = 1 ; this . commands = new Map ( ) ; this . eventListeners = new Map ( ) ; this . closeRequested = false ; this . _setSocket ( socket ) ; } get connected ( ) { return ( this . socket !== undefined && this . socket . readyState == this . socket . OPEN ) ; } private _setSocket ( socket ) { this . socket = socket ; this . haVersion = socket . haVersion ; socket . addEventListener ( "message" , this . _handleMessage ) ; socket . addEventListener ( "close" , this . _handleClose ) ; const oldSubscriptions = this . oldSubscriptions ; if ( oldSubscriptions ) { this . oldSubscriptions = undefined ; oldSubscriptions . forEach ( ( info ) => { if ( "subscribe" in info && info . subscribe ) { info . subscribe ( ) . then ( ( unsub ) => { info . unsubscribe = unsub ; info . resolve ( ) ; } ) ; } } ) ; } const queuedMessages = this . _queuedMessages ; if ( queuedMessages ) { this . _queuedMessages = undefined ; for ( const queuedMsg of queuedMessages ) { queuedMsg . resolve ( ) ; } } this . fireEvent ( "ready" ) ; } addEventListener ( eventType , callback ) { let listeners = this . eventListeners . get ( eventType ) ; if ( ! listeners ) { listeners = [ ] ; this . eventListeners . set ( eventType , listeners ) ; } listeners . push ( callback ) ; } removeEventListener ( eventType , callback ) { const listeners = this . eventListeners . get ( eventType ) ; if ( ! listeners ) { return ; } const index = listeners . indexOf ( callback ) ; if ( index !== - 1 ) { listeners . splice ( index , 1 ) ; } } fireEvent ( eventType , eventData ? ) { ( this . eventListeners . get ( eventType ) || [ ] ) . forEach ( ( callback ) => callback ( this , eventData ) ) ; } suspendReconnectUntil ( suspendPromise ) { this . suspendReconnectPromise = suspendPromise ; } suspend ( ) { if ( ! this . suspendReconnectPromise ) { throw new Error ( "Suspend promise not set" ) ; } if ( this . socket ) { this . socket . close ( ) ; } } param force reconnect ( force = false ) { if ( ! this . socket ) { return ; } if ( ! force ) { this . socket . close ( ) ; return ; } this . socket . removeEventListener ( "message" , this . _handleMessage ) ; this . socket . removeEventListener ( "close" , this . _handleClose ) ; this . socket . close ( ) ; this . _handleClose ( ) ; } close ( ) { this . closeRequested = true ; if ( this . socket ) { this . socket . close ( ) ; } } param callback param eventType returns async subscribeEvents < EventType > ( callback , eventType ? ) { return this . subscribeMessage ( callback , messages . subscribeEvents ( eventType ) ) ; } ping ( ) { return this . sendMessagePromise ( messages . ping ( ) ) ; } sendMessage ( message , commandId ? ) { if ( ! this . connected ) { throw ERR_CONNECTION_LOST ; } if ( DEBUG ) { console . log ( "Sending" , message ) ; } if ( this . _queuedMessages ) { if ( commandId ) { throw new Error ( "Cannot queue with commandId" ) ; } this . _queuedMessages . push ( { resolve : ( ) => this . sendMessage ( message ) } ) ; return ; } if ( ! commandId ) { commandId = this . _genCmdId ( ) ; } message . id = commandId ; this . socket ! . send ( JSON . stringify ( message ) ) ; } sendMessagePromise < Result > ( message ) { return new Promise ( ( resolve , reject ) => { if ( this . _queuedMessages ) { this . _queuedMessages ! . push ( { reject , resolve : async ( ) => { try { resolve ( await this . sendMessagePromise ( message ) ) ; } catch ( err ) { reject ( err ) ; } } , } ) ; return ; } const commandId = this . _genCmdId ( ) ; this . commands . set ( commandId , { resolve , reject } ) ; this . sendMessage ( message , commandId ) ; } ) ; } param message param callback param options . resubscribe returns async subscribeMessage < Result > ( callback , subscribeMessage , options ? ) { if ( this . _queuedMessages ) { await new Promise ( ( resolve , reject ) => { this . _queuedMessages ! . push ( { resolve , reject } ) ; } ) ; } let info ; await new Promise ( ( resolve , reject ) => { const commandId = this . _genCmdId ( ) ; info = { resolve , reject , callback , subscribe : options ?. resubscribe !== false ? ( ) => this . subscribeMessage ( callback , subscribeMessage ) : undefined , unsubscribe : async ( ) => { if ( this . connected ) { await this . sendMessagePromise ( messages . unsubscribeEvents ( commandId ) ) ; } this . commands . delete ( commandId ) ; } , } ; this . commands . set ( commandId , info ) ; try { this . sendMessage ( subscribeMessage , commandId ) ; } catch ( err ) { } } ) ; return ( ) => info . unsubscribe ( ) ; } private _handleMessage = ( event ) => { const message = JSON . parse ( event . data ) ; if ( DEBUG ) { console . log ( "Received" , message ) ; } const info = this . commands . get ( message . id ) ; switch ( message . type ) { case "event" : if ( info ) { ( info as SubscribeEventCommmandInFlight < any > ) . callback ( message . event ) ; } else { console . warn ( `Received event for unknown subscription ${ message . id }. Unsubscribing.` ) ; this . sendMessagePromise ( messages . unsubscribeEvents ( message . id ) ) ; } break ; case "result" : if ( info ) { if ( message . success ) { info . resolve ( message . result ) ; if ( ! ( "subscribe" in info ) ) { this . commands . delete ( message . id ) ; } } else { info . reject ( message . error ) ; this . commands . delete ( message . id ) ; } } break ; case "pong" : if ( info ) { info . resolve ( ) ; this . commands . delete ( message . id ) ; } else { console . warn ( `Received unknown pong response ${ message . id }` ) ; } break ; default : if ( DEBUG ) { console . warn ( "Unhandled message" , message ) ; } } } ; private _handleClose = async ( ) => { const oldCommands = this . commands ; this . commandId = 1 ; this . oldSubscriptions = this . commands ; this . commands = new Map ( ) ; this . socket = undefined ; oldCommands . forEach ( ( info ) => { if ( ! ( "subscribe" in info ) ) { info . reject ( messages . error ( ERR_CONNECTION_LOST , "Connection lost" ) ) ; } } ) ; if ( this . closeRequested ) { return ; } this . fireEvent ( "disconnected" ) ; const options = { ... this . options , setupRetry : 0 } ; const reconnect = ( tries ) => { setTimeout ( async ( ) => { if ( this . closeRequested ) { return ; } if ( DEBUG ) { console . log ( "Trying to reconnect" ) ; } try { const socket = await options . createSocket ( options ) ; this . _setSocket ( socket ) ; } catch ( err ) { if ( this . _queuedMessages ) { const queuedMessages = this . _queuedMessages ; this . _queuedMessages = undefined ; for ( const msg of queuedMessages ) { if ( msg . reject ) { msg . reject ( ERR_CONNECTION_LOST ) ; } } } if ( err === ERR_INVALID_AUTH ) { this . fireEvent ( "reconnect-error" , err ) ; } else { reconnect ( tries + 1 ) ; } } } , Math . min ( tries , 5 ) * 1000 ) ; } ; if ( this . suspendReconnectPromise ) { await this . suspendReconnectPromise ; this . suspendReconnectPromise = undefined ; this . _queuedMessages = [ ] ; } reconnect ( 0 ) ; } ; private _genCmdId ( ) { return ++ this . commandId ; } }