Baf 1155/use async client

CMLibStorage.cmake

@@ -1,3 +1,7 @@
+FIND_PACKAGE(CMLIB REQUIRED COMPONENTS CMCONF)
+
+CMCONF_INIT_SYSTEM(FLEET_PROTOCOL)
+
 SET(STORAGE_LIST DEP)
 
 SET(STORAGE_LIST_DEP "https://github.com/bacpack-system/package-tracker.git")

CMakeLists.txt

@@ -6,7 +6,7 @@ FIND_PACKAGE(CMLIB
         REQUIRED
 )
 
-SET(BRINGAUTO_MODULE_GATEWAY_VERSION 1.3.5)
+SET(BRINGAUTO_MODULE_GATEWAY_VERSION 1.4.0)
 
 SET(MODULE_GATEWAY_MINIMUM_LOGGER_VERBOSITY "DEBUG" CACHE STRING "Minimum logger verbosity level for module-gateway")
 
@@ -16,7 +16,7 @@ CMDEF_COMPILE_DEFINITIONS(
         "MODULE_GATEWAY_MINIMUM_LOGGER_VERBOSITY=\"${MODULE_GATEWAY_MINIMUM_LOGGER_VERBOSITY}\""
 )
 SET(CMAKE_INSTALL_RPATH "$ORIGIN/../${CMDEF_LIBRARY_INSTALL_DIR}")
-SET(CMAKE_CXX_STANDARD 20)
+SET(CMAKE_CXX_STANDARD 23)
 
 INCLUDE(CheckPIESupported)
 CHECK_PIE_SUPPORTED()
@@ -47,13 +47,16 @@ SET(Protobuf_USE_STATIC_LIBS ON)
 
 FIND_PACKAGE(Boost 1.74 REQUIRED CONFIG)
 FIND_PACKAGE(Protobuf 3.21.12 REQUIRED)
-FIND_PACKAGE(cxxopts 3.0.0 REQUIRED)
-FIND_PACKAGE(nlohmann_json 3.2.0 REQUIRED)
+FIND_PACKAGE(cxxopts 3.1.1 REQUIRED)
+FIND_PACKAGE(nlohmann_json 3.10.5 REQUIRED)
 FIND_PACKAGE(PahoMqttCpp REQUIRED)
 FIND_PACKAGE(eclipse-paho-mqtt-c REQUIRED)
 FIND_PACKAGE(libbringauto_logger 2.0.0 REQUIRED)
 FIND_PACKAGE(fleet-protocol-interface 2.0.0 REQUIRED)
 FIND_PACKAGE(ZLIB 1.2.11 REQUIRED)
+FIND_PACKAGE(fleet-protocol-cxx-helpers-static 1.2.0 REQUIRED)
+FIND_PACKAGE(aeron 1.48.6 REQUIRED)
+FIND_PACKAGE(async-function-execution-shared 1.0.0 REQUIRED)
 
 FILE(GLOB_RECURSE source_files "source/*")
 ADD_LIBRARY(module-gateway-lib STATIC "${source_files}")
@@ -71,6 +74,8 @@ TARGET_LINK_LIBRARIES(module-gateway-lib PUBLIC
                         eclipse-paho-mqtt-c::paho-mqtt3as
                         PahoMqttCpp::paho-mqttpp3
                         ZLIB::ZLIB
+                        fleet-protocol-cxx-helpers-static::fleet-protocol-cxx-helpers-static
+                        async-function-execution-shared::async-function-execution-shared
                         ${CMAKE_DL_LIBS}
 )
 

README.md

