/********************************************************************************
 * Copyright (c) 2021 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
 ********************************************************************************/ import { MBusForm } from "./mbus" ; import { Content } from "@node-wot/core" ; import { Readable } from "stream" ; const MbusMaster = require ( "node-mbus" ) ; const configDefaults = { operationTimeout : 10000 , connectionRetryTime : 10000 , maxRetries : 5 , } ; /**
 * MBusConnection represents a client connected to a specific host and port
 */ export class MBusConnection { host : string ; port : number ; client : any ; connecting : boolean ; connected : boolean ; timer : NodeJS . Timer ; currentTransaction : MBusTransaction ; queue : Array < MBusTransaction > ; config : { connectionTimeout ? : number ; operationTimeout ? : number ; connectionRetryTime ? : number ; maxRetries ? : number ; } ; constructor ( host , port , config = configDefaults ) { this . host = host ; this . port = port ; this . client = new MbusMaster ( { host : this . host , port : this . port , timeout : config . connectionTimeout , autoConnect : false , } ) ; this . connecting = false ; this . connected = false ; this . timer = null ; this . currentTransaction = null ; this . queue = new Array < MBusTransaction > ( ) ; this . config = Object . assign ( configDefaults , config ) ; } param op enqueue ( op ) { for ( const t of this . queue ) { if ( op . unitId === t . unitId ) { t . inform ( op ) ; return ; } } const transaction = new MBusTransaction ( op . unitId , op . base ) ; transaction . inform ( op ) ; this . queue . push ( transaction ) ; } async connect ( ) { if ( ! this . connecting && ! this . connected ) { console . debug ( "[binding-mbus]" , "Trying to connect to" , this . host ) ; this . connecting = true ; for ( let retry = 0 ; retry < this . config . maxRetries ; retry ++ ) { if ( this . client . connect ( ( error ) => { if ( error != null ) console . warn ( "[binding-mbus]" , "Cannot connect to" , this . host , "reason" , error , ` retry in ${ this . config . connectionRetryTime }ms` ) ; } ) ) { this . connecting = false ; this . connected = true ; console . debug ( "[binding-mbus]" , "MBus connected to " + this . host ) ; return ; } else { this . connecting = false ; if ( retry >= this . config . maxRetries - 1 ) { throw new Error ( "Max connection retries" ) ; } await new Promise ( ( resolve , reject ) => setTimeout ( resolve , this . config . connectionRetryTime ) ) ; } } } } async execute ( op ) { this . trigger ( ) ; return op . execute ( ) ; } /**
     * Trigger work on this connection.
     *
     * If the MBusConnection is unconnected, connect it and retrigger.
     * If no transaction is currently being processed but transactions are waiting,
     * start the next transaction.
     * Retrigger after success or failure.
     */ async trigger ( ) { console . debug ( "[binding-mbus]" , "MBusConnection:trigger" ) ; if ( ! this . connecting && ! this . connected ) { try { await this . connect ( ) ; this . trigger ( ) ; } catch ( error ) { console . warn ( "[binding-mbus]" , "cannot reconnect to m-bus server" ) ; this . queue . forEach ( ( transaction ) => { transaction . operations . forEach ( ( op ) => { op . failed ( error ) ; } ) ; } ) ; } } else if ( this . connected && this . currentTransaction == null && this . queue . length > 0 ) { this . currentTransaction = this . queue . shift ( ) ; try { await this . executeTransaction ( this . currentTransaction ) ; this . currentTransaction = null ; this . trigger ( ) ; } catch ( error ) { console . warn ( "[binding-mbus]" , "transaction failed:" , error ) ; this . currentTransaction = null ; this . trigger ( ) ; } } } async executeTransaction ( transaction ) { console . debug ( "[binding-mbus]" , "Execute read operation on unit" , transaction . unitId ) ; try { const result = await this . readMBus ( transaction ) ; console . debug ( "[binding-mbus]" , "Got result from read operation on unit" , transaction . unitId ) ; transaction . operations . forEach ( ( op ) => op . done ( op . base , result ) ) ; } catch ( error ) { console . warn ( "[binding-mbus]" , "read operation failed on unit" , transaction . unitId , error ) ; transaction . operations . forEach ( ( op ) => op . failed ( error ) ) ; throw error ; } } public close ( ) { this . mbusstop ( ) ; } async readMBus ( transaction ) { return new Promise < unknown > ( ( resolve , reject ) => { console . debug ( "[binding-mbus]" , "Invoking read transaction" ) ; if ( this . timer ) { clearTimeout ( this . timer ) ; } this . timer = global . setTimeout ( ( ) => this . mbusstop ( ) , this . config . operationTimeout ) ; this . client . getData ( transaction . unitId , ( error , data ) => { if ( error !== null ) reject ( error ) ; resolve ( data ) ; } ) ; } ) ; } private mbusstop ( ) { console . debug ( "[binding-mbus]" , "Closing unused connection" ) ; this . client . close ( ( err ) => { if ( err === null ) { console . debug ( "[binding-mbus]" , "session closed" ) ; this . connecting = false ; this . connected = false ; } else { console . error ( "[binding-mbus]" , "cannot close session " + err ) ; } } ) ; clearInterval ( this . timer ) ; this . timer = null ; } } /**
 * MBusTransaction represents a raw M-Bus operation performed on a MBusConnection
 */ class MBusTransaction { unitId : number ; base : number ; operations : Array < PropertyOperation > ; constructor ( unitId , base ) { this . unitId = unitId ; this . base = base ; this . operations = new Array < PropertyOperation > ( ) ; } param op inform ( op ) { this . operations . push ( op ) ; } } /**
 * PropertyOperation represents a read or write operation on a property
 */ export class PropertyOperation { unitId : number ; base : number ; resolve : ( value ? : Content | PromiseLike < Content > ) => void ; reject : ( reason ? : Error ) => void ; constructor ( form ) { this . unitId = form [ "mbus:unitID" ] ; this . base = form [ "mbus:offset" ] ; } /**
     * Trigger execution of this operation.
     *
     */ async execute ( ) { return new Promise ( ( resolve , reject ) => { this . resolve = resolve ; this . reject = reject ; } ) ; } param base param result param data done ( base ? , result ? ) { console . debug ( "[binding-mbus]" , "Operation done" ) ; let payload = "" ; if ( base === - 1 ) { payload = result . SlaveInformation ; } else { result . DataRecord . forEach ( ( dataRec ) => { if ( base === dataRec . id ) { payload = dataRec ; } } ) ; } const resp = { body : Readable . from ( JSON . stringify ( payload ) ) , type : "application/json" , } ; this . resolve ( resp ) ; } param reason failed ( reason ) { console . warn ( "[binding-mbus]" , "Operation failed:" , reason ) ; this . reject ( new Error ( reason ) ) ; } }