EventManager.hpp

#ifndef HAVE_EVENT_MANAGER_H
#define HAVE_EVENT_MANAGER_H 1

#include <string>
#include <map>
#include <thread>
#include <mutex>
#include <condition_variable>

// network I/O
#include <boost/asio.hpp>

#include <amqpcpp.h>
#include "rdf_model.hpp"
#include "Item.hpp"
#include "Uri.hpp"
#include "PersistenceService.hpp"
#include "AnalysisService.hpp"
#include "Event.pb.h"

namespace mico {
    namespace event {

        static const std::string EXCHANGE_SERVICE_REGISTRY = "service_registry";
        static const std::string EXCHANGE_SERVICE_DISCOVERY = "service_discovery";

        static const std::string QUEUE_CONTENT_INPUT  = "content_input";

        static const std::string QUEUE_CONTENT_OUTPUT = "content_output";

        static const std::string QUEUE_CONFIG_REQUEST = "config_request";


        class AnalysisConsumer;
        class DiscoveryConsumer;
        class RabbitConnectionHandler;
        class ConfigurationClient;

        // Bug: https://github.com/CopernicaMarketingSoftware/AMQP-CPP/issues/25
        // Got fixed with version 2.2.0
        class AMQPCPPOnCloseBugfix {
        private:
            bool firstCall;
            std::mutex firstCallMutex;

        public:
            AMQPCPPOnCloseBugfix();
            bool isFirstCall();
        };

        class AnalysisResponse
        {
        private:
          AnalysisService& m_service;
          const AMQP::Message &m_message;
          AMQP::Channel* m_channel;

        public:
          AnalysisResponse(AnalysisService& service, const AMQP::Message &message, AMQP::Channel* channel):
            m_service(service), m_message(message), m_channel(channel) {}

          void sendFinish(      std::shared_ptr< mico::persistence::model::Item > i);
          void sendErrorMessage(std::shared_ptr< mico::persistence::model::Item > i, const mico::event::model::ErrorCodes& errcode, const std::string& msg, const std::string& desc);
          void sendProgress(    std::shared_ptr< mico::persistence::model::Item > i, const mico::persistence::model::URI& part, const float& progress);
          void sendNew(         std::shared_ptr< mico::persistence::model::Item > i, const mico::persistence::model::URI& part);
        };

        class EventManagerException {
        private:
            std::string message;
        public:
            EventManagerException(std::string message) : message(message) {};

            const std::string& getMessage() { return message; };
        };

        class EventManager : public AMQP::ConnectionHandler
        {
        private:
            boost::asio::
            io_service io_service; 

            boost::asio::ip::tcp::
            socket socket;         

            std::string host;    
            int    rabbitPort;   
            int    marmottaPort; 

            std::string user;         
            std::string password;     

            bool connected;              
            bool unavailable;            

            mico::persistence::
            PersistenceService *persistence = NULL;  

            size_t recv_len;
            char*  recv_buf;


            RabbitConnectionHandler* rabbit; 

            AMQP::Connection*   connection;  
            AMQP::Channel*      channel;     

            std::map<AnalysisService*, AnalysisConsumer*> services; 

            std::thread receiver;        
            std::mutex m;                
            std::condition_variable cv;  

            void doConnect();            
            void doRead();               

            std::string stripPatchVersion(std::string ver); 

            ConfigurationClient *configurationClient = NULL; //<! configuration service

            AMQPCPPOnCloseBugfix amqpWorkaround;

            std::vector<char> intermediateReceiveBuffer; 

        public:

            EventManager(const std::string& host) : EventManager(host, "mico", "mico") {};

            EventManager(const std::string& host, const std::string& user, const std::string& password) : EventManager(host, 5672, 8080, user, password) {};

            EventManager(const std::string& host, int rabbitPort, int marmottaPort, const std::string& user, const std::string& password);

            virtual ~EventManager();

            mico::persistence::
            PersistenceService* getPersistenceService();

            void onData(AMQP::Connection *connection, const char *data, size_t size);

            void onError(AMQP::Connection *connection, const char *message);

            void onConnected(AMQP::Connection *connection);

            void onClosed(AMQP::Connection *connection);

            void registerService(AnalysisService* service);


            void unregisterService(AnalysisService* service);

            void injectItem(std::shared_ptr< mico::persistence::model::Item > item);
        };

    }
}
#endif