channel.hpp

浏览该文件的文档.

 
#pragma once
#include "muduo/net/TcpConnection.h"
#include "muduo/protobuf/codec.h"
#include "muduo/protobuf/dispatcher.h"
#include "../common/logger.hpp"
#include "../common/helper.hpp"
#include "../common/msg.pb.h"
#include "../common/protocol.pb.h"
#include "consumer.hpp"
#include <mutex>
#include <condition_variable>
 
namespace XuMQ
{
    using MessagePtr = std::shared_ptr<google::protobuf::Message>;         
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;               
    using basicConsumeResponsePtr = std::shared_ptr<basicConsumeResponse>; 
    using basicResponsePtr = std::shared_ptr<basicResponse>;               
    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>; 
        Channel(const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec)
            : _cid(UUIDHelper::uuid()), _conn(conn), _codec(codec) {}
        ~Channel()
        {
            basicCancel();
        }
        bool openChannel()
        {
            std::string rid = UUIDHelper::uuid();
            openChannelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn, req);
            basicResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }
        void closeChannel()
        {
            std::string rid = UUIDHelper::uuid();
            closeChannelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn, req);
            waitResponse(rid);
        }
        bool declareExchange(const std::string &name,
                             ExchangeType type,
                             bool durable,
                             bool auto_delete,
                             google::protobuf::Map<std::string, std::string> &args)
        {
            // 构造一个声明交换机的请求对象
            declareExchangeRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(name);
            req.set_exchange_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            // 发送请求
            _codec->send(_conn, req);
            // 等待服务器响应
            basicResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }
        void deleteExchange(const std::string &name)
        {
            deleteExchangeRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(name);
            // 发送请求
            _codec->send(_conn, req);
            waitResponse(rid);
        }
        bool declareQueue(const std::string &qname,
                          bool qdurable,
                          bool qexclusive,
                          bool qauto_delete,
                          google::protobuf::Map<std::string, std::string> &qargs)
        {
            declareQueueRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            req.set_exclusive(qexclusive);
            req.set_durable(qdurable);
            req.set_auto_delete(qauto_delete);
            req.mutable_args()->swap(qargs);
            _codec->send(_conn, req);
            basicResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }
        void deleteQueue(const std::string &qname)
        {
            deleteQueueRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            _codec->send(_conn, req);
            waitResponse(rid);
        }
        bool queueBind(const std::string &ename, const std::string &qname, const std::string &key)
        {
            queueBindRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            req.set_exchange_name(ename);
            req.set_binding_key(key);
            _codec->send(_conn, req);
            basicResponsePtr resp = waitResponse(rid);
            return resp->ok();
        }
        void queueUnBind(const std::string &ename, const std::string &qname)
        {
            queueUnBindRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            req.set_exchange_name(ename);
            _codec->send(_conn, req);
            waitResponse(rid);
        }
        void basicPublish(const std::string &ename, const BasicProperties *bp, const std::string &body)
        {
            basicPublishRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_body(body);
            if (bp != nullptr)
            {
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                req.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn, req);
            waitResponse(rid);
        }
        void basicAck(const std::string &msg_id)
        {
            if (_consumer.get() == nullptr) // 消费者不存在 无法进行确认
            {
                error(logger, "消息确认时 当前信道未设置消费者!");
                return;
            }
            basicAckRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->qname);
            req.set_msg_id(msg_id);
            _codec->send(_conn, req);
            waitResponse(rid);
        }
        bool basicConsume(const std::string &tag, const std::string &qname, bool auto_ack, const ConsumerCallback &cb)
        {
            if (_consumer.get() != nullptr) // 不为空时 说明消费者已经存在 不需要再创建
                return false;
            
            basicConsumeRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            req.set_consumer_tag(tag);
            req.set_auto_ack(auto_ack);
            _codec->send(_conn, req);
            basicResponsePtr resp = waitResponse(rid);
            debug(logger,"请求创建一个消费者 消费者tag为%s 消费者所在队列是%s",tag.c_str(), qname.c_str());
            if (resp->ok() == false)
            {
                error(logger, "添加订阅失败!");
                return false;
            }
            _consumer = std::make_shared<Consumer>(tag, qname, auto_ack, cb);
            return true;
        }
        void basicCancel()
        {
            if (_consumer.get() == nullptr) // 消费者为空 无法取消订阅
                return;
            basicCancelRequest req;
            std::string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->qname);
            req.set_consumer_tag(_consumer->tag);
            _codec->send(_conn, req);
            waitResponse(rid);
            _consumer.reset();
        }
 
        void putBasicResponse(const basicResponsePtr &resp)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _basic_resp.insert(std::make_pair(resp->rid(), resp));
            _cv.notify_all();
        }
        void consume(const basicConsumeResponsePtr &resp)
        {
            if (_consumer.get() == nullptr)
            {
                warn(logger, "消息处理时未找到订阅者信息!");
                return;
            }
            if (_consumer->tag != resp->consumer_tag())
            {
                error(logger, "推送消息中消费者标识与信道消费者标识不一致!");
                return;
            }
            _consumer->callback(resp->consumer_tag(), resp->mutable_properties(), resp->body());
        }
 
        std::string cid()
        {
            return _cid;
        }
 
    private:
        basicResponsePtr waitResponse(const std::string &rid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cv.wait(lock, [&rid, this]
                     { return _basic_resp.find(rid) != _basic_resp.end(); });
            basicResponsePtr resp = _basic_resp[rid];
            _basic_resp.erase(rid);
            return resp;
        }
 
    private:
        std::string _cid;                                              
        muduo::net::TcpConnectionPtr _conn;                            
        ProtobufCodecPtr _codec;                                       
        Consumer::ptr _consumer;                                       
        std::mutex _mutex;                                             
        std::condition_variable _cv;                                   
        std::unordered_map<std::string, basicResponsePtr> _basic_resp; 
    };
    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>; 
        ChannelManager() {}
        Channel::ptr create(const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto channel = std::make_shared<Channel>(conn, codec);
            _channels.insert(std::make_pair(channel->cid(), channel));
            return channel;
        }
        void remove(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(cid);
        }
        Channel::ptr get(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if (it == _channels.end())
            {
                return Channel::ptr();
            }
            return it->second;
        }
 
    private:
        std::mutex _mutex;                                       
        std::unordered_map<std::string, Channel::ptr> _channels; 
    };
}