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 "ContentItem.hpp"
#include "PersistenceService.hpp"
#include "AnalysisService.hpp"

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";


        class AnalysisConsumer;
        class DiscoveryConsumer;
        class RabbitConnectionHandler;

        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;  

            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();               
        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() { return persistence; };

            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 injectContentItem(const mico::persistence::ContentItem& item);

        };

    }
}
#endif