@@ -29,10 +29,17 @@ connection is broken and as soon as the connection is up, then error aggregated
 - [cmlib](https://github.com/cmakelib/cmakelib)
 
 - [protobuf](https://github.com/protocolbuffers/protobuf/tree/main/src) >= v3.21.12
-- [cxxopts](https://github.com/jarro2783/cxxopts) >= v3.0.0
-- [boost](https://github.com/boostorg/boost) >= v1.74.0
-- [nlohmann-json](https://github.com/nlohmann/json) >= v3.2.0
-- [ba-logger](https://github.com/bringauto/ba-logger) >= v1.2.0
+- [cxxopts](https://github.com/jarro2783/cxxopts) >= v3.1.1
+- [boost](https://github.com/boostorg/boost) >= v1.86.0
+- [nlohmann-json](https://github.com/nlohmann/json) >= v3.10.5/
+- [pahomqtt](https://github.com/eclipse-paho/paho.mqtt.c) >= v1.3.9
+- [pahomqttcpp](https://github.com/eclipse-paho/paho.mqtt.cpp) >= v1.3.2
+- [zlib](https://github.com/madler/zlib) >= v1.2.11
+- [ba-logger](https://github.com/bringauto/ba-logger) >= v2.0.0
+- [fleet-protocol-interface](https://github.com/bringauto/fleet-protocol) >= v2.0.0
+- [fleet-protocol-cpp](https://github.com/bringauto/fleet-protocol-cpp) >= v1.1.1
+- [aeron](https://github.com/aeron-io/aeron) >= v1.48.6
+- [async-function-execution](https://github.com/bringauto/async-function-execution) >= 0.1.0
 - g++ >= 10 or other compiler with c++20 support
 
 ## Build
@@ -50,7 +57,7 @@ make
 
 ### Arguments
 
-* required arguments:
+* Required arguments:
   * `-c <string> | --config-path=<string>`path to json configuration file ([Configs Readme](./configs/README.md))
 * All arguments:
   * `-h | --help` print help

cmake/Dependencies.cmake

@@ -3,13 +3,15 @@ SET(CMAKE_FIND_USE_CMAKE_SYSTEM_PATH FALSE)
 BA_PACKAGE_LIBRARY(protobuf                 v4.21.12)
 BA_PACKAGE_LIBRARY(fleet-protocol-interface v2.0.0 NO_DEBUG ON)
 BA_PACKAGE_LIBRARY(nlohmann-json            v3.10.5 NO_DEBUG ON)
-BA_PACKAGE_LIBRARY(cxxopts                  v3.0.5 NO_DEBUG ON)
+BA_PACKAGE_LIBRARY(cxxopts                  v3.1.1 NO_DEBUG ON)
 BA_PACKAGE_LIBRARY(boost                    v1.86.0)
-BA_PACKAGE_LIBRARY(ba-logger                v2.0.0
-)
+BA_PACKAGE_LIBRARY(ba-logger                v2.0.0)
 BA_PACKAGE_LIBRARY(pahomqttc                v1.3.9)
 BA_PACKAGE_LIBRARY(pahomqttcpp              v1.3.2)
 BA_PACKAGE_LIBRARY(zlib                     v1.2.11 OUTPUT_PATH_VAR ZLIB_DIR)
+BA_PACKAGE_LIBRARY(fleet-protocol-cpp       v1.2.0)
+BA_PACKAGE_LIBRARY(aeron                    v1.48.6)
+BA_PACKAGE_LIBRARY(async-function-execution v1.0.0)
 
 IF (BRINGAUTO_TESTS)
     BA_PACKAGE_LIBRARY(gtest v1.12.1)

include/bringauto/external_client/ErrorAggregator.hpp

@@ -1,6 +1,6 @@
 #pragma once
 
-#include <bringauto/modules/ModuleManagerLibraryHandler.hpp>
+#include <bringauto/modules/IModuleManagerLibraryHandler.hpp>
 #include <bringauto/structures/DeviceIdentification.hpp>
 
 #include <filesystem>
@@ -25,7 +25,7 @@ class ErrorAggregator {
 	 * @return OK if initialization was successful
 	 * @return NOT_OK if an error occurred
 	 */
-	int init_error_aggregator(const std::shared_ptr <modules::ModuleManagerLibraryHandler> &library);
+	int init_error_aggregator(const std::shared_ptr <modules::IModuleManagerLibraryHandler> &library);
 
 	/**
 	 * @short Clean up.
@@ -111,7 +111,7 @@ class ErrorAggregator {
 		modules::Buffer lastStatus {};
 	};
 
-	std::shared_ptr <modules::ModuleManagerLibraryHandler> module_ {};
+	std::shared_ptr <modules::IModuleManagerLibraryHandler> module_ {};
 
 	/**
 	 * @brief Map of devices states, key is device identification converted to string

include/bringauto/modules/Buffer.hpp

@@ -18,7 +18,8 @@ namespace bringauto::modules {
  */
 struct Buffer final {
 
-	friend class ModuleManagerLibraryHandler;
+	friend class ModuleManagerLibraryHandlerLocal;
+	friend class ModuleManagerLibraryHandlerAsync;
 
 	Buffer() = default;
 	Buffer(const Buffer& buff) = default;

include/bringauto/modules/IModuleManagerLibraryHandler.hpp

@@ -0,0 +1,78 @@
+#pragma once
+
+#include <bringauto/modules/Buffer.hpp>
+
+#include <filesystem>
+
+
+
+namespace bringauto::modules {
+
+/**
+ * @brief Class used to load and handle library created by module maintainer
+ */
+class IModuleManagerLibraryHandler {
+public:
+	explicit IModuleManagerLibraryHandler() = default;
+
+	virtual ~IModuleManagerLibraryHandler() = default;
+
+	/**
+	 * @brief Load library created by a module maintainer
+	 *
+	 * @param path path to the library
+	 */
+	virtual void loadLibrary(const std::filesystem::path &path) = 0;
+
+	virtual int getModuleNumber() = 0;
+
+	virtual int isDeviceTypeSupported(unsigned int device_type) = 0;
+
+	virtual int sendStatusCondition(const Buffer &current_status, const Buffer &new_status, unsigned int device_type) = 0;
+
+	/**
+	 * @short After executing the respective module function, an error might be thrown when allocating the buffer.
+	 * 
+	 * @see fleet-protocol/lib/module_maintainer/module_gateway/include/module_manager.h
+	 */
+	virtual int generateCommand(Buffer &generated_command, const Buffer &new_status,
+								const Buffer &current_status, const Buffer &current_command,
+								unsigned int device_type) = 0;
+
+	/**
+	 * @short After executing the respective module function, an error might be thrown when allocating the buffer.
+	 * 
+	 * @see fleet-protocol/lib/module_maintainer/module_gateway/include/module_manager.h
+	 */
+	virtual int aggregateStatus(Buffer &aggregated_status, const Buffer &current_status,
+								const Buffer &new_status, unsigned int device_type) = 0;
+
+	/**
+	 * @short After executing the respective module function, an error might be thrown when allocating the buffer.
+	 * 
+	 * @see fleet-protocol/lib/module_maintainer/module_gateway/include/module_manager.h
+	 */
+	virtual int aggregateError(Buffer &error_message, const Buffer &current_error_message, const Buffer &status,
+						  	   unsigned int device_type) = 0;
+
+	/**
+	 * @short After executing the respective module function, an error might be thrown when allocating the buffer.
+	 * 
+	 * @see fleet-protocol/lib/module_maintainer/module_gateway/include/module_manager.h
+	 */
+	virtual int generateFirstCommand(Buffer &default_command, unsigned int device_type) = 0;
+
+	virtual int statusDataValid(const Buffer &status, unsigned int device_type) = 0;
+
+	virtual int commandDataValid(const Buffer &command, unsigned int device_type) = 0;
+
+	/**
+	 * @brief Constructs a buffer with the given size
+	 * 
+	 * @param size size of the buffer
+	 * @return a new Buffer object
+	 */
+	virtual Buffer constructBuffer(std::size_t size = 0) = 0;
+};
+
+}

include/bringauto/modules/ModuleManagerLibraryHandlerAsync.hpp

@@ -0,0 +1,123 @@
+#pragma once
+
+#include <bringauto/modules/IModuleManagerLibraryHandler.hpp>
+#include <bringauto/settings/Constants.hpp>
+
+#include <bringauto/async_function_execution/AsyncFunctionExecutor.hpp>
+#include <bringauto/fleet_protocol/cxx/AsyncModuleFunctionDefinitions.hpp>
+#include <boost/process.hpp>
+
+#include <mutex>
+
+
+
+namespace bringauto::modules {
+
+/**
+ * @brief Class used to load and handle library created by module maintainer
+ */
+class ModuleManagerLibraryHandlerAsync : public IModuleManagerLibraryHandler {
+public:
+	explicit ModuleManagerLibraryHandlerAsync(const std::filesystem::path &moduleBinaryPath, const int moduleNumber);
+
+	~ModuleManagerLibraryHandlerAsync() override;
+
+	/**
+	 * @brief Load library created by a module maintainer
+	 *
+	 * @param path path to the library
+	 */
+	void loadLibrary(const std::filesystem::path &path) override;
+
+	int getModuleNumber() override;
+
+	int isDeviceTypeSupported(unsigned int device_type) override;
+
+	int	sendStatusCondition(const Buffer &current_status, const Buffer &new_status, unsigned int device_type) override;
+
+	/**
+	 * @short After executing the respective module function, an error might be thrown when allocating the buffer.
+	 * 
+	 * @see fleet-protocol/lib/module_maintainer/module_gateway/include/module_manager.h
+	 */
+	int generateCommand(Buffer &generated_command, const Buffer &new_status,
+						const Buffer &current_status, const Buffer &current_command,
+						unsigned int device_type) override;
+
+	/**
+	 * @short After executing the respective module function, an error might be thrown when allocating the buffer.
+	 * 
+	 * @see fleet-protocol/lib/module_maintainer/module_gateway/include/module_manager.h
+	 */
+	int aggregateStatus(Buffer &aggregated_status, const Buffer &current_status,
+						const Buffer &new_status, unsigned int device_type) override;
+
+	/**
+	 * @short After executing the respective module function, an error might be thrown when allocating the buffer.
+	 * 
+	 * @see fleet-protocol/lib/module_maintainer/module_gateway/include/module_manager.h
+	 */
+	int	aggregateError(Buffer &error_message, const Buffer &current_error_message, const Buffer &status,
+						  unsigned int device_type) override;
+
+	/**
+	 * @short After executing the respective module function, an error might be thrown when allocating the buffer.
+	 * 
+	 * @see fleet-protocol/lib/module_maintainer/module_gateway/include/module_manager.h
+	 */
+	int generateFirstCommand(Buffer &default_command, unsigned int device_type) override;
+
+	int statusDataValid(const Buffer &status, unsigned int device_type) override;
+
+	int commandDataValid(const Buffer &command, unsigned int device_type) override;
+
+	/**
+	 * @brief Constructs a buffer with the given size
+	 * 
+	 * @param size size of the buffer
+	 * @return a new Buffer object
+	 */
+	Buffer constructBuffer(std::size_t size = 0) override;
+
+private:
+
+	int allocate(struct buffer *buffer_pointer, size_t size_in_bytes) const;
+
+	void deallocate(struct buffer *buffer) const;
+
+	/**
+	 * @brief Constructs a buffer with the same raw c buffer as provided
+	 * 
+	 * @param buffer c buffer to be used
+	 * @return a new Buffer object
+	 */
+	Buffer constructBufferByTakeOwnership(struct ::buffer& buffer);
+
+	std::function<void(struct buffer *)> deallocate_ {};
+
+	/// Path to the module binary
+	std::filesystem::path moduleBinaryPath_ {};
+	/// Process of the module binary
+	boost::process::child moduleBinaryProcess_ {};
+
+	/// TODO find a way to not need this
+	std::mutex getModuleNumberMutex_ {};
+	std::mutex isDeviceTypeSupportedMutex_ {};
+	std::mutex sendStatusConditionMutex_ {};
+	std::mutex generateCommandMutex_ {};
+	std::mutex aggregateStatusMutex_ {};
+	std::mutex aggregateErrorMutex_ {};
+	std::mutex generateFirstCommandMutex_ {};
+	std::mutex statusDataValidMutex_ {};
+	std::mutex commandDataValidMutex_ {};
+
+	fleet_protocol::cxx::ModuleFunctionExecutor aeronClient {
+		async_function_execution::Config {
+			.isProducer = true,
+			.defaultTimeout = settings::AeronClientConstants::aeron_client_default_timeout,
+		},
+		fleet_protocol::cxx::moduleFunctionList
+	};
+};
+
+}