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>
#include "timestamp.h"

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

MQTTPusher::MQTTPusher(int brokerPort, const std::string& brokerHost, const std::string& sensorPattern, int qosLevel,
					   pluginVector_t& plugins, an_pluginVector_t& aPlugins, int maxNumberOfMessages, unsigned int maxInflightMsgNum, unsigned int maxQueuedMsgNum) :
		_qosLevel(qosLevel),
		_brokerPort(brokerPort),
		_brokerHost(brokerHost),
		_sensorPattern(sensorPattern),
		_plugins(plugins),
		_analyticsPlugins(aPlugins),
		_connected(false),
		_keepRunning(true),
		_msgCap(DISABLED),
		_doHalt(false),
		_halted(false),
		_maxNumberOfMessages(maxNumberOfMessages),
		_maxInflightMsgNum(maxInflightMsgNum),
		_maxQueuedMsgNum(maxQueuedMsgNum) {

	//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();
	_mosq = mosquitto_new(hostname, false, NULL);
	if (!_mosq) {
		perror(NULL);
		exit(EXIT_FAILURE);
	}

	mosquitto_threaded_set(_mosq, true);
	mosquitto_max_inflight_messages_set(_mosq, _maxInflightMsgNum);
	mosquitto_max_queued_messages_set(_mosq, _maxQueuedMsgNum);
}

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

void MQTTPusher::push() {
    int mosqErr;
    uint64_t idleTime = 0;
	//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!";
		}
	}

	//Performing auto-publish if necessary
	sendMappings();

	computeMsgRate();
	//collect sensor-data
	reading_t* reads = new reading_t[SensorBase::QUEUE_MAXLIMIT];
	std::size_t totalCount = 0; //number of messages
	while (_keepRunning || totalCount) {
		if (_doHalt) {
			_halted = true;
			sleep(2);
			continue;
		}
		_halted = false;
		
		//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) {
            if(getTimestamp() - idleTime >= PUSHER_IDLETIME) {
                idleTime = getTimestamp();
                totalCount = 0;
                // Push sensor data
                for (auto &p : _plugins) {
                    if (_doHalt) {
                        //for faster response
                        break;
                    }
                    for (const auto &g : p.configurator->getSensorGroups()) {
                        for (const auto &s : g->getSensors()) {
                            if (s->getSizeOfReadingQueue() >= g->getMinValues()) {
                                if (_msgCap == DISABLED || totalCount < (unsigned)_maxNumberOfMessages) {
                                    if (sendReadings(*s, reads, totalCount) > 0) {
                                        break;
                                    }
                                } else {
                                    break; //ultimately we will go to sleep 1 second
                                }
                            }
                        }
                    }
                }
                // Push output analytics sensors
                for (auto &p : _analyticsPlugins) {
                    if (_doHalt) {
                        break;
                    }
                    for (const auto &a : p.configurator->getAnalyzers()) {
                    	if(a->getStreaming()) {
							for (const auto &u : a->getUnits()) {
								for (const auto &s : u->getBaseOutputs()) {
									if (s->getSizeOfReadingQueue() >= a->getMinValues()) {
										if (_msgCap == DISABLED || totalCount < (unsigned) _maxNumberOfMessages) {
											if (sendReadings(*s, reads, totalCount) > 0) {
												break;
											}
										} else {
											break;
										}
									}
								}
							}
						}
                    }
                }
            }

			if ((mosqErr = mosquitto_loop(_mosq, -1, 1)) != MOSQ_ERR_SUCCESS) {
				if (mosqErr == MOSQ_ERR_CONN_LOST) {
					LOGM(info) << "Disconnected.";
					_connected = false;
				} else {
					LOGM(error) << "Error in mosquitto_loop: " << mosquitto_strerror(mosqErr);
				}
			}
		}
	}
	mosquitto_disconnect(_mosq);
}

