main.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_TELSUB_MAIN_HPP
#define RTCTK_TELSUB_MAIN_HPP
 
#include "ciiException.hpp"
#include "rtctk/componentFramework/exceptions.hpp"
#include <boost/interprocess/exceptions.hpp>
#include <memory>
#include <memory_resource>
 
#include <rtctk/componentFramework/rtcComponentMain.hpp>
#include <rtctk/telSub/businessLogic.hpp>
#include <rtctk/telSub/correlator.hpp>
#include <rtctk/telSub/ddsWaitSet.hpp>
#include <rtctk/telSub/shmPublisher.hpp>
 
namespace rtctk::telSub {
 
template <class UserTopicType, class DataBlender>
void Main(const rtctk::componentFramework::Args& args, DataBlender&& blender) {
    auto factory =
        [blender = std::forward<DataBlender>(blender)](
            const std::string& name,
            componentFramework::ServiceContainer& services) -> std::unique_ptr<BusinessLogic> {
        // Operational Logic factory.
        // This takes ownership of `blender` and instances created from factory will only receive a
        // reference. (TBD) (this requires a stateless DataBlender).
        // If DataBlender is stateful with configuration params then this means that new instances
        // should be created for each new instance of OperationalLogic?
        auto op_logic_factory =
            [&, blender = std::move(blender)](const OperationalLogicFactoryParams& params)
            -> std::unique_ptr<OperationalLogicIf> {
            // Allocate a block of memory used by Correlator and DdsWaitSet
            // The resource and memory ownership is then transferred to OperationalLogic which is
            // responsible for destroying them last.
            // clang-format off
            size_t size = (params.dds_params.m_topics.size() + 1u) * (
                           /* AgnosticDataSamples contains two vectors of dds sequences
                            owned by DdsWaitSet */
                           sizeof(DdsSampleSeq) +
                           sizeof(DdsInfoSeq) +
                           /* DataSamplesView contains vector of views owned by Correlator */
                           sizeof(DataSampleView) +
                           /* Vector of indices in Correlator */
                           sizeof(size_t));
            // clang-format on
            std::unique_ptr<std::byte[]> mem(new std::byte[size]);
            // Both DdsWaitSet and Correlator only allocates once so we can use a simple
            // monotonic_buffer_resource.
            auto resource = std::make_unique<std::pmr::monotonic_buffer_resource>(mem.get(), size);
            // Required by OperationalLogic
            auto& metrics = services.Get<componentFramework::ComponentMetricsIf>();
            auto& alerts = services.Get<componentFramework::AlertServiceIf>();
 
            auto dds_subscriber = std::make_unique<DdsWaitSet>(params.dds_params, resource.get());
            auto correlator = std::make_unique<Correlator>(
                params.correlator_params, std::move(dds_subscriber), metrics, resource.get());
            try {
                auto writer = ipcq::Writer<UserTopicType>(params.shm_params.topic_name.c_str(),
                                                          params.shm_params.capacity,
                                                          params.shm_params.mem_policy);
                auto shm_publisher = MakeShmPublisher<UserTopicType>(std::move(writer), blender);
                return std::make_unique<OperationalLogic>(params.operational_params,
                                                          std::move(correlator),
                                                          std::move(shm_publisher),
                                                          metrics,
                                                          alerts,
                                                          std::move(resource),
                                                          std::move(mem));
            } catch (boost::interprocess::interprocess_exception& e) {
                throw CII_MAKE_EXCEPTION_WITH_NESTED(
                    RtctkException,
                    e,
                    fmt::format("Failed to create SHM queue {}, does it already exist?",
                                params.shm_params.topic_name.c_str()));
            }
        };
 
        return std::make_unique<BusinessLogic>(name, services, std::move(op_logic_factory));
    };
    componentFramework::RunAsRtcComponent<BusinessLogic>(args, std::move(factory));
}
 
}  // namespace rtctk::telSub
 
#endif  // RTCTK_TELSUB_MAIN_HPP