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 <ciiLogManager.hpp>
#include <rad/actionCallback.hpp>
#include <rad/exceptions.hpp>
#include <rad/logger.hpp>
#include <rad/mal/publisher.hpp>
#include <rad/mal/replier.hpp>
#include <rad/mal/utils.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 <log4cplus/hierarchy.h>
 
#include <fmt/format.h>
 
#include <functional>
#include <memory>
 
#include <daq/conversion.hpp>
#include <daq/daqController.hpp>
#include <daq/dpmClient.hpp>
#include <daq/error/report.hpp>
#include <daq/manager.hpp>
#include <daq/workspace.hpp>
#include <daqif/uri.hpp>
 
class DaqService {
public:
    DaqService(std::string name,
               daq::ManagerParams manager_params,
               std::filesystem::path const& workspace,
               std::string const& output_path,
               rad::IoExecutor& executor,
               std::shared_ptr<mal::Mal> mal,
               rad::cii::Replier& replier,
               rad::cii::Publisher<daqif::DaqStatus>& publisher,
               rad::SMAdapter& state_machine,
               daq::DpmClientParams const& dpm_params)
        : m_name(std::move(name))
        , m_workspace(workspace)
        , m_mal(std::move(mal))
        , m_event_log(std::make_shared<daq::ObservableEventLog>())
        , m_dpm_client(
              std::make_shared<daq::DpmClientImpl>(executor.get_io_context(), *m_mal, dpm_params))
        , m_daq_factory(executor.get_io_context(), *m_mal, m_dpm_client)
        , m_manager(executor,
                    std::move(manager_params),
                    m_workspace,
                    m_event_log,
                    m_daq_factory,
                    log4cplus::Logger::getInstance(server::LOGGER_NAME_MANAGER))
        , m_service(std::make_shared<OcmDaqService>(
              executor.get_io_context(), *m_mal, m_manager, m_name, output_path, m_event_log))
        , 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); });
 
        // Try restoring from workspace
        try {
            m_manager.RestoreFromWorkspace();
        } catch (...) {
            daq::error::NestedExceptionReporter r(std::current_exception());
            LOG4CPLUS_ERROR(server::GetLogger(),
                            "Failed to restore state from OCM workspace at "
                                << m_workspace.GetPath().native() << ":\n"
                                << r);
        }
    }
 
    void Register() {
        LOG4CPLUS_DEBUG(server::GetLogger(), "RegisterService");
        m_replier.RegisterService(m_name,
                                  std::static_pointer_cast<daqif::AsyncOcmDaqControl>(m_service));
    }
    void Unregister() {
        LOG4CPLUS_DEBUG(server::GetLogger(), "UnregisterService");
        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;
                LOG4CPLUS_DEBUG(server::GetLogger(), "Calling ProcessEvent");
                m_state_machine.ProcessEvent(Events::AnyDaqActive{});
                LOG4CPLUS_DEBUG(server::GetLogger(), "Calling ProcessEvent done");
            } 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
                    LOG4CPLUS_DEBUG(server::GetLogger(), "Calling ProcessEvent");
                    m_state_machine.ProcessEvent(Events::AllDaqInactive{});
                    LOG4CPLUS_DEBUG(server::GetLogger(), "Calling ProcessEvent done");
                }
            }
            // publish state
            *m_sample << status.GetStatus();
            m_publisher.Publish(*m_sample);
            LOG4CPLUS_DEBUG(server::GetLogger(),
                            fmt::format("Publishing DAQ status for DAQ {} done", status.GetId()));
        } catch (std::exception const& e) {
            LOG4CPLUS_ERROR(server::GetLogger(),
                            fmt::format("Failed to publish status: {}", e.what()));
        }
    }
 
