Merge branch 'supermucng'

.gitignore

+# eclipse files
+.cproject
+.project
+.settings
+
+#binaries
+**/*.so
+**/*.o
+**/*.dylib
+
+#log files
+*.log

Makefile

@@ -19,7 +19,8 @@ CXXFLAGS =	-std=c++11 -DBOOST_DATE_TIME_POSIX_TIME_STD_CONFIG -DBOOST_NETWORK_EN
 LIBS = -L../deps/mosquitto_build/lib -L$(DCDBDEPLOYPATH)/lib/ -ldl -lmosquitto -lboost_system -lboost_thread -lboost_log_setup -lboost_log -lboost_regex -lpthread -lcrypto -lssl -lcppnetlib-server-parsers -lcppnetlib-uri -rdynamic
 OBJS = src/dcdbpusher.o src/Configuration.o src/MQTTPusher.o src/HttpsServer.o
 
-PLUGINS = procfs pdu sysfs ipmi bacnet snmp tester
+PLUGINS = procfs pdu sysfs ipmi bacnet snmp gpfsmon tester
+ 
 ifeq ($(OS),Darwin)
 	BACNET_PORT = bsd
 	LIBEXT = dylib
@@ -69,9 +70,13 @@ install_lib: $(PLUGIN_LIBS)
 install_conf: $(foreach p,global $(PLUGINS),config/$(p).conf) 
 	install -m 644 $^ $(DCDBDEPLOYPATH)/etc/
 
-install: $(TARGET) install_lib install_conf
+install: $(TARGET) install_lib 
 	install $(TARGET) $(DCDBDEPLOYPATH)/bin/
-
+	@echo "Done with installation."
+	@echo "====================================="
+	@echo "To copy the configuration files type:"
+	@echo "				     > make install_conf"
+	
 src/Sensor.o: CXXFLAGS+= $(PLUGINFLAGS)
 src/SensorGroup.o: CXXFLAGS+= $(PLUGINFLAGS)
 src/sensors/%.o: CXXFLAGS+= $(PLUGINFLAGS) -I$(DCDBDEPSPATH)/bacnet-stack-$(BACNET-STACK_VERSION)/include -I$(DCDBDEPSPATH)/bacnet-stack-$(BACNET-STACK_VERSION)/ports/$(BACNET_PORT)
@@ -101,5 +106,8 @@ libdcdbplugin_procfs.$(LIBEXT): src/sensors/procfs/ProcfsSensorGroup.o src/senso
 libdcdbplugin_tester.$(LIBEXT): src/sensors/tester/TesterSensorGroup.o src/sensors/tester/TesterConfigurator.o
 	$(CXX) $(LIBFLAGS)$@ -o $@ $^ -L$(DCDBDEPLOYPATH)/lib/ -lboost_log -lboost_system
 	
+libdcdbplugin_gpfsmon.$(LIBEXT): src/sensors/gpfsmon/GpfsmonSensorGroup.o src/sensors/gpfsmon/GpfsmonConfigurator.o
+	$(CXX) $(LIBFLAGS)$@ -o $@ $^ -L$(DCDBDEPLOYPATH)/lib/ -lboost_log -lboost_system
+
 #libdcdbplugin_opa.$(LIBEXT): src/sensors/opa/OpaSensorGroup.o src/sensors/opa/OpaConfigurator.o
 #	$(CXX) $(LIBFLAGS)$@ -o $@ $^ -L$(DCDBDEPLOYPATH)/lib/ -lboost_log -lboost_system -lopamgt -libverbs -libumad -lssl

README.md

@@ -68,6 +68,7 @@ Please have a look at the provided `config/global.conf` example to get familiar
 | mqttprefix | To not rewrite a full MQTT-topic for every sensor one can specify here a consistent prefix.
 | sensorpattern | pattern used to perform automatic sensor name publishing. See the corresponding [section](#autopublish) for more information.
 | threads | Specify how many threads should be created to handle the sensors async. Default value of threads is 1. Note that the MQTTPusher always starts an extra thread. So the actual number of started threads is always one more than defined here. Specifying not enough threads can result in a delay for some sensors until they are read.
+| maxMsgNum | To avoid publishing too many MQTT messages at once you can define here a maximum count of values that are published in one turn. After reaching this limit the MQTTPusher will be forced to sleep for a short time before continuing.
 | verbosity | Level of detail in the logfile (dcdb.log). Set to a value between 5 (all log-messages, default) and 0 (only fatal messages). NOTE: level of verbosity for the command-line log can be set via the -v flag independently when invoking dcdbpusher.
 | daemonize | Set to 'true' if dcdbpusher should run detached as daemon. Default is false.
 | tempdir | One can specify a writeable directory where dcdbpusher can write its temporary and logging files to. Default is the current (' ./ ' ) directory.
@@ -190,6 +191,7 @@ All the different plugins share some same general principles in common regarding
         * __default__     (One can define the name of a template group (see below) whose values and sensors should be used as default)
     3. Sensors hold only those attributes which are necessary to uniquely identify the target sensor. Common base attributes:
         * __mqttsuffix__  (to make its [mqtt-topic](#mqttTopic) unique)
+        * __delta__ (identifies a monotonic sensor. If set to "on", differences between successive readings are collected)
 5. Be aware that naming of sensor/group/entity is not fixed. A plugin developer can name them as he likes, e.g. counter/multicounter/host.
 6. It is possible to define template groups or entities in the config file, but not template sensors (as a sensor should only consists of attributs which make him unique this would not be too useful). To specify a template group/entity simply prefix its definition with `template_` (see the example below). You can reference them later by using the `default` attribute. A template entity can consist of groups and these in turn can consist of sensors. When using a template, all of its attribute values are copied to the actual sensor. Copied attributes can be overwritten in the actual entity/sensor (some of them even should be overwritten, e.g. the mqttPart). However, groups/sensors associated with a template are copied to the actual entity/group and can NOT be overwritten. One can specify further groups/sensors which are then added to those copied from the template. Template entitys/groups itself or sensors within them are never used in live operation of the plugin. They are purely cosmetic for convenient configuration.
  
@@ -316,6 +318,9 @@ Explanation of the values specific for the perfevent plugin:
 | cpus | One can define a comma-separated list of cpu numbers (also value ranges can be specified, e.g. 2-4 equals 2,3,4). The hardware counter will then be only opened on the specified cpus.
 
 
+> NOTE     As perfevent counters are usually always monotonic, the delta attribute is by default set to true for all sensors. One has to explicitly set delta to "off" for a sensor to overwrite this behaviour.
+
+
 ### type and config <a name="perfTypeConfig"></a>
 
 (see the [perf_event_open man-page](http://man7.org/linux/man-pages/man2/perf_event_open.2.html) for more detailed explanations)
@@ -350,9 +355,10 @@ Explanation of the values specific for the perfevent plugin:
 | PERF_TYPE_HW_CACHE | | not yet implemented
 | | | |
 | PERF_TYPE_RAW | | user can define architecture-specific raw events here.
-| " | *XXXX* | Config must be an hex value (without 0x prefix). See <sup>[2](#fn2)</sup>
+| " | *XXXX* | Config must be a raw event config value, see <sup>[2](#fn2)</sup>
 | | | |
 | PERF_TYPE_BREAKPOINT | --- | config not required, any values will be ignored. However config must still be specified (even if empty)
+|<Custom>|<Custom>| dynamic PMU event, see <sup>[3](#fn3)</sup>
 
 #### Footnotes <a name="perfFootnotes"></a>
 
@@ -378,6 +384,7 @@ The existence of the perf_event_paranoid file is the official method for determi
 
 <a name="fn2">**2**</a>: &ensp; If type is *PERF_TYPE_RAW*, then a custom "raw" config value is needed. Most CPUs support events that are not covered by the "generalized" events. These are implementation defined; see your CPU manual (for example the Intel Volume 3B documentation or the AMD BIOS and Kernel Developer Guide). The libpfm4 library can be used to translate from the name in the architectural manual to the raw hex value perf_event_open() expects in this field.
 
+<a name="fn3">**3**</a>: &ensp; Custom type and Config values can be specified to use the PMU of a specific device. The necessary configuration parameters can be obtained from the type and config files the respective in /sys/devices/<device> tree.
 ## snmp <a name="snmp"></a>
 
 The SNMP plugin enables dcdbpusher to talk with devices which have an SNMP agent running and query requests from them. A SNMP sensor corresponds to a single value as identified by the unique OID. Sensors are aggregated by connections. See the exemplary snmp.conf file in the `config/` directory.
@@ -511,7 +518,7 @@ Explanation of the values specific for the ProcFS plugin:
 
 | Value | Explanation |
 |:----- |:----------- |
-| type | The type of the file parsed by the sensor group. Can be either "vmstat", "meminfo", "procstat" or "procstat_perc"
+| type | The type of the file parsed by the sensor group. Can be either "vmstat", "meminfo", "procstat" or "sar"
 | path | Path of the file, if different from the default path in the /proc filesystem
 | mqttPart | The mqttPart works similarly as in the Perf-event plugin. For sensors associated to metrics that are core-specific (e.g. some of those in /proc/stat) the mqttPart is replaced with the CPU id. For all other metrics that are system-wide, the mqttPart is used as it is.
 | mqttStart | Base MQTT suffix that is automatically incremented to generate topics for sensors associated to metrics in the same file. Note that this parameter is used only if automatic MQTT topic generation is enabled, when no sensors are explicitly defined.
@@ -537,6 +544,7 @@ The "type" field can be inferred for each sensor by simply checking the underlyi
 * col_guest_nice
 
 Additional CPU-related metrics (that may be introduced in future versions of the Linux kernel) are not supported by the DCDB ProcFS plugin.
+Note that for /proc/meminfo instances, an additional synthetic sensor of type "MemUsed" can be defined. This sensor will automatically extract the amount of used memory from the MemTotal and MemFree values present in meminfo files.
 
 ## Writing own plugins <a name="writingOwnPlugins"></a>
 First make sure you read the [plugins](#plugins) section.

config/global.conf

@@ -2,6 +2,7 @@ global {
 	mqttBroker localhost:1883
 	mqttprefix /00112233445566778899AABB
 	threads 24
+	maxMsgNum 100
 	verbosity 5
 	daemonize false
 	tempdir	.

config/gpfsmon.conf

+;comments in config files are indicated by a semicolon
+
+global {
+	mqttPrefix	/FF112233445566778899AABBFFFF
+	
+	;add here other global attributes for your plugin
+}
+
+group template_a1 {
+	interval    1000
+	minValues	10
+	sensor IOBYTESREAD {
+		metric		IOBYTESREAD
+		mqttsuffix	0000
+	}
+	
+	sensor IOBYTESWRITE {
+		metric		IOBYTESWRITE
+		mqttsuffix	0001
+	}
+	
+	sensor IOREADS {
+		metric		IOREADS
+		mqttsuffix	0002
+	}
+}
+
+group io01 {
+	default		a1
+	
+	sensor IOWRITES {
+		metric		IOWRITES
+		mqttsuffix	0003
+	}
+}
+
+

pluginGenerator/SensorGroup_cpp.sh

@@ -57,6 +57,14 @@ void ${PLUGIN_NAME}SensorGroup::stop() {
 	LOG(info) << "Sensorgroup " << _groupName << " stopped.";
 }
 
+void ${PLUGIN_NAME}SensorGroup::init(boost::asio::io_service& io) {
+	SensorGroupTemplate::init(io);
+	/**
+  	 * TODO Initialize here Sensor Group or remove this member function completely
+	 * The SensorGroupTemplate::init(io); will then be called
+	 */
+}
+
 void ${PLUGIN_NAME}SensorGroup::read() {
 	reading_t reading;
 	reading.timestamp = getTimestamp();
@@ -68,25 +76,21 @@ void ${PLUGIN_NAME}SensorGroup::read() {
 							 *  Read a value for every sensor affiliated with this group and store
 							 *  it with the appropriate sensor.
 							 */ 0;							 
-			s->storeReading(reading, _cacheIndex);
+			s->storeReading(reading);
 #ifdef DEBUG
-			LOG(debug) << _groupName << "::" << s->getName() << ": \"" << reading.value << "\"";
+			LOG(debug) << _groupName << "::" << s->getName() << " raw reading: \"" << reading.value << "\"";
 #endif
 		}
 	} catch (const std::exception& e) {
 		LOG(error) << "Sensorgroup" << _groupName << " could not read value: " << e.what();
-		return;
 	}
