Connection管理类实现(模块六)

news2025/1/12 8:49:01

目录

类功能

类定义

类实现

编译


本文使用了自定的Any类

Any类的简单实现-CSDN博客

类功能

类定义

// DISCONECTED -- 连接关闭状态  CONNECTING -- 连接建立成功-待处理状态
// CONNECTED -- 连接建立完成,各种设置已完成,可以通信状态    DISCONNECTING -- 待关闭状态
typedef enum
{
    DISCONECTED,
    CONNECTING,
    CONNECTED,
    DISCONNECTING
} ConnStatu;
using PtrConnection = std::shared_ptr<Connection>;
class Connection
{
private:
    uint64_t _conn_id; // 连接的唯一ID,便于连接的管理和查找
    // uint64_t _timer_id; // 定时器ID,必须是唯一的,这块是为了简化操作使用conn_id作为定时器
    int _sockfd;                   // 连接关联的文件描述符
    bool _enable_inactive_release; // 连接是否启动非活跃的判断标志,默认为false
    EventLoop *_loop;              // 连接所关联的一个EventLoop
    ConnStatu _statu;              // 连接状态
    Socket _socket;                // 套接字操作管理
    Channel _channel;              // 连接的事件管理
    Buffer _in_buffer;             // 输入缓冲区--存放从socket中读取到的数据
    Buffer _out_buffer;            // 输出缓冲区--存放要发送给对端的数据
    Any _context;                  // 请求的接收处理上下文

    /*这四个回调函数,是让服务器模块来设置的(其实服务器模块的处理回调也是组件使用者设置的)*/
    /*换句话来说,这几个回调都是组件使用者使用的*/
    using ConnectedCallback = std::function<void(const PtrConnection &)>;
    using MessageCallback = std::function<void(const PtrConnection &, Buffer *)>;
    using ClosedCallback = std::function<void(const PtrConnection &)>;
    using AnyEventCallback = std::function<void(const PtrConnection &)>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closef_callback;
    AnyEventCallback _event_callback;
    /*组件内的连接关闭回调--组件内设置的,因为服务器组件内会把所有的连接管理起来,一旦某个连接要关闭*/
    /*就应该从管理的地方移除掉自己的信息*/
    ClosedCallback _server_closed_callback;

private:
    /*五个channel的事件回调函数*/
    void HandleRead();        // 描述符可读事件触发后调用的函数,接收socket数据放到接收缓冲区中,然后调用_message_callback
    void HandleWrite();       // 描述符可写事件触发后调用的函数,将发送缓冲区中的数据进行发送
    void HandleClose();       // 描述符触发挂断事件
    void HandleError();       // 描述符触发出错事件
    void HandleEvent();       // 描述符触发任意事件
    void EstablishedInLoop(); // 连接获取之后,所处的状态要进行各种设置(给channel设置事件回调,启动读监控)
    void ReleaseInLoop();     // 这个接口才是实际的释放接口
    void SendInLoop(char *data, size_t len);
    void ShutdownInLoop();
    void EnableInactiveReleaseInLoop(int sec);
    void CancelInactiveReleaseInLoop();
    void UpgradeInLoop(const Context &context,
                       const ConnectedCallback &conn,
                       const MessageCallback &msg,
                       const ClosedCallback &closed,
                       const AnyEventCallback &event);

public:
    Connection(EventLoop *loop, uint64_t conn_id, int sockfd);
    ~Connection();
    int Fd();                            // 获取管理的文件描述符
    int Id();                            // 获取连接ID
    bool Connected();                    // 是否处于CONNECTED状态
    void SetContext(const Any &context); // 设置上下文--连接建立完成时进行调用
    Any *GetContext();                   // 获取上下文,返回的是指针
    void SetConnectedCallback(const ConnectedCallback &cb);
    void SetMessageCallback(const MessageCallback &cb);
    void SetClosedCallback(const ClosedCallback &cb);
    void SetAnyEventCallback(const AnyEventCallback &cb);
    void Established();                  // 连接建立就绪后,进行channel回调设置,启动读监控,调用_connect_callback
    void Send(char *data, size_t len);   // 发送数据,将数据发送到发送缓冲区,启动写事件监控
    void Shutdown();                     // 提供给组件使用者的关闭接口--并不实际关闭,需要判断有没有数据待处理
    void EnableInactiveRelease(int sec); // 启动非活跃销毁,并定义多长时间无通信就是非活跃,添加定时任务
    void CancelInactiveRelease();        // 取消非活跃销毁
    // 切换协议--重置上下文以及阶段性处理函数
    void Upgrade(const Context, const ConnectedCallback &conn,
                 const ClosedCallback &closed, const AnyEventCallback &event);
};

