52 lines
1.4 KiB
JavaScript
52 lines
1.4 KiB
JavaScript
const mqtt = require('mqtt')
|
|
const { MongoClient } = require('mongodb')
|
|
const _ = require('lodash')
|
|
const dbClient = new MongoClient("mongodb://admin:03112546@mongo:27017/Smart_Garden", { useUnifiedTopology: true })
|
|
|
|
export default async function startMqttObserver() {
|
|
console.log("setup connections...")
|
|
// setup mongodb
|
|
dbClient.connect(err => {
|
|
if (err) {console.log("mongodb connection error")} else {console.log("mongodb connected.")}
|
|
const db = dbClient.db("Smart_Garden")
|
|
const sensorData = db.collection('sensorData');
|
|
console.log("collection retrieved.")
|
|
// setup mqtt client
|
|
let mqttClient = mqtt.connect('mqtt://mqtt.timovolkmann.de')
|
|
mqttClient.on('connect', onMqttConnect(mqttClient))
|
|
mqttClient.on('message', messageCallback(sensorData))
|
|
})
|
|
}
|
|
|
|
function messageCallback(collection) {
|
|
return function (topic, message) {
|
|
let topicElements = topic.split('/');
|
|
const type = topicElements.pop();
|
|
const id = topicElements.pop();
|
|
|
|
var date = new Date;
|
|
|
|
let doc = {
|
|
device_id: id,
|
|
timestamp: date,
|
|
}
|
|
try {
|
|
doc = _.merge(doc, JSON.parse(message));
|
|
} catch {
|
|
|
|
}
|
|
if (type === 'data') {
|
|
console.log(doc);
|
|
collection.insertOne(doc)
|
|
}
|
|
}
|
|
}
|
|
|
|
function onMqttConnect(mqttClient) {
|
|
return function () {
|
|
mqttClient.subscribe('smartgarden/updates/#', function (err) {
|
|
if (err) console.log(err);
|
|
console.log("Sucessfully subscribed...");
|
|
})
|
|
}
|
|
} |