binding.hpp

浏览该文件的文档.

 
#pragma once
#include "../common/logger.hpp"
#include "../common/helper.hpp"
#include "../common/msg.pb.h"
#include <iostream>
#include <unordered_map>
#include <mutex>
#include <memory>
 
namespace XuMQ
{
    struct Binding
    {
        using ptr = std::shared_ptr<Binding>; 
        std::string exchange_name;            
        std::string msgqueue_name;            
        std::string binding_key;              
        Binding() {}
        Binding(const std::string &ename, const std::string &qname, const std::string &key)
            : exchange_name(ename), msgqueue_name(qname), binding_key(key) {}
    };
    using MsgQueueBindingMap = std::unordered_map<std::string, Binding::ptr>;
    using BindingMap = std::unordered_map<std::string, MsgQueueBindingMap>;
    class BindingMapper
    {
    public:
        BindingMapper(const std::string &dbfile)
            : _sql_helper(dbfile)
        {
            std::string path = FileHelper::parentDirectory(dbfile);
            FileHelper::createDirectory(path);
            assert(_sql_helper.open());
            createTable();
        }
        void createTable()
        {
            const char *CREATE_TABLE = "create table if not exists binding_table(exchange_name varchar(32), msgqueue_name varchar(32), binding_key varchar(128));";
            bool ret = _sql_helper.exec(CREATE_TABLE, nullptr, nullptr);
            if (ret == false)
            {
                fatal(logger, "创建绑定信息数据库表失败!");
                abort();
            }
        }
        void removeTable()
        {
            const char *DROP_TABLE = "drop table if exists binding_table;";
            bool ret = _sql_helper.exec(DROP_TABLE, nullptr, nullptr);
            if (ret == false)
            {
                fatal(logger, "删除绑定信息数据库表失败!");
                abort();
            }
        }
        bool insert(Binding::ptr &binding)
        {
            const char *INSERT_SQL = "insert into binding_table values('%s', '%s', '%s');";
            char sql_str[4096] = {0};
            sprintf(sql_str, INSERT_SQL, binding->exchange_name.c_str(),
                    binding->msgqueue_name.c_str(), binding->binding_key.c_str());
            bool ret = _sql_helper.exec(sql_str, nullptr, nullptr);
            if (ret == false)
            {
                error(logger, "数据库: 插入绑定信息失败!");
                return false;
            }
            return true;
        }
        bool remove(const std::string &ename, const std::string &qname)
        {
            const char *DELETE_SQL = "delete from binding_table where exchange_name = '%s' \
            and msgqueue_name = '%s';";
            char sql_str[4096] = {0};
            sprintf(sql_str, DELETE_SQL, ename.c_str(), qname.c_str());
            bool ret = _sql_helper.exec(sql_str, nullptr, nullptr);
            if (ret == false)
            {
                error(logger, "数据库: 删除绑定信息失败!");
                return false;
            }
            return true;
        }
        bool removeExchangeBindings(const std::string &ename)
        {
            const char *DELETE_SQL = "delete from binding_table where exchange_name = '%s';";
            char sql_str[4096] = {0};
            sprintf(sql_str, DELETE_SQL, ename.c_str());
            bool ret = _sql_helper.exec(sql_str, nullptr, nullptr);
            if (ret == false)
            {
                error(logger, "数据库: 删除交换机绑定信息失败!");
                return false;
            }
            return true;
        }
        bool removeQueueBindings(const std::string &qname)
        {
            const char *DELETE_SQL = "delete from binding_table where msgqueue_name = '%s';";
            char sql_str[4096] = {0};
            sprintf(sql_str, DELETE_SQL, qname.c_str());
            bool ret = _sql_helper.exec(sql_str, nullptr, nullptr);
            if (ret == false)
            {
                error(logger, "数据库: 删除消息队列绑定信息失败!");
                return false;
            }
            return true;
        }
        BindingMap recovery()
        {
            BindingMap result;
            const char *SELECT_SQL = "select * from binding_table;";
            _sql_helper.exec(SELECT_SQL, selectCallback, &result);
            return result;
        }
 
    private:
        static int selectCallback(void *arg, int numcol, char **row, char **fields)
        {
            BindingMap *result = (BindingMap *)arg;
            Binding::ptr bp = std::make_shared<Binding>(row[0], row[1], row[2]);
            MsgQueueBindingMap &qmap = (*result)[bp->exchange_name];
            qmap.insert(std::make_pair(bp->msgqueue_name, bp));
            return 0;
        }
 
    private:
        SqliteHelper _sql_helper; 
    };
    class BindingManager
    {
    public:
        using ptr = std::shared_ptr<BindingManager>; 
        BindingManager(const std::string &dbfile) : _mapper(dbfile) {
            _bindings = _mapper.recovery();
        }
        bool bind(const std::string &ename, const std::string &qname, const std::string &key, bool durable)
        {
            // 加锁 构造一个队列信息绑定对象 添加映射关系
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _bindings.find(ename);
            if (it != _bindings.end() && it->second.find(qname) != it->second.end()) // 绑定信息已经存在
                return true;
            Binding::ptr bp = std::make_shared<Binding>(ename, qname, key);
            if (durable)
            {
                bool ret = _mapper.insert(bp);
                if (ret == false)
                    return false;
            }
            auto &qbmap = _bindings[ename];
            qbmap.insert(std::make_pair(qname, bp));
            return true;
        }
        void unbind(const std::string &ename, const std::string &qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto eit = _bindings.find(ename);
            if (eit == _bindings.end()) // 没有交换机的绑定信息
                return;
            auto qit = eit->second.find(qname); // 没有交换机对应队列的绑定信息
            if (qit == eit->second.end())
                return;
            _mapper.remove(ename, qname);
            _bindings[ename].erase(qname);
        }
        void removeExchangeBindings(const std::string &ename)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.removeExchangeBindings(ename);
            _bindings.erase(ename);
        }
        void removeMsgQueueBindings(const std::string &qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.removeQueueBindings(qname);
            for (auto &binding : _bindings) // 遍历所有交换机
                binding.second.erase(qname);
        }
        MsgQueueBindingMap getExchangeBindings(const std::string &ename)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto eit = _bindings.find(ename);
            if (eit == _bindings.end())
                return MsgQueueBindingMap();
            return eit->second;
        }
 
        Binding::ptr getBinding(const std::string &ename, const std::string &qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto eit = _bindings.find(ename);
            if (eit == _bindings.end())
                return Binding::ptr();
            auto qit = eit->second.find(qname);
            if (qit == eit->second.end())
                return Binding::ptr();
            return qit->second;
        }
        bool exists(const std::string &ename, const std::string &qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto eit = _bindings.find(ename);
            if (eit == _bindings.end())
                return false;
            auto qit = eit->second.find(qname);
            if (qit == eit->second.end())
                return false;
            return true;
        }
        size_t size()
        {
            size_t total_size = 0;
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto &it : _bindings)
                total_size += it.second.size();
            return total_size;
        }
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.removeTable();
            _bindings.clear();
        }
 
    private:
        std::mutex _mutex;     
        BindingMapper _mapper; 
        BindingMap _bindings;  
    };
}