node.h

//========================================================================
#ifndef comfrm_ifc_NodeH
#define comfrm_ifc_NodeH
//========================================================================
#include "anymap.h"
#include "stringtools.h"

#include <string>
#include <unordered_map>
#include <boost/asio.hpp>
#include <boost/thread/mutex.hpp>
//========================================================================
namespace com {
    namespace interproc {
        class Message;
        class NodeProtocol;
        class NodeConnection;
        struct NodeHandshake;

        struct MAC_DLLEXPORT NodeAddress
        {
            NodeAddress()
            {}

            NodeAddress(const std::string& p_machineIp, const std::string& p_appSid)
                : m_machineIp(boost::asio::ip::address::from_string(p_machineIp))
                , m_appSid(p_appSid)
            {}

            NodeAddress(const boost::asio::ip::address& p_machineIp, const std::string& p_appSid)
                : m_machineIp(p_machineIp)
                , m_appSid(p_appSid)
            {}

            NodeAddress(const std::string& p_appAddress)
            {
                std::string machineIpStr;
                com::str::paramValueSplit(p_appAddress, '/', machineIpStr, m_appSid);
                m_machineIp = boost::asio::ip::address::from_string(machineIpStr);
            }

            std::string toString() const
            {
                return m_machineIp.to_string() + "/" + m_appSid;
            }

            friend bool operator==(const NodeAddress& p_addr1, const NodeAddress& p_addr2);

            boost::asio::ip::address m_machineIp;
            std::string m_appSid; // unique application's id
        };

        // TODO: доделать!!!
        /*
        struct stc_NodeData
        {
            std::shared_ptr< NodeConnection > m_nodeConn;
            boost::function< void (Message, CommunicationError) > m_callback;
        };
        */
        //========================================================================
        //template<>
        //class MAC_DLLEXPORT cls_Node< Message, NodeAddress >
        // TODO: должен быть thread-safe!
    }
}

// Определяем хэш структуры адрес для его использования в качестве ключа в unordered_map
namespace std
{
    template <>
    struct hash< com::interproc::NodeAddress >
    {
        size_t operator()(const com::interproc::NodeAddress& p_addr) const
        {
            if (p_addr.m_machineIp.is_v4())
                    return ((hash<unsigned long>()(p_addr.m_machineIp.to_v4().to_ulong()) >> 1)
                             ^ (hash<string>()(p_addr.m_appSid) << 1));
            else
            {
                // TODO: поддерживается только v4
                return ((hash<unsigned long>()(0) >> 1)
                         ^ (hash<string>()(p_addr.m_appSid) << 1));
            }
        }
    };
}

namespace com {
    namespace interproc {
        class MAC_DLLEXPORT Node
        {
        public:
/*
            enum StandartDestination
            {
                sdSelf,
                sdCaller,
                sdRouter,
            };
*/

            enum CommunicationError
            {
                ceSuccess = 0, 
                ceNodeNotFound, 
                ceCannotConnectToMasterNode, 
                ceNetworkError, 
                ceUnknownError, 
            };

            enum NodeStatus
            {
                nsMaster, 
                nsSlave, 
                nsNotSet, 
            };

            typedef std::function< void (std::shared_ptr< Message >, CommunicationError) > OnReplyClb;

            struct RemoteNodeData
            {
                RemoteNodeData(std::shared_ptr< NodeConnection > p_conn, OnReplyClb p_callback)
                    : m_conn(p_conn)
                    , m_callback(p_callback)
                {}

                std::shared_ptr< NodeConnection > m_conn;
                OnReplyClb m_callback;
            };

            // TODO: временно
            struct BufferedMessage
            {
                BufferedMessage(const std::vector< char >& p_bytearray, const NodeAddress& p_nodeAddr, OnReplyClb p_callback)
                    : m_bytearray(p_bytearray)
                    , m_nodeAddr(p_nodeAddr)
                    , m_callback(p_callback)
                {}

                std::vector< char > m_bytearray;
                NodeAddress m_nodeAddr;
                OnReplyClb m_callback;
            };

            static const unsigned short m_port = 8221;

        public:
            Node(const std::string& p_appId, boost::asio::io_service& p_service);
            ~Node();

            CommunicationError deploy();

            void setDefaultCallback(OnReplyClb p_callback);

            void send(const Message& p_message, OnReplyClb p_callback = 0);

            const NodeAddress& getAddr() const;

            NodeStatus getStatus() const;

        protected:
            void setStatus(NodeStatus p_status);
            void onNewMessageToSend(const Message& p_message, OnReplyClb p_callback);
            void onAccepted(std::shared_ptr< NodeConnection > p_conn, const boost::system::error_code & p_error);
            void onConnectionClosed(std::shared_ptr< NodeConnection > p_conn);
            void onMessageReceived(std::shared_ptr< NodeConnection > p_nodeConn, std::vector< char >& p_message);
            void handshake(std::shared_ptr< NodeConnection > p_nodeConn, NodeHandshake p_handshake);
            void proxyMessage(std::shared_ptr< NodeConnection > p_nodeConn, std::vector< char >& p_message);
            void onAnyMapReceived(std::shared_ptr< NodeConnection > p_nodeConn, std::string p_appSid,
                                          std::shared_ptr< Anymap > p_anymap);
            void sendBytearrayToNode(const std::vector< char >& p_bytearray, const NodeAddress& p_nodeAddr, OnReplyClb p_callback = 0);
            void runService();

        protected:
            NodeAddress m_selfAddr;
            std::string m_masterAppSid; // TODO: временно, не лучшее решение хранить здесь
            OnReplyClb m_defaultCallback;
            NodeStatus m_status;
            unsigned int m_threadsCount;
            std::shared_ptr< boost::asio::ip::tcp::acceptor > m_acceptor;
            // TODO: переделать на unordered_multimap, так как может оказаться, что несколько нод с одинаковыми именами на одной машине
            // TODO: решение с ... временное. Сейчас коллбэк для адреса задается в методе send, что не совсем корректно.
            // Необходимо иметь возможность задавать коллбэк на каждое отдельное сообщение.
            std::unordered_map< NodeAddress, RemoteNodeData > m_establishedNodes;
            std::set< std::shared_ptr< NodeConnection > > m_anonymousNodes;
            std::vector< BufferedMessage > m_msgBuffer;
            boost::asio::io_service& m_service;
            NodeProtocol* m_protocol;
            boost::mutex m_mutex;
        };
    }
}

//========================================================================
#endif
//========================================================================