Caliper plugin: store recv. data in the backend

dcdbpusher/includes/ConfiguratorInterface.h

@@ -135,7 +135,7 @@ public:
      * @details True if locked, false if failed to lock.
      */
     bool trylockSensors() {
-        return _mutex.lock();
+        return _mutex.try_lock();
     }
 
     /**
@@ -145,7 +145,7 @@ public:
      *          locked before.
      */
     void unlockSensors() {
-        return _mutex.lock();
+        _mutex.lock();
     }
 
     /**

dcdbpusher/sensors/caliper/CaliperConfigurator.cpp

@@ -40,7 +40,5 @@ void CaliperConfigurator::sensorBase(CaliperSensorBase& s, CFG_VAL config) {
 }
 
 void CaliperConfigurator::sensorGroup(CaliperSensorGroup& s, CFG_VAL config) {
-    ADD {
-		//no group attributes currently
-	}
+    s.setGlobalMqttPrefix(_mqttPrefix);
 }

dcdbpusher/sensors/caliper/CaliperSensorGroup.cpp

@@ -110,46 +110,63 @@ void CaliperSensorGroup::read() {
 
     const size_t bufSize = 2048;
     char buf[bufSize];
-    const ssize_t nrec = recv(_connection, (void *) buf, bufSize, MSG_DONTWAIT);
-    LOG(debug) << _groupName << ": Receiving message...";
 
-    if (nrec == 0) {
-        close(_connection);
-        _connection = -1;
-        LOG(debug) << _groupName << ": Connection closed";
-        return;
-    } else if (nrec == -1) {
-        if (errno != EAGAIN && errno != EWOULDBLOCK) {
-            LOG(error) << _groupName << ": Recv failed: " << strerror(errno);
+    //retrieve all messages currently available at the socket
+    while(true) {
+        const ssize_t nrec = recv(_connection, (void *) buf, bufSize, MSG_DONTWAIT);
+#ifdef DEBUG
+        LOG(debug) << _groupName << ": Receiving message...";
+#endif
+
+        //nrec==0 indicates that the connection was closed. Probably because Caliper terminated
+        if (nrec == 0) {
+            close(_connection);
+            _connection = -1;
+            LOG(debug) << _groupName << ": Connection closed";
+            return;
+        //nrec==-1 indicates an error during recv()
+        //if errno==EAGAIN or errno==EWOULDBLOCK there are currently no more messages available to receive
+        } else if (nrec == -1) {
+            if (errno != EAGAIN && errno != EWOULDBLOCK) {
+                LOG(error) << _groupName << ": Recv failed: " << strerror(errno);
+            }
+            return;
         }
-        return;
-    } else {
-        std::string timestamp(buf);
-        std::string function(&(buf[timestamp.length()+1]));
-        LOG(info) << _groupName << ": Received value " << function << " (" << timestamp << ")";
-    }
 
+        //actual message processing
+        std::string timestamp(buf);
+        std::string feName(&(buf[timestamp.length()+1])); //function OR event name
 
-/*    reading_t reading;
-	reading.timestamp = getTimestamp();
+        reading_t reading;
+        reading.value     = 1;
+        reading.timestamp = std::stoull(timestamp);
+#ifdef DEBUG
+        LOG(debug) << _groupName << ": Received value " << feName << " (" << timestamp << ")";
+#endif
 
-	try {
-		for(auto s : _sensors) {
-			reading.value =  *
-							 *  TODO
-							 *  Read a value for every sensor affiliated with this group and store
-							 *  it with the appropriate sensor.
-							 * 0;
-			s->storeReading(reading);
+        S_Ptr s;
+        auto it = _sensorIndex.find(feName);
+        if(it != _sensorIndex.end()) {
+            //we encountered this function or event name already
+            s = it->second;
+        } else {
+            //unknown function or event name --> create a new sensor
+            s = std::make_shared<CaliperSensorBase>(feName);
+            s->setMqtt(_globalMqttPrefix + _mqttPart + feName);
+            s->setName(s->getMqtt());
+            s->initSensor(_interval);
+
+            _sensors.push_back(s);
+            _baseSensors.push_back(s);
+            _sensorIndex.insert(std::make_pair(feName, s));
+        }
+        s->storeReading(reading);
 #ifdef DEBUG
-			LOG(debug) << _groupName << "::" << s->getName() << " raw reading: \"" << 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();
-	}*/
+    }
 }
 
 void CaliperSensorGroup::printGroupConfig(LOG_LEVEL ll) {
-
+    //nothing special to print
 }

dcdbpusher/sensors/caliper/CaliperSensorGroup.h

@@ -31,6 +31,8 @@
 
 #include "CaliperSensorBase.h"
 
+#include <unordered_map>
+
 /**
  * @brief SensorGroupTemplate specialization for this plugin.
  *
@@ -48,11 +50,16 @@ public:
 	 
 	void printGroupConfig(LOG_LEVEL ll) final override;
 
+	void setGlobalMqttPrefix(const std::string& prefix) { _globalMqttPrefix = prefix; }
+
 private:
-	void read() override;
+	void read() final override;
 
 	int _socket;
 	int _connection;
+	std::string _globalMqttPrefix;
+
+	std::unordered_map<std::string, S_Ptr> _sensorIndex; ///< Additional sensor storage for fast lookup
 };
 
 #endif /* CALIPER_CALIPERSENSORGROUP_H_ */