main.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup server
 * @copyright ESO - European Southern Observatory
 * @author
 *
 * @brief main source file.
 */
 
#include "actionMgr.hpp"
#include "actionsStd.hpp"
#include "dataContext.hpp"
#include "dbInterface.hpp"
#include "logger.hpp"
#include "ocmDaqService.hpp"
#include "stdCmdsImpl.hpp"
 
#include <Stdif.hpp>
#include <events.rad.hpp>
 
#include <rad/actionCallback.hpp>
#include <rad/dbAdapterRedis.hpp>
#include <rad/exceptions.hpp>
#include <rad/mal/replier.hpp>
#include <rad/mal/utils.hpp>
#include <rad/mal/publisher.hpp>
#include <rad/smAdapter.hpp>
 
#include <scxml4cpp/Context.h>
#include <scxml4cpp/EventQueue.h>
 
#include <boost/asio.hpp>
#include <boost/exception/diagnostic_information.hpp>
#include <boost/filesystem.hpp>
 
#include <fmt/format.h>
 
#include <functional>
#include <memory>
 
#include <daq/manager.hpp>
#include <daq/daqController.hpp>
#include <ocmif/conversion.hpp>
#include <ocmif/uri.hpp>
 
 
class DaqService {
public:
    DaqService(std::string name,
               std::string instrument,
               std::string const& output_path,
               rad::IoExecutor& executor,
               std::shared_ptr<mal::Mal> mal,
               rad::cii::Replier& replier,
               rad::cii::Publisher<ocmif::DaqStatus>& publisher,
               rad::SMAdapter& state_machine)
        : m_name(std::move(name))
        , m_manager(executor, std::move(instrument))
        , m_mal(std::move(mal))
        , m_service(std::make_shared<OcmDaqService>(
              executor.get_io_context(), *m_mal, m_manager, output_path))
        , m_replier(replier)
        , m_publisher(publisher)
        , m_state_machine(state_machine)
        , m_is_active(false) {
        m_sample = m_publisher.CreateTopic();
        m_connection = m_manager.GetStatusSignal().ConnectObserver(
            [this](auto const& status) { this->DaqStatusUpdate(status); });
    }
    void Register() {
        m_replier.RegisterService(m_name, std::static_pointer_cast<ocmif::AsyncOcmDaq>(m_service));
    }
    void Unregister() {
        m_replier.UnregisterService(m_name);
    }
 
    /**
     * Is notified of any DAQ status change and will post events to SM on DAQ activity flank
     * changes.
     *
     * @note Current implementation assumes all DAQs begin in state NotStarted. When ocm supports
     * loading old status from persistent storage this will no longer work as number of active daqs
     * is not starting at zero and state is not starting at NotStarted.
     */
    void DaqStatusUpdate(daq::ObservableStatus const& status) {
        try {
            LOG4CPLUS_DEBUG(server::GetLogger(),
                            fmt::format("Publishing DAQ status for DAQ {}", status.GetId()));
            auto state = status.GetState();
            if (state == daq::State::NotStarted && !m_is_active) {
                // New data acquisition
                m_is_active = true;
                m_state_machine.ProcessEvent(Events::AnyDaqActive{});
            } else if (m_is_active && daq::IsFinalState(status.GetState())) {
                // At least one DAQ is completed. Check if all DAQs are completed.
                auto daqs = m_manager.GetDaqControllers();
                auto all_final =
                    std::all_of(daqs.begin(), daqs.end(), [](auto const shrd_daq) -> bool {
                        assert(shrd_daq);
                        return daq::IsFinalState(shrd_daq->GetState());
                    });
                if (all_final) {
                    m_is_active = false;
                    // Active -> Idle
                    m_state_machine.ProcessEvent(Events::AllDaqInactive{});
                }
            }
            // publish state
            *m_sample << status.GetStatus();
            m_publisher.Publish(*m_sample);
        } catch (std::exception const& e) {
            LOG4CPLUS_ERROR(server::GetLogger(),
                            fmt::format("Failed to publish status: {}", e.what()));
        }
    }
 
private:
    std::string m_name;
    daq::ManagerImpl m_manager;
    std::shared_ptr<mal::Mal> m_mal;
    std::shared_ptr<OcmDaqService> m_service;
    rad::cii::Replier& m_replier;
    rad::cii::Publisher<ocmif::DaqStatus>& m_publisher;
    rad::SMAdapter& m_state_machine;
    bool m_is_active;
 
    boost::signals2::connection m_connection;
    std::shared_ptr<ocmif::DaqStatus> m_sample;
};
 
 
/**
 * Application main.
 *
 * @param[in] argc Number of command line options.
 * @param[in] argv Command line options.
 */
