Merge pull request #1404 from fledge-iot/2.5.0RC

2.5.0RC

C/common/config_category.cpp

@@ -590,6 +590,8 @@ string ConfigCategory::getItemAttribute(const string& itemName,
 					return m_items[i]->m_listSize;
 				case ITEM_TYPE_ATTR:
 					return m_items[i]->m_listItemType;
+				case LIST_NAME_ATTR:
+				    return m_items[i]->m_listName;
 				default:
 					throw new ConfigItemAttributeNotFound();
 			}
@@ -664,6 +666,9 @@ bool ConfigCategory::setItemAttribute(const string& itemName,
 				case ITEM_TYPE_ATTR:
 					m_items[i]->m_listItemType = value;
 					return true;
+				case LIST_NAME_ATTR:
+					m_items[i]->m_listName = value;
+					return true;
 				default:
 					return false;
 			}
@@ -1321,6 +1326,17 @@ ConfigCategory::CategoryItem::CategoryItem(const string& name,
 			throw new runtime_error("ListSize configuration item property is not a string");
 		}
 	}
+	if (item.HasMember("listName"))
+	{
+		if (item["listName"].IsString())
+		{
+			m_listName = item["listName"].GetString();
+		}
+		else
+		{
+			throw new runtime_error("ListName configuration item property is not a string");
+		}
+	}
 
 	std::string m_typeUpperCase = m_type;
 	for (auto & c: m_typeUpperCase) c = toupper(c);
@@ -1607,6 +1623,7 @@ ConfigCategory::CategoryItem::CategoryItem(const CategoryItem& rhs)
 	m_bucketProperties = rhs.m_bucketProperties;
 	m_listSize = rhs.m_listSize;
 	m_listItemType = rhs.m_listItemType;
+	m_listName = rhs.m_listName;
 }
 
 /**
@@ -1727,6 +1744,10 @@ ostringstream convert;
 		{
 			convert << ", \"items\" : \"" << m_listItemType << "\"";
 		}
+		if (!m_listName.empty())
+		{
+			convert << ", \"listName\" : \"" << m_listName << "\"";
+		}
 	}
 	convert << " }";
 
@@ -1822,6 +1843,11 @@ ostringstream convert;
 	{
 		convert << ", \"items\" : \"" << m_listItemType << "\"";
 	}
+	if (!m_listName.empty())
+	{
+	    convert << ", \"listName\" : \"" << m_listName << "\"";
+	}
+
 
 	if (m_itemType == StringItem ||
 	    m_itemType == EnumerationItem ||

C/common/include/config_category.h

@@ -141,7 +141,8 @@ class ConfigCategory {
 					RULE_ATTR,
 					BUCKET_PROPERTIES_ATTR,
 					LIST_SIZE_ATTR,
-					ITEM_TYPE_ATTR
+					ITEM_TYPE_ATTR,
+					LIST_NAME_ATTR
 					};
 		std::string			getItemAttribute(const std::string& itemName,
 								 ItemAttribute itemAttribute) const;
@@ -192,6 +193,7 @@ class ConfigCategory {
 				std::string	m_bucketProperties;
 				std::string	m_listSize;
 				std::string	m_listItemType;
+				std::string	m_listName;
 		};
 		std::vector<CategoryItem *>	m_items;
 		std::string			m_name;

C/common/include/readingset_circularbuffer.h

@@ -0,0 +1,43 @@
+#ifndef _READINGSETCIRCULARBUFFER_H
+#define _READINGSETCIRCULARBUFFER_H
+/*
+ * Fledge ReadingSet Circular Buffer.
+ *
+ * Copyright (c) 2024 Dianomic Systems
+ *
+ * Released under the Apache 2.0 Licence
+ *
+ * Author: Devki Nandan Ghildiyal
+ */
+#include <reading_set.h>
+#include <mutex>
+#include <vector>
+#include <memory>
+
+/**
+ * Reading set circular buffer class
+ *
+ * Reading set circular buffer is a data structure to hold ReadingSet
+ * passed to a plugin.
+ */
+class ReadingSetCircularBuffer {
+	public:
+		ReadingSetCircularBuffer(unsigned long maxBufferSize=10);
+		~ReadingSetCircularBuffer();
+
+		void	insert(ReadingSet*);
+		void	insert(ReadingSet&);
+		std::vector<std::shared_ptr<ReadingSet>> extract(bool isExtractSingleElement=true);
+
+	private:
+		std::mutex	m_mutex;
+		unsigned long	m_maxBufferSize;
+		unsigned long	m_nextReadIndex;
+		void appendReadingSet(const std::vector<Reading *>& readings);
+		ReadingSetCircularBuffer (const ReadingSetCircularBuffer&) = delete;
+		ReadingSetCircularBuffer&	operator=(const ReadingSetCircularBuffer&) = delete;
+		std::vector<std::shared_ptr<ReadingSet>> m_circularBuffer;
+};
+
+#endif
+

