rtcComponentMain.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_COMPONENTFRAMEWORK_RTCCOMPONENTMAIN_HPP
#define RTCTK_COMPONENTFRAMEWORK_RTCCOMPONENTMAIN_HPP
 
#include <exception>
#include <rtctk/componentFramework/alertService.hpp>
#include <rtctk/componentFramework/commandReplier.hpp>
#include <rtctk/componentFramework/componentMetrics.hpp>
#include <rtctk/componentFramework/exceptions.hpp>
#include <rtctk/componentFramework/logger.hpp>
#include <rtctk/componentFramework/malDdsEventService.hpp>
#include <rtctk/componentFramework/oldbIf.hpp>
#include <rtctk/componentFramework/rtcComponentArgs.hpp>
#include <rtctk/componentFramework/runtimeRepoIf.hpp>
#include <rtctk/componentFramework/serviceContainer.hpp>
#include <rtctk/componentFramework/serviceDiscovery.hpp>
#include <rtctk/componentFramework/serviceRegistryIf.hpp>
#include <rtctk/componentFramework/stateMachineEngine.hpp>
#include <rtctk/componentFramework/statePublisher.hpp>
#include <rtctk/componentFramework/telegrafAdapter.hpp>
 
#include <mal/Cii.hpp>
 
#include <fmt/format.h>
 
#include <boost/core/demangle.hpp>
 
#include <boost/stacktrace.hpp>
#include <iostream>
#include <memory>
#include <string>

void RtcComponentMain(const rtctk::componentFramework::Args& args);
 
namespace rtctk::componentFramework {

template <class BL, class BLF>
void RunAsRtcComponent(const Args& args, BLF factory) {
    try {
        auto& logger = GetLogger("rtctk");
 
        auto component_name = args.GetComponentName();
        LOG4CPLUS_INFO(logger, "RtcComponent '" << component_name << "' started.");
        using boost::core::demangle;
        LOG4CPLUS_DEBUG(logger, "BusinessLogic '" << demangle(typeid(BL).name()) << "'");
 
        {
            ServiceContainer services;
 
            // Service Discovery
            auto svc_disc_endpoint = args.GetServiceDiscEndpoint();
            services.Add<ServiceDiscovery>(
                std::make_unique<ServiceDiscovery>(svc_disc_endpoint, component_name));
            ServiceDiscovery& svc_disc = services.Get<ServiceDiscovery>();
 
            // Runtime repository
            auto rtr_endpoint = svc_disc.GetRuntimeRepoEndpoint();
            auto rtr_adapter = std::shared_ptr(RuntimeRepoIf::CreateAdapter(rtr_endpoint));
            services.Add<RuntimeRepoIf>(rtr_adapter);
 
            // OLDB
            auto oldb_endpoint = svc_disc.GetOldbEndpoint();
            auto oldb_adapter = std::shared_ptr(OldbIf::CreateAdapter(oldb_endpoint));
            services.Add<OldbIf>(oldb_adapter);
 
            // Telegraf
            auto telegraf_endpoint = svc_disc.GetTelegrafEndpoint();
            auto telegraf_adapter =
                std::make_shared<TelegrafAdapter>(telegraf_endpoint, component_name);
            services.Add<TelegrafAdapter>(telegraf_adapter);
 
            // Telemetry service
            auto component_metrics = std::make_shared<ComponentMetrics>(
                *oldb_adapter,
                DataPointPath(fmt::format("/{}/metrics", component_name)),
                *rtr_adapter,
                DataPointPath(fmt::format("/{}/dynamic/metrics", component_name)),
                GetLogger("rtctk.metrics"),
                telegraf_adapter.get());
            services.Add<ComponentMetricsIf>(component_metrics);
 
            // Event Service
            auto event_service = std::make_shared<MalDdsEventService>();
            services.Add<EventServiceIf>(event_service);
 
            // Alert service and standard observers
            auto alert_service = std::make_shared<AlertService>();
            alert_service->AddObserver(std::make_unique<AlertLogger>());
            alert_service->AddObserver(
                std::make_unique<AlertOldbPublisher>(component_name, *oldb_adapter));
            alert_service->AddObserver(
                std::make_unique<AlertEventPublisher>(component_name, *event_service));
            alert_service->AddObserver(
                std::make_unique<AlertTelegrafPublisher>(component_name, *telegraf_adapter));
            services.Add<AlertServiceIf>(alert_service);
            // for RR endpoint we do not want to be checked if the process is there or not (as we
            // are just starting it up)
            auto rr_endpoint =
                svc_disc.GetReqRepEndpoint("", ServiceRegistryIf::COMPONENT_TYPE, false);
            CommandReplier replier(rr_endpoint);
 
            // ... and sam for PubSub endpoint we do not want to be checked if the process is there
            // or not (as we are just starting it up)
            auto ps_endpoint =
                svc_disc.GetPubSubEndpoint("", ServiceRegistryIf::COMPONENT_TYPE, false);
            StatePublisher publisher(ps_endpoint, component_name);
 
            StateMachineEngine engine;
 
            engine.RegisterStateChangeHandler([&](const std::string& state) {
                try {
                    LOG4CPLUS_DEBUG(logger, "Entering State: " << state);
                    publisher.PublishState(state);
 
                    // save to OLDB
                    OldbIf& oldb = services.Get<OldbIf>();
                    DataPointPath dp_name = DataPointPath("/" + component_name + "/state");
                    if (not oldb.DataPointExists(dp_name)) {
                        oldb.CreateDataPoint<std::string>(dp_name);
                    }
                    oldb.SetDataPoint(dp_name, state);
 
                    // save to InfluxDB
                    TelegrafAdapter& telegraf = services.Get<TelegrafAdapter>();
                    auto points =
                        ParseStateToInfluxPoints(component_name, state, telegraf.BaseTags());
                    telegraf.SendPoints(points);
                } catch (...) {
                    LOG4CPLUS_ERROR(logger,
                                    NestedExceptionPrinter(WrapWithNested(
                                        RtctkException("StateChangeHandler failed"))));
                }
            });
 
            static_assert(std::is_base_of_v<typename BL::ComponentType::BizLogicIf, BL>,
                          "BusinessLogic must implement BusinessLogicInterface");
 
            static_assert(
                std::is_invocable_v<BLF, const std::string&, ServiceContainer&>,
                "Factory must be invocable with 'const std::string&' and 'ServiceContainer&'");
 
            static_assert(
                std::is_same_v<std::invoke_result_t<BLF, const std::string&, ServiceContainer&>,
                               std::unique_ptr<BL>>,
                "Factory must return type 'std::unique_ptr<BusinessLogic>'");
 
            // Invoking user factory here -> catch and rethrow RtctkException
            std::unique_ptr<BL> biz_logic;
            try {
                biz_logic = invoke(factory, component_name, services);
                if (!biz_logic) {
                    CII_THROW(RtctkException,
                              "BusinessLogic factory did not return a valid object");
                }
            } catch (const std::exception& ex) {
                CII_THROW_WITH_NESTED(RtctkException, ex, "BusinessLogic factory failed");
            }
 
            typename BL::ComponentType::InputStage input_stage(replier, engine);
 
            typename BL::ComponentType::OutputStage output_stage(engine, *biz_logic);
 
            typename BL::ComponentType::ModelBuilder model_builder(engine);
 
            if (auto emf = args.GetModelExportFile(); emf) {
                model_builder.ExportModel(emf->path().to_string());
            }
 
            model_builder.RegisterModel();
            input_stage.Start();
            component_metrics->StartMonitoring();
            engine.Work();
        }
 
        LOG4CPLUS_INFO(logger, "RtcComponent '" << component_name << "' terminated.");
 
    } catch (const std::exception& ex) {
        CII_THROW_WITH_NESTED(RtctkException, ex, "RunAsRtcComponent() failed");
    }
}

template <class BL>
void RunAsRtcComponent(const Args& args) {
    static_assert(
        std::is_constructible_v<BL, const std::string&, ServiceContainer&>,
        "BusinessLogic must be constructible with 'const std::string&' and 'ServiceContainer&'");
 
    RunAsRtcComponent<BL>(args, std::make_unique<BL, const std::string&, ServiceContainer&>);
}
 
}  // namespace rtctk::componentFramework
 
