queue.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 MsgQueue
    {
        using ptr = std::shared_ptr<MsgQueue>;             
        std::string name;                                  
        bool durable;                                      
        bool exclusive;                                    
        bool auto_delete;                                  
        google::protobuf::Map<std::string, std::string> args; 
        MsgQueue() {}
        MsgQueue(const std::string &qname,
                 bool qdurable,
                 bool qexclusive,
                 bool qauto_delete,
                 const google::protobuf::Map<std::string, std::string> &qargs) : name(qname), durable(qdurable), exclusive(qexclusive),
                                                                        auto_delete(qauto_delete), args(qargs)
        {
        }
        void setArgs(const std::string &str_args)
        {
            std::vector<std::string> sub_args;
            StrHelper::split(str_args, "&", sub_args);
            for (auto &str : sub_args)
            {
                size_t pos = str.find('=');
                std::string key = str.substr(0, pos);
                std::string value = str.substr(pos + 1);
                args[key] = value;
            }
        }
        std::string getArgs()
        {
            std::string result;
            for (auto &arg : args)
            {
                result += (arg.first + "=" + arg.second + "&");
            }
            if (!result.empty())
                result.pop_back(); // 去除最后一个'&'
            return result;
        }
    };
    using QueueMap = std::unordered_map<std::string, MsgQueue::ptr>; 
    class MsgQueueMapper
    {
    public:
        MsgQueueMapper(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 queue_table(name varchar(32) primary key,\
            durable tinyint, exclusive tinyint,\
            auto_delete tinyint, args 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 queue_table;";
            bool ret = _sql_helper.exec(DROP_TABLE, nullptr, nullptr);
            if (ret == false)
            {
                fatal(logger, "删除消息队列数据库表失败!");
                abort();
            }
        }
        bool insert(MsgQueue::ptr &queue)
        {
            const char *INSERT_SQL = "insert into queue_table values('%s', %d, %d, %d, '%s');";
            char sql_str[4096] = {0};
            sprintf(sql_str, INSERT_SQL, queue->name.c_str(), queue->durable, queue->exclusive,
                    queue->auto_delete, queue->getArgs().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 &name)
        {
            const char *DELETE_SQL = "delete from queue_table where name = '%s';";
            char sql_str[4096] = {0};
            sprintf(sql_str, DELETE_SQL, name.c_str());
            bool ret = _sql_helper.exec(sql_str, nullptr, nullptr);
            if (ret == false)
            {
                error(logger, "数据库: 删除消息队列失败!");
                return false;
            }
            return true;
        }
        QueueMap recovery()
        {
            QueueMap result;
            const char *SELECT_SQL = "select * from queue_table;";
            _sql_helper.exec(SELECT_SQL, selectCallback, &result);
            return result;
        }
 
    private:
        static int selectCallback(void *arg, int numcol, char **row, char **fields)
        {
            QueueMap *result = (QueueMap *)arg;
            auto mqp = std::make_shared<MsgQueue>();
            mqp->name = row[0];
            mqp->durable = (bool)std::stoi(row[1]);
            mqp->exclusive = (bool)std::stoi(row[2]);
            mqp->auto_delete = (bool)std::stoi(row[3]);
            if (row[4])
                mqp->setArgs(row[4]);
            result->insert(std::make_pair(mqp->name, mqp));
            return 0;
        }
 
    private:
        SqliteHelper _sql_helper; 
    };
    class MsgQueueManager
    {
    public:
        using ptr = std::shared_ptr<MsgQueueManager>; 
        MsgQueueManager(const std::string &dbfile)
            : _mapper(dbfile)
        {
            _queues = _mapper.recovery();
        }
        bool declareQueue(const std::string &qname,
                          bool qdurable,
                          bool qexclusive,
                          bool qauto_delete,
                          const google::protobuf::Map<std::string, std::string> &qargs)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queues.find(qname);
            if (it != _queues.end())
                return true;
            auto mqp = std::make_shared<MsgQueue>(qname, qdurable, qexclusive, qauto_delete, qargs);
            if (qdurable)
            {
                bool ret = _mapper.insert(mqp);
                if (ret == false)
                    return false;
            }
            _queues.insert(std::make_pair(qname, mqp));
            return true;
        }
        void deleteQueue(const std::string &name)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queues.find(name);
            if (it == _queues.end())
                return;
            _queues.erase(name);
            if (it->second->durable == true)
                _mapper.remove(name);
        }
        MsgQueue::ptr selectQueue(const std::string &name)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queues.find(name);
            if (it == _queues.end())
                return MsgQueue::ptr();
            return it->second;
        }
        QueueMap &allQueue()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _queues;
        }
        bool exists(const std::string &name)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queues.find(name);
            if (it == _queues.end())
                return false;
            return true;
        }
        size_t size()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _queues.size();
        }
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.removeTable();
            _queues.clear();
        }
 
    private:
        std::mutex _mutex;      
        MsgQueueMapper _mapper; 
        QueueMap _queues;       
    };
}