handle retained messages, introduc missing event detector

src/main.ts

@@ -6,23 +6,33 @@ import * as cmdargs from 'command-line-args'
 
 import * as EspThermToJson from './espthermtojson'
 import * as MongoSave from './mongosave'
+import * as MissingEventDetector from './missingeventdetector'
 
-log.info("Dispatcher starting")
 
-const optionDefinitions = [
+const OPTION_DEFINITIONS = [
     { name: 'verbose', alias: 'v', type: Boolean },
     { name: 'config', alias: 'c', type: String, defaultValue: '~/mqttDispatcher.conf' }
 ];
-const OPTIONS = cmdargs(optionDefinitions)
+const OPTIONS = cmdargs(OPTION_DEFINITIONS)
 const CONFIG = JSON.parse(fs.readFileSync(OPTIONS.config, "utf8"))
 
+
+
+
+log.info("Dispatcher starting")
+
 let dispatcher = new mqtt.MqttDispatcher(CONFIG.brokerUrl, 
     CONFIG.brokerUser, CONFIG.brokerPass, CONFIG.brokerCa)
 
-dispatcher.register('IoT/espThermometer2/#', 'toJson', EspThermToJson.espThermToJson)
+const ESP_THERM_TOPIC : string = 'IoT/espThermometer2/#'
+dispatcher.register(ESP_THERM_TOPIC, 'toJson', EspThermToJson.espThermToJson)
 
-let mongo : MongoSave.MongoSave = new MongoSave.MongoSave(CONFIG.mongoDbUrl)
-dispatcher.register('IoT/espThermometer2/#', 'MongoSave', mongo);
+let missingeventdetector : MissingEventDetector.MissingEventDetector = 
+    new MissingEventDetector.MissingEventDetector()
+dispatcher.register(ESP_THERM_TOPIC, 'MissingEventDetector', missingeventdetector)
+
+// let mongo : MongoSave.MongoSave = new MongoSave.MongoSave(CONFIG.mongoDbUrl)
+// dispatcher.register(ESP_THERM_TOPIC, 'MongoSave', mongo);
 
 
 dispatcher.exec()

src/missingeventdetector.ts

@@ -0,0 +1,71 @@
+import * as log from './log'
+import * as CallChain from './callchain'
+import * as Processor from './processor'
+
+
+const CHECK_PERIOD : number = 10  // seconds
+
+class ClientEntry {
+    client : string
+    lastEvent : number
+    delay : number
+    count : number
+    delaySum : number
+    avgDelay : number
+}
+
+class MissingEventProcessor extends Processor.AProcessor {
+    private timer : NodeJS.Timer
+    private clientMap : Map<string, ClientEntry> = new Map<string, ClientEntry>()
+
+    constructor() {
+        super("MissingEventProcessor")
+        this.timer = setInterval(() => {
+            this.clientMap.forEach((value : ClientEntry, key : string) : void => {
+                let currentTime : number = new Date().getTime()
+                let elapsedTime : number = currentTime - value.lastEvent
+                log.info(`Checking ${key}, elapsed: ${elapsedTime / 1000}, avg. delay: ${value.avgDelay / 1000}`)
+            })
+        }, CHECK_PERIOD * 1000)
+    }
+
+    protected process(message : any) : void {
+        let client = message.metadata.client
+        log.info(`Event for client ${client}`)
+        let currentTime : number = new Date().getTime()
+        if (this.clientMap.has(client)) {
+            let clientEntry = <ClientEntry>this.clientMap.get(client)
+            clientEntry.client = client
+            clientEntry.delay = currentTime - clientEntry.lastEvent
+            clientEntry.lastEvent = currentTime
+            clientEntry.count += 1
+            clientEntry.delaySum += clientEntry.delay
+            clientEntry.avgDelay = clientEntry.delaySum / clientEntry.count
+            this.clientMap.set(client, clientEntry)
+            log.info(`Entry for ${client} updated`)
+        } else {
+            let clientEntry : ClientEntry = new ClientEntry()
+            clientEntry.client = client
+            clientEntry.lastEvent = currentTime
+            clientEntry.delay = 0
+            clientEntry.count = 0
+            clientEntry.delaySum = 0
+            clientEntry.avgDelay = 0
+            this.clientMap.set(client, clientEntry)
+            log.info(`Entry for ${client} inserted`)
+        }
+    }
+}
+
+export class MissingEventDetector extends CallChain.ABaseChainItem {
+    private missingEventProcessor : MissingEventProcessor = new MissingEventProcessor()
+
+    constructor() {
+        super("MissingEventDetector")
+    }
+
+    protected func(message : any) : any {
+        this.missingEventProcessor.in(message)
+        return message
+    }
+}

src/mqttdispatcher.ts

@@ -84,8 +84,11 @@ export class MqttDispatcher implements IDispatcher {
                 this.mqttClient.subscribe(topicHandler.topic)
             }
         })
-        this.mqttClient.on('message', (topic: string, payload: Buffer): void => {
+        this.mqttClient.on('message', (topic: string, payload: Buffer, packet : Mqtt.IPublishPacket): void => {
             log.info(`message received, topic ${topic}, payload ${payload}`)
+            if (packet.retain) {
+                log.info("IS RETAINED")
+            }
             for (let topicHandler of this.topicHandlers) {
                 if (this.topicMatch(topicHandler.topic, topic)) {
                     log.info(`received topic ${topic} matches registered topic ${topicHandler.topic}`);