private:
    std::string m_name;
    daq::WorkspaceImpl m_workspace;
    std::shared_ptr<mal::Mal> m_mal;
    std::shared_ptr<daq::ObservableEventLog> m_event_log;
    std::shared_ptr<daq::DpmClientImpl> m_dpm_client;
    daq::DaqControllerFactoryDefault m_daq_factory;
    daq::ManagerImpl m_manager;
    std::shared_ptr<OcmDaqService> m_service;
    rad::cii::Replier& m_replier;
    rad::cii::Publisher<daqif::DaqStatus>& m_publisher;
    rad::SMAdapter& m_state_machine;
    bool m_is_active;
 
    boost::signals2::connection m_connection;
    std::shared_ptr<daqif::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 {
        /* Read only configuration */
        server::Config config;
        if (config.ParseOptions(argc, argv) == false) {
            // request for help
            return EXIT_SUCCESS;
        }
        /*
         *  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);
 
        config.LoadConfig();
 
        elt::log::CiiLogManager::Configure(
            rad::GetDefaultLogProperties(config.GetProcName() + ".log", config.GetLogLevel()));
        if (auto const& file = config.GetLogProperties(); !file.empty()) {
            auto resolved = rad::Helper::FindFile(config.GetLogProperties());
            if (resolved.empty()) {
                LOG4CPLUS_ERROR(server::GetLogger(),
                                "Configured log property file not found: '"
                                    << file << "', $CFGPATH=" << rad::Helper::GetEnvVar("CFGPATH"));
                return EXIT_FAILURE;
            }
            elt::log::CiiLogManager::Configure(resolved);
        }
 
        if (config.m_out_path.empty()) {
            auto const msg =
                "Output path not configured. Define environment "
                "variable $DATAROOT or configuration " +
                server::KEY_CONFIG_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);
        }
 
        auto workspace_root = config.GetWorkspace();
        LOG4CPLUS_INFO(server::GetLogger(),
                       fmt::format("Using workspace directory: {}", workspace_root.native()));
 
        /*
         * 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(daqif::MakeServerUri(config.GetMsgReplierEndpoint()));
        auto std_status_publisher = std::make_unique<rad::cii::Publisher<stdif::Status>>(
            daqif::MakeServiceUri(config.GetPubEndpoint(), "std/status"));
        auto daq_status_publisher = std::make_unique<rad::cii::Publisher<daqif::DaqStatus>>(
            daqif::MakeServiceUri(config.GetPubEndpoint(), "daq/status"));
 
        /* Runtime DB */
        rad::OldbAsyncWriter oldb_writer(config.GetDbTimeout(), std::chrono::milliseconds(100));
 
        /* Runtime data context */
        server::DataContext data_ctx(config, oldb_writer);
        data_ctx.UpdateDb();
 
        /*
         * 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);
 
        daq::ManagerParams manager_params;
        manager_params.instrument_id = config.m_instrument_id;
        manager_params.acquiring_stale_age = config.m_stale_acquiring;
        manager_params.merging_stale_age = config.m_stale_merging;
 
        // @todo: Load DPM RR URI and timeout
        DaqService daq_service("daq",
                               manager_params,
                               workspace_root,
                               config.m_out_path,
                               executor,
                               mal,
                               mal_replier,
                               *daq_status_publisher,
                               state_machine,
                               config.GetDpmClientParams());
 
        // 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);
            sample->setSource(config.GetProcName());
            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();
        oldb_writer.StartWriter();
        io_ctx.run();
        state_machine.Stop();
    } catch (rad::Exception& e) {
        LOG4CPLUS_ERROR(server::GetLogger(), e.what());
 
        std::cout << "Exit main()\n";
        return EXIT_FAILURE;
    } catch (...) {
        LOG4CPLUS_ERROR(server::GetLogger(), boost::current_exception_diagnostic_information());
        std::cout << "Exit main()\n";
        return EXIT_FAILURE;
    }
 
    // to avoid valgrind warnings on potential memory loss
    // google::protobuf::ShutdownProtobufLibrary();
 
    LOG4CPLUS_INFO(server::GetLogger(), "Application ocm-server terminated.");
    std::cout << "Exit main()\n";
    return EXIT_SUCCESS;
}