dds-commander/src/AgentChannel.h

Go to the documentation of this file.

// Copyright 2014 GSI, Inc. All rights reserved.
//
//
//
#ifndef __DDS__CAgentChannel__
#define __DDS__CAgentChannel__
// DDS
#include "ServerChannelImpl.h"
// STD
#include <chrono>

namespace dds
{
    namespace commander_cmd
    {
        enum EAgentState
        {
            unknown = 0,
            idle,     // no tasks are assigned
            executing // assigned task is executing
        };
        const std::array<std::string, 3> g_agentStates = { { "unknown", "idle", "executing" } };

        using slotID_t = uint64_t;
        using taskID_t = uint64_t;
        struct SSlotInfo
        {
            slotID_t m_id{ 0 }; 
            taskID_t m_taskID{ 0 };
            EAgentState m_state{ EAgentState::unknown };
        };

        struct SAgentInfo
        {
            using SlotContainer_t = std::map<slotID_t, SSlotInfo>;

            SAgentInfo()
                : m_id(0)
                , m_startUpTime(0)
            {
            }
            void addSlot(const SSlotInfo& _slot)
            {
                std::lock_guard<std::mutex> lock(m_mtxSlot);
                m_slots.insert(std::make_pair(_slot.m_id, _slot));
            }
            SlotContainer_t& getSlots()
            {
                return m_slots;
            }

            SSlotInfo& getSlotByID(slotID_t _slotID)
            {
                std::lock_guard<std::mutex> lock(m_mtxSlot);
                auto it = m_slots.find(_slotID);
                if (it == m_slots.end())
                {
                    std::stringstream ss;
                    ss << "getSlotByID: slot " << _slotID << " does not exist.";
                    throw std::runtime_error(ss.str());
                }

                return it->second;
            }

            // We use unique ID because we need to identify a new agent after shutdown of the system on the same host.
            // We have to distinguish between new and old agent.
            uint64_t m_id;
            protocol_api::SHostInfoCmd m_remoteHostInfo;
            std::chrono::milliseconds m_startUpTime;

          private:
            std::mutex m_mtxSlot;
            SlotContainer_t m_slots;
        };

        class CAgentChannel : public protocol_api::CServerChannelImpl<CAgentChannel>
        {
            CAgentChannel(boost::asio::io_context& _context, uint64_t _protocolHeaderID = 0);

          public:
            BEGIN_MSG_MAP(CAgentChannel)
                MESSAGE_HANDLER(cmdREPLY_HOST_INFO, on_cmdREPLY_HOST_INFO)
                //====> replay on the "submit" command request
                MESSAGE_HANDLER(cmdSUBMIT, on_cmdSUBMIT)
                MESSAGE_HANDLER_DISPATCH(cmdUSER_TASK_DONE)
                //====> replay on the "info" command request
                // - get Agents Info command
                MESSAGE_HANDLER_DISPATCH(cmdGET_AGENTS_INFO)
                MESSAGE_HANDLER_DISPATCH(cmdGET_IDLE_AGENTS_COUNT)
                //
                MESSAGE_HANDLER_DISPATCH(cmdREPLY_ID)
                MESSAGE_HANDLER(cmdBINARY_ATTACHMENT_RECEIVED, on_cmdBINARY_ATTACHMENT_RECEIVED)
                MESSAGE_HANDLER_DISPATCH(cmdTRANSPORT_TEST)
                MESSAGE_HANDLER(cmdREPLY, on_cmdREPLY)
                // - Topology commands
                MESSAGE_HANDLER_DISPATCH(cmdUPDATE_TOPOLOGY)
                // - Agents commands
                MESSAGE_HANDLER_DISPATCH(cmdGET_LOG)
                MESSAGE_HANDLER_DISPATCH(cmdUPDATE_KEY)
                // Watchdog
                MESSAGE_HANDLER(cmdWATCHDOG_HEARTBEAT, on_cmdWATCHDOG_HEARTBEAT)
                MESSAGE_HANDLER_DISPATCH(cmdGET_PROP_LIST)
                MESSAGE_HANDLER_DISPATCH(cmdGET_PROP_VALUES)
                // custom command
                MESSAGE_HANDLER_DISPATCH(cmdCUSTOM_CMD)
                // TASK SLOTS
                MESSAGE_HANDLER(cmdREPLY_ADD_SLOT, on_cmdREPLY_ADD_SLOT)
            END_MSG_MAP()

          public:
            uint64_t getId() const;
            void setId(uint64_t _id);

            protocol_api::SHostInfoCmd getRemoteHostInfo(const dds::protocol_api::SSenderInfo& _sender);
            std::chrono::milliseconds getStartupTime(const dds::protocol_api::SSenderInfo& _sender);

            SAgentInfo& getAgentInfo();

          private:
            // Message Handlers
            bool on_cmdSUBMIT(protocol_api::SCommandAttachmentImpl<protocol_api::cmdSUBMIT>::ptr_t _attachment,
                              const protocol_api::SSenderInfo& _sender);
            bool on_cmdREPLY_HOST_INFO(
                protocol_api::SCommandAttachmentImpl<protocol_api::cmdREPLY_HOST_INFO>::ptr_t _attachment,
                const protocol_api::SSenderInfo& _sender);
            bool on_cmdBINARY_ATTACHMENT_RECEIVED(
                protocol_api::SCommandAttachmentImpl<protocol_api::cmdBINARY_ATTACHMENT_RECEIVED>::ptr_t _attachment,
                const protocol_api::SSenderInfo& _sender);
            bool on_cmdREPLY(protocol_api::SCommandAttachmentImpl<protocol_api::cmdREPLY>::ptr_t _attachment,
                             const protocol_api::SSenderInfo& _sender);
            bool on_cmdWATCHDOG_HEARTBEAT(
                protocol_api::SCommandAttachmentImpl<protocol_api::cmdWATCHDOG_HEARTBEAT>::ptr_t _attachment,
                const protocol_api::SSenderInfo& _sender);
            bool on_cmdREPLY_ADD_SLOT(
                protocol_api::SCommandAttachmentImpl<protocol_api::cmdREPLY_ADD_SLOT>::ptr_t _attachment,
                const protocol_api::SSenderInfo& _sender);

            std::string _remoteEndIDString();

          private:
            std::string m_sCurrentTopoFile;
            SAgentInfo m_info;
            std::mutex m_mtxInfo;
        };
    } // namespace commander_cmd
} // namespace dds
#endif /* defined(__DDS__CAgentChannel__) */