initial code

package.json

@@ -19,6 +19,7 @@
     "@types/moment": "^2.13.0",
     "@types/mqtt": "0.0.34",
     "@types/node": "^7.0.14",
+    "@types/request": "^2.47.1",
     "typescript": "^2.3.1"
   },
   "dependencies": {
@@ -26,6 +27,7 @@
     "command-line-args": "^5.0.2",
     "commander": "^2.15.1",
     "moment": "^2.22.1",
-    "mqtt": "^2.6.2"
+    "mqtt": "^2.6.2",
+    "request": "^2.87.0"
   }
 }

src/log.ts

@@ -0,0 +1,51 @@
+import * as moment from 'moment'
+
+
+enum Level {
+    All,
+    NoDebug,
+    NoDebugNoInfo,
+    NoDebugNoInfoNoWarning
+}
+
+var level = Level.NoDebug
+
+function timestamp(): string {
+    return moment().format('HH:mm:ss.SSS')
+}
+
+export function setLevel(value: string): void {
+    switch (value) {
+        case 'info':  level = Level.NoDebug; break
+        case 'warn':  level = Level.NoDebugNoInfo; break
+        case 'error': level = Level.NoDebugNoInfoNoWarning; break
+        default:      level = Level.All
+    }
+}
+
+
+export function info(message: string): void {
+    if (level < Level.NoDebugNoInfo) {
+        console.log(`${timestamp()} [ II ] ${message}`)
+    }
+}
+
+export function warn(message: string): void {
+    if (level < Level.NoDebugNoInfoNoWarning) {
+        console.log(`${timestamp()} [ WW ] ${message}`)
+    }
+}
+
+export function error(message: string): void {
+    console.log(`${timestamp()} [ EE ] ${message}`)
+}
+
+export function success(message: string): void {
+    console.log(`${timestamp()} [ OK ] ${message}`)
+}
+
+export function debug(message: string): void {
+    if (level < Level.NoDebug) {
+        console.log(`${timestamp()} [ DB ] ${message}`)
+    }
+}

src/main.ts

@@ -0,0 +1,66 @@
+import * as log from './log'
+import * as Mqtt from 'mqtt'
+import { post } from 'request'
+
+
+const BROKER : string = 'mqtt://localhost'
+const INFLUXDB : string = 'http://172.16.3.15:8086/write?db=smarthome'
+const VERBOSE : boolean = true
+
+
+
+const TOPIC_PARSERS : { [key : string] : (msg: string) => string } =     {
+    'IoT/espThermometer2/measurement': (msg : string) => {
+        let jsonData = JSON.parse(msg)
+        return `temperature,location=${jsonData.location} value=${jsonData.temperature}\nbattery,location=${jsonData.location},device=thermometer value=${jsonData.battery}`
+    }
+}
+
+
+
+
+let mqttClient = Mqtt.connect(BROKER)
+mqttClient.on('offline', () => { log.warn("MQTT client is offline") })
+mqttClient.on('reconnect', () => { log.warn("MQTT client is reconnecting") })
+mqttClient.on('close', () => { log.warn("MQTT connection closed") })
+
+mqttClient.on('connect', () => {
+    log.info("MQTT broker connected")
+    for (let topic in TOPIC_PARSERS) {
+        mqttClient.subscribe(topic)
+        log.info(`Subscribed to ${topic}`)
+    }
+})
+
+mqttClient.on('message', (topic : string, messageBuf : Buffer) => {
+    let message = messageBuf.toString('UTF-8')
+    if (VERBOSE) {
+        log.info(`Message received: topic ${topic}, payload ${message}`)
+    }
+    if (topic in TOPIC_PARSERS) {
+        try {
+            let lineProt = TOPIC_PARSERS[topic](message)
+            log.info(`parsed message: ${lineProt}`)
+
+            post({
+                url: INFLUXDB,
+                body: lineProt
+            }, (err, res, body) => {
+                if (err) {
+                    log.error(`Error when sending data to database: ${err}`)
+                } else {
+                    log.info(`Response from database: ${res}, ${body}`)
+                }
+            })
+
+        } catch (e) {
+            log.error(`Error ${e} when parsing ${topic}, ${message}`)
+        }
+    }
+})
+
+
+
+
+
+log.info("MqttToInflux started")