https://telecominfraproject.atlassian.net/browse/WIFI-7831

Signed-off-by: stephb9959 <[email protected]>
Telecominfraproject · Sep 13, 2023 · caf1ec9 · caf1ec9
1 parent 6b49c68
commit caf1ec9
Show file tree

Hide file tree

Showing 21 changed files with 237 additions and 80 deletions.
diff --git a/build b/build
@@ -1 +1 @@
-18
+19
diff --git a/src/AP_WS_Connection.cpp b/src/AP_WS_Connection.cpp
@@ -231,10 +231,7 @@ namespace OpenWifi {
 			Details.set(uCentralProtocol::TIMESTAMP, Utils::Now());
 			Details.set(uCentralProtocol::UUID,uuid);
 			Disconnect.set(uCentralProtocol::DISCONNECTION, Details);
-			Poco::JSON::Stringifier Stringify;
-			std::ostringstream OS;
-			Stringify.condense(Disconnect, OS);
-			KafkaManager()->PostMessage(KafkaTopics::CONNECTION, SerialNumber, std::make_shared<std::string>(OS.str()));
+			KafkaManager()->PostMessage(KafkaTopics::CONNECTION, SerialNumber, Disconnect);
 		} catch (...) {
 		}
 	}
@@ -725,10 +722,7 @@ namespace OpenWifi {
 					PingDetails.set(uCentralProtocol::UUID, uuid_);
 					PingDetails.set("locale", State_.locale);
 					PingObject.set(uCentralProtocol::PING, PingDetails);
-					Poco::JSON::Stringifier Stringify;
-					std::ostringstream OS;
-					Stringify.condense(PingObject, OS);
-					KafkaManager()->PostMessage(KafkaTopics::CONNECTION, SerialNumber_, std::make_shared<std::string>(OS.str()));
+					KafkaManager()->PostMessage(KafkaTopics::CONNECTION, SerialNumber_,PingObject);
 				}
 				return;
 			} break;

diff --git a/src/AP_WS_Process_alarm.cpp b/src/AP_WS_Process_alarm.cpp
@@ -25,7 +25,7 @@ namespace OpenWifi {
 				Poco::JSON::Stringifier Stringify;
 				std::ostringstream OS;
 				Stringify.condense(ParamsObj, OS);
-				KafkaManager()->PostMessage(KafkaTopics::ALERTS, SerialNumber_, std::make_shared<std::string>(OS.str()));
+				KafkaManager()->PostMessage(KafkaTopics::ALERTS, SerialNumber_, OS.str());
 			}
 		}
 	}

diff --git a/src/AP_WS_Process_connect.cpp b/src/AP_WS_Process_connect.cpp
@@ -32,9 +32,7 @@ namespace OpenWifi {
 			Event.set("type", "device.firmware_change");
 			Event.set("timestamp", Utils::Now());
 			Event.set("payload", EventDetails);
-			std::ostringstream OS;
-			Event.stringify(OS);
-			KafkaManager()->PostMessage(KafkaTopics::DEVICE_EVENT_QUEUE, SerialNumber, std::make_shared<std::string>(OS.str()));
+			KafkaManager()->PostMessage(KafkaTopics::DEVICE_EVENT_QUEUE, SerialNumber, Event);
 		}
 	}
 
@@ -51,9 +49,7 @@ namespace OpenWifi {
 			Event.set("type", "device.not_provisioned");
 			Event.set("timestamp", Utils::Now());
 			Event.set("payload", EventDetails);
-			std::ostringstream OS;
-			Event.stringify(OS);
-			KafkaManager()->PostMessage(KafkaTopics::DEVICE_EVENT_QUEUE, SerialNumber, std::make_shared<std::string>(OS.str()));
+			KafkaManager()->PostMessage(KafkaTopics::DEVICE_EVENT_QUEUE, SerialNumber, Event);
 		}
 	}
 