-	
-	_cacheIndex = (_cacheIndex + 1) % _cacheSize;
 }
 
 void ${PLUGIN_NAME}SensorGroup::readAsync() {
 	uint64_t now = getTimestamp();
 	read();
 	if (_timer && _keepRunning) {
-		uint64_t next = now + MS_TO_NS(_interval);
-		_timer->expires_at(timestamp2ptime(next));
+		_timer->expires_at(nextReadingTime());
 		_pendingTasks++;
 		_timer->async_wait(std::bind(&${PLUGIN_NAME}SensorGroup::readAsync, this));
 	}

pluginGenerator/config.sh

@@ -4,7 +4,7 @@ cat << EOF > ${PLUGIN_NAME_LWC}.conf
 ;comments in config files are indicated by a semicolon
 
 global {
-	mqttPrefix	/FF112233445566778899AABBFFFF
+	mqttPrefix	/FF112233445566778899AABB
 	
 	;add here other global attributes for your plugin
 }
@@ -36,7 +36,7 @@ group g1 {
 	mqttprefix	01
 	default		def1
 	
-	;sensor temp1 is took over from def1 and does not need to be redefined
+	;sensor temp1 is taken from def1 and does not need to be redefined
 	sensor gSens {
 		mqttsuffix	00
 	}

src/Configuration.cpp

@@ -29,6 +29,7 @@ Configuration::Configuration(const std::string& cfgFilePath) :
 	_global.brokerHost = "";
 	_global.brokerPort = 1883;
 	_global.threads	   = 1;
+	_global.maxMsgNum  = 100;
 	_global.logLevelFile = boost::log::trivial::trace;
 	_global.logLevelCmd  = boost::log::trivial::info;
 
@@ -94,6 +95,8 @@ bool Configuration::readGlobal() {
 				_global.qosLevel = 1;
 		} else if (boost::iequals(global.first, "threads")) {
 			_global.threads = stoi(global.second.data());
+		} else if (boost::iequals(global.first, "maxMsgNum")) {
+			_global.maxMsgNum = stoull(global.second.data());
 		} else if (boost::iequals(global.first, "daemonize")) {
 			if (global.second.data() == "true") {
 				_global.daemonize = 1;
@@ -278,8 +281,8 @@ bool Configuration::readPlugins() {
 					}
 
 					//check if an MQTT-suffix was assigned twice
-					for(auto g : dynLib.configurator->getSensorGroups()) {
-						for(auto s : g->getSensors()) {
+					for(const auto& g : dynLib.configurator->getSensorGroups()) {
+						for(const auto& s : g->getSensors()) {
 							bool ok = checkMqtt(s->getMqtt());
 							if(!ok) {
 								LOG(error) << "Problematic MQTT-Topics! Please check your config files";

src/Configuration.h

@@ -21,6 +21,7 @@
 		int qosLevel;
 		std::string brokerHost;
 		uint32_t threads;
+		unsigned int maxMsgNum;
 		boost::log::trivial::severity_level logLevelFile;
 		boost::log::trivial::severity_level logLevelCmd;
 		pluginSettings_t pluginSettings;

src/HttpsServer.cpp

@@ -203,7 +203,7 @@ void HttpsServer::requestHandler::operator()(server::request const &request, ser
 							time -= S_TO_NS(std::stoul(p.second));
 						}
 					}
-					
+
 					//process actual request
 					response = "Plugin not found!";
 					connection->set_status(server::connection::not_found);
@@ -211,10 +211,10 @@ void HttpsServer::requestHandler::operator()(server::request const &request, ser
 					for(auto& p : _httpsServer._plugins) {
 						if (p.id == pathStrs[0]) {
 							response = "Sensor not found!";
-							for(auto g : p.configurator->getSensorGroups()) {
-								for(auto s : g->getSensors()) {
+							for(const auto& g : p.configurator->getSensorGroups()) {
+								for(const auto& s : g->getSensors()) {
 									if (s->getName() == sensor) {
-										response = pathStrs[0] + "::" + sensor + _httpsServer.calcAvg(s, g->getCacheSize(), time);
+										response = pathStrs[0] + "::" + sensor + _httpsServer.calcAvg(*s, time);
 										connection->set_status(server::connection::ok);
 										break;
 									}
@@ -250,7 +250,7 @@ void HttpsServer::requestHandler::operator()(server::request const &request, ser
 		if (action == "start") {
 			for(auto& p : _httpsServer._plugins) {
 				if (p.id == pathStrs[0]) {
-					for(auto g : p.configurator->getSensorGroups()) {
+					for(const auto& g : p.configurator->getSensorGroups()) {
 						g->start();
 					}
 					response = "Plugin " + pathStrs[0] + ": Sensors started";
@@ -261,7 +261,7 @@ void HttpsServer::requestHandler::operator()(server::request const &request, ser
 		} else if (action == "stop") {
 			for(auto& p : _httpsServer._plugins) {
 				if (p.id == pathStrs[0]) {
-					for(auto g : p.configurator->getSensorGroups()) {
+					for(const auto& g : p.configurator->getSensorGroups()) {
 						g->stop();
 					}
 					response = "Plugin " + pathStrs[0] + ": Sensors stopped";
@@ -287,8 +287,8 @@ void HttpsServer::requestHandler::operator()(server::request const &request, ser
 						response = "Plugin " + pathStrs[0] + ": Could not reload configuration";
 						connection->set_status(server::connection::internal_server_error);
 					}
-					
-					for(auto g : p.configurator->getSensorGroups()) {
+
+					for(const auto& g : p.configurator->getSensorGroups()) {
 						g->init(_httpsServer._io);
 						g->start();
 					}
@@ -340,12 +340,12 @@ HttpsServer::~HttpsServer() {
 	delete _server;
 }
 
-std::string HttpsServer::calcAvg(SensorBase* const s, unsigned cacheSize, uint64_t time) {
-	uint64_t avg = 0;
-	const reading_t * const cache = s->getCache();
+std::string HttpsServer::calcAvg(SensorBase& s, uint64_t time) {
+	int64_t avg = 0;
+	const reading_t * const cache = s.getCache();
 
 	unsigned count = 0;
-	for(unsigned i = 0; i < cacheSize; i++) {
+	for(unsigned i = 0; i < s.getCacheSize(); i++) {
 		if (cache[i].timestamp > time) {
 			avg += cache[i].value;
 			count++;

src/HttpsServer.h

@@ -94,7 +94,7 @@ private:
 	 *
 	 * @return		Response message of the form " Average of last *count* values is *avg*"
 	 */
-	std::string calcAvg(SensorBase* const s, unsigned cacheSize, uint64_t time);
+	std::string calcAvg(SensorBase& s, uint64_t time);
 
 	/*
 	 * Check if the authkey is valid and has the permission requiredPerm associated

src/MQTTPusher.cpp

@@ -14,15 +14,16 @@
 
 #define LOGM(sev) LOG(sev) << "Mosquitto: "
 
-MQTTPusher::MQTTPusher(int brokerPort, const std::string& brokerHost, const std::string& mqttPrefix,
-        const std::string& sensorPattern, int qosLevel, pluginVector_t& plugins) :
-		_qosLevel(qosLevel),
-        _brokerPort(brokerPort),
+MQTTPusher::MQTTPusher(int brokerPort, const std::string& brokerHost, const std::string& sensorPattern, int qosLevel, pluginVector_t& plugins, unsigned int maxNumberOfMessages) :
+                _brokerPort(brokerPort),
 		_brokerHost(brokerHost),
 		_sensorPattern(sensorPattern),
-		_plugins(plugins),
+                _qosLevel(qosLevel),
+                _plugins(plugins),
+                _maxNumberOfMessages(maxNumberOfMessages),
 		_connected(false),
 		_keepRunning(true),
+		_overrideMsgCap(true),
 		_doHalt(false),
 		_halted(false) {
 
@@ -79,39 +80,45 @@ void MQTTPusher::push() {
 	//Performing sensor name auto-publish if necessary
 	sendMappings();
 
-    //collect sensor-data
-    reading_t *reads = new reading_t[1024];
-    std::size_t totalCount = 0;
-    while (_keepRunning || totalCount) {
-        if (_doHalt) {
-            _halted = true;
-            sleep(2);
-            continue;
-        }
-        _halted = false;
-        totalCount = 0;
-        for (auto &p : _plugins) {
-            if (_doHalt) {
-                //for faster response
-                break;
-            }
-            for (auto g : p.configurator->getSensorGroups()) {
-                for (auto s : g->getSensors()) {
-                    if (s->getSizeOfReadingQueue() >= g->getMinValues()) {
-                        sendReadings(s, reads, totalCount);
-                    }
-                }
-            }
-        }
-        int mosqErr;
-        if ((mosqErr = mosquitto_loop(_mosq, -1, 1)) != MOSQ_ERR_SUCCESS) {
-            LOGM(error) << "Error in mosquitto_loop: " << mosquitto_strerror(mosqErr);
-        }
-    }
+	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;
+		totalCount = 0;
+		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(_overrideMsgCap || totalCount < _maxNumberOfMessages){
+							sendReadings(*s, reads, totalCount);
+						} else {
+							break; //ultimately we will go to sleep 1 second
+						}
+					}
+				}
+			}
+		}
+		int mosqErr;
+		if ((mosqErr = mosquitto_loop(_mosq, -1, 1)) != MOSQ_ERR_SUCCESS) {
+			LOGM(error) << "Error in mosquitto_loop: " << mosquitto_strerror(mosqErr);
+		}
+		sleep(1);
+	}
 	mosquitto_disconnect(_mosq);
 }
 
-void MQTTPusher::sendReadings(SensorBase* s, reading_t* reads, std::size_t& totalCount) {
+void MQTTPusher::sendReadings(SensorBase& s, reading_t* reads, std::size_t& totalCount) {
 	//there was a (unintended) disconnect in the meantime --> reconnect
 	if (!_connected) {
 		LOGM(error) << "Lost connection. Reconnecting...";
@@ -127,10 +134,11 @@ void MQTTPusher::sendReadings(SensorBase* s, reading_t* reads, std::size_t& tota
 	if (_connected) {
 
 		//get all sensor values out of its queue
-		std::size_t count = s->popReadingQueue(reads, 1024);
-		totalCount+= count;
+		std::size_t count = s.popReadingQueue(reads, SensorBase::QUEUE_MAXLIMIT);
+		//totalCount+= count;
+		totalCount+= 1;
 #ifdef DEBUG
-		LOGM(debug) << "Sending " << count << " values from " << s->getName();
+		LOGM(debug) << "Sending " << count << " values from " << s.getName();
 #endif
 
 #if DEBUG
@@ -139,13 +147,13 @@ void MQTTPusher::sendReadings(SensorBase* s, reading_t* reads, std::size_t& tota
 		}
 #endif
 		//try to send them to the broker
-		if (mosquitto_publish(_mosq, NULL, (s->getMqtt()).c_str(), sizeof(reading_t)*count, reads, _qosLevel, false) != MOSQ_ERR_SUCCESS) {
-
+		if (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
 			LOGM(error) << "Could not send message! Trying again later";
 			_connected = false;
-			s->pushReadingQueue(reads, count);
-			totalCount -= count;
+			s.pushReadingQueue(reads, count);
+			//totalCount -= count;
+			totalCount -= 1;
 			sleep(5);
 		}
 	}
@@ -184,3 +192,14 @@ bool MQTTPusher::sendMappings() {
     LOGM(info) << "Sensor name auto-publish performed for all 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 += g->getSensors().size() * ( 1000 / g->getInterval() ) / g->getMinValues();
+	// The formula below assumes the pusher's sleep time is 1 sec; if not, change accordingly
+	_overrideMsgCap = _maxNumberOfMessages == 0 || msgRate > _maxNumberOfMessages;
+	if( _overrideMsgCap && _maxNumberOfMessages != 0 )
+		LOGM(error) << "Cannot enforce max rate of " << _maxNumberOfMessages << " msg/s lower than actual " << msgRate << " msg/s!";
+}

src/MQTTPusher.h

@@ -21,8 +21,7 @@
  */
 class MQTTPusher {
 public:
-	MQTTPusher(int brokerPort, const std::string& _brokerHost, const std::string& mqttPrefix,
-			const std::string& sensorPattern, int qosLevel, pluginVector_t& plugins);
+	MQTTPusher(int brokerPort, const std::string& brokerHost, const std::string& sensorPattern, int qosLevel, pluginVector_t& plugins, unsigned int maxNumberOfMessages);
 	virtual ~MQTTPusher();
 
 	void push();
@@ -40,6 +39,7 @@ public:
 	}
 
 	void cont() {
+		computeMsgRate();
 		_doHalt = false;
 	}
 
@@ -48,8 +48,9 @@ public:
 	}
 
 private:
-	void sendReadings(SensorBase* s, reading_t* reads, std::size_t& totalCount);
-    bool sendMappings();
+	void sendReadings(SensorBase& s, reading_t* reads, std::size_t& totalCount);
+	bool sendMappings();
+	void computeMsgRate();
 
 	int _qosLevel;
 	int _brokerPort;
@@ -59,8 +60,10 @@ private:
 	struct mosquitto* _mosq;
 	bool _connected;
 	bool _keepRunning;
+	bool _overrideMsgCap;
 	bool _doHalt;
 	bool _halted;
+	unsigned int _maxNumberOfMessages;
 
 	boost::log::sources::severity_logger<boost::log::trivial::severity_level> lg;
 };

src/dcdbpusher.cpp

@@ -63,7 +63,7 @@ void sigHandler(int sig) {
 	LOG(info) << "Stopping sensors...";
 	for(auto& p : _configuration->getPlugins()) {
 		LOG(info) << "Stop \"" << p.id << "\" plugin";
-		for(auto g : p.configurator->getSensorGroups()) {
+		for(const auto& g : p.configurator->getSensorGroups()) {
 			g->stop();
 		}
 	}
@@ -266,17 +266,11 @@ int main(int argc, char** argv) {
 	LOG(info) << "  Certificate, private key and DH-param file not printed.";
 #endif
 
-	//MQTTPusher and Https server get their own threads
-	_mqttPusher = new MQTTPusher(globalSettings.brokerPort, globalSettings.brokerHost, pluginSettings.mqttPrefix,
-			pluginSettings.sensorPattern, globalSettings.qosLevel, _configuration->getPlugins());
-	_httpsServer = new HttpsServer(restAPISettings, _configuration->getPlugins(), _mqttPusher, io);
-	_configuration->readAuthkeys(_httpsServer);
-
 	//Init all sensors
 	LOG(info) << "Init sensors...";
 	for(auto& p : _configuration->getPlugins()) {
 		LOG(info) << "Init \"" << p.id << "\" plugin";
-		for(auto g : p.configurator->getSensorGroups()) {
+		for(const auto& g : p.configurator->getSensorGroups()) {
 			LOG(debug) << " -Group: " << g->getGroupName();
 			g->init(io);
 		}
@@ -286,7 +280,7 @@ int main(int argc, char** argv) {
 	LOG(info) << "Starting sensors...";
 	for(auto& p : _configuration->getPlugins()) {
 		LOG(info) << "Start \"" << p.id << "\" plugin";
-		for(auto g : p.configurator->getSensorGroups()) {
+		for(const auto& g : p.configurator->getSensorGroups()) {
 			g->start();
 		}
 	}
@@ -317,7 +311,12 @@ int main(int argc, char** argv) {
 	for(size_t i = 0; i < globalSettings.threads; i++) {
 		threads.create_thread(bind(static_cast< size_t (boost::asio::io_service::*) () >(&boost::asio::io_service::run), &io));
 	}
-	
+
+	//MQTTPusher and Https server get their own threads
+	_mqttPusher = new MQTTPusher(globalSettings.brokerPort, globalSettings.brokerHost, pluginSettings.sensorPattern, globalSettings.qosLevel, _configuration->getPlugins(), globalSettings.maxMsgNum);
+	_httpsServer = new HttpsServer(restAPISettings, _configuration->getPlugins(), _mqttPusher, io);
+	_configuration->readAuthkeys(_httpsServer);
+
 	boost::thread mqttThread(bind(&MQTTPusher::push, _mqttPusher));
 	boost::thread restThread(bind(&HttpsServer::run, _httpsServer));
 

src/includes/ConfiguratorInterface.h

@@ -32,7 +32,7 @@ public:
 	virtual bool	readConfig(std::string cfgPath)	= 0;
 	virtual bool	reReadConfig() 					= 0;
 	virtual void	setGlobalSettings(const pluginSettings_t& pluginSettings) = 0;
-	virtual std::vector<SensorGroupInterface*>& getSensorGroups()	= 0;
+	virtual std::vector<SGroupPtr>& getSensorGroups()	= 0;
 
 protected:
 	boost::log::sources::severity_logger<boost::log::trivial::severity_level> lg;

src/includes/ConfiguratorTemplate.h

@@ -2,7 +2,7 @@
  * ConfiguratorTemplate.h
  *
  *  Created on: 13.01.2018
- *      Author: Micha Mueller
+ *      Author: Micha Mueller, Carla Guillen
  */
 
 #ifndef SRC_CONFIGURATORTEMPLATE_H_
@@ -24,6 +24,10 @@
 #include <sstream>
 #include <iomanip>
 
+#include <iostream>
+#include <sstream>
+#include <iomanip>
+
 //#define STRCMP(node,str) boost::iequals(node.first,str) //DEPRECATED
 #define CFG_VAL	boost::property_tree::iptree&
 
@@ -45,6 +49,14 @@ protected:
 	typedef std::map<std::string, SGroup*> sGroupMap_t;
 	typedef std::map<std::string, SEntity*> sEntityMap_t;
 
+	using SB_Ptr = std::shared_ptr<SBase>;
+	using SG_Ptr = std::shared_ptr<SGroup>;
+
+	const char COMMA = ',';
+	const char OPEN_SQBRKET = '[';
+	const char CLOSE_SQBRKET = ']';
+	const char DASH = '-';
+
 public:
 	ConfiguratorTemplate() :
 		_entityName("INVALID"),
@@ -57,9 +69,6 @@ public:
 	ConfiguratorTemplate(const ConfiguratorTemplate&) = delete;
 
 	virtual ~ConfiguratorTemplate() {
-		for (auto g : _sensorGroups) {
-			delete g;
-		}
 		for (auto e : _sensorEntitys) {
 			delete e;