application.hpp

Go to the documentation of this file.

 
#ifndef CCF_CONTROL_APPLICATION_HPP_
#define CCF_CONTROL_APPLICATION_HPP_
 
#include <mal/Mal.hpp>
 
#include <rad/mal/replier.hpp>
#include <rad/dbAdapterRedis.hpp>
 
#include <Recif.hpp>
 
#include <ccf/common/base.hpp>
#include <ccf/common/frameQueue.hpp>
#include <ccf/common/recipeBase.hpp>
#include <ccf/common/comAdptBase.hpp>
#include <ccf/common/appBase.hpp>
#include <ccf/control/config.hpp>
#include <ccf/control/dataContext.hpp>
#include <ccf/control/actionMgr.hpp>
#include <ccf/control/stdActions.hpp>
#include <ccf/control/stdCmdsImpl.hpp>
#include <ccf/control/dcsCmdsImpl.hpp>
#include <ccf/control/recCmdsImpl.hpp>
#include <ccf/control/comAdptSim.hpp>
#include <ccf/control/monitorThread.hpp>
#include <ccf/control/acqThread.hpp>
#include <ccf/control/procThread.hpp>
#include <ccf/control/pubThread.hpp>
#include <ccf/control/recWaitRequest.hpp>
 
namespace ccf::control {
 
   // @{
   const std::string MON_THR_TAG    = "MonThr";
   const std::string ACQ_THR_TAG    = "AcqThr";
   const std::string PROC_THR_TAG   = "ProcThr::{}";
   const std::string PUB_THR_TAG    = "PubThr::{}::{}";
   const std::string INPUT_Q_TAG    = "InputQueue";
   const std::string OUTPUT_Q_TAG   = "OutputQueue::{}";
   // @}
 
   // @{
   const std::string THREAD_SIG_TERM           = "SigTerm";
   const std::string THREAD_SIG_SETUP          = "SigSetup";
   const std::string THREAD_SIG_NEW_DATA       = "SigNewData";
   const std::string THREAD_SIG_PUBLISH        = "SigPublish";
   const std::string THREAD_SIG_START          = "SigStart";
   // @}
 
   //const std::string THREAD_SYNC_ACK_OK      = "SigAckOk";
   //const std::string THREAD_SYNC_ACK_ERROR   = "SigAckError";
   //const std::string THREAD_SYNC_UNKNOWN     = "SigUnknown";
 
   // @{
   const std::string ACTIVE_REC_MUTEX          = "ActiveRecMutex";
   const std::string REC_HISTORY_MUTEX         = "RecHistoryMutex";
   const std::string MUTEX_REC_WAIT_REQUESTS   = "RecWaitRequests";
   // @}
 
   template <typename T>
   void CcfRequestRejectHandler(const rad::AnyEvent& event, const std::string& state) {
     LOG4CPLUS_TRACE_METHOD(Logger(), __PRETTY_FUNCTION__);
     auto req = rad::GetPayloadNothrow<T>(event);
     if (req == nullptr) {
       CCFERROR(Logger(), event.GetId() << " has no associated request!");
       return;
     }
     //req->SetException(rad::Exception("Request rejected in state " + state));
     req->SetException(elt::mal::MalException("Request rejected in state " + state));
   }
 
class Application: public ccf::common::AppBase {
 public:
 
    static Application* s_app_instance;
 
    static Application& Instance();
 
    virtual ~Application();
 
    elt::mal::Mal& GetMal();
 
    //Config& GetGetCfg().
 
    int Execute(int argc, char* argv[]);
 
    void CreateThreads();
 
    void LoadInitSetup();
 
    void HandleSetup();
 
    void Dismantle();
    
    bool GetAcquisitionActive();
 
    void StartAcquistion();
 
    void StopAcquistion();
 
    void CheckStopAcquistion();
 
    void IncrementFrameCounter();
 
    std::string GetState();
 
    void SetRequestStatus(const std::string& request,
                          const std::string& status);
 
    std::string GetRequestStatus(const std::string& request) const;
 
    std::string ListAdapters();
 
