MQTTPusher.cpp

/*
 * MQTTPusher.cpp
 *
 *  Created on: 13.12.2017
 *      Author: Michael Ott (original), Micha Mueller
 */

#include "MQTTPusher.h"
#include <iostream>
#include <string>
#include <unistd.h>

#define LOGM(sev) LOG(sev) << "Mosquitto: "

extern volatile int keepRunning;

MQTTPusher::MQTTPusher(int brokerPort, const std::string& brokerHost,
		const std::string& mqttPrefix, sensorVector_t& sensors) :
		_brokerPort(brokerPort), _brokerHost(brokerHost),
		_sensors(sensors),_connected(false) {

	//first print some info
	int mosqMajor, mosqMinor, mosqRevision;
	mosquitto_lib_version(&mosqMajor, &mosqMinor, &mosqRevision);
	LOGM(info) << mosqMajor << "." << mosqMinor << "." << mosqRevision;
	char hostname[256];
	if (gethostname(hostname, 255) != 0) {
		LOG(fatal) << "Cannot get hostname";
		exit(EXIT_FAILURE);
	}
	hostname[255] = '\0';
	LOG(info) << "Hostname: " << hostname;
	//enough information

	//init mosquitto-struct
	mosquitto_lib_init();
	std::string clientID(mqttPrefix);
	_mosq = mosquitto_new(mqttPrefix.c_str(), false, NULL);
	if (!_mosq) {
		perror(NULL);
		exit(EXIT_FAILURE);
	}
}

MQTTPusher::~MQTTPusher() {
	if(_connected) {
		mosquitto_disconnect(_mosq);
	}
}

void MQTTPusher::push() {

	//connect to broker (if necessary)
	while (keepRunning && !_connected) {
		if (mosquitto_connect(_mosq, _brokerHost.c_str(), _brokerPort, 1000) != MOSQ_ERR_SUCCESS) {
			LOGM(error) << "Could not connect to MQTT broker " << _brokerHost << ":" << _brokerPort;
			sleep(1);
		} else {
			_connected = true;
			LOGM(info) << "Connection established!";
		}
	}

	if (mosquitto_loop_start(_mosq) != MOSQ_ERR_SUCCESS) {
		LOGM(fatal) << "Setup failed";
		keepRunning = 0;
		mosquitto_disconnect(_mosq);
		return;
	}

	//collect sensor-data
	reading_t* reads = new reading_t[1024];
	std::size_t totalCount = 0;
	while (keepRunning || totalCount) {
		totalCount = 0;
		for(auto s : _sensors) {
			if (s->getSizeOfReadingQueue() >= s->getMinValues()) {

				//there was a (unintended) disconnect in the meantime --> reconnect
				if (!_connected) {
					LOGM(error) << "Lost connection. Reconnecting...";
					if (mosquitto_reconnect(_mosq) != MOSQ_ERR_SUCCESS) {
						LOGM(error) << "Could not reconnect to MQTT broker " << _brokerHost << ":" << _brokerPort << std::endl;
						sleep(5);
					} else {
						_connected = true;
						LOGM(info) << "Connection established!";
					}
				}

				if (_connected) {

					//get all sensor values out of its queue
					std::size_t count = s->popReadingQueue(reads, 1024);
					totalCount+= count;
#ifdef DEBUG
					LOGM(debug) << "Sending " << count << " values from " << s->getName();
#endif

#if DEBUG
					for (std::size_t i=0; i<count; i++) {
						LOG(debug) << "  " << reads[i].timestamp << " " << reads[i].value;
					}
#endif
					//try to send them to the broker
					if (mosquitto_publish(_mosq, NULL, (s->getMqtt()).c_str(), sizeof(reading_t)*count, reads, 1, false) != MOSQ_ERR_SUCCESS) {

						//could not send them --> push the sensor values back into the queue
						LOGM(error) << "Could not send message! Trying again later";
						_connected = false;
						s->pushReadingQueue(reads, count);
						totalCount -= count;
						sleep(5);
						break;
					}
				}
			}
		}
		sleep(1);
	}
	mosquitto_disconnect(_mosq);
	mosquitto_loop_stop(_mosq, false);
}