类实现

class Connection;
// DISCONNECTED -- 连接关闭状态  CONNECTING -- 连接建立成功-待处理状态
// CONNECTED -- 连接建立完成,各种设置已完成,可以通信状态    DISCONNECTING -- 待关闭状态
typedef enum
{
    DISCONNECTED,
    CONNECTING,
    CONNECTED,
    DISCONNECTING
} ConnStatu;
using PtrConnection = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection>
{
private:
    uint64_t _conn_id; // 连接的唯一ID,便于连接的管理和查找
    // uint64_t _timer_id; // 定时器ID,必须是唯一的,这块是为了简化操作使用conn_id作为定时器
    int _sockfd;                   // 连接关联的文件描述符
    bool _enable_inactive_release; // 连接是否启动非活跃的判断标志,默认为false
    EventLoop *_loop;              // 连接所关联的一个EventLoop
    ConnStatu _statu;              // 连接状态
    Socket _socket;                // 套接字操作管理
    Channel _channel;              // 连接的事件管理
    Buffer _in_buffer;             // 输入缓冲区--存放从socket中读取到的数据
    Buffer _out_buffer;            // 输出缓冲区--存放要发送给对端的数据
    Any _context;                  // 请求的接收处理上下文

    /*这四个回调函数,是让服务器模块来设置的(其实服务器模块的处理回调也是组件使用者设置的)*/
    /*换句话来说,这几个回调都是组件使用者使用的*/
    using ConnectedCallback = std::function<void(const PtrConnection &)>;
    using MessageCallback = std::function<void(const PtrConnection &, Buffer *)>;
    using ClosedCallback = std::function<void(const PtrConnection &)>;
    using AnyEventCallback = std::function<void(const PtrConnection &)>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _event_callback;
    /*组件内的连接关闭回调--组件内设置的,因为服务器组件内会把所有的连接管理起来,一旦某个连接要关闭*/
    /*就应该从管理的地方移除掉自己的信息*/
    ClosedCallback _server_closed_callback;

private:
    /*五个channel的事件回调函数*/
    // 描述符可读事件触发后调用的函数,接收socket数据放到接收缓冲区中,然后调用_message_callback
    void HandleRead()
    {
        // 1.接收socket的数据,放到缓冲区
        char buf[65536];
        ssize_t ret = _socket.NonBlockRecv(buf, 65536);
        if (ret < 0)
        {
            // 出错了,不能直接关闭连接
            return ShutdownInLoop();
        }
        // 这里的等于0表示的是没有读取到数据,而并不是连接断开了,连接断开返回的是-1
        // 将数据放入输入缓冲区,写入之后顺便将写偏移向后移动
        _in_buffer.WriteAndPush(buf, ret);
        // 2.调用message_callback进行业务处理
        if (_in_buffer.ReadAbleSize() > 0)
        {
            // shard_from_this--从当前对象自身获取自身的shared_ptr管理对象
            return _message_callback(shared_from_this(), &_in_buffer);
        }
    }
    // 描述符可写事件触发后调用的函数,将发送缓冲区中的数据进行发送
    void HandleWrite()
    {
        // _out_buffer中保存的就是要发送的数据
        ssize_t ret = _socket.NonBlockSend(_out_buffer.ReadPosition(), _out_buffer.ReadAbleSize());
        if (ret < 0)
        {
            // 发送错误就应该关闭连接了
            if (_in_buffer.ReadAbleSize() > 0)
            {
                _message_callback(shared_from_this(), &_in_buffer);
            }
        }
        _out_buffer.MoveReadOffset(ret); // 千万不要忘了,将读偏移向后移动
        if (_out_buffer.ReadAbleSize() == 0)
        {
            _channel.DisableWrite(); // 没有数据待发送,关闭写事件监控
            // 如果当前是连接待关闭状态,则有数据,发送完数据释放连接,没有数据则直接释放
            if (_statu == DISCONNECTING)
            {
                return ReleaseInLoop(); // 这时候就是实际的关闭释放操作了
            }
        }
        return;
    }
    // 描述符触发挂断事件
    void HandleClose()
    {
        /*一旦连接挂断了,套接字就什么都干不了了,因此有数据待处理就处理一下,完毕关闭连接*/
        if (_in_buffer.ReadAbleSize() > 0)
        {
            _message_callback(shared_from_this(), &_in_buffer);
        }
        return ReleaseInLoop();
    }
    // 描述符触发出错事件
    void HandleError()
    {
        return HandleClose();
    }
    // 描述符触发任意事件: 1.刷新连接活跃度--延迟定时销毁任务 2.调用组件使用者的任意事件回调
    void HandleEvent()
    {
        if (_enable_inactive_release == true)
        {
            _loop->TimerRefresh(_conn_id);
        }
        if (_event_callback)
        {
            _event_callback(shared_from_this());
        }
    }
    // 连接获取之后,所处的状态要进行各种设置(给channel设置事件回调,启动读监控,调用回调函数)
    void EstablishedInLoop()
    {
        // 1.修改连接状态   2.启动读事件监控    3.调用回调函数
        assert(_statu == CONNECTING); // 当前状态必须一定是上层的半连接状态
        _statu = CONNECTED;           // 当前函数执行完毕,则连接进入已完成连接状态
        // 一旦启动读事件监控就有可能会立即触发读事件,如果这时候启动了非活跃连接销毁
        _channel.EnableRead();
        if (_connected_callback)
            _connected_callback(shared_from_this());
    }
    // 这个接口才是实际的释放接口
    void ReleaseInLoop()
    {
        // 1.修改连接状态,将其置为DISCONNECTED
        _statu = DISCONNECTED;
        // 2.移除连接的事件监控
        _channel.Remove();
        // 3.关闭描述符
        _socket.Close();
        // 4.如果当前定时器队列中还有定时销毁任务,则取消任务
        if (_loop->HasTimer(_conn_id))
            CancelInactiveReleaseInLoop();
        // 5.调用关闭回调函数,避免先移除服务器管理的连接信息导致Connection被释放,再去处理会出错,因此先调用用户的回调函数
        if (_closed_callback)
            _closed_callback(shared_from_this());
        // 移除服务器内部管理的连接信息
        if (_server_closed_callback)
            _server_closed_callback(shared_from_this());
    }
    // 这个并不是实际的发送接口,而只是把数据放到了发送缓冲区,启动了可写事件监控
    void SendInLoop(char *data, size_t len)
    {
        if (_statu == DISCONNECTED)
            return;
        _out_buffer.WriteAndPush(data, len);
        if (_channel.WriteAble() == false)
        {
            _channel.EnableWrite();
        }
    }
    // 这个关闭操作并非实际的连接释放操作,需要判断还有没有数据待处理,待发送
    void ShutdownInLoop()
    {
        _statu = DISCONNECTING; // 设置连接为半关闭状态
        if (_in_buffer.ReadAbleSize() > 0)
        {
            if (_message_callback)
                _message_callback(shared_from_this(), &_in_buffer);
        }
        // 要么就是写入数据的时候出错关闭,要么就是没有待发送数据,直接关闭
        if (_out_buffer.ReadAbleSize() > 0)
        {
            if (_channel.WriteAble() == false)
            {
                _channel.EnableWrite();
            }
        }
        if (_out_buffer.ReadAbleSize() == 0)
        {
            ReleaseInLoop();
        }
    }
    // 启动非活跃连接超时释放规则
    void EnableInactiveReleaseInLoop(int sec)
    {
        // 1.将判断标志 _enable_inactive_release 置为true
        _enable_inactive_release = true;
        // 2.如果当前定时销毁任务已经存在,那就刷新一下延迟即可
        if (_loop->HasTimer(_conn_id))
        {
            return _loop->TimerRefresh(_conn_id);
        }
        // 3.如果不存在定时销毁任务,则新增
        _loop->TimerAdd(_conn_id, sec, std::bind(&Connection::ReleaseInLoop, this));
    }
    void CancelInactiveReleaseInLoop()
    {
        _enable_inactive_release = false;
        if (_loop->HasTimer(_conn_id))
        {
            _loop->TimerCancel(_conn_id);
        }
    }
    void UpgradeInLoop(const Any &context,
                       const ConnectedCallback &conn,
                       const MessageCallback &msg,
                       const ClosedCallback &closed,
                       const AnyEventCallback &event)
    {
        _context = context;
        _connected_callback = conn;
        _message_callback = msg;
        _closed_callback = closed;
        _event_callback = event;
    }

public:
    Connection(EventLoop *loop, uint64_t conn_id, int sockfd) : _conn_id(conn_id), _sockfd(sockfd),
                                                                _enable_inactive_release(false), _loop(loop), _statu(CONNECTING), _socket(_sockfd),
                                                                _channel(loop, _sockfd)
    {
        _channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
        _channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));
        _channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
        _channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
        _channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
    }
    ~Connection() { DBG_LOG("RELEASE CONNECTION:%p", this); }
    // 获取管理的文件描述符
    int Fd() { return _sockfd; }
    // 获取连接ID
    int Id() { return _conn_id; }
    // 是否处于CONNECTED状态
    bool Connected() { return (_statu == CONNECTED); }
    // 设置上下文--连接建立完成时进行调用
    void SetContext(const Any &context) { _context = context; }
    // 获取上下文,返回的是指针
    Any *GetContext() { return &_context; }
    void SetConnectedCallback(const ConnectedCallback &cb) { _connected_callback = cb; }
    void SetMessageCallback(const MessageCallback &cb) { _message_callback = cb; }
    void SetClosedCallback(const ClosedCallback &cb) { _closed_callback = cb; }
    void SetAnyEventCallback(const AnyEventCallback &cb) { _event_callback = cb; }
    void SetSrvClosedCallback(const ClosedCallback &cb) { _server_closed_callback = cb; }
    // 连接建立就绪后,进行channel回调设置,启动读监控,调用_connect_callback
    void Established()
    {
        _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this));
    }
    // 发送数据,将数据发送到发送缓冲区,启动写事件监控
    void Send(char *data, size_t len)
    {
        _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, data, len));
    }
    // 提供给组件使用者的关闭接口--并不实际关闭,需要判断有没有数据待处理
    void Shutdown()
    {
        _loop->RunInLoop(std::bind(&Connection::ShutdownInLoop, this));
    }
    // 启动非活跃销毁,并定义多长时间无通信就是非活跃,添加定时任务
    void EnableInactiveRelease(int sec)
    {
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec));
    }
    // 取消非活跃销毁
    void CancelInactiveRelease()
    {
        _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this));
    }
    // 切换协议--重置上下文以及阶段性处理函数--而是这个接口必须在EventLoop线程中立即执行
    // 防备新的事件触发后,处理的时候,切换任务还没有被执行--会导致数据使用原协议处理了
    void Upgrade(const Any &context, const ConnectedCallback &conn, const MessageCallback &msg,
                 const ClosedCallback &closed, const AnyEventCallback &event)
    {
        _loop->AssertInLoop();
        _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, context, conn, msg, closed, event));
    }
};

