/********************************************************************************
 * Copyright (c) 2018 Contributors to the Eclipse Foundation
 *
 * See the NOTICE file(s) distributed with this work for additional
 * information regarding copyright ownership.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v. 2.0 which is available at
 * http://www.eclipse.org/legal/epl-2.0, or the W3C Software Notice and
 * Document License (2015-05-13) which is available at
 * https://www.w3.org/Consortium/Legal/2015/copyright-software-and-document.
 *
 * SPDX-License-Identifier: EPL-2.0 OR W3C-20150513
 ********************************************************************************/ /**
 * MQTT Broker Server
 */ import { IPublishPacket } from "mqtt" ; import * as mqtt from "mqtt" ; import * as url from "url" ; import * as TD from "@node-wot/td-tools" ; import { ProtocolServer , Servient , ExposedThing , ContentSerdes , ProtocolHelpers , Content } from "@node-wot/core" ; import { InteractionOptions } from "wot-typescript-definitions" ; export default class MqttBrokerServer implements ProtocolServer { readonly scheme : string = "mqtt" ; private port = - 1 ; private address : string = undefined ; private user : string = undefined ; private psw : string = undefined ; private clientId : string = undefined ; private protocolVersion : number = undefined ; private brokerURI : string = undefined ; private readonly things : Map < string , ExposedThing > = new Map < string , ExposedThing > ( ) ; private broker : mqtt . MqttClient ; private rejectUnauthorized : boolean ; constructor ( uri , user ? , psw ? , clientId ? , protocolVersion ? , rejectUnauthorized ? ) { if ( uri !== undefined ) { if ( uri . indexOf ( "://" ) === - 1 ) { uri = this . scheme + "://" + uri ; } this . brokerURI = uri ; } if ( user !== undefined ) { this . user = user ; } if ( psw !== undefined ) { this . psw = psw ; } if ( clientId !== undefined ) { this . clientId = clientId ; } if ( protocolVersion !== undefined ) { this . protocolVersion = protocolVersion ; } this . rejectUnauthorized = rejectUnauthorized ; } public async expose ( thing ) { if ( this . broker === undefined ) { return ; } let name = thing . title ; if ( this . things . has ( name ) ) { const suffix = name . match ( /.+_([0-9]+)$/ ) ; if ( suffix !== null ) { name = name . slice ( 0 , - suffix [ 1 ] . length ) + ( 1 + parseInt ( suffix [ 1 ] ) ) ; } else { name = name + "_2" ; } } console . debug ( "[binding-mqtt]" , `MqttBrokerServer at ${ this . brokerURI } exposes '${ thing . title }' as unique '${ name }/*'` ) ; this . things . set ( name , thing ) ; for ( const propertyName in thing . properties ) { const topic = encodeURIComponent ( name ) + "/properties/" + encodeURIComponent ( propertyName ) ; const property = thing . properties [ propertyName ] ; if ( ! property . writeOnly ) { const href = this . brokerURI + "/" + topic ; const form = new TD . Form ( href , ContentSerdes . DEFAULT ) ; form . op = [ "readproperty" , "observeproperty" , "unobserveproperty" ] ; property . forms . push ( form ) ; console . debug ( "[binding-mqtt]" , `MqttBrokerServer at ${ this . brokerURI } assigns '${ href }' to property '${ propertyName }'` ) ; const observeListener = async ( data ) => { let content ; try { content = ContentSerdes . get ( ) . valueToContent ( data , property . data ) ; } catch ( err ) { console . warn ( "[binding-mqtt]" , `MqttServer cannot process data for Property '${ propertyName }': ${ err . message }` ) ; thing . handleUnobserveProperty ( propertyName , observeListener , { formIndex : property . forms . length - 1 , } ) ; return ; } console . debug ( "[binding-mqtt]" , `MqttBrokerServer at ${ this . brokerURI } publishing to Property topic '${ propertyName }' ` ) ; const buffer = await ProtocolHelpers . readStreamFully ( content . body ) ; this . broker . publish ( topic , buffer ) ; } ; thing . handleObserveProperty ( propertyName , observeListener , { formIndex : property . forms . length - 1 } ) ; } if ( ! property . readOnly ) { const href = this . brokerURI + "/" + topic + "/writeproperty" ; this . broker . subscribe ( topic + "/writeproperty" ) ; const form = new TD . Form ( href , ContentSerdes . DEFAULT ) ; form . op = [ "writeproperty" ] ; thing . properties [ propertyName ] . forms . push ( form ) ; console . debug ( "[binding-mqtt]" , `MqttBrokerServer at ${ this . brokerURI } assigns '${ href }' to property '${ propertyName }'` ) ; } } for ( const actionName in thing . actions ) { const topic = encodeURIComponent ( name ) + "/actions/" + encodeURIComponent ( actionName ) ; this . broker . subscribe ( topic ) ; const href = this . brokerURI + "/" + topic ; const form = new TD . Form ( href , ContentSerdes . DEFAULT ) ; form . op = [ "invokeaction" ] ; thing . actions [ actionName ] . forms . push ( form ) ; console . debug ( "[binding-mqtt]" , `MqttBrokerServer at ${ this . brokerURI } assigns '${ href }' to Action '${ actionName }'` ) ; } this . broker . on ( "message" , async ( receivedTopic , rawPayload , packet ) => { const segments = receivedTopic . split ( "/" ) ; let payload ; if ( rawPayload instanceof Buffer ) { payload = rawPayload ; } else if ( typeof rawPayload === "string" ) { payload = Buffer . from ( rawPayload ) ; } if ( segments . length === 4 ) { console . debug ( "[binding-mqtt]" , `MqttBrokerServer at ${ this . brokerURI } received message for '${ receivedTopic }'` ) ; const thing = this . things . get ( segments [ 1 ] ) ; if ( thing ) { if ( segments [ 2 ] === "actions" ) { const action = thing . actions [ segments [ 3 ] ] ; let value ; if ( action ) { if ( "properties" in packet && "contentType" in packet . properties ) { try { value = ContentSerdes . get ( ) . contentToValue ( { type : packet . properties . contentType , body : payload } , action . input ) ; } catch ( err ) { console . warn ( `MqttBrokerServer at ${ this . brokerURI } cannot process received message for '${ segments [ 3 ] }': ${ err . message }` ) ; } } else { try { value = JSON . parse ( payload . toString ( ) ) ; } catch ( err ) { console . warn ( `MqttBrokerServer at ${ this . brokerURI }, packet has no Content Type and does not parse as JSON, relaying raw (string) payload.` ) ; value = payload . toString ( ) ; } } } const options = { formIndex : ProtocolHelpers . findRequestMatchingFormIndex ( action . forms , this . scheme , this . brokerURI , "application/json" ) , } ; try { const output = await thing . handleInvokeAction ( segments [ 3 ] , value , options ) ; if ( output ) { console . warn ( `MqttBrokerServer at ${ this . brokerURI } cannot return output '${ segments [ 3 ] }'` ) ; } } catch ( err ) { console . error ( `MqttBrokerServer at ${ this . brokerURI } got error on invoking '${ segments [ 3 ] }': ${ err . message }` ) ; } return ; } } } else if ( segments . length === 5 && segments [ 4 ] === "writeproperty" ) { const thing = this . things . get ( segments [ 1 ] ) ; if ( thing ) { if ( segments [ 2 ] === "properties" ) { const property = thing . properties [ segments [ 3 ] ] ; if ( property ) { if ( ! property . readOnly ) { let contentType = ContentSerdes . DEFAULT ; if ( "contentType" in packet . properties ) { contentType = packet . properties . contentType ; } const options = { formIndex : ProtocolHelpers . findRequestMatchingFormIndex ( property . forms , this . scheme , this . brokerURI , contentType ) , } ; try { thing . handleWriteProperty ( segments [ 3 ] , JSON . parse ( payload . toString ( ) ) , options ) ; } catch ( err ) { console . error ( "[binding-mqtt]" , `MqttBrokerServer at ${ this . brokerURI } got error on writing to property '${ segments [ 3 ] }': ${ err . message }` ) ; } return ; } else { console . warn ( "[binding-mqtt]" , `MqttBrokerServer at ${ this . brokerURI } received message for readOnly property at '${ receivedTopic }'` ) ; return ; } } } } return ; } console . warn ( "[binding-mqtt]" , `MqttBrokerServer at ${ this . brokerURI } received message for invalid topic '${ receivedTopic }'` ) ; } ) ; for ( const eventName in thing . events ) { const topic = encodeURIComponent ( name ) + "/events/" + encodeURIComponent ( eventName ) ; const event = thing . events [ eventName ] ; const href = this . brokerURI + "/" + topic ; const form = new TD . Form ( href , ContentSerdes . DEFAULT ) ; form . op = [ "subscribeevent" , "unsubscribeevent" ] ; event . forms . push ( form ) ; console . debug ( "[binding-mqtt]" , `MqttBrokerServer at ${ this . brokerURI } assigns '${ href }' to Event '${ eventName }'` ) ; const eventListener = async ( content ) => { if ( ! content ) { console . warn ( "[binding-mqtt]" , `HttpServer on port ${ this . getPort ( ) } cannot process data for Event ${ eventName }` ) ; thing . handleUnsubscribeEvent ( eventName , eventListener , { formIndex : event . forms . length - 1 } ) ; return ; } console . debug ( "[binding-mqtt]" , `MqttBrokerServer at ${ this . brokerURI } publishing to Event topic '${ eventName }' ` ) ; const buffer = await ProtocolHelpers . readStreamFully ( content . body ) ; this . broker . publish ( topic , buffer ) ; } ; thing . handleSubscribeEvent ( eventName , eventListener , { formIndex : event . forms . length - 1 } ) ; } this . broker . publish ( name , JSON . stringify ( thing . getThingDescription ( ) ) , { retain : true } ) ; } public destroy ( thingId ) { console . debug ( "[binding-mqtt]" , `MqttBrokerServer on port ${ this . getPort ( ) } destroying thingId '${ thingId }'` ) ; return new Promise < boolean > ( ( resolve , reject ) => { let removedThing ; for ( const name of Array . from ( this . things . keys ( ) ) ) { const expThing = this . things . get ( name ) ; if ( expThing != null && expThing . id != null && expThing . id === thingId ) { this . things . delete ( name ) ; removedThing = expThing ; } } if ( removedThing ) { console . info ( "[binding-mqtt]" , `MqttBrokerServer succesfully destroyed '${ removedThing . title }'` ) ; } else { console . info ( "[binding-mqtt]" , `MqttBrokerServer failed to destroy thing with thingId '${ thingId }'` ) ; } resolve ( removedThing !== undefined ) ; } ) ; } public start ( servient ) { return new Promise < void > ( ( resolve , reject ) => { if ( this . brokerURI === undefined ) { console . warn ( "[binding-mqtt]" , `No broker defined for MQTT server binding - skipping` ) ; resolve ( ) ; } else { if ( this . psw === undefined ) { console . debug ( "[binding-mqtt]" , `MqttBrokerServer trying to connect to broker at ${ this . brokerURI }` ) ; } else if ( this . clientId === undefined ) { console . debug ( "[binding-mqtt]" , `MqttBrokerServer trying to connect to secured broker at ${ this . brokerURI }` ) ; } else if ( this . protocolVersion === undefined ) { console . debug ( "[binding-mqtt]" , `MqttBrokerServer trying to connect to secured broker at ${ this . brokerURI } with client ID ${ this . clientId }` ) ; } else { console . debug ( "[binding-mqtt]" , `MqttBrokerServer trying to connect to secured broker at ${ this . brokerURI } with client ID ${ this . clientId }` ) ; } this . broker = mqtt . connect ( this . brokerURI , { username : this . user , password : this . psw , clientId : this . clientId , protocolVersion : this . protocolVersion , rejectUnauthorized : this . rejectUnauthorized , } ) ; this . broker . on ( "connect" , ( ) => { console . info ( "[binding-mqtt]" , `MqttBrokerServer connected to broker at ${ this . brokerURI }` ) ; const parsed = new url . URL ( this . brokerURI ) ; this . address = parsed . hostname ; const port = parseInt ( parsed . port ) ; this . port = port > 0 ? port : 1883 ; resolve ( ) ; } ) ; this . broker . on ( "error" , ( error ) => { console . error ( "[binding-mqtt]" , `MqttBrokerServer could not connect to broker at ${ this . brokerURI }` ) ; reject ( error ) ; } ) ; } } ) ; } public stop ( ) { return new Promise < void > ( ( resolve , reject ) => { if ( this . broker === undefined ) resolve ( ) ; this . broker . unsubscribe ( "*" ) ; } ) ; } public getPort ( ) { return this . port ; } public getAddress ( ) { return this . address ; } }