int MQTTPusher::sendReadings(SensorBase& s, reading_t* reads, std::size_t& totalCount) {
	//get all sensor values out of its queue
	std::size_t count = s.popReadingQueue(reads, SensorBase::QUEUE_MAXLIMIT);
	//totalCount+= count;
	totalCount+= 1;
#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
	int rc;
	if ((rc = mosquitto_publish(_mosq, NULL, (s.getMqtt()).c_str(), sizeof(reading_t)*count, reads, _qosLevel, false)) != MOSQ_ERR_SUCCESS) {
		//could not send them --> push the sensor values back into the queue
		if (rc == MOSQ_ERR_NOMEM) {
			LOGM(info) << "Can\'t queue additional messages";
		} else {
			LOGM(error) << "Could not send message! Trying again later";
			_connected = false;
		}
		s.pushReadingQueue(reads, count);
		//totalCount -= count;
		totalCount -= 1;
		return 1;
	}
	return 0;
}

bool MQTTPusher::sendMappings() {
	if(_sensorPattern == "")
		return false;

	std::string topic, name;
	unsigned int publishCtr=0;
	// Performing auto-publish for sensors
	for(auto& p: _plugins)
		for(auto& g: p.configurator->getSensorGroups())
			for(auto& s: g->getSensors()) {
				topic = std::string(DCDB_MAP) + s->getMqtt();
				name  = s->getName();

				// Try to send mapping to the broker
				if (mosquitto_publish(_mosq, NULL, topic.c_str(), name.length(), name.c_str(), _qosLevel, false) != MOSQ_ERR_SUCCESS) {
					LOGM(error) << "Broker not reachable! Only " << publishCtr << " sensors were published.";
					_connected = false;
					return true;
				}
				else
					publishCtr++;
			}

	// Performing auto-publish for analytics output sensors
	for(auto& p: _analyticsPlugins)
		for(auto& a: p.configurator->getAnalyzers())
			if(a->getStreaming())
				for(auto& u: a->getUnits())
					for(auto& s: u->getBaseOutputs()) {
						topic = std::string(DCDB_MAP) + s->getMqtt();
						name  = s->getName();
	
						// Try to send mapping to the broker
						if (mosquitto_publish(_mosq, NULL, topic.c_str(), name.length(), name.c_str(), _qosLevel, false) != MOSQ_ERR_SUCCESS) {
							LOGM(error) << "Broker not reachable! Only " << publishCtr << " sensors were published.";
							_connected = false;
							return true;
						}
						else
							publishCtr++;
					}
	LOGM(info) << "Sensor name auto-publish performed for all " << publishCtr << " sensors!";
	return true;
}

void MQTTPusher::computeMsgRate() {
	// Computing number of sent MQTT messages per second
	float msgRate = 0;
	for(auto& p : _plugins)
		for(const auto& g : p.configurator->getSensorGroups())
			msgRate += (float)g->getSensors().size() * ( 1000.0f / (float)g->getInterval() ) / (float)g->getMinValues();
	for(auto& p : _analyticsPlugins)
		for(const auto& a : p.configurator->getAnalyzers())
			for(const auto& u : a->getUnits())
				msgRate += (float)u->getBaseOutputs().size() * ( 1000.0f / (float)a->getInterval() ) / (float)a->getMinValues();
	// The formula below assumes the pusher's sleep time is 1 sec; if not, change accordingly
	if(_maxNumberOfMessages >= 0 && _msgCap != MINIMUM) {
		_msgCap = _maxNumberOfMessages == 0 || msgRate > _maxNumberOfMessages ? DISABLED : ENABLED;
		if (_msgCap == DISABLED && _maxNumberOfMessages > 0)
			LOGM(warning) << "Cannot enforce max rate of " << _maxNumberOfMessages << " msg/s lower than actual " << msgRate << " msg/s!";
		else if(_maxNumberOfMessages > 0)
			LOGM(info) << "Enforcing message cap of " << _maxNumberOfMessages << " msg/s against actual " << msgRate << " msg/s.";
		else
			LOGM(info) << "No message cap enforced. Predicted message rate is " << msgRate << " msg/s.";
	} else {
		_msgCap = MINIMUM;
		_maxNumberOfMessages = msgRate + 10;
		LOGM(info) << "Enforcing message cap of " << _maxNumberOfMessages << " msg/s against actual " << msgRate << " msg/s.";
	}
}