MqttDispatcher/dist/mongosave.js
2017-08-28 12:44:10 +02:00

54 lines
2.1 KiB
JavaScript

"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
const CallChain = require("./callchain");
const log = require("./log");
const MongoDB = require("mongodb");
class MongoSave extends CallChain.AAsyncBaseChainItem {
constructor(url) {
super('MongoSave');
this.url = url;
this.mongoClient = new MongoDB.MongoClient();
this.connectPending = false;
}
func(message, finished) {
if (!this.dbh) {
log.info("MongoSave: Not database connection yet");
if (!this.connectPending) {
this.connectPending = true;
this.mongoClient.connect(this.url)
.then((db) => {
log.info("MongoSave: Successfully opened MongoDB connect");
this.dbh = db;
})
.catch((err) => {
log.error(`MongoSave: Failure when opening MongoDB connect: ${err}`);
this.dbh = undefined;
});
}
else {
log.info("MongoSave: Connecting to database is pending");
}
}
if (this.dbh) {
log.info("MongoSave: Database handle is available");
let coll = this.dbh.collection("iot");
coll.insertOne(message)
.then((res) => {
log.info(`MongoSave: Successfully wrote one item in database: ${res.insertedId}`);
let nextValue = { id: res.insertedId, payload: message };
finished(nextValue);
})
.catch((err) => {
log.error(`MongoSave: Failure when trying to write one item in database: ${err}`);
log.error("MongoSave: Chain interrupted");
});
}
else {
log.error(`MongoSave: No database connection yet, drop message ${message}`);
log.error("MongoSave: Chain interrupted");
}
log.info(`MongoSave: Returning from ${this.label}`);
}
}
exports.MongoSave = MongoSave;
//# sourceMappingURL=mongosave.js.map