C/common/include/storage_client.h

@@ -104,6 +104,7 @@ class StorageClient {
 								std::vector<std::string> keyValues, const std::string& operation, const std::string& callbackUrl);
 		void		registerManagement(ManagementClient *mgmnt) { m_management = mgmnt; };
 		bool 		createSchema(const std::string&);
+		bool		deleteHttpClient();
 
 	private:
 		void		handleUnexpectedResponse(const char *operation,

C/common/json_utils.cpp

@@ -96,25 +96,27 @@ string escaped = subject;
 std::string JSONunescape(const std::string& input)
 {
 	std::string output;
-	output.reserve(input.size());
+	size_t inputSize = input.size();
+	output.reserve(inputSize);
 
-	for (size_t i = 0; i < input.size(); ++i)
+	for (size_t i = 0; i < inputSize; ++i)
 	{
 		// skip leading or trailing "
-		if ((i == 0 || i == input.size() -1) && input[i] == '"')
+		if ((i == 0 || i == inputSize -1) && input[i] == '"')
 		{
 			continue;
 		}
 
-		// \\" -> \"
-		if (input[i] == '\\' && i + 2 < input.size() && input[i + 1] == '\\' && input[i + 2] == '"')
+		// \\\" -> \"
+		if (input[i] == '\\' && i + 3 < inputSize && input[i + 1] == '\\' && input[i + 2] == '\\' && input[i + 3] == '"')
 		{
 			output.push_back('\\');
 			output.push_back('"');
-			i += 2;
+			i += 3;
 		}
+		// \\" -> \"
 		// \" -> "
-		else if (input[i] == '\\' && i + 1 < input.size() && input[i + 1] == '"')
+		else if (input[i] == '\\' && i + 1 < inputSize && input[i + 1] == '"')
 		{
 			output.push_back('"');
 			++i;

C/common/readingset_circularbuffer.cpp

@@ -0,0 +1,133 @@
+/*
+ * Fledge ReadingSet Circular Buffer.
+ *
+ * Copyright (c) 2024 Dianomic Systems
+ *
+ * Released under the Apache 2.0 Licence
+ *
+ * Author: Devki Nandan Ghildiyal
+ */
+#include <readingset_circularbuffer.h>
+#include <logger.h>
+
+
+using namespace std;
+using namespace rapidjson;
+
+/**
+ * Construct an empty reading set circular buffer
+ *
+ * @param maxBufferSize Maximum size of the ReadingSet circular buffer. It should be atleast one.
+ */
+ReadingSetCircularBuffer::ReadingSetCircularBuffer(unsigned long maxBufferSize)
+{
+	if ( maxBufferSize <= 0)
+	{
+		maxBufferSize = 1;
+		Logger::getLogger()->warn("Minimum size of ReadingSetCircularBuffer cannot be less than one, setting buffer size to 1");
+	}
+	m_maxBufferSize = maxBufferSize;
+	m_nextReadIndex = 0;
+}
+
+/**
+ * Destructor for a result set
+ */
+ReadingSetCircularBuffer::~ReadingSetCircularBuffer()
+{
+	lock_guard<mutex> guard(m_mutex);
+	/* Delete the readings */
+	m_circularBuffer.clear();
+}
+
+/**
+ * Insert a ReadingSet into circular buffer
+ *
+ * @param readings		Reference for ReadingSet to be inserted into circular buffer
+ */
+void ReadingSetCircularBuffer::insert(ReadingSet& readings)
+{
+	appendReadingSet(readings.getAllReadings());
+}
+
+/**
+ * Insert a ReadingSet into circular buffer
+ *
+ * @param readings		Pointer for ReadingSet to be inserted into circular buffer
+ */
+void ReadingSetCircularBuffer::insert(ReadingSet* readings)
+{
+	appendReadingSet(readings->getAllReadings());
+}
+
+/**
+ * Internal implementation for inserting ReadingSet into the circular buffer
+ *
+ * @param readings		appends ReadingSet into the circular buffer
+ */
+void ReadingSetCircularBuffer::appendReadingSet(const std::vector<Reading *>& readings)
+{
+	lock_guard<mutex> guard(m_mutex);
+	bool isBufferFull = (m_circularBuffer.size() == m_maxBufferSize);
+
+	//Check if there is space available to insert a new ReadingSet
+	if (isBufferFull)
+	{
+		Logger::getLogger()->info("ReadingSetCircularBuffer buffer is full, removing first element");
+		// Make space for new ReadingSet and adjust m_nextReadIndex
+		m_circularBuffer.erase(m_circularBuffer.begin() + 0);
+		m_nextReadIndex--;
+	}	
+
+	std::vector<Reading *> *newReadings = new std::vector<Reading *>;
+
+	// Iterate over all the readings in ReadingSet
+	for (auto const &reading : readings)
+	{
+		newReadings->emplace_back(new Reading(*reading));
+	}
+	// Insert ReadingSet into buffer
+	m_circularBuffer.push_back(std::make_shared<ReadingSet>(newReadings));
+
+}
+
+/**
+ * Fetch the vector of ReadingSet from circular buffer
+ *
+ * @param isExtractSingleElement True to extract single ReadingSet otherwise for extract entire buffer 
+ * @return		Return a vector of shared pointer to ReadingSet
+ */
+std::vector<std::shared_ptr<ReadingSet>> ReadingSetCircularBuffer::extract(bool isExtractSingleElement)
+{
+
+    lock_guard<mutex> guard(m_mutex);
+    bool isNoDataToRead = m_circularBuffer.empty() || (m_nextReadIndex == m_circularBuffer.size());
+    std::vector<std::shared_ptr<ReadingSet>> bufferedItem;
+    // Check for empty buffer
+    if (isNoDataToRead)
+    {
+	Logger::getLogger()->info("There is no more data to read in ReadingSet circualr buffer");
+	return  bufferedItem;
+    }
+
+    // Return single item from buffer
+    if (isExtractSingleElement)
+    {
+        bufferedItem.emplace_back(m_circularBuffer[m_nextReadIndex]);
+        m_nextReadIndex++;
+        return  bufferedItem;
+    }
+
+    // Return Entire buffer
+    if(m_nextReadIndex == 0)
+    {
+        m_nextReadIndex = m_circularBuffer.size();
+        return m_circularBuffer;
+    }
+    // Send remaining items in the buffer
+    for (int i = m_nextReadIndex; i <  m_circularBuffer.size(); i ++)
+        bufferedItem.emplace_back(m_circularBuffer[i]);
+
+    m_nextReadIndex =  m_circularBuffer.size();
+	return bufferedItem;
+}

C/common/result_set.cpp

@@ -55,7 +55,7 @@ ResultSet::ResultSet(const std::string& json)
 					{
 						type = NUMBER_COLUMN;
 					}
-					else if (itr->value.IsNumber())
+          				else if (itr->value.IsNumber())
 					{
 						type = INT_COLUMN;
 					}

C/common/storage_client.cpp

@@ -83,6 +83,30 @@ StorageClient::~StorageClient()
 	}
 }
 
+
+/**
+ * Delete HttpClient object for current thread
+ */
+bool StorageClient::deleteHttpClient()
+{
+	std::thread::id thread_id = std::this_thread::get_id();
+
+	lock_guard<mutex> guard(sto_mtx_client_map);
+
+	if(m_client_map.find(thread_id) == m_client_map.end())
+		return false;
+
+	ostringstream ss;
+	ss << thread_id;
+	Logger::getLogger()->debug("Storage client deleting HttpClient object @ %p for thread %s", m_client_map[thread_id], ss.str().c_str());
+
+	delete m_client_map[thread_id];
+	m_client_map.erase(thread_id);
+
+	return true;
+}
+
+
 /**
  * Creates a HttpClient object for each thread
  * it stores/retrieves the reference to the HttpClient and the associated thread id in a map
@@ -316,7 +340,7 @@ ReadingSet *StorageClient::readingFetch(const unsigned long readingId, const uns
 	try {
 
 		char url[256];
-		snprintf(url, sizeof(url), "/storage/reading?id=%ld&count=%ld",
+		snprintf(url, sizeof(url), "/storage/reading?id=%lu&count=%lu",
 				readingId, count);
 
 		auto res = this->getHttpClient()->request("GET", url);