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

Signed-off-by: stephb9959 <stephane.bourque@gmail.com>

build

@@ -1 +1 @@
-20
+22

src/framework/ConfigurationValidator.cpp

@@ -8037,26 +8037,31 @@ bool ExternalValijsonFormatChecker(const std::string &format, const std::string
 			return true;
 		if (results)
 			results->pushError(context, fmt::format("{} is not a valid CIDR IPv4 block", value));
+        return false;
 	} else if (format == "uc-cidr6") {
 		if (IsCIDRv6(value))
 			return true;
 		if (results)
 			results->pushError(context, fmt::format("{} is not a valid CIDR IPv6 block", value));
+        return false;
 	} else if (format == "uc-cidr") {
 		if (IsCIDR(value))
 			return true;
 		if (results)
 			results->pushError(context, fmt::format("{} is not a valid CIDR block", value));
+        return false;
 	} else if (format == "uc-mac") {
 		if (std::regex_match(value, mac_regex))
 			return true;
 		if (results)
 			results->pushError(context, fmt::format("{} is not a valid MAC address", value));
+        return false;
 	} else if (format == "uc-timeout") {
 		if (std::regex_match(value, uc_timeout_regex))
 			return true;
 		if (results)
 			results->pushError(context, fmt::format("{} is not a valid timeout value", value));
+        return false;
 	} else if (format == "uc-host") {
 		if (IsIP(value))
 			return true;
@@ -8064,18 +8069,22 @@ bool ExternalValijsonFormatChecker(const std::string &format, const std::string
 			return true;
 		if (results)
 			results->pushError(context, fmt::format("{} is not a valid hostname", value));
+        return false;
 	} else if (format == "fqdn" || format == "uc-fqdn") {
 		if (std::regex_match(value, host_regex))
 			return true;
 		if (results)
 			results->pushError(context, fmt::format("{} is not a valid FQDN", value));
+        return false;
 	} else if (format == "uc-base64") {
-		std::string s{value};
+        bool valid =  value.size() % 4 == 0 && std::all_of(value.begin(), value.end(), [](char c) {
-		Poco::trimInPlace(s);
+            return std::isalnum(c) || c == '+' || c == '/' || c == '=';
-		if ((s.size() % 4 == 0) && std::regex_match(s, b64_regex))
+        });
+        if (valid)
             return true;
 		if (results)
-			results->pushError(context, fmt::format("{} is not a valid base 64 value", value));
+			results->pushError(context, fmt::format("{} is not a valid base 64 value", "..."));
+        return false;
 	} else if (format == "uri") {
 		try {
 			Poco::URI uri(value);
@@ -8084,6 +8093,7 @@ bool ExternalValijsonFormatChecker(const std::string &format, const std::string
 		}
 		if (results)
 			results->pushError(context, fmt::format("{} is not a valid URL", value));
+        return false;
 	} else if (format == "uc-portrange") {
 		try {
 			if (IsPortRangeIsValid(value))
@@ -8092,11 +8102,13 @@ bool ExternalValijsonFormatChecker(const std::string &format, const std::string
 		}
 		if (results)
 			results->pushError(context, fmt::format("{} is not a valid post range", value));
+        return false;
 	} else if (format == "ip") {
 		if (IsIP(value))
 			return true;
 		if (results)
 			results->pushError(context, fmt::format("{} is not a valid IP address", value));
+        return false;
 	}
 	return true;
 }

src/framework/KafkaManager.cpp

@@ -79,8 +79,10 @@ namespace OpenWifi {
 		Utils::SetThreadName("Kafka:Prod");
 		cppkafka::Configuration Config(
 			{{"client.id", MicroServiceConfigGetString("openwifi.kafka.client.id", "")},
-			 {"metadata.broker.list",
+			 {"metadata.broker.list",MicroServiceConfigGetString("openwifi.kafka.brokerlist", "")},
-			  MicroServiceConfigGetString("openwifi.kafka.brokerlist", "")}});
+			 {"send.buffer.bytes", KafkaManager()->KafkaManagerMaximumPayloadSize() }
+			}
+ 		);
 
 		AddKafkaSecurity(Config);
 
@@ -275,6 +277,7 @@ namespace OpenWifi {
 	int KafkaManager::Start() {
 		if (!KafkaEnabled_)
 			return 0;
+		MaxPayloadSize_ = MicroServiceConfigGetInt("openwifi.kafka.max.payload", 2500000);
 		ConsumerThr_.Start();
 		ProducerThr_.Start();
 		return 0;

src/framework/KafkaManager.h

@@ -94,11 +94,14 @@ namespace OpenWifi {
 			return ConsumerThr_.UnregisterTopicWatcher(Topic,Id);
 		}
 
+		std::uint64_t KafkaManagerMaximumPayloadSize() const { return MaxPayloadSize_; }
+
 	  private:
 		bool KafkaEnabled_ = false;
 		std::string SystemInfoWrapper_;
 		KafkaProducer ProducerThr_;
 		KafkaConsumer ConsumerThr_;
+		std::uint64_t MaxPayloadSize_ = 2500000;
 
 		void PartitionAssignment(const cppkafka::TopicPartitionList &partitions);
 		void PartitionRevocation(const cppkafka::TopicPartitionList &partitions);

src/framework/MicroService.cpp

@@ -1,4 +1,5 @@
 //
+//
 // Created by stephane bourque on 2022-10-26.
 //
 
@@ -480,6 +481,16 @@ namespace OpenWifi {
         KafkaManager()->RegisterTopicWatcher(KafkaTopics::SERVICE_EVENTS, F);
     }
 
+    void MicroService::StopEverything([[maybe_unused]] Poco::Util::Application &self) {
+        LoadConfigurationFile();
+        InitializeLoggingSystem();
+
+        Types::TopicNotifyFunction F = [this](const std::string &Key, const std::string &Payload) {
+            this->BusMessageReceived(Key, Payload);
+        };
+        KafkaManager()->RegisterTopicWatcher(KafkaTopics::SERVICE_EVENTS, F);
+    }
+
     void MicroService::initialize([[maybe_unused]] Poco::Util::Application &self) {
 #ifndef USE_MEDUSA_CLIENT
         StartEverything(self);

src/framework/MicroService.h

@@ -113,6 +113,7 @@ namespace OpenWifi {
 		void LoadMyConfig();
 		void initialize(Poco::Util::Application &self) override;
         void StartEverything(Poco::Util::Application &self);
+        void StopEverything(Poco::Util::Application &self);
         void uninitialize() override;
 		void reinitialize(Poco::Util::Application &self) override;
 		void defineOptions(Poco::Util::OptionSet &options) override;
@@ -168,6 +169,15 @@ namespace OpenWifi {
 
         inline void SetConfigContent(const std::string &Content) { ConfigContent_ = Content; }
 
+        inline std::optional<OpenWifi::Types::MicroServiceMeta> GetPrivateEndPointServiceKey( const std::string & ServicePrivateEndPoint ) {
+            std::lock_guard   G(InfraMutex_);
+            auto K = Services_.find(ServicePrivateEndPoint);
+            if(K==end(Services_)) {
+                return std::nullopt;
+            }
+            return K->second;
+        }
+
 	  private:
 		static MicroService *instance_;
 		bool HelpRequested_ = false;

src/framework/MicroServiceFuncs.cpp

@@ -133,4 +133,8 @@ namespace OpenWifi {
 		return MicroService::instance().Hash();
 	}
 
+    std::optional<OpenWifi::Types::MicroServiceMeta> MicroServicePrivateAccessKey(const std::string &servicePrivateEndPoint) {
+        return MicroService::instance().GetPrivateEndPointServiceKey(servicePrivateEndPoint);
+    }
+
 } // namespace OpenWifi

src/framework/MicroServiceFuncs.h

@@ -23,6 +23,8 @@ namespace OpenWifi {
 	std::string MicroServiceConfigGetString(const std::string &Key,
 											const std::string &DefaultValue);
 	std::string MicroServiceAccessKey();
+    std::optional<OpenWifi::Types::MicroServiceMeta> MicroServicePrivateAccessKey(const std::string &servicePrivateEndPoint);
+
     bool MicroServiceConfigGetBool(const std::string &Key, bool DefaultValue);
 	std::uint64_t MicroServiceConfigGetInt(const std::string &Key, std::uint64_t DefaultValue);
 	std::string MicroServicePrivateEndPoint();