int main(int argc, char* argv[]) {
    namespace fs = boost::filesystem;
 
    auto log_initializer = rad::LogInitializer();
    LOG4CPLUS_INFO(server::GetLogger(), "Application ocm-server started.");
 
    try {
        /*
         *  Load CII/MAL middleware here because it resets
         *  the log4cplus configuration!
         */
        elt::mal::CiiFactory& factory = elt::mal::CiiFactory::getInstance();
        auto mal = elt::mal::loadMal("zpb", {});
        factory.registerMal("zpb", mal);
 
        /* Read only configuration */
        server::Config config;
        if (config.ParseOptions(argc, argv) == false) {
            // request for help
            return EXIT_SUCCESS;
        }
        config.LoadConfig();
        log_initializer.Configure(rad::Helper::FindFile(config.GetLogProperties()));
 
        if (config.m_out_path.empty()) {
            auto const msg = "Output path not configured. Define environment "
                             "variable $DATAROOT or configuration cfg.dataroot";
            LOG4CPLUS_ERROR(server::GetLogger(), msg);
            std::cerr << msg << std::endl;
            return -1;
        }
        // Create out path if necessary with permissive permissions
        if (!fs::is_directory(config.m_out_path)) {
            LOG4CPLUS_INFO(server::GetLogger(),
                           fmt::format("Output dir '{}' is not a directory. Creating it now",
                                       config.m_out_path));
            fs::create_directories(config.m_out_path);
            fs::permissions(config.m_out_path, fs::others_read | fs::owner_all | fs::group_all);
        }
 
 
        /*
         * LAN 2020-07-09 EICSSW-717
         * Create CII/MAL replier as soon as possible to avoid problems when
         * an exceptions is thrown from and Action/Guard.
         */
        rad::cii::Replier mal_replier(ocmif::MakeServerUri(config.GetMsgReplierEndpoint()));
        auto std_status_publisher =
                std::make_unique<rad::cii::Publisher<stdif::StatusTopic>>(
                                    ocmif::MakeServiceUri(config.GetPubEndpoint(), "std/status"));
        auto daq_status_publisher =
                std::make_unique<rad::cii::Publisher<ocmif::DaqStatus>>(
                                    ocmif::MakeServiceUri(config.GetPubEndpoint(), "daq/status"));
 
        /* Runtime DB */
        rad::DbAdapterRedis redis_db;
 
        /* Runtime data context */
        server::DataContext data_ctx(config, redis_db);
 
        /*
         * Create event loop
         */
 
        // event loop
        boost::asio::io_context io_ctx;
        rad::IoExecutor executor(io_ctx);
 
        /*
         * State Machine related objects
         */
        // SM event queue and context
        scxml4cpp::EventQueue external_events;
        scxml4cpp::Context state_machine_ctx;
 
        // State Machine facade
        rad::SMAdapter state_machine(io_ctx, &state_machine_ctx, external_events);
 
        DaqService daq_service("daq",
                               config.m_instrument_id,
                               config.m_out_path,
                               executor,
                               mal,
                               mal_replier,
                               *daq_status_publisher,
                               state_machine);
 
 
        // actions and activities
        server::ActionMgr action_mgr;
        action_mgr.CreateActions(io_ctx, state_machine, data_ctx);
        action_mgr.CreateActivities(state_machine, data_ctx);
        action_mgr.AddAction(new rad::ActionCallback("OcmDaq.RegisterService",
                                                     [&](auto) { daq_service.Register(); }));
        action_mgr.AddAction(new rad::ActionCallback("OcmDaq.UnregisterService",
                                                     [&](auto) { daq_service.Unregister(); }));
 
        // Load SM model
        state_machine.Load(
            config.GetSmScxmlFilename(), &action_mgr.GetActions(), &action_mgr.GetActivities());
 
        // Register handlers to reject events
        state_machine.RegisterDefaultRequestRejectHandler<Events::Init>();
        state_machine.RegisterDefaultRequestRejectHandler<Events::Enable>();
        state_machine.RegisterDefaultRequestRejectHandler<Events::Disable>();
 
        // Register publisher to export state information
        state_machine.SetStatusPublisher([&](std::string const& status) {
            auto ics_status = server::MakeStatusString(state_machine.GetActiveStates())
                .value_or("NotOperational;Undefined");
 
            auto sample = std_status_publisher->CreateTopic();
            sample->setStatus(ics_status);
            std_status_publisher->Publish(*sample);
            data_ctx.GetDbInterface().SetControlState(status);
        });
        /*
         * Register CII/MAL replier
         */
        mal_replier.RegisterService<stdif::AsyncStdCmds>(
            "std", std::make_shared<server::StdCmdsImpl>(state_machine));
 
        /*
         * Start event loop
         */
        state_machine.Start();
        io_ctx.run();
        state_machine.Stop();
    } catch (rad::Exception& e) {
        LOG4CPLUS_ERROR(server::GetLogger(), e.what());
        return EXIT_FAILURE;
    } catch (...) {
        LOG4CPLUS_ERROR(server::GetLogger(), boost::current_exception_diagnostic_information());
        return EXIT_FAILURE;
    }
 
    // to avoid valgrind warnings on potential memory loss
    //google::protobuf::ShutdownProtobufLibrary();
 
    LOG4CPLUS_INFO(server::GetLogger(), "Application ocm-server terminated.");
    return EXIT_SUCCESS;
}