编译

符合预期,联合调试会将其放在下一篇中

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/1471316.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

每日五道java面试题之spring篇(六)

目录&#xff1a; 第一题 ApplicationContext通常的实现是什么&#xff1f;第二题 什么是Spring的依赖注入&#xff1f;第三题 依赖注入的基本原则第四题 依赖注入有什么优势&#xff1f;第五题 有哪些不同类型的依赖注入实现方式&#xff1f; 第一题 ApplicationContext通常的…

基于频率增强的数据增广的视觉语言导航方法(VLN论文阅读)

基于频率增强的数据增广的视觉语言导航方法&#xff08;VLN论文阅读&#xff09; 摘要 视觉和语言导航&#xff08;VLN&#xff09;是一项具有挑战性的任务&#xff0c;它需要代理基于自然语言指令在复杂的环境中导航。 在视觉语言导航任务中&#xff0c;之前的研究主要是在空间…

pycharm如何设置滚轮缩放代码大小?

左上角的File找到设置&#xff0c;或者快捷键ctrlalts。 弹出对话框&#xff0c;手动输入mouse&#xff0c;点击general&#xff0c;勾选改变字体大小&#xff0c;ok确认

转前端了!!

大家好&#xff0c;我是冰河~~ 没错&#xff0c;为了更好的设计和开发分布式IM即时通讯系统&#xff0c;也为了让大家能够直观的体验到分布式IM即时通讯系统的功能&#xff0c;冰河开始转战前端了。也就是说&#xff0c;整个项目从需求立项到产品设计&#xff0c;从架构设计到…

