import { Client , connect , IClientOptions , ISubscriptionMap } from "mqtt" ; import { CommunicationBinding , CommunicationBindingJoinOptions , CommunicationBindingLogLevel , CommunicationBindingOptions , CommunicationBindingWithOptions , CommunicationEventLike , CommunicationEventType , CommunicationState , } from "../.." ; import { MqttTopic } from "./mqtt-topic" ; /**
 * Options provided by the MQTT communication binding.
 */ export interface MqttBindingOptions extends CommunicationBindingOptions { remarks brokerUrl : string ; /**
     * Keep alive interval in seconds (optional).
     *
     * Defaults to 60 seconds, set to 0 to disable.
     */ keepalive ? : number ; /**
     * Interval in milliseconds between two reconnection attempts (optional).
     *
     * Defaults to 1000 ms. Disable auto reconnect by setting to 0.
     */ reconnectPeriod ? : number ; /**
     * Time in milliseconds to wait for a connection acknowledgement message
     * from the broker (optional).
     *
     * Defaults to 30000 ms. If no CONNACK is received within the given time,
     * the connection is aborted.
     */ connectTimeout ? : number ; /**
     * The username required by your MQTT broker (optional).
     */ username ? : string ; /**
     * The password required by your MQTT broker (optional).
     */ password ? : string ; remarks tlsOptions ? : { /**
         * Private keys in PEM format (optional).
         */ key ? : string | string [ ] | Buffer | Buffer [ ] | Array < { pem : string | Buffer ; passphrase ? : string ; } > ; /**
         * Cert chains in PEM format (optional).
         */ cert ? : string | string [ ] | Buffer | Buffer [ ] ; /**
         * Optionally override the trusted CA certificates in PEM format (optional).
         */ ca ? : string | string [ ] | Buffer | Buffer [ ] ; /**
         * PFX or PKCS12 encoded private key and certificate chain. pfx is an
         * alternative to providing `key` and `cert` individually. PFX is
         * usually encrypted, if it is, `passphrase` will be used to decrypt it.
         */ pfx ? : string | string [ ] | Buffer | Buffer [ ] | Array < { buf : string | Buffer ; passphrase ? : string ; } > ; /**
         * Shared passphrase used for a single private key and/or a PFX.
         */ passphrase ? : string ; remarks rejectUnauthorized ? : boolean ; /**
         * Any other option supported by
         * [`tls.connect()`](https://nodejs.org/api/tls.html#tls_tls_createsecurecontext_options).
         */ [ option ] : any ; } ; /**
     * WebSocket specific connection options (optional).
     *
     * Default is {}. Only used for WebSocket connections.
     *
     * For possible options have a look at:
     * https://github.com/websockets/ws/blob/master/doc/ws.md.
     */ wsOptions ? : object ; /**
     * For WebSocket ws/wss protocols only (optional). Can be used to implement
     * signing urls or authentication options which upon reconnect can have
     * become expired.
     *
     * For details, see
     * [here](https://github.com/mqttjs/MQTT.js#refresh-authentication-options--signed-urls-with-transformwsurl-websocket-only).
     */ transformWsUrl ? : ( url : string , options : any , client : any ) => string ; } /**
 * Defines a communication binding for transmitting Coaty communication events
 * via the MQTT publish-subscribe messaging protocol.
 *
 * This binding is compatible with any MQTT broker that supports the MQTT v3.1.1
 * or v5.0 protocols. The binding itself connects to a broker using the MQTT
 * v3.1.1 protocol.
 *
 * This binding provides the following MQTT-specific publication options for Raw
 * events:
 * - retain: false | true (defaults to false)
 *
 * If specified, these options overwrite the default values supplied by the
 * binding.
 */ export class MqttBinding extends CommunicationBinding < MqttBindingOptions > { private _joinOptions : CommunicationBindingJoinOptions ; private _pendingPublicationItems : PublicationItem [ ] ; private _issuedSubscriptionItems : SubscriptionItem [ ] ; private _isDrainingPublications : boolean ; private _client : Client ; private _clientEventHandlers : Map < string , ( ... args : any [ ] ) => void > ; private _clientIdLogItem : string ; private _qos : 0 | 1 | 2 ; param options static withOptions ( options ) { return { type : MqttBinding , options } ; } get apiName ( ) : string { return "MQTT" ; } get apiVersion ( ) { return 3 ; } createIoRoute ( ioSourceId ) { return MqttTopic . getTopicName ( this . apiVersion , this . options . namespace , CommunicationEventType . IoValue , undefined , ioSourceId , undefined ) ; } protected onInit ( ) { this . _initClientEventHandlers ( ) ; this . _reset ( ) ; } protected onJoin ( joinOptions ) { this . _joinOptions = joinOptions ; const mqttClientOpts = { ... this . options , ... this . options . tlsOptions } ; delete ( mqttClientOpts as any ) . namespace ; delete ( mqttClientOpts as any ) . shouldEnableCrossNamespacing ; delete ( mqttClientOpts as any ) . logLevel ; delete ( mqttClientOpts as any ) . brokerUrl ; delete ( mqttClientOpts as any ) . tlsOptions ; delete mqttClientOpts . properties ; delete mqttClientOpts . reschedulePings ; delete mqttClientOpts . incomingStore ; delete mqttClientOpts . outgoingStore ; delete ( mqttClientOpts as any ) . customHandleAcks ; delete ( mqttClientOpts as any ) . authPacket ; this . _qos = 0 ; mqttClientOpts . will = this . _getLastWill ( ) ; const clientId = this . _brokerClientId ; mqttClientOpts . clientId = clientId ; this . _clientIdLogItem = `[${ clientId }] ` ; mqttClientOpts . clean = true ; mqttClientOpts . queueQoSZero = false ; mqttClientOpts . resubscribe = false ; mqttClientOpts . reschedulePings = false ; mqttClientOpts . protocolId = "MQTT" ; mqttClientOpts . protocolVersion = 4 ; let connectionUrl = this . options . brokerUrl ; const isWebPlatform = new Function ( "try {return this===window;}catch(e){return false;}" ) ( ) ; if ( isWebPlatform ) { connectionUrl = connectionUrl . replace ( /^mqtt/ , "ws" ) ; } this . _client = connect ( connectionUrl , mqttClientOpts ) ; this . log ( CommunicationBindingLogLevel . info , "Client connecting to " , this . options . brokerUrl ) ; this . _registerClientEventHandlers ( ) ; } protected onUnjoin ( ) { if ( ! this . _client ) { this . _reset ( ) ; return Promise . resolve ( ) ; } return new Promise < void > ( resolve => { this . _addPublicationItem ( this . _joinOptions . unjoinEvent , publishFailed => { this . _client . end ( publishFailed , ( ) => { this . _deregisterClientEventHandlers ( ) ; this . _reset ( ) ; resolve ( ) ; } ) ; } ) ; this . _drainPublications ( ) ; } ) ; } protected onPublish ( eventLike ) { if ( eventLike . eventType === CommunicationEventType . Raw && ( MqttTopic . isCoatyTopicLike ( eventLike . eventTypeFilter ) || ! MqttTopic . isValidTopic ( eventLike . eventTypeFilter , false ) ) ) { this . log ( CommunicationBindingLogLevel . error , "Raw publication topic is invalid: " , eventLike . eventTypeFilter ) ; return ; } this . _addPublicationItem ( eventLike ) ; this . _drainPublications ( ) ; } protected onSubscribe ( eventLike ) { if ( eventLike . eventType === CommunicationEventType . Raw && ( MqttTopic . isCoatyTopicLike ( eventLike . eventTypeFilter ) || ! MqttTopic . isValidTopic ( eventLike . eventTypeFilter , true ) ) ) { this . log ( CommunicationBindingLogLevel . error , "Raw subscription topic is invalid: " , eventLike . eventTypeFilter ) ; return ; } if ( eventLike . eventType === CommunicationEventType . IoValue && ( ( eventLike . isExternalIoRoute && MqttTopic . isCoatyTopicLike ( eventLike . eventTypeFilter ) ) || ! MqttTopic . isValidTopic ( eventLike . eventTypeFilter , false ) ) ) { this . log ( CommunicationBindingLogLevel . error , "IO route topic is invalid: " , eventLike . eventTypeFilter ) ; return ; } this . _addSubscriptionItem ( eventLike ) ; } protected onUnsubscribe ( eventLike ) { this . _removeSubscriptionItem ( eventLike ) ; } protected log ( logLevel , arg1 , arg2 ? , arg3 ? , arg4 ? ) { super . log ( logLevel , this . _clientIdLogItem , arg1 , arg2 , arg3 , arg4 ) ; } private _reset ( ) { this . _client = undefined ; this . _clientIdLogItem = "[---] " ; this . _joinOptions = undefined ; this . _isDrainingPublications = false ; this . _pendingPublicationItems = [ ] ; this . _issuedSubscriptionItems = [ ] ; this . emit ( "communicationState" , CommunicationState . Offline ) ; } private _initClientEventHandlers ( ) { if ( this . _clientEventHandlers ) { return ; } this . _clientEventHandlers = new Map ( ) ; this . _clientEventHandlers . set ( "connect" , ( connack ) => { this . log ( CommunicationBindingLogLevel . info , "Client connected" ) ; this . emit ( "communicationState" , CommunicationState . Online ) ; this . _subscribeItems ( this . _issuedSubscriptionItems ) ; const joinEvents = this . _joinOptions . joinEvents ; for ( let i = joinEvents . length - 1 ; i >= 0 ; i -- ) { this . _addPublicationItem ( joinEvents [ i ] , undefined , true , true ) ; } this . _isDrainingPublications = false ; this . _drainPublications ( ) ; } ) ; this . _clientEventHandlers . set ( "reconnect" , ( ) => { this . log ( CommunicationBindingLogLevel . info , "Client reconnecting..." ) ; } ) ; this . _clientEventHandlers . set ( "close" , ( ) => { this . log ( CommunicationBindingLogLevel . info , "Client disconnected" ) ; this . _isDrainingPublications = true ; this . emit ( "communicationState" , CommunicationState . Offline ) ; } ) ; this . _clientEventHandlers . set ( "offline" , ( ) => { this . log ( CommunicationBindingLogLevel . info , "Client offline" ) ; this . _isDrainingPublications = true ; this . emit ( "communicationState" , CommunicationState . Offline ) ; } ) ; this . _clientEventHandlers . set ( "error" , ( error ) => { this . log ( CommunicationBindingLogLevel . info , "Client error on connect: " , error ) ; this . emit ( "communicationState" , CommunicationState . Offline ) ; } ) ; this . _clientEventHandlers . set ( "message" , ( topic , payload , packet ) => { this . _dispatchMessage ( topic , payload ) ; } ) ; } private _registerClientEventHandlers ( ) { this . _client . on ( "connect" , this . _clientEventHandlers . get ( "connect" ) ) ; this . _client . on ( "reconnect" , this . _clientEventHandlers . get ( "reconnect" ) ) ; this . _client . on ( "close" , this . _clientEventHandlers . get ( "close" ) ) ; this . _client . on ( "offline" , this . _clientEventHandlers . get ( "offline" ) ) ; this . _client . on ( "error" , this . _clientEventHandlers . get ( "error" ) ) ; this . _client . on ( "message" , this . _clientEventHandlers . get ( "message" ) ) ; } private _deregisterClientEventHandlers ( ) { this . _client . removeListener ( "connect" , this . _clientEventHandlers . get ( "connect" ) ) ; this . _client . removeListener ( "reconnect" , this . _clientEventHandlers . get ( "reconnect" ) ) ; this . _client . removeListener ( "close" , this . _clientEventHandlers . get ( "close" ) ) ; this . _client . removeListener ( "offline" , this . _clientEventHandlers . get ( "offline" ) ) ; this . _client . removeListener ( "error" , this . _clientEventHandlers . get ( "error" ) ) ; this . _client . removeListener ( "message" , this . _clientEventHandlers . get ( "message" ) ) ; } private _getTopicFor ( eventLike ) { if ( eventLike . eventType === CommunicationEventType . Raw ) { return eventLike . eventTypeFilter ; } if ( eventLike . eventType === CommunicationEventType . IoValue ) { return eventLike . eventTypeFilter ; } return MqttTopic . getTopicName ( this . apiVersion , this . options . namespace , eventLike . eventType , eventLike . eventTypeFilter , eventLike . sourceId , eventLike . correlationId ) ; } private _getTopicFilterFor ( eventLike ) { if ( eventLike . eventType === CommunicationEventType . Raw ) { return eventLike . eventTypeFilter ; } if ( eventLike . eventType === CommunicationEventType . IoValue ) { return eventLike . eventTypeFilter ; } return MqttTopic . getTopicFilter ( this . apiVersion , this . options . shouldEnableCrossNamespacing ? undefined : this . options . namespace , eventLike . eventType , eventLike . eventTypeFilter , eventLike . correlationId ) ; } private _getLastWill ( ) { try { const unjoinEventLike = this . _joinOptions . unjoinEvent ; return { topic : this . _getTopicFor ( unjoinEventLike ) , payload : this . _encodePayload ( unjoinEventLike ) , qos : this . _qos , retain : false , } ; } catch ( error ) { this . log ( CommunicationBindingLogLevel . error , "Last will message cannot be composed: " , error ) ; return undefined ; } } /** Assign own client id because the MQTT.JS default value
     * is based on a weak random calculation:
     * 'mqttjs_' + Math.random().toString(16).substr(2, 8)
     *
     * MQTT Spec 3.1:
     * The Server MUST allow ClientIds which are between 1 and 23 UTF-8 encoded
     * bytes in length, and that contain only the characters
     * "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ".
     * The Server MAY allow ClientId’s that contain more than 23 encoded bytes.
     * The Server MAY allow ClientId’s that contain characters not included in the list given above.
     */ private get _brokerClientId ( ) { const id = this . _joinOptions . agentId ; return `Coaty${ id . replace ( /-/g , "" ) . substr ( 0 , 18 ) }` ; } private _dispatchMessage ( topicName , payload ) { try { const topic = MqttTopic . createByName ( topicName ) ; if ( topic === undefined || topic . eventType === CommunicationEventType . IoValue ) { this . _issuedSubscriptionItems . forEach ( item => { if ( ( item . eventType === CommunicationEventType . Raw && MqttTopic . matches ( topicName , item . topicFilter ) ) || ( item . eventType === CommunicationEventType . IoValue && topicName === item . topicFilter ) ) { this . log ( CommunicationBindingLogLevel . debug , "Inbound message as " , CommunicationEventType [ item . eventType ] , " on " , topicName ) ; this . emit ( "inboundEvent" , { eventType : item . eventType , eventTypeFilter : topicName , sourceId : topic ?. sourceId , correlationId : item . topicFilter , data : item . shouldDecodeData ? this . _decodePayload ( payload ) : payload , } ) ; } } ) ; return ; } this . log ( CommunicationBindingLogLevel . debug , "Inbound message on " , topicName ) ; this . emit ( "inboundEvent" , { eventType : topic . eventType , eventTypeFilter : topic . eventTypeFilter , sourceId : topic . sourceId , correlationId : topic . correlationId , data : this . _decodePayload ( payload ) , } ) ; } catch ( error ) { this . log ( CommunicationBindingLogLevel . error , "Inbound message error on topic " , topicName , ": " , error ) ; } } param eventLike returns throws private _encodePayload ( eventLike ) { if ( eventLike . isDataRaw ) { return eventLike . data ; } return JSON . stringify ( eventLike . data ) ; } private _decodePayload ( payload ) { return JSON . parse ( payload . toString ( ) ) ; } private _addPublicationItem ( eventLike , callback ? , shouldAddFirst = false , once = false ) { try { const topic = this . _getTopicFor ( eventLike ) ; const payload = this . _encodePayload ( eventLike ) ; const options = eventLike . options ; if ( once && this . _pendingPublicationItems . some ( i => i . topic === topic ) ) { callback && callback ( true ) ; return ; } if ( shouldAddFirst ) { this . _pendingPublicationItems . unshift ( { topic , payload , options , callback } ) ; } else { this . _pendingPublicationItems . push ( { topic , payload , options , callback } ) ; } } catch ( error ) { this . log ( CommunicationBindingLogLevel . error , "Publication message cannot be composed: " , error ) ; callback && callback ( true ) ; } } private _drainPublications ( ) { if ( this . _isDrainingPublications ) { return ; } this . _isDrainingPublications = true ; this . _doDrainPublications ( ) ; } private _doDrainPublications ( ) { if ( ! this . _client || this . _pendingPublicationItems . length === 0 ) { this . log ( CommunicationBindingLogLevel . debug , "Stop draining publications" ) ; this . _isDrainingPublications = false ; return ; } const pendingItem = this . _pendingPublicationItems [ 0 ] ; const { topic , payload , options , callback } = pendingItem ; this . log ( CommunicationBindingLogLevel . debug , "Publishing on " , topic ) ; this . _client . publish ( topic , payload , { qos : this . _qos , retain : options ?. retain || false } , err => { if ( err ) { this . log ( CommunicationBindingLogLevel . debug , "Error publishing on " , topic , err ) ; this . _isDrainingPublications = false ; this . _pendingPublicationItems . forEach ( item => item . callback && item . callback ( true ) ) ; } else { this . log ( CommunicationBindingLogLevel . debug , "Published on " , topic ) ; const index = this . _pendingPublicationItems . indexOf ( pendingItem ) ; if ( index !== - 1 ) { this . _pendingPublicationItems . splice ( index , 1 ) ; callback && callback ( false ) ; } this . _doDrainPublications ( ) ; } } ) ; } private _addSubscriptionItem ( eventLike ) { const topicFilter = this . _getTopicFilterFor ( eventLike ) ; const item = { eventType : eventLike . eventType , topicFilter , shouldDecodeData : ! eventLike . isDataRaw , } ; const index = this . _issuedSubscriptionItems . findIndex ( i => i . eventType === eventLike . eventType && i . topicFilter === topicFilter ) ; if ( index === - 1 ) { this . _issuedSubscriptionItems . push ( item ) ; } else { this . _issuedSubscriptionItems [ index ] = item ; } this . _subscribeItems ( item ) ; } private _removeSubscriptionItem ( eventLike ) { const topicFilter = this . _getTopicFilterFor ( eventLike ) ; const index = this . _issuedSubscriptionItems . findIndex ( i => i . eventType === eventLike . eventType && i . topicFilter === topicFilter ) ; if ( index === - 1 ) { return ; } const items = this . _issuedSubscriptionItems . splice ( index , 1 ) ; let otherEventType ; if ( eventLike . eventType === CommunicationEventType . Raw ) { otherEventType = CommunicationEventType . IoValue ; } else if ( eventLike . eventType === CommunicationEventType . IoValue ) { otherEventType = CommunicationEventType . Raw ; } if ( otherEventType !== undefined && this . _issuedSubscriptionItems . some ( i => i . eventType === otherEventType && i . topicFilter === topicFilter ) ) { return ; } this . _unsubscribeItems ( items ) ; } private _subscribeItems ( items ) { if ( this . _client ?. connected ) { const subsMap = { } ; const subsOpts = { qos : this . _qos } ; if ( Array . isArray ( items ) ) { items . forEach ( item => { subsMap [ item . topicFilter ] = subsOpts ; this . log ( CommunicationBindingLogLevel . debug , "Subscribed on " , item . topicFilter ) ; } ) ; } else { subsMap [ items . topicFilter ] = subsOpts ; this . log ( CommunicationBindingLogLevel . debug , "Subscribed on " , items . topicFilter ) ; } this . _client . subscribe ( subsMap ) ; } } private _unsubscribeItems ( items ) { if ( this . _client ?. connected ) { if ( Array . isArray ( items ) ) { const topicFilters = items . map ( i => i . topicFilter ) ; this . _client . unsubscribe ( topicFilters ) ; this . log ( CommunicationBindingLogLevel . debug , "Unsubscribed on " , topicFilters ) ; } else { this . _client . unsubscribe ( items . topicFilter ) ; this . log ( CommunicationBindingLogLevel . debug , "Unsubscribed on " , items . topicFilter ) ; } } } } /** Represents an item to be published. */ interface PublicationItem { topic : string ; payload : any ; options : { [ key ] : any ; } ; callback ? : ( publishFailed : boolean ) => void ; } /** Represents an item to be subscribed or unsubscribed. */ interface SubscriptionItem { eventType : CommunicationEventType ; topicFilter : string ; shouldDecodeData : boolean ; }