@@ -285,15 +281,11 @@ namespace OpenWifi {
 			GWWebSocketNotifications::DeviceConnected(Notification);
 
 			if (KafkaManager()->Enabled()) {
-				Poco::JSON::Stringifier Stringify;
-
 				ParamsObj->set(uCentralProtocol::CONNECTIONIP, CId_);
 				ParamsObj->set("locale", State_.locale);
 				ParamsObj->set(uCentralProtocol::TIMESTAMP, Utils::Now());
 				ParamsObj->set(uCentralProtocol::UUID, uuid_);
-				std::ostringstream OS;
-				Stringify.condense(ParamsObj, OS);
-				KafkaManager()->PostMessage(KafkaTopics::CONNECTION, SerialNumber_, std::make_shared<std::string>(OS.str()));
+				KafkaManager()->PostMessage(KafkaTopics::CONNECTION, SerialNumber_, *ParamsObj);
 			}
 		} else {
 			poco_warning(

diff --git a/src/AP_WS_Process_event.cpp b/src/AP_WS_Process_event.cpp
@@ -34,11 +34,8 @@ namespace OpenWifi {
 					FullEvent.set("type", EventType);
 					FullEvent.set("timestamp", EventTimeStamp);
 					FullEvent.set("payload", EventPayload);
-
-					std::ostringstream OS;
-					FullEvent.stringify(OS);
 					KafkaManager()->PostMessage(KafkaTopics::DEVICE_EVENT_QUEUE, SerialNumber_,
-												std::make_shared<std::string>(OS.str()));
+												FullEvent);
 				}
 			}
 		} catch (const Poco::Exception &E) {

diff --git a/src/AP_WS_Process_healthcheck.cpp b/src/AP_WS_Process_healthcheck.cpp
@@ -60,11 +60,7 @@ namespace OpenWifi {
 
 			SetLastHealthCheck(Check);
 			if (KafkaManager()->Enabled()) {
-				Poco::JSON::Stringifier Stringify;
-				std::ostringstream OS;
-				ParamsObj->set("timestamp", Utils::Now());
-				Stringify.condense(ParamsObj, OS);
-				KafkaManager()->PostMessage(KafkaTopics::HEALTHCHECK, SerialNumber_, std::make_shared<std::string>(OS.str()));
+				KafkaManager()->PostMessage(KafkaTopics::HEALTHCHECK, SerialNumber_, *ParamsObj);
 			}
 		} else {
 			poco_warning(Logger_, fmt::format("HEALTHCHECK({}): Missing parameter", CId_));

diff --git a/src/AP_WS_Process_state.cpp b/src/AP_WS_Process_state.cpp
@@ -56,10 +56,7 @@ namespace OpenWifi {
 											State_.Associations_5G, State_.Associations_6G);
 
 			if (KafkaManager()->Enabled()) {
-				Poco::JSON::Stringifier Stringify;
-				std::ostringstream OS;
-				Stringify.condense(ParamsObj, OS);
-				KafkaManager()->PostMessage(KafkaTopics::STATE, SerialNumber_, std::make_shared<std::string>(OS.str()));
+				KafkaManager()->PostMessage(KafkaTopics::STATE, SerialNumber_, *ParamsObj);
 			}
 
 			GWWebSocketNotifications::SingleDevice_t N;

diff --git a/src/AP_WS_Process_telemetry.cpp b/src/AP_WS_Process_telemetry.cpp
@@ -27,7 +27,7 @@ namespace OpenWifi {
 				std::ostringstream SS;
 				Payload->stringify(SS);
 				auto now = Utils::Now();
-				auto KafkaPayload = std::make_shared<std::string>(SS.str());
+				auto KafkaPayload = SS.str();
 				if (ParamsObj->has("adhoc")) {
 					KafkaManager()->PostMessage(KafkaTopics::DEVICE_TELEMETRY, SerialNumber_,
 												KafkaPayload);
@@ -39,7 +39,7 @@ namespace OpenWifi {
 						// std::endl;
 						TelemetryWebSocketPackets_++;
 						State_.websocketPackets = TelemetryWebSocketPackets_;
-						TelemetryStream()->NotifyEndPoint(SerialNumberInt_, *KafkaPayload);
+						TelemetryStream()->NotifyEndPoint(SerialNumberInt_, KafkaPayload);
 					} else {
 						StopWebSocketTelemetry(CommandManager()->Next_RPC_ID());
 					}

diff --git a/src/AP_WS_Process_wifiscan.cpp b/src/AP_WS_Process_wifiscan.cpp
@@ -21,11 +21,7 @@ namespace OpenWifi {
 
 		if (ParamsObj->has(uCentralProtocol::SERIAL) && ParamsObj->has(uCentralProtocol::DATA)) {
 			if (KafkaManager()->Enabled()) {
-				auto Data = ParamsObj->get(uCentralProtocol::DATA);
-				Poco::JSON::Stringifier Stringify;
-				std::ostringstream OS;
-				Stringify.condense(ParamsObj, OS);
-				KafkaManager()->PostMessage(KafkaTopics::WIFISCAN, SerialNumber_, std::make_shared<std::string>(OS.str()));
+				KafkaManager()->PostMessage(KafkaTopics::WIFISCAN, SerialNumber_, *ParamsObj);
 			}
 		}
 	}

diff --git a/src/AP_WS_Server.cpp b/src/AP_WS_Server.cpp
@@ -233,10 +233,7 @@ namespace OpenWifi {
 		FullEvent.set("timestamp", now);
 		FullEvent.set("payload", KafkaNotification);
 
-		std::ostringstream OS;
-		FullEvent.stringify(OS);
-		KafkaManager()->PostMessage(KafkaTopics::DEVICE_EVENT_QUEUE, "system",
-									std::make_shared<std::string>(OS.str()));
+		KafkaManager()->PostMessage(KafkaTopics::DEVICE_EVENT_QUEUE, "system", FullEvent);
 	}
 
 	void AP_WS_Server::Stop() {

diff --git a/src/GWKafkaEvents.cpp b/src/GWKafkaEvents.cpp
@@ -14,8 +14,7 @@ namespace OpenWifi {
 			Event.set("payload", payload_);
 			std::ostringstream OS;
 			Event.stringify(OS);
-			auto payload = std::make_shared<std::string>(OS.str());
-			KafkaManager()->PostMessage(KafkaTopics::DEVICE_EVENT_QUEUE, Utils::IntToSerialNumber(serialNumber_), payload);
+			KafkaManager()->PostMessage(KafkaTopics::DEVICE_EVENT_QUEUE, Utils::IntToSerialNumber(serialNumber_), OS.str());
 		}
 	}
 

diff --git a/src/RESTAPI/RESTAPI_device_commandHandler.cpp b/src/RESTAPI/RESTAPI_device_commandHandler.cpp
@@ -1024,7 +1024,7 @@ namespace OpenWifi {
 		RESTAPI_RPC::WaitForCommand(CMD_RPC, APCommands::Commands::wifiscan, false, Cmd, Params,
 									*Request, *Response, timeout, nullptr, this, Logger_);
 		if (Cmd.ErrorCode == 0) {
-			KafkaManager()->PostMessage(KafkaTopics::WIFISCAN, SerialNumber_, std::make_shared<std::string>(Cmd.Results));
+			KafkaManager()->PostMessage(KafkaTopics::WIFISCAN, SerialNumber_, Cmd.Results);
 		}
 	}
 
@@ -1069,7 +1069,7 @@ namespace OpenWifi {
 										Logger_);
 			if (Cmd.ErrorCode == 0) {
 				KafkaManager()->PostMessage(KafkaTopics::DEVICE_EVENT_QUEUE, SerialNumber_,
-											std::make_shared<std::string>(Cmd.Results));
+											Cmd.Results);
 			}
 			return;
 		}
@@ -1126,9 +1126,6 @@ namespace OpenWifi {
 		BadRequest(RESTAPI::Errors::MissingOrInvalidParameters);
 	}
 
-#define DBGLINE                                                                                    \
-	{ std::cout << __LINE__ << std::endl; }
-
 	void RESTAPI_device_commandHandler::Rtty(
 		const std::string &CMD_UUID, uint64_t CMD_RPC, std::chrono::milliseconds timeout,
 		[[maybe_unused]] const GWObjects::DeviceRestrictions &Restrictions) {

diff --git a/src/framework/EventBusManager.cpp b/src/framework/EventBusManager.cpp
@@ -14,18 +14,18 @@ namespace OpenWifi {
 	void EventBusManager::run() {
 		Running_ = true;
 		Utils::SetThreadName("fmwk:EventMgr");
-		auto Msg = std::make_shared<std::string>(MicroServiceMakeSystemEventMessage(KafkaTopics::ServiceEvents::EVENT_JOIN));
+		auto Msg = (MicroServiceMakeSystemEventMessage(KafkaTopics::ServiceEvents::EVENT_JOIN));
 		KafkaManager()->PostMessage(KafkaTopics::SERVICE_EVENTS, MicroServicePrivateEndPoint(), Msg,
 									false);
 		while (Running_) {
 			Poco::Thread::trySleep((unsigned long)MicroServiceDaemonBusTimer());
 			if (!Running_)
 				break;
-			Msg = std::make_shared<std::string>(MicroServiceMakeSystemEventMessage(KafkaTopics::ServiceEvents::EVENT_KEEP_ALIVE));
+			Msg = (MicroServiceMakeSystemEventMessage(KafkaTopics::ServiceEvents::EVENT_KEEP_ALIVE));
 			KafkaManager()->PostMessage(KafkaTopics::SERVICE_EVENTS, MicroServicePrivateEndPoint(),
 										Msg, false);
 		}
-		Msg = std::make_shared<std::string>(MicroServiceMakeSystemEventMessage(KafkaTopics::ServiceEvents::EVENT_LEAVE));
+		Msg = (MicroServiceMakeSystemEventMessage(KafkaTopics::ServiceEvents::EVENT_LEAVE));
 		KafkaManager()->PostMessage(KafkaTopics::SERVICE_EVENTS, MicroServicePrivateEndPoint(), Msg,
 									false);
 	};

diff --git a/src/framework/KafkaManager.cpp b/src/framework/KafkaManager.cpp
@@ -180,7 +180,7 @@ namespace OpenWifi {
 							Consumer.async_commit(Msg);
 						continue;
 					}
-					KafkaManager()->Dispatch(Msg.get_topic().c_str(), Msg.get_key(), std::make_shared<std::string>(Msg.get_payload()));
+					KafkaManager()->Dispatch(Msg.get_topic().c_str(), Msg.get_key(), Msg.get_payload());
 					if (!AutoCommit)
 						Consumer.async_commit(Msg);
 				}
@@ -213,7 +213,7 @@ namespace OpenWifi {
 	}
 
 	void KafkaProducer::Produce(const char *Topic, const std::string &Key,
-								std::shared_ptr<std::string> Payload) {
+								const std::string &Payload) {
 		std::lock_guard G(Mutex_);
 		Queue_.enqueueNotification(new KafkaMessage(Topic, Key, Payload));
 	}
@@ -276,7 +276,7 @@ namespace OpenWifi {
 	}
 
 	void KafkaDispatcher::Dispatch(const char *Topic, const std::string &Key,
-								   const std::shared_ptr<std::string> Payload) {
+								   const std::string & Payload) {
 		std::lock_guard G(Mutex_);
 		auto It = Notifiers_.find(Topic);
 		if (It != Notifiers_.end()) {
@@ -333,20 +333,29 @@ namespace OpenWifi {
 	}
 
 	void KafkaManager::PostMessage(const char *topic, const std::string &key,
-								   const std::shared_ptr<std::string> PayLoad, bool WrapMessage) {
+								   const std::string & PayLoad, bool WrapMessage) {
 		if (KafkaEnabled_) {
 			ProducerThr_.Produce(topic, key, WrapMessage ? WrapSystemId(PayLoad) : PayLoad);
 		}
 	}
 
+	void KafkaManager::PostMessage(const char *topic, const std::string &key,
+					 const Poco::JSON::Object &Object, bool WrapMessage) {
+		if (KafkaEnabled_) {
+			std::ostringstream ObjectStr;
+			Object.stringify(ObjectStr);
+			ProducerThr_.Produce(topic, key, WrapMessage ? WrapSystemId(ObjectStr.str()) : ObjectStr.str());
+		}
+	}
+
+
 	void KafkaManager::Dispatch(const char *Topic, const std::string &Key,
-								const std::shared_ptr<std::string> Payload) {
+								const std::string &Payload) {
 		Dispatcher_.Dispatch(Topic, Key, Payload);
 	}
 
-	[[nodiscard]] const std::shared_ptr<std::string> KafkaManager::WrapSystemId(const std::shared_ptr<std::string> PayLoad) {
-		*PayLoad = SystemInfoWrapper_ + *PayLoad + "}";
-		return PayLoad;
+	[[nodiscard]] std::string KafkaManager::WrapSystemId(const std::string & PayLoad) {
+		return SystemInfoWrapper_ + PayLoad + "}";
 	}
 
 	uint64_t KafkaManager::RegisterTopicWatcher(const std::string &Topic,

diff --git a/src/framework/KafkaManager.h b/src/framework/KafkaManager.h
@@ -6,7 +6,7 @@
 
 #include "Poco/Notification.h"
 #include "Poco/NotificationQueue.h"
-
+#include "Poco/JSON/Object.h"
 #include "framework/KafkaTopics.h"
 #include "framework/OpenWifiTypes.h"
 #include "framework/SubSystemServer.h"
@@ -18,25 +18,25 @@ namespace OpenWifi {
 
 	class KafkaMessage : public Poco::Notification {
 	  public:
-		KafkaMessage(const char * Topic, const std::string &Key, std::shared_ptr<std::string> Payload)
+		KafkaMessage(const char * Topic, const std::string &Key, const std::string &Payload)
 			: Topic_(Topic), Key_(Key), Payload_(Payload) {}
 
 		inline const char * Topic() { return Topic_; }
 		inline const std::string &Key() { return Key_; }
-		inline const std::string &Payload() { return *Payload_; }
+		inline const std::string &Payload() { return Payload_; }
 
 	  private:
 		const char *Topic_;
 		std::string Key_;
-		std::shared_ptr<std::string> Payload_;
+		std::string Payload_;
 	};
 
 	class KafkaProducer : public Poco::Runnable {
 	  public:
 		void run() override;
 		void Start();
 		void Stop();
-		void Produce(const char *Topic, const std::string &Key, std::shared_ptr<std::string> Payload);
+		void Produce(const char *Topic, const std::string &Key, const std::string & Payload);
 
 	  private:
 		std::recursive_mutex Mutex_;
@@ -63,7 +63,7 @@ namespace OpenWifi {
 		void Stop();
 		auto RegisterTopicWatcher(const std::string &Topic, Types::TopicNotifyFunction &F);
 		void UnregisterTopicWatcher(const std::string &Topic, int Id);
-		void Dispatch(const char *Topic, const std::string &Key, const std::shared_ptr<std::string> Payload);
+		void Dispatch(const char *Topic, const std::string &Key, const std::string & Payload);
 		void run() override;
 		void Topics(std::vector<std::string> &T);
 
@@ -92,9 +92,12 @@ namespace OpenWifi {
 		void Stop() override;
 
 		void PostMessage(const char *topic, const std::string &key,
-						 std::shared_ptr<std::string> PayLoad, bool WrapMessage = true);
-		void Dispatch(const char *Topic, const std::string &Key, std::shared_ptr<std::string> Payload);
-		[[nodiscard]] const std::shared_ptr<std::string> WrapSystemId(std::shared_ptr<std::string> PayLoad);
+						 const std::string &PayLoad, bool WrapMessage = true);
+		void PostMessage(const char *topic, const std::string &key,
+						 const Poco::JSON::Object &Object, bool WrapMessage = true);
+
+		void Dispatch(const char *Topic, const std::string &Key, const std::string &Payload);
+		[[nodiscard]] std::string WrapSystemId(const std::string & PayLoad);
 		[[nodiscard]] inline bool Enabled() const { return KafkaEnabled_; }
 		uint64_t RegisterTopicWatcher(const std::string &Topic, Types::TopicNotifyFunction &F);
 		void UnregisterTopicWatcher(const std::string &Topic, uint64_t Id);

diff --git a/src/framework/RESTAPI_Handler.h b/src/framework/RESTAPI_Handler.h
@@ -584,7 +584,27 @@ namespace OpenWifi {
             return ReturnRawJSON(os.str());
         }
 
-		inline void ReturnRawJSON(const std::string &json_doc) {
+        template<class T> void ReturnObject(const std::vector<T> &Objects) {
+            Poco::JSON::Array   Arr;
+            for(const auto &Object:Objects) {
+                Poco::JSON::Object O;
+                Object.to_json(O);
+                Arr.add(O);
+            }
+            std::ostringstream os;
+            Arr.stringify(os);
+            return ReturnRawJSON(os.str());
+        }
+
+        template<class T> void ReturnObject(const T &Object) {
+            Poco::JSON::Object  O;
+            Object.to_json(O);
+            std::ostringstream os;
+            O.stringify(os);
+            return ReturnRawJSON(os.str());
+        }
+
+        inline void ReturnRawJSON(const std::string &json_doc) {
 			PrepareResponse();
 			if (Request != nullptr) {
 				//   can we compress ???

diff --git a/src/framework/ow_constants.h b/src/framework/ow_constants.h
@@ -40,6 +40,7 @@ namespace OpenWifi {
 	};
 }
 
+#define DBGLINE     std::cout << __LINE__ << ":" << __FILE__ << ", " << __func__ << std::endl;
 namespace OpenWifi::RESTAPI::Errors {
 	struct msg {
 		uint64_t err_num;