ubuntu20.04中配置Pyrep和CoppeliaSim

ubuntu20.04中配置Pyrep和CoppeliaSim 在Ubuntu20.04中配置 Pyrep &#xff0c;实现应用Python语言的机器人在 Vrep&#xff08;CoppeliaSim&#xff09;中的虚拟仿真 一、安装CoppeliaSim 4.1 1.1 下载适配Ubuntu20.04的CoppeliaSim 4.1软件 下载链接&#xff1a;https://…

【前端素材】推荐优质后台管理系统Dashy平台模板(附源码)

一、需求分析 后台管理系统&#xff08;或称作管理后台、管理系统、后台管理平台&#xff09;是一种专门用于管理网站、应用程序或系统后台运营的软件系统。它通常由一系列功能模块组成&#xff0c;为管理员提供了管理、监控和控制网站或应用程序的各个方面的工具和界面。以下…

Nginx实现平滑升级

平滑升级 本篇目标&#xff1a;将现有的 nginx 1.22.0 版本升级为 1.24.0 //查看现有版本 [root12 ~]# nginx -v nginx version: nginx/1.22.01、首先在官网下载软件包&#xff0c;地址&#xff1a;nginx: download 2、把要 1.24.0 拖进 /opt 目录后&#xff0c;解压&#xf…

