automatic reconnect to database

src/main.ts

@@ -1,5 +1,6 @@
 import * as Mqtt from 'mqtt'
 import * as Mongo from 'mongodb'
+import * as Events from 'events'
 import * as Queue from './queue'
 import * as MqttMessage from './mqtt_message'
 
@@ -22,51 +23,66 @@ options
   .parse(process.argv)
 
 
+class MqttMongo extends Events.EventEmitter {
 
-var dbHandle : Mongo.Db;
+  private options : any
-var dbReady : boolean = false;
 
-Mongo.MongoClient.connect(options['database'], 
+  constructor(options : any) {
-{
+    super()
-  'server': {
+    this.options = options
-    'reconnectTries': 5,
+    this.on('reconnectDatabase', this.connectToDatabase)
-     'reconnectInterval': 1000,
-     'socketOptions': {
-       'autoReconnect': false
   }
-   }  
+
-})
+  private dbHandle : Mongo.Db;
+  private dbReady : boolean = false;
+  private queue : Queue.Queue<MqttMessage.MqttMessage> = new Queue.Queue<MqttMessage.MqttMessage>()
+  private mqttClient : Mqtt.Client
+  private heartbeatTimer : NodeJS.Timer
+  private uptime : number = 0
+
+  connectToDatabase() {
+    console.info("About to connect to database")
+    Mongo.MongoClient.connect(this.options['database'], 
+    {
+//      'server': {
+//        'reconnectTries': 5,
+//        'reconnectInterval': 1000,
+//        'socketOptions': {
+//          'autoReconnect': false
+//        }
+//      }  
+    })
       .then(
         (tmpDbHandle: Mongo.Db) => { 
-      dbHandle = tmpDbHandle
+          this.dbHandle = tmpDbHandle
-      dbReady = true; 
+          this.dbReady = true; 
           console.info("Connected to database")
-      dbHandle.on('reconnectFailed', (err : any) => { console.warn(`Error on database ${err}`) })
+          this.dbHandle.on('reconnectFailed', (err : any) => { console.warn(`Error on database ${err}`) })
-      dbHandle.on('reconnect', () => { 
+          this.dbHandle.on('reconnect', () => { 
             console.info("Reconnect on database") 
-        dbReady = true
+            this.dbReady = true
-        queue.knock()
+            this.queue.knock()
           })
-      dbHandle.on('timeout', () => { console.warn("Timeout on database") })
+          this.dbHandle.on('timeout', () => { console.warn("Timeout on database") })
-      dbHandle.on('close', () => { 
+          this.dbHandle.on('close', () => { 
             console.info("Close on database") 
-        dbReady = false
+            this.dbReady = false
           })
         },
         (err: String) => {
-      console.error("Unable to connect to database: %s", err)
+          console.error(`Unable to connect to database: ${err}`)
-      process.exit(1)
+          this.dbReady = false
+        }
+    )
   }
-)
 
-
+  connectToBroker() {
-var queue = new Queue.Queue<MqttMessage.MqttMessage>()
+    this.queue.on('data', () => {
-queue.on('data', () => {
+      if (this.dbReady || true) { 
-  if (dbReady) { 
+        while (! this.queue.isEmpty()) {
-    while (! queue.isEmpty()) {
+          var msg : MqttMessage.MqttMessage = this.queue.deq()
-      var msg : MqttMessage.MqttMessage = queue.deq()
           console.info(`Something in the queue: ${JSON.stringify(msg)}`)
-      var coll = dbHandle.collection(options['collection'])
+          var coll = this.dbHandle.collection(options['collection'])
           coll.insertOne(msg.getMessage())
             .then(
               (r) => { 
@@ -74,9 +90,10 @@ queue.on('data', () => {
               },
               (err) => { 
                 console.error(`Error when trying to insert into database ${err}`)
-            if (! dbReady) {
+                if (! this.dbReady) {
                   console.info("Error occured while database connection is lost, re-enqueue msg.")
-              queue.reenq(msg)
+                  this.queue.reenq(msg)
+                  this.emit('reconnectDatabase')
                 } else {
                   console.error(`Message ${JSON.stringify(msg.getMessage())} is lost`)
                 }
@@ -86,48 +103,59 @@ queue.on('data', () => {
       } else {
         // console.info("Database currently not available, not reading from stream")
       }
-})
+    })
   
+    this.mqttClient = Mqtt.connect(options['broker'])
+    this.mqttClient.on('offline', () => { console.warn("MQTT client is offline") })
+    this.mqttClient.on('reconnect', () => { console.warn("MQTT client is reconnecting") })
+    this.mqttClient.on('close', () => { console.warn("MQTT connection closed") })
 
-var mqttClient = Mqtt.connect(options['broker'])
+    this.mqttClient.on('connect', () => {
-mqttClient.on('offline', () => { console.warn("MQTT client is offline") })
-mqttClient.on('reconnect', () => { console.warn("MQTT client is reconnecting") })
-mqttClient.on('close', () => { console.warn("MQTT connection closed") })
-
-mqttClient.on('connect', () => {
       console.info("Connected to MQTT broker")
-  mqttClient.subscribe(options['topic'])
+      this.mqttClient.subscribe(options['topic'])
-  mqttClient.subscribe('MqttMongo/Command')
+      this.mqttClient.subscribe('MqttMongo/Command')
-  mqttClient.publish('MqttMongo/Status', 'hello, started up')
+      this.mqttClient.publish('MqttMongo/Status', 'hello, started up')
-})
+    })
 
-var msgCnt : number = 0
+    var msgCnt : number = 0
-mqttClient.on('message', (topic : string, message : string) => {
+    this.mqttClient.on('message', (topic : string, message : string) => {
       msgCnt++;
       console.info(`Message received ${msgCnt}, topic ${topic}, payload ${message}`)
 
       if (topic == "MqttMongo/Command" && message == "shutdown") {
-    console.info("Shutting down MqttMongo")
+        this.shutdown()
-    clearInterval(uptimeInterval)
-    mqttClient.end()
-    dbHandle.close()
       } else {
         try {
           var mqttMessage = new MqttMessage.MqttMessage(topic, message)
-      queue.enq(mqttMessage)
+          this.queue.enq(mqttMessage)
         } catch (e) {
           console.error(`Error while parsing Mqtt message, topic '${topic}', message '${message}' , ${e.toString()}`)
         }
       }
-})
+    })
+  }
 
+  setupHeartbeat() {
+    this.heartbeatTimer = setInterval(() => {
+      this.uptime++
+      this.mqttClient.publish('MqttMongo/Status', `{'Uptime': ${this.uptime}}`)
+      if (! this.dbReady) {
+        this.emit("reconnectDatabase")
+      }
+    }, 1000)
+  }
 
-var uptime : number = 0
+  shutdown() {
-var uptimeInterval = setInterval(() => {
+    console.info("Shutting down MqttMongo")
-  uptime++
+    clearInterval(this.heartbeatTimer)
-  mqttClient.publish('MqttMongo/Status', `{'Uptime': ${uptime}}`)
+    this.mqttClient.end()
-}, 1000)
+    this.dbHandle.close()
-
+  }
+}
 
+var mqttMongo : MqttMongo = new MqttMongo(options)
+mqttMongo.connectToDatabase()
+mqttMongo.connectToBroker()
+mqttMongo.setupHeartbeat()
 
 console.info("MqttMongo started")