Merge pull request #2 from bringauto/first_release

First release

Author: MarioIvancik

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

@@ -4,7 +4,7 @@ PROJECT(async-function-execution C CXX)
 SET(CMAKE_POSITION_INDEPENDENT_CODE ON)
 SET(CMAKE_CXX_STANDARD 20)
 
-SET(ASYNC_FUNCTION_EXECUTION_VERSION 0.0.0)
+SET(ASYNC_FUNCTION_EXECUTION_VERSION 0.1.0)
 
 FIND_PACKAGE(CMLIB
         COMPONENTS CMDEF CMUTIL
@@ -20,6 +20,7 @@ OPTION(BRINGAUTO_SAMPLES OFF)
 OPTION(BRINGAUTO_PACKAGE "Package creation" OFF)
 OPTION(BRINGAUTO_INSTALL "Enable install" OFF)
 OPTION(BRINGAUTO_SYSTEM_DEP "Enable system dependencies" OFF)
+OPTION(BRINGAUTO_TESTS "Enable tests" OFF)
 
 IF (NOT BRINGAUTO_SYSTEM_DEP)
     FIND_PACKAGE(CMLIB
@@ -36,25 +37,28 @@ IF (BRINGAUTO_PACKAGE)
     ENDIF ()
 ENDIF ()
 
-FIND_PACKAGE(aeron REQUIRED)
+FIND_PACKAGE(aeron         1.48.6 REQUIRED)
+FIND_PACKAGE(nlohmann_json 3.2.0  REQUIRED)
 
 FILE(GLOB_RECURSE source_files ${CMAKE_CURRENT_LIST_DIR}/source/*)
 
 CMDEF_ADD_LIBRARY(
         LIBRARY_GROUP ${ASYNC_FUNCTION_EXECUTION_TARGET_NAME}
-        TYPE STATIC
+        TYPE SHARED
         SOURCES ${source_files}
         INCLUDE_DIRECTORIES "${CMAKE_CURRENT_LIST_DIR}/include/"
         INSTALL_INCLUDE_DIRECTORIES "include/"
         VERSION ${ASYNC_FUNCTION_EXECUTION_VERSION}
 )
 
-ADD_LIBRARY(${ASYNC_FUNCTION_EXECUTION_ALIAS_NAME} ALIAS "${ASYNC_FUNCTION_EXECUTION_TARGET_NAME}-static")
+# I guess this is not needed when we use package?
+#ADD_LIBRARY(${ASYNC_FUNCTION_EXECUTION_ALIAS_NAME} ALIAS "${ASYNC_FUNCTION_EXECUTION_TARGET_NAME}-shared")
 
-TARGET_LINK_LIBRARIES(${ASYNC_FUNCTION_EXECUTION_TARGET_NAME}-static PUBLIC
+TARGET_LINK_LIBRARIES(${ASYNC_FUNCTION_EXECUTION_TARGET_NAME}-shared PUBLIC
     aeron::aeron
     aeron::aeron_client
     aeron::aeron_driver
+    nlohmann_json::nlohmann_json
 )
 
 IF(BRINGAUTO_TESTS)
@@ -65,14 +69,14 @@ ENDIF(BRINGAUTO_TESTS)
 
 IF (BRINGAUTO_INSTALL)
     CMDEF_INSTALL(
-            TARGET ${ASYNC_FUNCTION_EXECUTION_TARGET_NAME}-static
-            NAMESPACE ${ASYNC_FUNCTION_EXECUTION_TARGET_NAME}-static::
+            TARGET ${ASYNC_FUNCTION_EXECUTION_TARGET_NAME}-shared
+            NAMESPACE ${ASYNC_FUNCTION_EXECUTION_TARGET_NAME}-shared::
     )
 ENDIF ()
 
 IF (BRINGAUTO_PACKAGE)
     CMDEF_PACKAGE(
-            MAIN_TARGET ${ASYNC_FUNCTION_EXECUTION_TARGET_NAME}-static
+            MAIN_TARGET ${ASYNC_FUNCTION_EXECUTION_TARGET_NAME}-shared
             VERSION ${ASYNC_FUNCTION_EXECUTION_VERSION}
     )
     SET(CPACK_GENERATOR ZIP)

README.md

@@ -12,7 +12,7 @@ Both sides need to have all functions defined the same way. Example implementati
 using namespace bringauto::async_function_execution;
 
 FunctionDefinition FunctionAdd {
-	FunctionId { 1 }, // id can be 0-999 and has to be unique
+	FunctionId { 1 }, // id can be 0-255 and has to be unique
 	Return { int {} }, // return type of the function
 	Arguments { int {}, int {}, int {} } // individual function argument types
 };

cmake/Dependencies.cmake

@@ -1,5 +1,8 @@
 SET(CMAKE_FIND_USE_CMAKE_SYSTEM_PATH FALSE)
 
+BA_PACKAGE_LIBRARY(nlohmann-json v3.10.5 NO_DEBUG ON)
+BA_PACKAGE_LIBRARY(aeron         v1.48.6)
+
 IF (BRINGAUTO_TESTS)
     BA_PACKAGE_LIBRARY(gtest v1.12.1)
 ENDIF ()

examples/CMakeLists.txt

@@ -2,4 +2,4 @@ CMAKE_MINIMUM_REQUIRED(VERSION 3.25 FATAL_ERROR)
 
 ADD_EXECUTABLE(example main.cpp)
 
-TARGET_LINK_LIBRARIES(example PUBLIC async-function-execution-static)
+TARGET_LINK_LIBRARIES(example PUBLIC async-function-execution-shared)

include/bringauto/async_function_execution/AsyncFunctionExecutor.hpp

@@ -2,6 +2,7 @@
 
 #include <bringauto/async_function_execution/clients/AeronClient.hpp>
 #include <bringauto/async_function_execution/TimeoutIdleStrategy.hpp>
+#include <bringauto/async_function_execution/structures/Settings.hpp>
 
 #include <utility>
 #include <stdexcept>
@@ -18,7 +19,7 @@ namespace bringauto::async_function_execution {
 struct Config {
 	bool isProducer = true;
 	std::chrono::nanoseconds defaultTimeout = std::chrono::nanoseconds(0);
-	// TODO per function config map (for now only config will be timeout)
+	std::string_view functionConfigurations = "";
 };
 
 
@@ -27,7 +28,7 @@ struct Config {
  * Supported range is 0-255
  */
 struct FunctionId {
-	uint8_t value;
+	const uint8_t value;
 };
 
 
@@ -45,7 +46,7 @@ concept HasSerialize = requires(const T& t) {
  */
 template<typename T>
 struct Return {
-	T value;
+	const T value;
 	constexpr Return(T &&val) : value(std::forward<T>(val)) {}
 };
 
@@ -64,7 +65,7 @@ struct Return<void> {
  */
 template<typename... Args>
 struct Arguments {
-	std::tuple<Args...> values;
+	const std::tuple<Args...> values;
 	constexpr Arguments(Args &&...args) : values{std::forward<Args>(args)...} {}
 };
 
@@ -74,9 +75,9 @@ struct Arguments {
  */
 template<typename Ret, typename... Args>
 struct FunctionDefinition {
-	FunctionId id;
-	Return<Ret> returnType;
-	Arguments<Args...> argumentTypes;
+	const FunctionId id;
+	const Return<Ret> returnType;
+	const Arguments<Args...> argumentTypes;
 };
 
 
@@ -85,7 +86,7 @@ struct FunctionDefinition {
  */
 template<typename... Funcs>
 struct FunctionList {
-	std::tuple<Funcs...> functions;
+	const std::tuple<Funcs...> functions;
 	FunctionList(Funcs&&... funcs) : functions(std::forward<Funcs>(funcs)...) {}
 };
 
@@ -106,15 +107,21 @@ class AsyncFunctionExecutor {
 	AsyncFunctionExecutor(Config config,
 						  const FunctionList<Funcs...> &functions,
 						  std::unique_ptr<clients::ClientInterface> client = nullptr)
-			: client_(nullptr), config_(config), functions_(functions) {
+			: client_(nullptr), settings_(config.isProducer, config.defaultTimeout, config.functionConfigurations), functions_(functions) {
 		// Default client if none is provided
 		if (client) {
 			client_ = std::move(client);
 		} else {
 			client_ = std::make_unique<clients::AeronClient<TimeoutIdleStrategy>>(
-				"aeron:ipc", TimeoutIdleStrategy(config.defaultTimeout)
+				DEFAULT_AERON_CONNECTION, TimeoutIdleStrategy(settings_.defaultTimeout)
 			);
 		}
+
+		for (const auto& [funcId, _] : settings_.functionConfigs) {
+			if (!isFunctionDefined(FunctionId{funcId})) {
+				throw std::runtime_error("Warning: Function ID " + std::to_string(static_cast<int>(funcId)) + " in configuration is not defined in FunctionList.");
+			}
+		}
 	};
 
 	~AsyncFunctionExecutor() = default;
@@ -129,11 +136,11 @@ class AsyncFunctionExecutor {
 		std::vector<uint32_t> fromProducer;
 
 		std::apply([&](auto&&... funcDefs) {
-			(toProducer.push_back(funcDefs.id.value + 1000), ...);
+			(toProducer.push_back(funcDefs.id.value + MESSAGE_RETURN_CHANNEL_OFFSET), ...);
 			(fromProducer.push_back(funcDefs.id.value), ...);
 		}, std::get<0>(functions_.functions));
 
-		if (config_.isProducer) {
+		if (settings_.isProducer) {
 			client_->connect(toProducer, fromProducer);
 		} else {
 			client_->connect(fromProducer, toProducer);
@@ -152,17 +159,20 @@ class AsyncFunctionExecutor {
 	 */
 	template <typename Ret, typename... FArgs, typename... CallArgs>
 	Ret callFunc(const FunctionDefinition<Ret, FArgs...> &function, CallArgs&&... args) {
-		if (!config_.isProducer) {
+		if (!settings_.isProducer) {
 			throw std::runtime_error("Cannot call function in consumer mode");
 		}
 		if (sizeof...(CallArgs) < sizeof...(FArgs)) {
 			throw std::invalid_argument("Argument count mismatch");
 		}
+		if (!isFunctionDefined(function.id)) {
+			throw std::runtime_error("Function ID not defined");
+		}
 
 		auto messageBytes = serializeArgs(function.id, args...);
 		client_->sendMessage(function.id.value, messageBytes);
 
-		auto responseBytes = client_->waitForMessage(function.id.value + 1000);
+		auto responseBytes = client_->waitForMessage(function.id.value + MESSAGE_RETURN_CHANNEL_OFFSET);
 		if (responseBytes.empty()) {
 			throw std::runtime_error("No response received or timeout");
 		}
@@ -184,14 +194,14 @@ class AsyncFunctionExecutor {
 	 * @return A tuple containing the FunctionId and a span of argument bytes. Returns an empty tuple on error.
 	 */
 	std::tuple<FunctionId, std::span<const uint8_t>> pollFunction() {
-		if (config_.isProducer) {
+		if (settings_.isProducer) {
 			std::cerr << "Cannot start polling in producer mode." << std::endl;
-			return {}; // Error: Cannot start polling in producer mode
+			return std::make_tuple(FunctionId{}, std::span<const uint8_t>{}); // Error: Cannot start polling in producer mode
 		}
 
 		auto requestBytes = client_->waitForAnyMessage();
 		if (requestBytes.empty()) {
-			return {}; // Error: No message received or timeout
+			return std::make_tuple(FunctionId{}, std::span<const uint8_t>{}); // Error: No message received or timeout
 		}
 
 		auto [funcId, argBytes] = deserializeRequest(requestBytes);
@@ -209,10 +219,14 @@ class AsyncFunctionExecutor {
 	 */
 	template<typename Ret, typename... Args>
 	auto getFunctionArgs(const FunctionDefinition<Ret, Args...> &function, const std::span<const uint8_t> &argBytes) {
-		if (config_.isProducer) {
+		if (settings_.isProducer) {
 			throw std::runtime_error("Cannot get function arguments in producer mode");
 		}
 
+		if (!isFunctionDefined(function.id)) {
+			throw std::runtime_error("Function ID not defined");
+		}
+
 		if (argBytes.size() < 1) {
 			throw std::invalid_argument("Not enough data to read argument count");
 		}
@@ -227,7 +241,8 @@ class AsyncFunctionExecutor {
 		std::tuple<Args...> args;
 		auto extractArg = [&](auto &arg) {
 			if (pos >= argBytes.size()) throw std::invalid_argument("Unexpected end of data while reading argument size");
-			uint8_t len = argBytes[pos++];
+			uint16_t len = argBytes[pos] | (static_cast<uint16_t>(argBytes[pos + 1]) << 8);
+			pos += 2;
 			if (pos + len > argBytes.size()) throw std::invalid_argument("Unexpected end of data while reading argument content");
 
 			if constexpr (HasSerialize<decltype(arg)>) {
@@ -257,21 +272,34 @@ class AsyncFunctionExecutor {
 	 */
 	template<typename T>
 	int sendReturnMessage(const FunctionId &functionId, const T &returnValue) {
-		if (config_.isProducer) {
+		if (!isFunctionDefined(functionId)) {
+			std::cerr << "Function ID not defined." << std::endl;
+			return -1; // Error: Function ID not defined
+		}
+
+		if (settings_.isProducer) {
 			std::cerr << "Cannot send return message in producer mode." << std::endl;
 			return -1; // Error: Cannot send return message in producer mode
 		}
 
 		auto messageBytes = serializeReturn(functionId, returnValue);
-		return client_->sendMessage(functionId.value + 1000, messageBytes);
+		return client_->sendMessage(functionId.value + MESSAGE_RETURN_CHANNEL_OFFSET, messageBytes);
 	}
 
 private:
+	bool isFunctionDefined(const FunctionId &funcId) {
+		bool found = false;
+		std::apply([&](auto&&... funcDefs) {
+			((funcDefs.id.value == funcId.value ? found = true : false), ...);
+		}, std::get<0>(functions_.functions));
+		return found;
+	}
+
 	template<typename... Args>
 	std::span<const uint8_t> serializeArgs(const FunctionId &funcId, const Args&... args) {
 		serializationBuffer_.clear();
 		std::size_t totalSize = 2; // Function ID + Argument count
-		((totalSize += 1 + sizeof(args)), ...); // Each argument: size byte + data
+		((totalSize += 2 + sizeof(args)), ...); // Each argument: size byte + data
 		serializationBuffer_.reserve(totalSize);
 		serializationBuffer_.push_back(funcId.value);
 		serializationBuffer_.push_back(static_cast<uint8_t>(sizeof...(Args)));
@@ -283,7 +311,7 @@ class AsyncFunctionExecutor {
 	template<typename T>
 	std::span<const uint8_t> serializeReturn(const FunctionId &funcId, const T &returnValue) {
 		serializationBuffer_.clear();
-		std::size_t totalSize = 2 + sizeof(returnValue);
+		std::size_t totalSize = 3 + sizeof(returnValue);
 		serializationBuffer_.reserve(totalSize);
 		serializationBuffer_.push_back(funcId.value);
 		appendArg(serializationBuffer_, returnValue);
@@ -295,14 +323,18 @@ class AsyncFunctionExecutor {
 	void appendArg(std::vector<uint8_t>& buffer, const T& arg) {
 		if constexpr (HasSerialize<T>) {
 			auto bytes = arg.serialize();
-			if (bytes.size() > 255) {
+			if (bytes.size() > MAX_ARGUMENT_SIZE) {
 				throw std::invalid_argument("Serialized data too large");
 			}
-			buffer.push_back(static_cast<uint8_t>(bytes.size()));
+			uint16_t size = static_cast<uint16_t>(bytes.size());
+			buffer.push_back(static_cast<uint8_t>(size & 0xFF));
+			buffer.push_back(static_cast<uint8_t>((size >> 8) & 0xFF));
 			buffer.insert(buffer.end(), bytes.begin(), bytes.end());
 		} else {
 			static_assert(std::is_trivially_copyable_v<T>, "Argument type must be trivially copyable");
-			buffer.push_back(static_cast<uint8_t>(sizeof(arg)));
+			uint16_t size = static_cast<uint16_t>(sizeof(arg));
+			buffer.push_back(static_cast<uint8_t>(size & 0xFF));
+			buffer.push_back(static_cast<uint8_t>((size >> 8) & 0xFF));
 			buffer.insert(buffer.end(), reinterpret_cast<const uint8_t*>(&arg), reinterpret_cast<const uint8_t*>(&arg) + sizeof(T));
 		}
 	}
@@ -316,9 +348,9 @@ class AsyncFunctionExecutor {
 
 		T value;
 		if constexpr (HasSerialize<T>) {
-			value.deserialize(std::span {bytes.data() + 2, bytes.size() - 2});
+			value.deserialize(std::span {bytes.data() + 3, bytes.size() - 3});
 		} else {
-			std::memcpy(&value, bytes.data() + 2, sizeof(T));
+			std::memcpy(&value, bytes.data() + 3, sizeof(T));
 		}
 		return value;
 	}
@@ -337,7 +369,7 @@ class AsyncFunctionExecutor {
 	/// Buffer used for serialization of messages.
 	mutable std::vector<uint8_t> serializationBuffer_;
 	std::unique_ptr<clients::ClientInterface> client_;
-	Config config_;
+	structures::Settings settings_;
 	FunctionList<Funcs...> functions_;
 };
 

include/bringauto/async_function_execution/Constants.hpp

@@ -0,0 +1,15 @@
+#pragma once
+
+#include <string_view>
+
+
+
+namespace bringauto::async_function_execution {
+
+constexpr int POLL_FRAGMENTS_LIMIT = 10;
+constexpr int MESSAGE_RETURN_CHANNEL_OFFSET = 1000;
+constexpr int MAX_ARGUMENT_SIZE = 65535;
+
+constexpr std::string_view DEFAULT_AERON_CONNECTION = "aeron:ipc";
+	
+}