抖店是怎么运营做起来的?一文详解抖店的运营逻辑和流程,可收藏

我是王路飞。 很多人都知道现在的抖音有【商城】&#xff0c;进入之后就是一个个的抖音小店了&#xff0c;也知道抖店的红利。 但是抖店具体是怎么运营并且做起来的&#xff0c;就不太清楚了&#xff0c;因此很多新手明明眼馋抖店的红利&#xff0c;却又无从下手。 今天这篇…

YOLOv9尝鲜测试五分钟极简配置

pip安装python包&#xff1a; pip install yolov9pip在https://github.com/WongKinYiu/yolov9/tree/main中下载好权重文件yolov9-c.pt。 运行下面代码&#xff1a; import yolov9model yolov9.load("yolov9-c.pt", device"cpu") # load pretrained or c…

Spring综合漏洞利用工具

Spring综合漏洞利用工具 工具目前支持Spring Cloud Gateway RCE(CVE-2022-22947)、Spring Cloud Function SpEL RCE (CVE-2022-22963)、Spring Framework RCE (CVE-2022-22965) 的检测以及利用&#xff0c;目前仅为第一个版本&#xff0c;后续会添加更多漏洞POC&#xff0c;以及…

【Flink精讲】Flink性能调优:CPU核数与并行度

