host.hpp

浏览该文件的文档.

 
#pragma once
#include "exchange.hpp"
#include "queue.hpp"
#include "binding.hpp"
#include "message.hpp"
 
namespace XuMQ
{
    class VirtualHost
    {
    public:
        using ptr = std::shared_ptr<VirtualHost>;
        VirtualHost(const std::string hname, const std::string &basedir, const std::string &dbfile)
            : _emp(std::make_shared<ExchangeManager>(dbfile)),
              _mqmp(std::make_shared<MsgQueueManager>(dbfile)),
              _bmp(std::make_shared<BindingManager>(dbfile)),
              _mmp(std::make_shared<MessageManager>(basedir))
 
        {
            // 获取所有队列信息 通过队列信息恢复历史消息
            QueueMap qm = _mqmp->allQueue();
            for (auto &q : qm)
                _mmp->initQueueMessage(q.first);
        }
 
        bool declareExchange(const std::string &name,
                             ExchangeType type,
                             bool durable,
                             bool auto_delete,
                             const google::protobuf::Map<std::string, std::string> &args)
        {
            return _emp->declareExchange(name, type, durable, auto_delete, args);
        }
        void deleteExchange(const std::string &name)
        {
            // 先删除绑定信息
            _bmp->removeExchangeBindings(name);
            return _emp->deleteExchange(name);
        }
 
        bool declareQueue(const std::string &qname,
                          bool qdurable,
                          bool qexclusive,
                          bool qauto_delete,
                          const google::protobuf::Map<std::string, std::string> &qargs)
        {
            // 初始化队列消息句柄
            // 消息对立创建
            _mmp->initQueueMessage(qname);
            return _mqmp->declareQueue(qname, qdurable, qexclusive, qauto_delete, qargs);
        }
        void deleteQueue(const std::string &name)
        {
            // 删除队列消息
            _mmp->destroyQueueMessage(name);
            // 删除队列绑定信息
            _bmp->removeMsgQueueBindings(name);
            return _mqmp->deleteQueue(name);
        }
 
        bool bind(const std::string &ename, const std::string &qname, const std::string &key)
        {
            Exchange::ptr ep = _emp->selectExchange(ename);
            if (ep.get() == nullptr)
            {
                error(logger, "队列绑定失败, 交换机 %s 不存在", ename.c_str());
                return false;
            }
            MsgQueue::ptr mqp = _mqmp->selectQueue(qname);
            if (mqp.get() == nullptr)
            {
                error(logger, "队列绑定失败, 队列 %s 不存在", qname.c_str());
                return false;
            }
            return _bmp->bind(ename, qname, key, ep->durable & mqp->durable);
        }
        void unbind(const std::string &ename, const std::string &qname)
        {
            return _bmp->unbind(ename, qname);
        }
        MsgQueueBindingMap exchangeBindings(const std::string &ename)
        {
            return _bmp->getExchangeBindings(ename);
        }
        QueueMap allQueues()
        {
            return _mqmp->allQueue();
        }
 
        bool basicPublish(const std::string &qname, BasicProperties *bp, const std::string &body)
        {
            MsgQueue::ptr mqp = _mqmp->selectQueue(qname);
            if (mqp.get() == nullptr)
            {
                error(logger, "队列绑定失败, 队列 %s 不存在", qname.c_str());
                return false;
            }
            return _mmp->insert(qname, bp, body, mqp->durable);
        }
        MessagePtr basicConsume(const std::string &qname)
        {
            return _mmp->front(qname);
        }
        void basicAck(const std::string &qname, const std::string &msg_id)
        {
            _mmp->ack(qname, msg_id);
        }
 
        Exchange::ptr selectExchange(const std::string &ename)
        {
            return _emp->selectExchange(ename);
        }
 
        void clear()
        {
            _emp->clear();
            _mqmp->clear();
            _bmp->clear();
            _mmp->clear();
        }
 
        bool existsExchange(const std::string &name)
        {
            return _emp->exists(name);
        }
 
        bool existsQueue(const std::string &name)
        {
            return _mqmp->exists(name);
        }
 
        bool existsBinding(const std::string &ename, const std::string &qname)
        {
            return _bmp->exists(ename, qname);
        }
 
    private:
        std::string _host_name;
        ExchangeManager::ptr _emp;  
        MsgQueueManager::ptr _mqmp; 
        BindingManager::ptr _bmp;   
        MessageManager::ptr _mmp;   
    };
 
    class VirtualHostManager
    {
    public:
        using ptr = std::shared_ptr<VirtualHostManager>; 
        VirtualHostManager() {}
        
        bool declareVirtualHost(const std::string &hname,
                                const std::string &basedir,
                                const std::string &dbfile)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _vhosts.find(hname);
            if(it!=_vhosts.end())
                return true;
            auto host = std::make_shared<VirtualHost>(hname, basedir, dbfile);
            _vhosts.insert(std::make_pair(hname, host));
            return true;
        }
 
        void deleteVirtualHost(const std::string &hname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _vhosts.find(hname);
            if(it == _vhosts.end())
                return;
            _vhosts.erase(hname);
        }
 
        VirtualHost::ptr selectVirtualHost(const std::string &hname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _vhosts.find(hname);
            if(it == _vhosts.end())
                return VirtualHost::ptr();
            return it->second;
        }
 
        bool exists(const std::string &hname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _vhosts.find(hname);
            if(it == _vhosts.end())
                return false;
            return true;
        }
 
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _vhosts.clear();
        }
 
        size_t size()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _vhosts.size();
        }
    private:
        std::mutex _mutex;                                         
        std::unordered_map<std::string, VirtualHost::ptr> _vhosts; 
    };
}