#ifndef RTCTK_NOMAIN
namespace rtctk::componentFramework {

int Main(int argc, char* argv[]) {
    LogInitializer initializer;
    Args args{argc, argv};
 
    try {
        args.Parse();
    } catch (const CLI::ParseError& e) {
        return args.PrintHelpOrErrMsg(e);
    } catch (const std::exception& e) {
        fmt::print(stderr, "RtcComponentMain failed during argument parsing: {}\n", e.what());
        return EXIT_FAILURE;
    } catch (...) {
        fmt::print(stderr, "{}", NestedExceptionPrinter(std::current_exception()).Str());
        return EXIT_FAILURE;
    }
 
    try {
        auto name = args.GetComponentName();
        auto log_props_file = args.GetLogPropsFile();
 
        if (log_props_file.has_value()) {
            LogConfigure(name, *log_props_file);
        } else {
            auto debug_mode = args.GetDebugMode();
            auto level = debug_mode ? log4cplus::DEBUG_LOG_LEVEL : log4cplus::INFO_LOG_LEVEL;
            LogConfigure(name, level);
        }
    } catch (const std::exception& e) {
        fmt::print(stderr, "Failed to initialise log system: {}\n", e.what());
        return EXIT_FAILURE;
    }
 
    try {
        // Call user implementation of RtcComponentMain
        ::RtcComponentMain(args);
 
        return EXIT_SUCCESS;
 
    } catch (const std::exception& e) {
        LOG4CPLUS_FATAL(GetLogger("rtctk"), NestedExceptionPrinter(e));
    } catch (...) {
        LOG4CPLUS_FATAL(GetLogger("rtctk"), NestedExceptionPrinter(std::current_exception()));
    }
    return EXIT_FAILURE;
}
 
}  // namespace rtctk::componentFramework

int main(int argc, char* argv[]) {
    std::set_terminate([]() {
        using namespace rtctk::componentFramework;
 
        std::exception_ptr eptr{std::current_exception()};
        if (eptr) {
            LOG4CPLUS_FATAL(GetLogger("rtctk"),
                            "Unhandled Exception caught: "
                                << NestedExceptionPrinter(eptr) << "\n\nBacktrace:\n"
                                << boost::to_string(boost::stacktrace::stacktrace()));
        } else {
            LOG4CPLUS_FATAL(GetLogger("rtctk"),
                            "Exiting without exception.\nBacktrace:\n"
                                << boost::to_string(boost::stacktrace::stacktrace()));
        }
        std::abort();
    });  // std::set_terminate
 
    try {
        return rtctk::componentFramework::Main(argc, argv);
    } catch (...) {
        // Note: Use minimal formatting to avoid further exception leaks.
        // This handler is only here to avoid std::terminate due to exception-leaks.
        constexpr std::string_view msg =
            "Caught exception leaked from rtctk::componentFramework::Main\n";
        std::cerr.write(msg.data(), msg.size());
    }
}
#endif  // RTCTK_NOMAIN
 
#endif  // RTCTK_COMPONENTFRAMEWORK_RTCCOMPONENTMAIN_HPP