常见问题 举个例子 提交任务命令&#xff1a; bin/flink run \ -t yarn-per-job \ -d \ -p 5 \ 指定并行度 -Dyarn.application.queuetest \ 指定 yarn 队列 -Djobmanager.memory.process.size2048mb \ JM2~4G 足够 -Dtaskmanager.memory.process.size4096mb \ 单个 TM2~8G 足…

【机器人学导论笔记】三、操作臂正运动学

3.1 概述 操作臂正运动学研究操作臂的运动特性&#xff0c;主要涉及与运动有关的几何参数和时间参数。本章中&#xff0c;只研究静止状态下操作臂连杆的位置和姿态。 处理这些复杂的几何参数需要一些步骤&#xff1a;首先需要在操作臂的每个连杆上分别固接一个连杆坐标系&…

基于ELFBoard开发板的车牌识别系统

本项目采用的是ElfBoard ELF1开发板作为项目的核心板&#xff0c;主要实现的功能为通过USB 摄像头对车牌进行识别&#xff0c;如果识别成功则会亮绿灯&#xff0c;并将识别的车牌号上传到手机APP上面&#xff0c;车牌识别的实现是通过百度OCR进行实现&#xff0c;手机APP是用Ja…

《低功耗方法学》翻译——第十四章:电源切换网络设计

第十四章&#xff1a;电源切换网络设计 功率门控是在待机或休眠模式下降低漏电功率最有效的方法&#xff0c;但这种方法存在诸如休眠晶体管占用的硅面积、永久和虚拟电源网络的布线资源以及复杂的功率门控设计和实现过程等开销&#xff0c;影响设计风险和进度。 除了开销外&a…

2024年Facebook自动回复优化指南:提升客户满意度的策略(内含自动回复中英文模板)

在这个数字化的时代&#xff0c;快速响应已经成为企业在与客户沟通的必备要素。但是当经常面对大量的相同信息时&#xff0c;如何可以提高效率呢&#xff1f;目前很多社交媒体平台都内设了自动回复功能&#xff0c;像是Facebook。这个功能确保无论何时有人联系你&#xff0c;都…

大气颗粒物与VOCs PMF源解析

原文&#xff1a;大气颗粒物与VOCs PMF源 第一&#xff1a;PMF源解析技术及其输入文件准备 1、大气污染源解析方法有哪些&#xff1f; 2、这些方法各自应用的条件以及它们的优缺点&#xff1f; 3、大气颗粒物的基础知识及各组分的主要来源 大气颗粒物的来源&#xff1a;…

Oracle EBS GL 外币折算逻辑

背景 由于公司财务在10月份期间某汇率维护错误,导致帐套折算以后并合传送至合并帐套生成合并日记帐凭证的借贷金额特别大,但是财务核对的科目余额有没有问题,始终觉得合并日记帐生成会计分发有问题,需要我们给出外币折算逻辑。 基础设置 汇率 Path: GL->设置->币种-&…

Java核心-核心类与API(3)

话接上回&#xff0c;继续核心类与API的学习&#xff0c;这次介绍一下枚举类以及与系统、交互有关的类&#xff0c;需要了解并能使用即可。 一、枚举类 1、概述 枚举也称穷举&#xff0c;简单理解就是把所有可能一一列举出来&#xff08;穷尽所有可能&#xff09;。枚举是一…

matlab新能源汽车三自由度操纵稳定性分析及优化

1、内容简介 略 可以交流、咨询、答疑 55-新能源汽车三自由度操纵稳定性分析及优化 2、内容说明 略 摘 要 电动化是节能减排、寻求替代能源的最佳途径&#xff0c;已成为行业共识&#xff0c;论文基于江西科技学院桑塔纳轿车油改气项目&#xff0c;在拆除发动机、变速…

JAVA工程师面试专题-《Redis》篇

目录 一、基础 1、Redis 是什么 2、说一下你对redis的理解 3、Redis 为什么这么快&#xff1f; 4、项目中如何使用缓存&#xff1f; 5、为什么使用缓存&#xff1f; 6、Redis key 和value 可以存储最大值分别多是多少&#xff1f; 7、Redis和memcache有什么区别&#xf…