Feature/SE integration

.gitignore

@@ -35,9 +35,11 @@
 .idea
 
 /build
+/build-ex
 /cmake-build-*
 
 /.build
 /.venv
 /.vcpkg
 /.build*
+/vcpkg_installed

CMakeLists.txt

@@ -15,6 +15,7 @@ set(CMAKE_CXX_STANDARD 20)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
 set(CMAKE_CXX_EXTENSIONS OFF)
 set(CMAKE_POSITION_INDEPENDENT_CODE ON)
+set(CMAKE_EXPERIMENTAL_CXX_MODULE_CMAKE_API ON)
 
 project(cpp-jam
     VERSION 0.0.1
@@ -75,6 +76,12 @@ find_package(Boost.DI CONFIG REQUIRED)
 find_package(qtils CONFIG REQUIRED)
 find_package(prometheus-cpp CONFIG REQUIRED)
 
+if(CMAKE_CXX_COMPILER_ID STREQUAL "GNU")
+  add_compile_options(-fmodules-ts)
+elseif(CMAKE_CXX_COMPILER_ID MATCHES "Clang|AppleClang")
+  add_compile_options(-fmodules)
+endif()
+
 add_library(headers INTERFACE)
 target_include_directories(headers INTERFACE
     $<BUILD_INTERFACE:${PROJECT_SOURCE_DIR}/src_>

src/CMakeLists.txt

@@ -24,3 +24,8 @@ add_subdirectory(metrics)
 # Clocks and time subsystem
 add_subdirectory(clock)
 
+# Subscription Engine subsystem
+add_subdirectory(se)
+
+# Modules subsystem
+add_subdirectory(modules)

src/app/CMakeLists.txt

@@ -28,7 +28,9 @@ add_library(build_version
 
 add_library(app_configuration SHARED configuration.cpp)
 target_link_libraries(app_configuration
-    Boost::boost)
+    Boost::boost
+    fmt::fmt
+)
 
 add_library(app_configurator SHARED configurator.cpp)
 target_link_libraries(app_configurator

src/app/configuration.hpp

@@ -5,6 +5,7 @@
 
 #pragma once
 
+#include <filesystem>
 #include <string>
 
 #include <boost/asio/ip/tcp.hpp>
@@ -22,6 +23,7 @@ namespace jam::app {
 
     [[nodiscard]] std::string nodeVersion() const;
     [[nodiscard]] std::string nodeName() const;
+
     [[nodiscard]] std::optional<Endpoint> metricsEndpoint() const;
 
    private:

src/app/configurator.cpp

@@ -3,20 +3,19 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-#include "app/configuration.hpp"
+#include "app/configurator.hpp"
 
 #include <filesystem>
 #include <iostream>
 
 #include <boost/asio/ip/tcp.hpp>
+#include <boost/beast/core/error.hpp>
 #include <boost/program_options.hpp>
 #include <boost/program_options/value_semantic.hpp>
 #include <qtils/outcome.hpp>
 
 #include "app/build_version.hpp"
-#include "app/configurator.hpp"
-
-#include <boost/beast/core/error.hpp>
+#include "app/configuration.hpp"
 
 using Endpoint = boost::asio::ip::tcp::endpoint;
 
@@ -110,7 +109,7 @@ namespace jam::app {
         .add(metrics_options);
   }
 
-  outcome::result<bool> Configurator::step1() {
+  outcome::result<bool> Configurator::step1() {  // read min cli-args and config
     namespace po = boost::program_options;
     namespace fs = std::filesystem;
 
@@ -120,7 +119,7 @@ namespace jam::app {
 
     po::variables_map vm;
 
-    // first-run parse to read only general options and to lookup for "help",
+    // first-run parse to read-only general options and to lookup for "help",
     // "config" and "version". all the rest options are ignored
     try {
       po::parsed_options parsed = po::command_line_parser(argc_, argv_)
@@ -200,6 +199,7 @@ namespace jam::app {
   }
 
   outcome::result<void> Configurator::initGeneralConfig() {
+    // Init by config-file
     if (config_file_.has_value()) {
       auto section = (*config_file_)["general"];
       if (section.IsDefined()) {

src/app/configurator.hpp

@@ -5,6 +5,8 @@
 
 #pragma once
 
+#include <optional>
+
 #include <boost/program_options.hpp>
 #include <qtils/enum_error_code.hpp>
 #include <qtils/outcome.hpp>

src/app/impl/application_impl.cpp

@@ -16,16 +16,24 @@
 #include "log/logger.hpp"
 #include "metrics/histogram_timer.hpp"
 #include "metrics/metrics.hpp"
+#include "se/impl/subscription_manager.hpp"
 
 namespace jam::app {
 
+  SeHolder::SeHolder(SePtr se) : se_(std::move(se)) {}
+
+  SeHolder::~SeHolder() {
+    se_->dispose();
+  }
+
   ApplicationImpl::ApplicationImpl(
       std::shared_ptr<log::LoggingSystem> logsys,
       std::shared_ptr<Configuration> config,
       std::shared_ptr<StateManager> state_manager,
       std::shared_ptr<Watchdog> watchdog,
       std::shared_ptr<metrics::Exposer> metrics_exposer,
-      std::shared_ptr<clock::SystemClock> system_clock)
+      std::shared_ptr<clock::SystemClock> system_clock,
+      std::shared_ptr<SeHolder>)
       : logger_(logsys->getLogger("Application", "application")),
         app_config_(std::move(config)),
         state_manager_(std::move(state_manager)),

src/app/impl/application_impl.hpp

@@ -6,12 +6,13 @@
 
 #pragma once
 
-#include "app/application.hpp"
-
 #include <memory>
 
 #include <metrics/registry.hpp>
 
+#include "app/application.hpp"
+#include "se/subscription_fwd.hpp"
+
 namespace jam {
   class Watchdog;
 }  // namespace jam
@@ -41,14 +42,46 @@ namespace jam::metrics {
 
 namespace jam::app {
 
+  /**
+   * @brief RAII holder for subscription engine management
+   *
+   * SeHolder is responsible for managing the lifetime of subscription engine
+   * components. It ensures proper initialization and cleanup of the
+   * subscription system during application lifecycle.
+   */
+  struct SeHolder final {
+    using SePtr = std::shared_ptr<Subscription>;
+    SePtr se_;
+
+    // Disable copying - subscription engine should not be copied
+    SeHolder(const SeHolder &) = delete;
+    SeHolder &operator=(const SeHolder &) = delete;
+
+    // Disable moving - subscription engine should not be moved
+    SeHolder(SeHolder &&) = delete;
+    SeHolder &operator=(SeHolder &&) = delete;
+
+    /**
+     * @brief Constructs SeHolder with subscription engine instance
+     * @param se Shared pointer to subscription engine
+     */
+    SeHolder(SePtr se);
+
+    /**
+     * @brief Destructor ensures proper cleanup of subscription engine
+     */
+    ~SeHolder();
+  };
+
   class ApplicationImpl final : public Application {
    public:
     ApplicationImpl(std::shared_ptr<log::LoggingSystem> logsys,
                     std::shared_ptr<Configuration> config,
                     std::shared_ptr<StateManager> state_manager,
                     std::shared_ptr<Watchdog> watchdog,
                     std::shared_ptr<metrics::Exposer> metrics_exposer,
-                    std::shared_ptr<clock::SystemClock> system_clock);
+                    std::shared_ptr<clock::SystemClock> system_clock,
+                    std::shared_ptr<SeHolder>);
 
     void run() override;
 

src/app/impl/state_manager_impl.hpp

@@ -6,13 +6,12 @@
 
 #pragma once
 
-#include "app/state_manager.hpp"
-
 #include <condition_variable>
 #include <csignal>
 #include <mutex>
 #include <queue>
 
+#include "app/state_manager.hpp"
 #include "utils/ctor_limiters.hpp"
 
 namespace soralog {

src/executable/CMakeLists.txt

@@ -35,6 +35,7 @@ else ()
   add_executable(jam_node jam_node.cpp)
   target_link_libraries(jam_node ${LIBRARIES})
 endif ()
+add_dependencies(jam_node all_modules)
 
 #if (BACKWARD)
 #  add_backward(jam_node)

src/executable/jam_node.cpp

@@ -4,18 +4,24 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
+#include <chrono>
 #include <iostream>
+#include <memory>
+#include <optional>
+#include <variant>
 
+#include <qtils/final_action.hpp>
 #include <soralog/impl/configurator_from_yaml.hpp>
 #include <soralog/logging_system.hpp>
 
-#include <qtils/final_action.hpp>
-
 #include "app/application.hpp"
 #include "app/configuration.hpp"
 #include "app/configurator.hpp"
 #include "injector/node_injector.hpp"
+#include "loaders/impl/example_loader.hpp"
 #include "log/logger.hpp"
+#include "modules/module_loader.hpp"
+#include "se/subscription.hpp"
 
 using std::string_view_literals::operator""sv;
 
@@ -35,11 +41,29 @@ namespace {
   int run_node(std::shared_ptr<LoggingSystem> logsys,
                std::shared_ptr<Configuration> appcfg) {
     auto injector = std::make_unique<NodeInjector>(logsys, appcfg);
+    // Load modules
+    std::deque<std::unique_ptr<jam::loaders::Loader>> loaders;
+    {
+      auto logger = logsys->getLogger("Modules", "jam");
+      const std::string path("modules");
+
+      jam::modules::ModuleLoader module_loader(path);
+      auto modules = module_loader.get_modules();
+      if (modules.has_error()) {
+        SL_CRITICAL(logger, "Failed to load modules from path: {}", path);
+        return EXIT_FAILURE;
+      }
+
+      for (const auto &module : modules.value()) {
+        auto loader = injector->register_loader(module);
+        if (loader) {
+          loaders.emplace_back(std::move(loader));
+        }
+      }
+    }
 
     auto logger = logsys->getLogger("Main", jam::log::defaultGroupName);
-
     auto app = injector->injectApplication();
-
     SL_INFO(logger, "Node started. Version: {} ", appcfg->nodeVersion());
 
     app->run();

src/injector/CMakeLists.txt

@@ -15,4 +15,6 @@ target_link_libraries(node_injector
     application
     metrics
     clock
+    se_async
+    modules
 )

src/injector/node_injector.cpp

@@ -9,18 +9,28 @@
 
 #include "injector/node_injector.hpp"
 
+#include <memory>
+#include <stdexcept>
+#include <string>
+#include <vector>
+
 #include <boost/di.hpp>
 #include <boost/di/extension/scopes/shared.hpp>
+#include <loaders/impl/example_loader.hpp>
 
 #include "app/configuration.hpp"
 #include "app/impl/application_impl.hpp"
 #include "app/impl/state_manager_impl.hpp"
 #include "app/impl/watchdog.hpp"
 #include "clock/impl/clock_impl.hpp"
 #include "injector/bind_by_lambda.hpp"
+#include "loaders/loader.hpp"
 #include "log/logger.hpp"
 #include "metrics/impl/exposer_impl.hpp"
 #include "metrics/impl/prometheus/handler_impl.hpp"
+#include "modules/module.hpp"
+#include "se/impl/async_dispatcher_impl.hpp"
+#include "se/subscription_fwd.hpp"
 
 namespace {
   namespace di = boost::di;
@@ -29,7 +39,7 @@ namespace {
 
   template <typename C>
   auto useConfig(C c) {
-    return boost::di::bind<std::decay_t<C> >().to(
+    return boost::di::bind<std::decay_t<C>>().to(
         std::move(c))[boost::di::override];
   }
 
@@ -50,6 +60,7 @@ namespace {
         di::bind<log::LoggingSystem>.to(logsys),
         di::bind<metrics::Handler>.to<metrics::PrometheusHandler>(),
         di::bind<metrics::Exposer>.to<metrics::ExposerImpl>(),
+        di::bind<Dispatcher>.to<se::AsyncDispatcher<SubscriptionEngineHandlers::kTotalCount, kThreadPoolSize>>(),
         di::bind<metrics::Exposer::Configuration>.to([](const auto &injector) {
           return metrics::Exposer::Configuration{
               {boost::asio::ip::address_v4::from_string("127.0.0.1"), 7777}
@@ -92,10 +103,40 @@ namespace jam::injector {
   NodeInjector::NodeInjector(std::shared_ptr<log::LoggingSystem> logsys,
                              std::shared_ptr<app::Configuration> config)
       : pimpl_{std::make_unique<NodeInjectorImpl>(
-          makeNodeInjector(std::move(logsys), std::move(config)))} {}
+            makeNodeInjector(std::move(logsys), std::move(config)))} {}
 
   std::shared_ptr<app::Application> NodeInjector::injectApplication() {
     return pimpl_->injector_
-        .template create<std::shared_ptr<app::Application> >();
+        .template create<std::shared_ptr<app::Application>>();
+  }
+
+  std::unique_ptr<jam::loaders::Loader> NodeInjector::register_loader(
+      std::shared_ptr<modules::Module> module) {
+    auto logsys = pimpl_->injector_
+                      .template create<std::shared_ptr<log::LoggingSystem>>();
+    auto logger = logsys->getLogger("Modules", "jam");
+
+    if ("ExampleLoader" == module->get_loader_id()) {
+      auto loader =
+          pimpl_->injector_
+              .template create<std::unique_ptr<jam::loaders::ExampleLoader>>();
+      loader->start(module);
+
+      if (auto info = loader->module_info()) {
+        SL_INFO(logger, "> Module: {} [{}]", *info, module->get_path());
+      } else {
+        SL_ERROR(logger,
+                 "> No module info for: {} [{}]",
+                 module->get_loader_id(),
+                 module->get_path());
+      }
+      return std::unique_ptr<jam::loaders::Loader>(loader.release());
+    }
+
+    SL_CRITICAL(logger,
+                "> No loader found for: {} [{}]",
+                module->get_loader_id(),
+                module->get_path());
+    return {};
   }
 }  // namespace jam::injector