    template <class COM_ADPT_TYPE, class SIM_COM_ADPT_TYPE>
    void RegisterComAdapters(const COM_ADPT_TYPE& com_adapter_factory_obj,
                             const SIM_COM_ADPT_TYPE& sim_com_adapter_factory_obj) {
      LOG4CPLUS_TRACE_METHOD(Logger(), __PRETTY_FUNCTION__);
      // if (m_com_adapter != nullptr) {
      //    delete m_com_adapter;
      // }
      // if (m_sim_com_adapter != nullptr) {
      //    delete m_sim_com_adapter;
      // }
      //m_com_adapter = new(std::nothrow)COM_ADPT_TYPE;
      m_com_adapter = std::make_unique<COM_ADPT_TYPE>();
      ccf::AssertPtr(m_com_adapter.get(), "Communication Adapter", CCFLOC);
      //m_sim_com_adapter = new(std::nothrow)SIM_COM_ADPT_TYPE;
      m_sim_com_adapter = std::make_unique<SIM_COM_ADPT_TYPE>();
      ccf::AssertPtr(m_sim_com_adapter.get(), "Simulation Communication Adapter", CCFLOC);
    }
 
    ccf::common::ComAdptBase& GetComAdapter();
 
    void ConnectDevice();
 
    void DisconnectDevice();
 
    const std::vector<std::string>& GetProcThreadIds() const;
 
    const std::vector<std::string>& GetPubThreadIds(const std::string& proc_thread_id) const;
 
    const std::list<PubThread*>& GetPubThreads() const;
 
    Application(const Application&) = delete;              
    Application& operator=(const Application&) = delete;   
 
   // Recording handling
 
   void UpdateRecordingStatus();
 
   void SetActiveRecStatusObj(std::shared_ptr<recif::RecStatus>& active_rec);
 
   void UpdateActiveRecStatusObj(const ccf::common::PublisherStatus& sum_rec_pub_status);
 
   void UpdateRecStatusInDb();
 
   void GetActiveRecStatus(std::shared_ptr<recif::RecStatus>& active_rec) const;
 
   void MoveActiveRecStatusToHistory(const ccf::PubStatus status,
                                     std::shared_ptr<recif::RecStatus>& active_rec);
 
   void SetRecordingStatus(const ccf::PubStatus status);
 
   void GetRecStatus(const std::string& rec_id,
                     std::shared_ptr<recif::RecStatus>& active_rec) const;
 
   void GetMostRecentRecStatus(std::shared_ptr<recif::RecStatus>& active_rec) const;
 
   void AddRecWaitRequest(std::shared_ptr<RecWaitRequest>& req_wait_req);
 
   void CheckRecWaitRequests(const bool recording_completed=false);
 
   void CleanUpRecWaitRequests();
 
 private:
    Application();
 
    int8_t                               m_simulation;
    std::unique_ptr<rad::cii::Replier>   m_mal_replier;
    std::shared_ptr<elt::mal::Mal>       m_mal;
    std::unique_ptr<DataContext>         m_data_ctx;
    ActionMgr                            m_action_mgr;
    //StdActions*                          m_actions_std;
    bool                                 m_dismantle_invoked;
    
    std::map<std::string, std::string>   m_request_status_map;
 
    std::unique_ptr<ccf::common::ComAdptBase>       m_com_adapter;
    std::unique_ptr<ccf::common::ComAdptBase>       m_sim_com_adapter;
 
    std::map<std::string, std::shared_ptr<ccf::common::FrameQueue>> m_frame_queues;
 
    std::vector<std::string>                        m_proc_thread_names;
    std::vector<std::string>                        m_proc_thread_ids;
    std::map<std::string, std::vector<std::string>> m_pub_thread_ids;
 
    // For easy access. The threads are handled by the MPTK.
    std::list<PubThread*>                           m_pub_threads;
 
    // Acquisition control.
    ccf::ExpoMode                                   m_acq_setup_expo_mode;
    int64_t                                         m_acq_setup_nb_of_frames;
    int64_t                                         m_acq_status_frame_count;
 
    // Recording session control.
    std::shared_ptr<recif::RecStatus>               m_active_recording;
    int64_t                                         m_exp_nb_of_frames;
    int32_t                                         m_max_sz_rec_history;
    int32_t                                         m_rec_history_exp;
    // Maps "<rec id>" into the set of associated Recording Sessions.
    std::map<std::string, std::shared_ptr<recif::RecStatus>>  m_rec_history;
    std::vector<std::string>                                  m_rec_id_history;
   
    std::map<std::string, std::string>                        m_file_stored_to_pub_id;
 
   // Pending ReqWait Requests.
   std::map<std::string, std::shared_ptr<RecWaitRequest>>     m_rec_wait_requests;
};
 
inline Application& App() {
   return Application::Instance();
}
 
} // namespace ccf::control
 
#endif // CCFCONTROL_APPLICATION_HPP_