connection.hpp

浏览该文件的文档.

 
#pragma once
 
#include "muduo/protobuf/dispatcher.h"
#include "muduo/protobuf/codec.h"
 
#include "muduo/base/Mutex.h"
#include "muduo/base/Logging.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpClient.h"
#include "muduo/net/EventLoopThread.h"
#include "muduo/base/CountDownLatch.h"
#include "channel.hpp"
#include "worker.hpp"
#include "../common/logger.hpp"
 
namespace XuMQ
{
    class Connection
    {
    public:
        using ptr = std::shared_ptr<Connection>; 
        Connection(const std::string &sip, int sport, const AsyncWorker::ptr &worker)
            : _latch(1), _client(worker->_loopthread.startLoop(), muduo::net::InetAddress(sip, sport), "Client"),
              _dispatcher(std::bind(&Connection::onUnknowMessage, this,
                                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),
              _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher,
                                                               std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
              _worker(worker), _channels(std::make_shared<ChannelManager>())
        {
            _dispatcher.registerMessageCallback<basicResponse>(std::bind(&Connection::commonResponse, this, std::placeholders::_1,
                                                                         std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicConsumeResponse>(std::bind(&Connection::consumeResponse, this, std::placeholders::_1,
                                                                                std::placeholders::_2, std::placeholders::_3));
            _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec, std::placeholders::_1,
                                                 std::placeholders::_2, std::placeholders::_3));
            _client.setConnectionCallback(std::bind(&Connection::onConnection, this, std::placeholders::_1));
            _client.connect();
            _latch.wait();
        }
        Channel::ptr openChannel()
        {
            Channel::ptr channel = _channels->create(_conn, _codec);
            bool ret = channel->openChannel();
            if (ret == false)
            {
                error(logger, "打开信道失败!");
                return Channel::ptr();
            }
            return channel;
        }
        void closeChannel(const Channel::ptr &channel)
        {
            channel->closeChannel();
            _channels->remove(channel->cid());
        }
 
    private:
        void commonResponse(const muduo::net::TcpConnectionPtr &conn, const basicResponsePtr message, muduo::Timestamp)
        {
            // 找到信道
            Channel::ptr channel = _channels->get(message->cid());
            if (channel == nullptr)
            {
                error(logger, "未找到信道!");
                return;
            }
            // 将得到的响应对象 添加到信道的基础响应中
            channel->putBasicResponse(message);
        }
        void consumeResponse(const muduo::net::TcpConnectionPtr &conn, const basicConsumeResponsePtr message, muduo::Timestamp)
        {
            // 找到信道
            Channel::ptr channel = _channels->get(message->cid());
            if (channel == nullptr)
            {
                error(logger, "未找到信道!");
                return;
            }
            // 封装异步任务交给线程池
            _worker->_threadpool.push([channel, message]()
                                      { channel->consume(message); });
        }
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                _latch.countDown();
                _conn = conn;
            }
            else
            {
                _conn.reset();
            }
        }
        void onUnknowMessage(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
        {
            info(logger, "onUnknowMessage: %s", message->GetTypeName());
            conn->shutdown();
        }
 
    private:
        muduo::CountDownLatch _latch;       
        muduo::net::TcpConnectionPtr _conn; 
        muduo::net::TcpClient _client;      
        ProtobufDispatcher _dispatcher;     
        ProtobufCodecPtr _codec;            
        AsyncWorker::ptr _worker;           
        ChannelManager::ptr _channels;      
    };
 
}