import { deserializeMessage , IMessage , IMessageMetadata , IOutputResponse } from "@iota/iota.js" ; import { Converter , RandomHelper , ReadStream } from "@iota/util.js" ; import * as mqtt from "mqtt" ; import type { IMqttMilestoneResponse } from "../models/api/IMqttMilestoneResponse" ; import type { IMqttClient } from "../models/IMqttClient" ; import type { IMqttStatus } from "../models/IMqttStatus" ; /**
 * MQTT Client implementation for pub/sub communication.
 */ export class MqttClient implements IMqttClient { internal private readonly _endpoints : string [ ] ; internal private _endpointsIndex : number ; internal private readonly _keepAliveTimeoutSeconds : number ; internal private _client ? : mqtt . MqttClient ; internal private _lastMessageTime : number ; internal private _timerId ? : NodeJS . Timeout ; internal private readonly _subscriptions : { [ topic ] : { /**
             * Should we deserialize the data as JSON.
             */ isJson : boolean ; /**
             * The callback for the subscriptions.
             */ subscriptionCallbacks : { /**
                 * The id of the subscription.
                 */ subscriptionId : string ; param event param data callback ( event , data ) : void ; } [ ] ; } ; } ; internal private readonly _statusSubscriptions : { [ subscriptionId ] : ( data : IMqttStatus ) => void ; } ; param endpoints param keepAliveTimeoutSeconds constructor ( endpoints , keepAliveTimeoutSeconds = 30 ) { this . _endpoints = Array . isArray ( endpoints ) ? endpoints : [ endpoints ] ; this . _endpointsIndex = 0 ; this . _subscriptions = { } ; this . _statusSubscriptions = { } ; this . _lastMessageTime = - 1 ; this . _keepAliveTimeoutSeconds = keepAliveTimeoutSeconds ; } param callback returns public milestonesLatest ( callback ) { return this . internalSubscribe ( "milestones/latest" , true , callback ) ; } param callback returns public milestonesConfirmed ( callback ) { return this . internalSubscribe ( "milestones/confirmed" , true , callback ) ; } param messageId param callback returns public messageMetadata ( messageId , callback ) { return this . internalSubscribe ( `messages/${ messageId }/metadata` , true , callback ) ; } param outputId param callback returns public output ( outputId , callback ) { return this . internalSubscribe ( `outputs/${ outputId }` , true , callback ) ; } param addressBech32 param callback returns public addressOutputs ( addressBech32 , callback ) { return this . internalSubscribe ( `addresses/${ addressBech32 }/outputs` , true , callback ) ; } param addressEd25519 param callback returns public addressEd25519Outputs ( addressEd25519 , callback ) { return this . internalSubscribe ( `addresses/ed25519/${ addressEd25519 }/outputs` , true , callback ) ; } param callback returns public messagesRaw ( callback ) { return this . internalSubscribe < Uint8Array > ( "messages" , false , ( topic , raw ) => { callback ( topic , raw ) ; } ) ; } param callback returns public messages ( callback ) { return this . internalSubscribe < Uint8Array > ( "messages" , false , ( topic , raw ) => { callback ( topic , deserializeMessage ( new ReadStream ( raw ) ) , raw ) ; } ) ; } param index param callback returns public indexRaw ( index , callback ) { return this . internalSubscribe < Uint8Array > ( `messages/indexation/${ typeof index === "string" ? Converter . utf8ToHex ( index ) : Converter . bytesToHex ( index ) }` , false , ( topic , raw ) => { callback ( topic , raw ) ; } ) ; } param index param callback returns public index ( index , callback ) { return this . internalSubscribe < Uint8Array > ( `messages/indexation/${ typeof index === "string" ? Converter . utf8ToHex ( index ) : Converter . bytesToHex ( index ) }` , false , ( topic , raw ) => { callback ( topic , deserializeMessage ( new ReadStream ( raw ) ) , raw ) ; } ) ; } param callback returns public messagesMetadata ( callback ) { return this . internalSubscribe ( "messages/referenced" , true , callback ) ; } param transactionId param callback returns public transactionIncludedMessageRaw ( transactionId , callback ) { return this . internalSubscribe ( `transactions/${ transactionId }/included-message` , false , callback ) ; } param transactionId param callback returns public transactionIncludedMessage ( transactionId , callback ) { return this . internalSubscribe < Uint8Array > ( `transactions/${ transactionId }/included-message` , false , ( topic , raw ) => { callback ( topic , deserializeMessage ( new ReadStream ( raw ) ) , raw ) ; } ) ; } param customTopic param callback returns public subscribeRaw ( customTopic , callback ) { return this . internalSubscribe ( customTopic , false , callback ) ; } param customTopic param callback returns public subscribeJson < T > ( customTopic , callback ) { return this . internalSubscribe < T > ( customTopic , true , callback ) ; } param subscriptionId public unsubscribe ( subscriptionId ) { this . triggerStatusCallbacks ( { type : "subscription-remove" , message : subscriptionId , state : this . calculateState ( ) } ) ; if ( this . _statusSubscriptions [ subscriptionId ] ) { delete this . _statusSubscriptions [ subscriptionId ] ; } else { const topics = Object . keys ( this . _subscriptions ) ; for ( let i = 0 ; i < topics . length ; i ++ ) { const topic = topics [ i ] ; for ( let j = 0 ; j < this . _subscriptions [ topic ] . subscriptionCallbacks . length ; j ++ ) { if ( this . _subscriptions [ topic ] . subscriptionCallbacks [ j ] . subscriptionId === subscriptionId ) { this . _subscriptions [ topic ] . subscriptionCallbacks . splice ( j , 1 ) ; if ( this . _subscriptions [ topic ] . subscriptionCallbacks . length === 0 ) { delete this . _subscriptions [ topic ] ; this . mqttUnsubscribe ( topic ) ; } return ; } } } } } param callback returns public statusChanged ( callback ) { const subscriptionId = Converter . bytesToHex ( RandomHelper . generate ( 32 ) ) ; this . _statusSubscriptions [ subscriptionId ] = callback ; return subscriptionId ; } param customTopic param isJson param callback returns internal private internalSubscribe < T > ( customTopic , isJson , callback ) { let isNewTopic = false ; if ( ! this . _subscriptions [ customTopic ] ) { this . _subscriptions [ customTopic ] = { isJson , subscriptionCallbacks : [ ] } ; isNewTopic = true ; } const subscriptionId = Converter . bytesToHex ( RandomHelper . generate ( 32 ) ) ; this . _subscriptions [ customTopic ] . subscriptionCallbacks . push ( { subscriptionId , callback } ) ; this . triggerStatusCallbacks ( { type : "subscription-add" , message : subscriptionId , state : this . calculateState ( ) } ) ; if ( isNewTopic ) { this . mqttSubscribe ( customTopic ) ; } return subscriptionId ; } param topic internal private mqttSubscribe ( topic ) { if ( ! this . _client ) { this . mqttConnect ( ) ; } else { try { this . _client . subscribe ( topic ) ; } catch ( err ) { this . triggerStatusCallbacks ( { type : "error" , message : `Subscribe to topic ${ topic } failed on ${ this . _endpoints [ this . _endpointsIndex ] }` , state : this . calculateState ( ) , error : err } ) ; } } } param topic internal private mqttUnsubscribe ( topic ) { if ( this . _client ) { try { this . _client . unsubscribe ( topic ) ; } catch ( err ) { this . triggerStatusCallbacks ( { type : "error" , message : `Unsubscribe from topic ${ topic } failed on ${ this . _endpoints [ this . _endpointsIndex ] }` , state : this . calculateState ( ) , error : err } ) ; } } } internal private mqttConnect ( ) { if ( ! this . _client ) { try { this . _client = mqtt . connect ( this . _endpoints [ this . _endpointsIndex ] , { keepalive : 0 , reconnectPeriod : this . _keepAliveTimeoutSeconds * 1000 } ) ; this . _client . on ( "connect" , ( ) => { try { if ( this . _client ) { this . _client . subscribe ( Object . keys ( this . _subscriptions ) ) ; this . startKeepAlive ( ) ; this . triggerStatusCallbacks ( { type : "connect" , message : `Connection complete ${ this . _endpoints [ this . _endpointsIndex ] }` , state : this . calculateState ( ) } ) ; } } catch ( err ) { this . triggerStatusCallbacks ( { type : "error" , message : `Subscribe to topics failed on ${ this . _endpoints [ this . _endpointsIndex ] }` , state : this . calculateState ( ) , error : err } ) ; } } ) ; this . _client . on ( "message" , ( topic , message ) => { this . _lastMessageTime = Date . now ( ) ; this . triggerCallbacks ( topic , message ) ; } ) ; this . _client . on ( "error" , err => { this . triggerStatusCallbacks ( { type : "error" , message : `Error on ${ this . _endpoints [ this . _endpointsIndex ] }` , state : this . calculateState ( ) , error : err } ) ; this . nextClient ( ) ; } ) ; } catch ( err ) { this . triggerStatusCallbacks ( { type : "connect" , message : `Connection failed to ${ this . _endpoints [ this . _endpointsIndex ] }` , state : this . calculateState ( ) , error : err } ) ; this . nextClient ( ) ; } } } internal private mqttDisconnect ( ) { this . stopKeepAlive ( ) ; if ( this . _client ) { const localClient = this . _client ; this . _client = undefined ; try { localClient . unsubscribe ( Object . keys ( this . _subscriptions ) ) ; localClient . end ( ) ; } catch { } this . triggerStatusCallbacks ( { type : "disconnect" , message : `Disconnect complete ${ this . _endpoints [ this . _endpointsIndex ] }` , state : this . calculateState ( ) } ) ; } } param topic param data internal private triggerCallbacks ( topic , data ) { if ( this . _subscriptions [ topic ] ) { let decodedData = data ; if ( this . _subscriptions [ topic ] . isJson ) { try { decodedData = JSON . parse ( ( data as Buffer ) . toString ( ) ) ; } catch ( err ) { this . triggerStatusCallbacks ( { type : "error" , message : `Error decoding JSON for topic ${ topic }` , state : this . calculateState ( ) , error : err } ) ; } } for ( let i = 0 ; i < this . _subscriptions [ topic ] . subscriptionCallbacks . length ; i ++ ) { try { this . _subscriptions [ topic ] . subscriptionCallbacks [ i ] . callback ( topic , decodedData ) ; } catch ( err ) { this . triggerStatusCallbacks ( { type : "error" , message : `Triggering callback failed for topic ${ topic } on subscription ${ this . _subscriptions [ topic ] . subscriptionCallbacks [ i ] . subscriptionId }` , state : this . calculateState ( ) , error : err } ) ; } } } } param status internal private triggerStatusCallbacks ( status ) { const subscriptionIds = Object . keys ( this . _statusSubscriptions ) ; for ( let i = 0 ; i < subscriptionIds . length ; i ++ ) { this . _statusSubscriptions [ subscriptionIds [ i ] ] ( status ) ; } } internal private startKeepAlive ( ) { this . stopKeepAlive ( ) ; this . _lastMessageTime = Date . now ( ) ; this . _timerId = setInterval ( ( ) => this . keepAlive ( ) , ( this . _keepAliveTimeoutSeconds / 2 ) * 1000 ) ; } internal private stopKeepAlive ( ) { if ( this . _timerId !== undefined ) { clearInterval ( this . _timerId ) ; this . _timerId = undefined ; } } internal private keepAlive ( ) { if ( Date . now ( ) - this . _lastMessageTime > this . _keepAliveTimeoutSeconds * 1000 ) { this . mqttDisconnect ( ) ; this . nextClient ( ) ; this . mqttConnect ( ) ; } } returns internal private calculateState ( ) { let state = "disconnected" ; if ( this . _client ) { if ( this . _client . connected ) { state = "connected" ; } else if ( this . _client . disconnecting ) { state = "disconnecting" ; } else if ( this . _client . reconnecting ) { state = "connecting" ; } } return state ; } /**
     * If there has been a problem switch to the next client endpoint.
     */ private nextClient ( ) { this . _endpointsIndex ++ ; if ( this . _endpointsIndex >= this . _endpoints . length ) { this . _endpointsIndex = 0 ; } } }