channel.hpp

浏览该文件的文档.

 
#pragma once
#include "muduo/net/TcpConnection.h"
#include "muduo/protobuf/codec.h"
#include "muduo/protobuf/dispatcher.h"
#include "muduo/base/Atomic.h"
#include "muduo/base/Types.h"
 
#include "../common/logger.hpp"
#include "../common/helper.hpp"
#include "../common/msg.pb.h"
#include "../common/protocol.pb.h"
#include "../common/threadpool.hpp"
#include <google/protobuf/map.h>
#include "consumer.hpp"
#include "host.hpp"
#include "route.hpp"
 
namespace XuMQ
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;                   
    using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;         
    using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;       
    using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>; 
    using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;   
    using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;       
    using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;         
    using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;             
    using queueUnBindRequestPtr = std::shared_ptr<queueUnBindRequest>;         
    using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;       
    using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;               
    using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;         
    using basicConsumeRequestPtr = std::shared_ptr<basicConsumeRequest>;       
    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
        Channel(const std::string &id, const VirtualHost::ptr &host, const ConsumerManager::ptr &cmp,
                const ProtobufCodecPtr &codec, const muduo::net::TcpConnectionPtr &conn, const threadpool::ptr &pool)
            : _cid(id), _conn(conn), _codec(codec), _cmp(cmp), _host(host), _pool(pool) {}
        ~Channel()
        {
            if (_consumer.get() != nullptr)
                _cmp->remove(_consumer->tag, _consumer->qname);
        }
        void declareExchange(const declareExchangeRequestPtr &req)
        {
            bool ret = _host->declareExchange(req->exchange_name(), req->exchange_type(),
                                              req->durable(), req->auto_delete(), req->args());
            basicRespFunc(ret, req->rid(), req->cid());
        }
        void deleteExchange(const deleteExchangeRequestPtr &req)
        {
            _host->deleteExchange(req->exchange_name());
            basicRespFunc(true, req->rid(), req->cid());
        }
        void declareQueue(const declareQueueRequestPtr &req)
        {
            bool ret = _host->declareQueue(req->queue_name(), req->durable(),
                                           req->exclusive(), req->auto_delete(), req->args());
            if (ret == false)
                basicRespFunc(ret, req->rid(), req->cid());
            debug(logger, "声明队列成功 队列名称为%s", req->queue_name().c_str());
            _cmp->initQueueConsumer(req->queue_name()); // 初始化队列消费者管理句柄
            basicRespFunc(ret, req->rid(), req->cid());
        }
        void deleteQueue(const deleteQueueRequestPtr &req)
        {
            _cmp->destroyQueueConsumer(req->queue_name());
            _host->deleteQueue(req->queue_name());
            basicRespFunc(true, req->rid(), req->cid());
        }
        void queueBind(const queueBindRequestPtr &req)
        {
            bool ret = _host->bind(req->exchange_name(), req->queue_name(), req->binding_key());
            basicRespFunc(ret, req->rid(), req->cid());
        }
        void queueUnBind(const queueUnBindRequestPtr &req)
        {
            _host->unbind(req->exchange_name(), req->queue_name());
            basicRespFunc(true, req->rid(), req->cid());
        }
        void basicPublish(const basicPublishRequestPtr &req)
        {
            // 获取交换机
            Exchange::ptr ep = _host->selectExchange(req->exchange_name());
            if (ep.get() == nullptr)
                basicRespFunc(false, req->rid(), req->cid());
            // 获取指定交换机的绑定信息
            MsgQueueBindingMap mqbm = _host->exchangeBindings(req->exchange_name());
            BasicProperties *properties = nullptr;
            std::string routing_key;
 
            if (req->has_properties())
            {
                properties = req->mutable_properties();
                routing_key = properties->routing_key();
            }
            for (auto &binding : mqbm)
            {
                // 交换路由 找到对应的队列
                if (Router::route(ep->type, routing_key, binding.second->binding_key))
                {
                    // 将消息添加到队列中
                    _host->basicPublish(binding.first, properties, req->body());
                    // 向线程池中添加一个消息消费任务(向指定队列的订阅者推送消息)
                    auto task = std::bind(&Channel::consume, this, binding.first);
                    _pool->push(task);
                }
                basicRespFunc(true, req->rid(), req->cid());
            }
        }
        void basicAck(const basicAckRequestPtr &req)
        {
            _host->basicAck(req->queue_name(), req->msg_id());
            basicRespFunc(true, req->rid(), req->cid());
        }
        void basicConsume(const basicConsumeRequestPtr &req)
        {
            // 判断队列是否存在
            bool ret = _host->existsQueue(req->queue_name());
            if (ret == false)
                basicRespFunc(false, req->rid(), req->cid());
            // 创建队列消费者
            auto cb = std::bind(&Channel::callback, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
            // 创建消费者之后 信道的角色就是消费者
            _consumer = _cmp->create(req->consumer_tag(), req->queue_name(), req->auto_ack(), cb);
            if(_consumer==nullptr)
            {
                fatal(logger,"消费者创建失败！");
            }
            else
            {
                info(logger,"消费者创建成功！");
            }
            basicRespFunc(true, req->rid(), req->cid());
        }
        void basicCancel(const basicCancelRequestPtr &req)
        {
            _cmp->remove(req->consumer_tag(), req->queue_name());
            basicRespFunc(true, req->rid(), req->cid());
        }
 
    private:
        void basicRespFunc(bool ok, const std::string &rid, const std::string &cid)
        {
            basicResponse resp;
            resp.set_rid(rid);
            resp.set_cid(cid);
            resp.set_ok(ok);
            _codec->send(_conn, resp);
        }
        void consume(const std::string &qname)
        {
            MessagePtr mp = _host->basicConsume(qname);
            if (mp.get() == nullptr)
            {
                error(logger, "消费任务失败, 指定队列中没有消息: %s", qname.c_str());
                return;
            }
            Consumer::ptr cp = _cmp->choose(qname);
            if (cp.get() == nullptr)
            {
                error(logger, "消费任务失败, 指定队列中没有消费者: %s", qname.c_str());
                return;
            }
            cp->callback(cp->tag, mp->mutable_payload()->mutable_properties(), mp->payload().body());
            if (cp->auto_ack == true)
                _host->basicAck(qname, mp->payload().properties().id());
        }
        void callback(const std::string &tag, const BasicProperties *bp, const std::string &body)
        {
            basicConsumeResponse resp;
            resp.set_cid(_cid);
            resp.set_body(body);
            resp.set_consumer_tag(tag);
            if (bp)
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn, resp);
        }
 
    private:
        std::string _cid;                   
        Consumer::ptr _consumer;            
        muduo::net::TcpConnectionPtr _conn; 
        ProtobufCodecPtr _codec;            
        ConsumerManager::ptr _cmp;          
        VirtualHost::ptr _host;             
        threadpool::ptr _pool;              
    };
    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>; 
        
        ChannelManager() {}
        bool openChannel(const std::string &id, const VirtualHost::ptr &host, const ConsumerManager::ptr &cmp,
                         const ProtobufCodecPtr &codec, const muduo::net::TcpConnectionPtr &conn, const threadpool::ptr &pool)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(id);
            if (it != _channels.end())
                return false;
            auto channel = std::make_shared<Channel>(id, host, cmp, codec, conn, pool);
            _channels.insert(std::make_pair(id, channel));
            return true;
        }
        void closeChannel(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(id);
        }
        Channel::ptr getChannel(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(id);
            if (it == _channels.end())
                return Channel::ptr();
            return it->second;
        }
 
    private:
        std::mutex _mutex; 
        std::unordered_map<std::string, Channel::ptr> _channels; 
    };
}