Use signed int64_t for sensor readings (for consistency with Cassandra; partially reverts 0b53c240

CollectAgent/collectagent.cpp

@@ -90,7 +90,7 @@ struct httpHandler_t {
       boost::network::uri::query_map(uri, queries);
       int avg = atoi(queries.find("avg")->second.c_str());
 
-      uint64_t val = mySensorCache.getSensor(uri.path(), (uint64_t) avg);
+      int64_t val = mySensorCache.getSensor(uri.path(), (uint64_t) avg);
 
       data << val << "\n";
       //data << "Sid : " << sid.toString() << ", Value: " << val << "." << std::endl;
@@ -167,7 +167,7 @@ int mqttCallback(SimpleMQTTMessage *msg)
           //In the 64 bit message case, the collect agent provides a timestamp
           if (len == sizeof(uint64_t)) {
               payload = &buf;
-              payload->value = *((uint64_t *) msg->getPayload());
+              payload->value = *((int64_t *) msg->getPayload());
               payload->timestamp = Messaging::calculateTimestamp();
               len = sizeof(uint64_t) * 2;
           }

CollectAgent/sensorcache.cpp

@@ -27,7 +27,7 @@ const sensorCache_t& SensorCache::getSensorMap() {
     return sensorCache;
 }
 
-void SensorCache::storeSensor(SensorId sid, uint64_t ts, uint64_t val) {
+void SensorCache::storeSensor(SensorId sid, uint64_t ts, int64_t val) {
   sensorReading_t s = { val, ts };
   /* Remove the reserved bytes to leverage the standard find function */
   sid.setRsvd(0);
@@ -36,7 +36,7 @@ void SensorCache::storeSensor(SensorId sid, uint64_t ts, uint64_t val) {
   sensorCache[sid].store(s);
 }
 
-uint64_t SensorCache::getSensor(SensorId sid, uint64_t avg) {
+int64_t SensorCache::getSensor(SensorId sid, uint64_t avg) {
   /* Remove the reserved bytes to leverage the standard find function */
   sid.setRsvd(0);
   sensorCache_t::iterator it = sensorCache.find(sid);
@@ -59,7 +59,7 @@ uint64_t SensorCache::getSensor(SensorId sid, uint64_t avg) {
   }
 }
 
-uint64_t SensorCache::getSensor(std::string topic, uint64_t avg) {
+int64_t SensorCache::getSensor(std::string topic, uint64_t avg) {
   topic.erase(std::remove(topic.begin(), topic.end(), '/'), topic.end());
 
   size_t wp = topic.find("*");
@@ -189,7 +189,7 @@ bool CacheEntry::checkValid() {
     return true;
 }
 
-uint64_t CacheEntry::getAverage(uint64_t avg) {
+int64_t CacheEntry::getAverage(uint64_t avg) {
     TimeStamp ts;
 
     if (_cache.size() > 0) {

CollectAgent/sensorcache.h

@@ -16,7 +16,7 @@
 namespace DCDB {
 
 typedef struct {
-	uint64_t val;
+	int64_t val;
 	uint64_t timestamp;
 } sensorReading_t;
 
@@ -47,7 +47,7 @@ typedef struct {
 		*					The cache is considered valid if it is not outdated, that is, the latest reading is not
 		*					older than 5 times the average sampling rate.
 		*
-		* @return 			True if the cache is stil valid, False otherwise
+		* @return 			True if the cache is still valid, False otherwise
 		**/
 		bool checkValid();
 
@@ -60,7 +60,7 @@ typedef struct {
 		* @param avg    	length of the average aggregation window in nanoseconds.
 		* @return 			Average value of the last sensor readings.
 		**/
-		uint64_t getAverage(uint64_t avg);
+		int64_t getAverage(uint64_t avg);
 
 		/**
 		* @brief        	Searches for the input timestamp in the cache.
@@ -145,27 +145,33 @@ public:
     * @param val    The actual sensor reading.
 	* @return       Returns true if the topic string was valid and the data field of the object was populated.
 	**/
-	void storeSensor(SensorId sid, uint64_t ts, uint64_t val);
+	void storeSensor(SensorId sid, uint64_t ts, int64_t val);
 
 	/**
-    * @brief        Return a sensor reading from the SensorCache.
+    * @brief        Return a sensor reading or the average of the last readings
+    *               from the SensorCache.
     *
     * @param sid    The SensorId of the sensor to be looked up in the cache.
-    * @return       The sensor reading of the corresponding cache entry.
+    * @param avg    If avg > 0: denotes the length of the average aggregation window in nanoseconds.
+    * @return       If avg == 0 :The sensor reading of the corresponding cache entry.
+    *               If avg > 0 the average of the last readings is returned.
     * @throws       std::invalid_argument if the SensorId doesn't exist in the SensorCache.
     * @throws       std::out_of_range if the sid was found in the cache entry but is outdated.
     **/
-	uint64_t getSensor(SensorId sid, uint64_t avg=0);
+	int64_t getSensor(SensorId sid, uint64_t avg=0);
 
 	/**
-	* @brief        Return a sensor reading from the SensorCache.
+	* @brief        Return a sensor reading or the average of the last readings
+    *               from the SensorCache.
 	*
 	* @param topic  The topic of the sensor to be looked up in the cache. May contain wildcards.
-	* @return       The sensor reading of the corresponding cache entry.
+	* @param avg    If avg > 0: denotes the length of the average aggregation window in nanoseconds.
+	* @return       If avg == 0 :The sensor reading of the corresponding cache entry.
+    *               If avg > 0 the average of the last readings is returned.
 	* @throws       std::invalid_argument if the topic couldn't be found in the SensorCache.
 	* @throws       std::out_of_range if the topic was found in the cache entry but is outdated.
 	**/
-	uint64_t getSensor(std::string topic, uint64_t avg=0);
+	int64_t getSensor(std::string topic, uint64_t avg=0);
 
 	/**
     * @brief        Dump the contents of the SensorCache to stdout.