From 9d4c2902c306746e92417d7976aaf78a2a4fffbc Mon Sep 17 00:00:00 2001 From: ypc <15051963820@163.com> Date: Wed, 8 Jul 2026 13:35:23 +0800 Subject: [PATCH] =?UTF-8?q?feat(libcomm):=20=E9=80=9A=E8=AE=AF=E6=8A=BD?= =?UTF-8?q?=E8=B1=A1=E5=B1=82=E6=A8=A1=E5=9D=97=20=E2=80=94=20TCP/UDP/UART?= =?UTF-8?q?=20+=20epoll=20=E5=BC=82=E6=AD=A5=E4=BA=8B=E4=BB=B6=E5=BE=AA?= =?UTF-8?q?=E7=8E=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 纯 C 接口（myComm.h, extern "C"），7 个 API: comm_create / connect / disconnect / send / recv_register / state_register / destroy - 按功能拆分 5 个源文件: comm_internal.h + core/tcp/udp/uart.cpp - TCP 服务器: bind + listen + epoll accept 多客户端 - TCP 客户端: 非阻塞 connect + EPOLLOUT 检测完成 - UDP: bind + recvfrom/sendto - UART: open + termios 配置（波特率/数据位/停止位/校验） - comm_epoll_init() 幂等自初始化，comm_run() 无需预创建实例 - comm_run() / comm_stop() 事件循环控制 --- release/src/public/libcomm/makefile | 31 ++ src/public/libcomm/inc/myComm.h | 187 +++++++ src/public/libcomm/src/comm_core.cpp | 722 +++++++++++++++++++++++++ src/public/libcomm/src/comm_internal.h | 98 ++++ src/public/libcomm/src/comm_tcp.cpp | 217 ++++++++ src/public/libcomm/src/comm_uart.cpp | 228 ++++++++ src/public/libcomm/src/comm_udp.cpp | 133 +++++ 7 files changed, 1616 insertions(+) create mode 100644 release/src/public/libcomm/makefile create mode 100644 src/public/libcomm/inc/myComm.h create mode 100644 src/public/libcomm/src/comm_core.cpp create mode 100644 src/public/libcomm/src/comm_internal.h create mode 100644 src/public/libcomm/src/comm_tcp.cpp create mode 100644 src/public/libcomm/src/comm_uart.cpp create mode 100644 src/public/libcomm/src/comm_udp.cpp diff --git a/release/src/public/libcomm/makefile b/release/src/public/libcomm/makefile new file mode 100644 index 0000000..b7a6080 --- /dev/null +++ b/release/src/public/libcomm/makefile @@ -0,0 +1,31 @@ +include ./../../../linux.mk +L := $(notdir $(realpath $(CURDIR)/..)) +M := libcomm +O := $(LIB_REL)/$(M).a +S := $(SRC_ROOT_DIR)/$(L)/libcomm/src +I := -I$(SRC_ROOT_DIR)/$(L)/libcomm/inc -I$(SRC_ROOT_DIR)/$(L)/libcomm/src +B := $(CURDIR)/$(M)/obj +SRCS := $(wildcard $(S)/*.cpp) +OBJS := $(patsubst $(S)/%.cpp, $(B)/%.o, $(SRCS)) +F := $(CXX_FLAGS) $(I) + +.PHONY: all +all: + @mkdir -p $(ROOT_DIR)/release/inc + @$(MAKE) $(O) + +$(O): $(OBJS) + @mkdir -p $(dir $@) + $(AR) rcs $@ $^ + @echo "[$(M)] built" + +$(B)/%.o: $(S)/%.cpp + @mkdir -p $(dir $@) + $(CXX) $(F) -c $< -o $@ + +.PHONY: clean +clean: + rm -rf $(B) $(O) + +.PHONY: rebuild +rebuild: clean all diff --git a/src/public/libcomm/inc/myComm.h b/src/public/libcomm/inc/myComm.h new file mode 100644 index 0000000..98a99ce --- /dev/null +++ b/src/public/libcomm/inc/myComm.h @@ -0,0 +1,187 @@ +/** + * @file myComm.h + * @brief 通讯抽象层 — 纯 C 接口 + * @details 提供 TCP/UDP/UART 统一通讯抽象，支持： + * - TCP 服务器/客户端 + * - UDP 服务器/客户端 + * - UART 串口通讯 + * - 异步接收回调 + 状态变化回调 + * - 基于 epoll 的单线程事件循环 + */ +#ifndef _MY_COMM_H_ +#define _MY_COMM_H_ + +#include "myBase.h" + +#ifdef __cplusplus +extern "C" +{ +#endif + +/* === 通讯类型枚举 === */ +typedef enum +{ + COMM_TYPE_TCP_SERVER = 0, /** TCP 服务器（监听+accept） */ + COMM_TYPE_TCP_CLIENT, /** TCP 客户端 */ + COMM_TYPE_UDP_SERVER, /** UDP 服务器 */ + COMM_TYPE_UDP_CLIENT, /** UDP 客户端 */ + COMM_TYPE_UART /** UART 串口 */ +} enum_comm_type; + +/* === 通讯状态枚举 === */ +typedef enum +{ + COMM_STATE_DISCONNECTED = 0, /** 未连接 */ + COMM_STATE_CONNECTING, /** 连接中 */ + COMM_STATE_CONNECTED, /** 已连接 */ + COMM_STATE_ERROR /** 错误状态 */ +} enum_comm_state; + +/* === 错误码 === */ +#define COMM_OK 0 /** 成功 */ +#define COMM_ERR_PARAM -1 /** 参数错误 */ +#define COMM_ERR_MEM -2 /** 内存不足 */ +#define COMM_ERR_SOCKET -3 /** Socket 创建失败 */ +#define COMM_ERR_BIND -4 /** 绑定失败 */ +#define COMM_ERR_LISTEN -5 /** 监听失败 */ +#define COMM_ERR_CONNECT -6 /** 连接失败 */ +#define COMM_ERR_SEND -7 /** 发送失败 */ +#define COMM_ERR_NOT_FOUND -8 /** 实例不存在 */ +#define COMM_ERR_STATE -9 /** 状态错误 */ +#define COMM_ERR_INTERNAL -10 /** 内部错误 */ + +/* === 通讯参数结构体 === */ + +/** TCP 参数 */ +typedef struct +{ + char local_ip[64]; /** 本地 IP（服务器用） */ + int local_port; /** 本地端口（服务器用） */ + char remote_ip[64]; /** 远端 IP（客户端用） */ + int remote_port; /** 远端端口（客户端用） */ +} stru_tcp_para; + +/** UDP 参数 */ +typedef struct +{ + char local_ip[64]; /** 本地 IP */ + int local_port; /** 本地端口 */ + char remote_ip[64]; /** 远端 IP（客户端用） */ + int remote_port; /** 远端端口（客户端用） */ +} stru_udp_para; + +/** UART 参数 */ +typedef struct +{ + char device[128]; /** 串口设备路径（如 /dev/ttyS0） */ + int baudrate; /** 波特率 */ + int data_bits; /** 数据位（5/6/7/8） */ + int stop_bits; /** 停止位（1/2） */ + char parity; /** 校验位（'N'/'E'/'O'） */ +} stru_uart_para; + +/** 通讯参数联合体 */ +typedef union +{ + stru_tcp_para tcp; + stru_udp_para udp; + stru_uart_para uart; +} union_comm_para; + +/* === 回调函数类型 === */ + +/** + * @brief 接收数据回调 + * @param id 实例 ID + * @param fd 接收数据的文件描述符 + * @param data 数据缓冲区 + * @param len 数据长度 + */ +typedef void (*comm_recv_cb)(int id, int fd, const unsigned char *data, int len); + +/** + * @brief 状态变化回调 + * @param id 实例 ID + * @param fd 文件描述符 + * @param state 新状态（enum_comm_state） + */ +typedef void (*comm_state_cb)(int id, int fd, int state); + +/* === API 接口 === */ + +/** + * @brief 创建通讯实例 + * @param type 通讯类型（enum_comm_type） + * @param debug 调试模式（1=打印调试信息, 0=静默） + * @param para 通讯参数指针 + * @return >=0 成功返回实例 ID, <0 失败返回错误码 + */ +int comm_create(int type, int debug, const union_comm_para *para); + +/** + * @brief 建立连接 + * @details TCP 客户端发起 connect, TCP 服务器开始 listen, + * UDP 进行 bind, UART 打开串口。 + * @param id 实例 ID + * @return COMM_OK 成功, <0 失败 + */ +int comm_connect(int id); + +/** + * @brief 断开连接（不清除参数，可重新 comm_connect） + * @param id 实例 ID + * @return COMM_OK 成功, <0 失败 + */ +int comm_disconnect(int id); + +/** + * @brief 发送数据 + * @param id 实例 ID + * @param fd 目标 fd（TCP 服务器发送给指定客户端） + * @param data 数据缓冲区 + * @param len 数据长度 + * @return >=0 实际发送字节数, <0 失败 + */ +int comm_send(int id, int fd, const unsigned char *data, int len); + +/** + * @brief 注册接收数据回调 + * @param id 实例 ID + * @param cb 回调函数指针 + * @return COMM_OK 成功, <0 失败 + */ +int comm_recv_register(int id, comm_recv_cb cb); + +/** + * @brief 注册状态变化回调 + * @param id 实例 ID + * @param cb 回调函数指针 + * @return COMM_OK 成功, <0 失败 + */ +int comm_state_register(int id, comm_state_cb cb); + +/** + * @brief 销毁通讯实例并释放所有资源 + * @param id 实例 ID + * @return COMM_OK 成功, <0 失败 + */ +int comm_destroy(int id); + +/** + * @brief 启动通讯事件循环（阻塞调用，需在独立线程中调用） + * @details 内部使用 epoll 监听所有已注册 fd，分发读/写/错误事件。 + * 该函数阻塞直到 comm_stop() 被调用。 + * @return COMM_OK 正常退出, <0 失败 + */ +int comm_run(void); + +/** + * @brief 停止通讯事件循环 + */ +void comm_stop(void); + +#ifdef __cplusplus +} +#endif + +#endif /* _MY_COMM_H_ */ diff --git a/src/public/libcomm/src/comm_core.cpp b/src/public/libcomm/src/comm_core.cpp new file mode 100644 index 0000000..328c8e0 --- /dev/null +++ b/src/public/libcomm/src/comm_core.cpp @@ -0,0 +1,722 @@ +/** + * @file comm_core.cpp + * @brief 通讯抽象层 — 核心模块 + * @details 全局变量定义、epoll 工具函数、事件循环主循环、 + * comm_create/connect/disconnect/destroy/run/stop 实现。 + */ + +#include "comm_internal.h" + +#include + +/* === 全局变量定义 === */ +std::map g_instances; +std::map g_fd_map; +std::mutex g_mutex; +std::atomic g_running(false); +int g_epoll_fd = -1; +int g_next_id = 1; +int g_stop_event_fd = -1; + +/* ================================================================ + * 工具函数 + * ================================================================ */ + +/** + * @brief 初始化 epoll + eventfd（幂等，可重复调用） + * @return COMM_OK 成功, COMM_ERR_INTERNAL 失败 + */ +int comm_epoll_init(void) +{ + if (g_epoll_fd != -1) + { + return COMM_OK; /* 已初始化 */ + } + + g_epoll_fd = epoll_create1(0); + if (g_epoll_fd == -1) + { + return COMM_ERR_INTERNAL; + } + + g_stop_event_fd = eventfd(0, EFD_NONBLOCK); + if (g_stop_event_fd == -1) + { + close(g_epoll_fd); + g_epoll_fd = -1; + + return COMM_ERR_INTERNAL; + } + + struct epoll_event ev; + + memset(&ev, 0, sizeof(ev)); + ev.events = EPOLLIN; + ev.data.fd = g_stop_event_fd; + epoll_ctl(g_epoll_fd, EPOLL_CTL_ADD, g_stop_event_fd, &ev); + + return COMM_OK; +} + +/** + * @brief 设置 fd 为非阻塞模式 + */ +int set_nonblock(int fd) +{ + int flags = fcntl(fd, F_GETFL, 0); + + if (flags == -1) + { + return -1; + } + + return fcntl(fd, F_SETFL, flags | O_NONBLOCK); +} + +/** + * @brief 向 epoll 添加 fd + */ +int epoll_add_fd(int fd, int instance_id, int is_listen) +{ + struct epoll_event ev; + + memset(&ev, 0, sizeof(ev)); + ev.events = EPOLLIN | EPOLLERR | EPOLLHUP | EPOLLRDHUP; + ev.data.fd = fd; + + if (epoll_ctl(g_epoll_fd, EPOLL_CTL_ADD, fd, &ev) == -1) + { + return -1; + } + + stru_comm_fd_info info; + + info.fd = fd; + info.is_listen = is_listen; + info.instance_id = instance_id; + g_fd_map[fd] = info; + + return 0; +} + +/** + * @brief 从 epoll 移除 fd + */ +int epoll_del_fd(int fd) +{ + epoll_ctl(g_epoll_fd, EPOLL_CTL_DEL, fd, NULL); + g_fd_map.erase(fd); + + return 0; +} + +/** + * @brief 通知状态变化 + */ +void notify_state(stru_comm_instance *inst, int fd, int state) +{ + inst->state = state; + + if (inst->state_cb) + { + inst->state_cb(inst->id, fd, state); + } +} + +/** + * @brief 关闭客户端 fd 并清理 + */ +void close_client_fd(stru_comm_instance *inst, int fd) +{ + epoll_del_fd(fd); + close(fd); + + notify_state(inst, fd, COMM_STATE_DISCONNECTED); + + /* 从 client_fds 中移除 */ + for (size_t i = 0; i < inst->client_fds.size(); i++) + { + if (inst->client_fds[i] == fd) + { + inst->client_fds.erase(inst->client_fds.begin() + (long)i); + break; + } + } +} + +/** + * @brief 处理可写事件（TCP 客户端连接完成） + */ +void handle_write(int fd) +{ + struct epoll_event ev; + + memset(&ev, 0, sizeof(ev)); + ev.events = EPOLLIN | EPOLLERR | EPOLLHUP | EPOLLRDHUP; + ev.data.fd = fd; + epoll_ctl(g_epoll_fd, EPOLL_CTL_MOD, fd, &ev); +} + +/** + * @brief 处理错误事件 + */ +void handle_error(stru_comm_instance *inst, int fd) +{ + int error = 0; + socklen_t len = sizeof(error); + + getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &len); + + if (inst->debug) + { + MY_LOG_E("comm[%d] fd=%d error: %s", inst->id, fd, strerror(error)); + } + + notify_state(inst, fd, COMM_STATE_ERROR); +} + +/* ================================================================ + * TCP 数据读取 + * ================================================================ */ + +/** + * @brief 处理 TCP 可读事件 + */ +LOCAL void handle_tcp_read(stru_comm_instance *inst, int fd) +{ + unsigned char buf[COMM_RECV_BUF_SIZE]; + + while (1) + { + ssize_t n = recv(fd, buf, sizeof(buf), 0); + + if (n > 0) + { + if (inst->recv_cb) + { + inst->recv_cb(inst->id, fd, buf, (int)n); + } + } + else if (n == 0) + { + if (inst->debug) + { + MY_LOG_I("comm[%d] fd=%d connection closed by peer", inst->id, fd); + } + close_client_fd(inst, fd); + break; + } + else + { + if (errno == EAGAIN || errno == EWOULDBLOCK) + { + break; + } + + if (errno != EINTR) + { + if (inst->debug) + { + MY_LOG_E("comm[%d] fd=%d recv error: %s", inst->id, fd, strerror(errno)); + } + close_client_fd(inst, fd); + } + break; + } + } +} + +/* ================================================================ + * 事件循环 + * ================================================================ */ + +/** + * @brief epoll 事件循环工作函数 + */ +LOCAL void worker_loop(void) +{ + struct epoll_event events[COMM_MAX_EVENTS]; + + while (g_running) + { + int nfds = epoll_wait(g_epoll_fd, events, COMM_MAX_EVENTS, 100); + + if (nfds == -1) + { + if (errno == EINTR) + { + continue; + } + break; + } + + for (int i = 0; i < nfds; i++) + { + int fd = events[i].data.fd; + + /* 停止事件 */ + if (fd == g_stop_event_fd) + { + g_running = false; + break; + } + + std::lock_guard lock(g_mutex); + + /* 查找 fd 对应信息 */ + auto it = g_fd_map.find(fd); + if (it == g_fd_map.end()) + { + continue; + } + + int inst_id = it->second.instance_id; + int is_listen = it->second.is_listen; + + auto inst_it = g_instances.find(inst_id); + if (inst_it == g_instances.end()) + { + continue; + } + + stru_comm_instance *inst = inst_it->second; + + /* 错误/挂起 */ + if (events[i].events & (EPOLLERR | EPOLLHUP | EPOLLRDHUP)) + { + if (is_listen) + { + handle_error(inst, fd); + } + else + { + handle_error(inst, fd); + close_client_fd(inst, fd); + } + continue; + } + + /* 可写（TCP 客户端连接完成） */ + if (events[i].events & EPOLLOUT) + { + int error = 0; + socklen_t len = sizeof(error); + + getsockopt(fd, SOL_SOCKET, SO_ERROR, &error, &len); + + if (error == 0) + { + handle_write(fd); + notify_state(inst, fd, COMM_STATE_CONNECTED); + + if (inst->debug) + { + MY_LOG_I("comm[%d] TCP client connected fd=%d", inst->id, fd); + } + } + else + { + if (inst->debug) + { + MY_LOG_E("comm[%d] TCP connect failed: %s", inst->id, strerror(error)); + } + epoll_del_fd(fd); + close(fd); + inst->local_fd = -1; + notify_state(inst, -1, COMM_STATE_ERROR); + } + continue; + } + + /* 可读 */ + if (events[i].events & EPOLLIN) + { + if (is_listen) + { + handle_accept(inst); + } + else if (inst->type == COMM_TYPE_UDP_SERVER || inst->type == COMM_TYPE_UDP_CLIENT) + { + handle_udp_read(inst, fd); + } + else if (inst->type == COMM_TYPE_UART) + { + handle_uart_read(inst, fd); + } + else + { + handle_tcp_read(inst, fd); + } + } + } + } +} + +/* ================================================================ + * 对外 C 接口 + * ================================================================ */ + +/** + * @brief 创建通讯实例 + */ +int comm_create(int type, int debug, const union_comm_para *para) +{ + if (!para) + { + return COMM_ERR_PARAM; + } + + if (type < COMM_TYPE_TCP_SERVER || type > COMM_TYPE_UART) + { + return COMM_ERR_PARAM; + } + + std::lock_guard lock(g_mutex); + + /* 惰性初始化 epoll */ + int ret = comm_epoll_init(); + + if (ret != COMM_OK) + { + return ret; + } + + stru_comm_instance *inst = new stru_comm_instance(); + + if (!inst) + { + return COMM_ERR_MEM; + } + + inst->id = g_next_id++; + inst->type = type; + inst->debug = debug; + inst->state = COMM_STATE_DISCONNECTED; + inst->listen_fd = -1; + inst->local_fd = -1; + inst->recv_cb = NULL; + inst->state_cb = NULL; + + /* 拷贝参数 */ + switch (type) + { + case COMM_TYPE_TCP_SERVER: + strncpy(inst->local_ip, para->tcp.local_ip, sizeof(inst->local_ip) - 1); + inst->local_port = para->tcp.local_port; + break; + + case COMM_TYPE_TCP_CLIENT: + strncpy(inst->local_ip, para->tcp.local_ip, sizeof(inst->local_ip) - 1); + inst->local_port = para->tcp.local_port; + strncpy(inst->remote_ip, para->tcp.remote_ip, sizeof(inst->remote_ip) - 1); + inst->remote_port = para->tcp.remote_port; + break; + + case COMM_TYPE_UDP_SERVER: + strncpy(inst->local_ip, para->udp.local_ip, sizeof(inst->local_ip) - 1); + inst->local_port = para->udp.local_port; + break; + + case COMM_TYPE_UDP_CLIENT: + strncpy(inst->local_ip, para->udp.local_ip, sizeof(inst->local_ip) - 1); + inst->local_port = para->udp.local_port; + strncpy(inst->remote_ip, para->udp.remote_ip, sizeof(inst->remote_ip) - 1); + inst->remote_port = para->udp.remote_port; + break; + + case COMM_TYPE_UART: + strncpy(inst->uart_device, para->uart.device, sizeof(inst->uart_device) - 1); + inst->uart_baudrate = para->uart.baudrate; + inst->uart_data_bits = para->uart.data_bits; + inst->uart_stop_bits = para->uart.stop_bits; + inst->uart_parity = para->uart.parity; + break; + } + + g_instances[inst->id] = inst; + + if (debug) + { + MY_LOG_I("comm[%d] created type=%d", inst->id, type); + } + + return inst->id; +} + +/** + * @brief 建立连接 + */ +int comm_connect(int id) +{ + std::lock_guard lock(g_mutex); + + auto it = g_instances.find(id); + if (it == g_instances.end()) + { + return COMM_ERR_NOT_FOUND; + } + + stru_comm_instance *inst = it->second; + int ret = COMM_OK; + + switch (inst->type) + { + case COMM_TYPE_TCP_SERVER: + ret = tcp_server_bind(inst); + if (ret == COMM_OK) + { + notify_state(inst, inst->listen_fd, COMM_STATE_CONNECTED); + } + break; + + case COMM_TYPE_TCP_CLIENT: + ret = tcp_client_connect(inst); + break; + + case COMM_TYPE_UDP_SERVER: + case COMM_TYPE_UDP_CLIENT: + ret = udp_bind(inst); + if (ret == COMM_OK) + { + notify_state(inst, inst->local_fd, COMM_STATE_CONNECTED); + } + break; + + case COMM_TYPE_UART: + ret = uart_open(inst); + if (ret == COMM_OK) + { + notify_state(inst, inst->local_fd, COMM_STATE_CONNECTED); + } + break; + + default: + ret = COMM_ERR_PARAM; + break; + } + + return ret; +} + +/** + * @brief 断开连接 + */ +int comm_disconnect(int id) +{ + std::lock_guard lock(g_mutex); + + auto it = g_instances.find(id); + if (it == g_instances.end()) + { + return COMM_ERR_NOT_FOUND; + } + + stru_comm_instance *inst = it->second; + + /* 关闭所有客户端 fd */ + for (size_t i = 0; i < inst->client_fds.size(); i++) + { + int fd = inst->client_fds[i]; + + epoll_del_fd(fd); + close(fd); + notify_state(inst, fd, COMM_STATE_DISCONNECTED); + } + inst->client_fds.clear(); + + /* 关闭监听 fd */ + if (inst->listen_fd != -1) + { + epoll_del_fd(inst->listen_fd); + close(inst->listen_fd); + inst->listen_fd = -1; + } + + /* 关闭本地 fd */ + if (inst->local_fd != -1) + { + if (inst->type == COMM_TYPE_UART) + { + uart_close(inst); + } + else + { + epoll_del_fd(inst->local_fd); + close(inst->local_fd); + } + inst->local_fd = -1; + } + + inst->state = COMM_STATE_DISCONNECTED; + + if (inst->debug) + { + MY_LOG_I("comm[%d] disconnected", inst->id); + } + + return COMM_OK; +} + +/** + * @brief 发送数据 + */ +int comm_send(int id, int fd, const unsigned char *data, int len) +{ + if (!data || len <= 0) + { + return COMM_ERR_PARAM; + } + + std::lock_guard lock(g_mutex); + + auto it = g_instances.find(id); + if (it == g_instances.end()) + { + return COMM_ERR_NOT_FOUND; + } + + stru_comm_instance *inst = it->second; + + /* UDP 客户端使用 sendto */ + if (inst->type == COMM_TYPE_UDP_CLIENT) + { + return udp_sendto(inst, fd, data, len); + } + + ssize_t sent = send(fd, data, len, MSG_NOSIGNAL); + + if (sent == -1) + { + if (errno == EAGAIN || errno == EWOULDBLOCK) + { + return COMM_ERR_SEND; + } + + if (inst->debug) + { + MY_LOG_E("comm[%d] fd=%d send error: %s", inst->id, fd, strerror(errno)); + } + + return COMM_ERR_SEND; + } + + return (int)sent; +} + +/** + * @brief 注册接收回调 + */ +int comm_recv_register(int id, comm_recv_cb cb) +{ + std::lock_guard lock(g_mutex); + + auto it = g_instances.find(id); + if (it == g_instances.end()) + { + return COMM_ERR_NOT_FOUND; + } + + it->second->recv_cb = cb; + + return COMM_OK; +} + +/** + * @brief 注册状态回调 + */ +int comm_state_register(int id, comm_state_cb cb) +{ + std::lock_guard lock(g_mutex); + + auto it = g_instances.find(id); + if (it == g_instances.end()) + { + return COMM_ERR_NOT_FOUND; + } + + it->second->state_cb = cb; + + return COMM_OK; +} + +/** + * @brief 销毁通讯实例 + */ +int comm_destroy(int id) +{ + std::lock_guard lock(g_mutex); + + auto it = g_instances.find(id); + if (it == g_instances.end()) + { + return COMM_ERR_NOT_FOUND; + } + + stru_comm_instance *inst = it->second; + + /* 先断开所有连接 */ + for (size_t i = 0; i < inst->client_fds.size(); i++) + { + epoll_del_fd(inst->client_fds[i]); + close(inst->client_fds[i]); + } + inst->client_fds.clear(); + + if (inst->listen_fd != -1) + { + epoll_del_fd(inst->listen_fd); + close(inst->listen_fd); + } + + if (inst->local_fd != -1) + { + if (inst->type == COMM_TYPE_UART) + { + uart_close(inst); + } + else + { + epoll_del_fd(inst->local_fd); + close(inst->local_fd); + } + } + + g_instances.erase(it); + delete inst; + + return COMM_OK; +} + +/** + * @brief 启动事件循环（阻塞调用） + */ +int comm_run(void) +{ + /* 自初始化 epoll（不依赖 comm_create） */ + int ret = comm_epoll_init(); + + if (ret != COMM_OK) + { + return ret; + } + + g_running = true; + worker_loop(); + g_running = false; + + return COMM_OK; +} + +/** + * @brief 停止事件循环 + */ +void comm_stop(void) +{ + if (g_stop_event_fd != -1) + { + uint64_t val = 1; + + write(g_stop_event_fd, &val, sizeof(val)); + } + g_running = false; +} diff --git a/src/public/libcomm/src/comm_internal.h b/src/public/libcomm/src/comm_internal.h new file mode 100644 index 0000000..eb0efcd --- /dev/null +++ b/src/public/libcomm/src/comm_internal.h @@ -0,0 +1,98 @@ +/** + * @file comm_internal.h + * @brief libcomm 内部共享定义 + * @details 结构体、全局变量、内部函数声明，供 comm_core/tcp/udp/uart.cpp 共享。 + * 不对外暴露。 + */ +#ifndef _COMM_INTERNAL_H_ +#define _COMM_INTERNAL_H_ + +#include "myComm.h" + +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +/* === 常量 === */ +#define COMM_MAX_EVENTS 64 +#define COMM_RECV_BUF_SIZE 4096 +#define COMM_BACKLOG 128 + +/* === 内部 fd 信息 === */ +typedef struct +{ + int fd; + int is_listen; + int instance_id; +} stru_comm_fd_info; + +/* === 内部通讯实例 === */ +typedef struct +{ + int id; + int type; + int debug; + int state; + int listen_fd; + int local_fd; + + char local_ip[64]; + int local_port; + char remote_ip[64]; + int remote_port; + char uart_device[128]; + int uart_baudrate; + int uart_data_bits; + int uart_stop_bits; + char uart_parity; + + comm_recv_cb recv_cb; + comm_state_cb state_cb; + + std::vector client_fds; +} stru_comm_instance; + +/* === 全局状态（定义在 comm_core.cpp） === */ +extern std::map g_instances; +extern std::map g_fd_map; +extern std::mutex g_mutex; +extern std::atomic g_running; +extern int g_epoll_fd; +extern int g_next_id; +extern int g_stop_event_fd; + +/* === 内部函数声明 === */ +int comm_epoll_init(void); +int set_nonblock(int fd); +void notify_state(stru_comm_instance *inst, int fd, int state); +int epoll_add_fd(int fd, int instance_id, int is_listen); +int epoll_del_fd(int fd); +void close_client_fd(stru_comm_instance *inst, int fd); +void handle_write(int fd); +void handle_error(stru_comm_instance *inst, int fd); + +/* comm_tcp.cpp */ +void handle_accept(stru_comm_instance *inst); +int tcp_server_bind(stru_comm_instance *inst); +int tcp_client_connect(stru_comm_instance *inst); + +/* comm_udp.cpp */ +void handle_udp_read(stru_comm_instance *inst, int fd); +int udp_bind(stru_comm_instance *inst); +int udp_sendto(stru_comm_instance *inst, int fd, const unsigned char *data, int len); + +/* comm_uart.cpp */ +void handle_uart_read(stru_comm_instance *inst, int fd); +int uart_open(stru_comm_instance *inst); +void uart_close(stru_comm_instance *inst); + +#endif diff --git a/src/public/libcomm/src/comm_tcp.cpp b/src/public/libcomm/src/comm_tcp.cpp new file mode 100644 index 0000000..b16900c --- /dev/null +++ b/src/public/libcomm/src/comm_tcp.cpp @@ -0,0 +1,217 @@ +/** + * @file comm_tcp.cpp + * @brief 通讯抽象层 — TCP 服务器/客户端 + * @details TCP 服务器: socket → bind → listen → accept（epoll 驱动） + * TCP 客户端: socket → nonblock-connect → EPOLLOUT 检测完成 + */ + +#include "comm_internal.h" + +#include +#include +#include + +/** + * @brief 处理 TCP accept + */ +void handle_accept(stru_comm_instance *inst) +{ + struct sockaddr_in client_addr; + socklen_t addr_len = sizeof(client_addr); + char ip_str[INET_ADDRSTRLEN]; + + while (1) + { + int client_fd = accept(inst->listen_fd, (struct sockaddr *)&client_addr, &addr_len); + + if (client_fd == -1) + { + if (errno == EAGAIN || errno == EWOULDBLOCK) + { + break; + } + + if (inst->debug) + { + MY_LOG_E("comm[%d] accept error: %s", inst->id, strerror(errno)); + } + break; + } + + set_nonblock(client_fd); + + /* 禁用 Nagle 算法 */ + int opt = 1; + + setsockopt(client_fd, IPPROTO_TCP, TCP_NODELAY, &opt, sizeof(opt)); + + if (epoll_add_fd(client_fd, inst->id, 0) != 0) + { + close(client_fd); + break; + } + + inst->client_fds.push_back(client_fd); + + inet_ntop(AF_INET, &client_addr.sin_addr, ip_str, sizeof(ip_str)); + + if (inst->debug) + { + MY_LOG_I("comm[%d] TCP client connected: %s:%d fd=%d", + inst->id, ip_str, ntohs(client_addr.sin_port), client_fd); + } + + notify_state(inst, client_fd, COMM_STATE_CONNECTED); + } +} + +/** + * @brief TCP 服务器 bind + listen + */ +int tcp_server_bind(stru_comm_instance *inst) +{ + int fd; + struct sockaddr_in addr; + int opt = 1; + + fd = socket(AF_INET, SOCK_STREAM, 0); + if (fd == -1) + { + return COMM_ERR_SOCKET; + } + + setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)); + set_nonblock(fd); + + memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_port = htons(inst->local_port); + + if (inst->local_ip[0] != '\0') + { + inet_pton(AF_INET, inst->local_ip, &addr.sin_addr); + } + else + { + addr.sin_addr.s_addr = INADDR_ANY; + } + + if (bind(fd, (struct sockaddr *)&addr, sizeof(addr)) == -1) + { + close(fd); + + return COMM_ERR_BIND; + } + + if (listen(fd, COMM_BACKLOG) == -1) + { + close(fd); + + return COMM_ERR_LISTEN; + } + + if (epoll_add_fd(fd, inst->id, 1) != 0) + { + close(fd); + + return COMM_ERR_INTERNAL; + } + + inst->listen_fd = fd; + + if (inst->debug) + { + MY_LOG_I("comm[%d] TCP server listening on %s:%d", + inst->id, inst->local_ip, inst->local_port); + } + + return COMM_OK; +} + +/** + * @brief TCP 客户端 connect（非阻塞） + */ +int tcp_client_connect(stru_comm_instance *inst) +{ + int fd; + struct sockaddr_in addr; + + fd = socket(AF_INET, SOCK_STREAM, 0); + if (fd == -1) + { + return COMM_ERR_SOCKET; + } + + set_nonblock(fd); + + /* 禁用 Nagle 算法 */ + int opt = 1; + + setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &opt, sizeof(opt)); + + /* 如果指定了本地 IP:Port，先 bind */ + if (inst->local_ip[0] != '\0' && inst->local_port > 0) + { + struct sockaddr_in local_addr; + + memset(&local_addr, 0, sizeof(local_addr)); + local_addr.sin_family = AF_INET; + local_addr.sin_port = htons(inst->local_port); + inet_pton(AF_INET, inst->local_ip, &local_addr.sin_addr); + + if (bind(fd, (struct sockaddr *)&local_addr, sizeof(local_addr)) == -1) + { + if (inst->debug) + { + MY_LOG_E("comm[%d] TCP client local bind failed: %s", inst->id, strerror(errno)); + } + } + } + + memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_port = htons(inst->remote_port); + inet_pton(AF_INET, inst->remote_ip, &addr.sin_addr); + + int ret = connect(fd, (struct sockaddr *)&addr, sizeof(addr)); + + if (ret == -1 && errno != EINPROGRESS) + { + close(fd); + + return COMM_ERR_CONNECT; + } + + /* 注册到 epoll，监听 EPOLLOUT 检测连接完成 */ + struct epoll_event ev; + + memset(&ev, 0, sizeof(ev)); + ev.events = EPOLLIN | EPOLLOUT | EPOLLERR | EPOLLHUP | EPOLLRDHUP; + ev.data.fd = fd; + + if (epoll_ctl(g_epoll_fd, EPOLL_CTL_ADD, fd, &ev) == -1) + { + close(fd); + + return COMM_ERR_INTERNAL; + } + + stru_comm_fd_info info; + + info.fd = fd; + info.is_listen = 0; + info.instance_id = inst->id; + g_fd_map[fd] = info; + + inst->local_fd = fd; + + if (inst->debug) + { + MY_LOG_I("comm[%d] TCP client connecting to %s:%d", + inst->id, inst->remote_ip, inst->remote_port); + } + + notify_state(inst, fd, COMM_STATE_CONNECTING); + + return COMM_OK; +} diff --git a/src/public/libcomm/src/comm_uart.cpp b/src/public/libcomm/src/comm_uart.cpp new file mode 100644 index 0000000..4407e3b --- /dev/null +++ b/src/public/libcomm/src/comm_uart.cpp @@ -0,0 +1,228 @@ +/** + * @file comm_uart.cpp + * @brief 通讯抽象层 — UART 串口通讯 + * @details 串口设备打开、波特率/数据位/停止位/校验位配置、 + * 非阻塞读写、epoll 集成。 + */ + +#include "comm_internal.h" + +#include + +/** + * @brief 波特率常量 → termios 速度标志 + */ +LOCAL speed_t uart_get_baudrate(int baudrate) +{ + switch (baudrate) + { + case 9600: return B9600; + case 19200: return B19200; + case 38400: return B38400; + case 57600: return B57600; + case 115200: return B115200; + case 230400: return B230400; + case 460800: return B460800; + case 921600: return B921600; + default: return B115200; + } +} + +/** + * @brief 数据位数 → termios CS 标志 + */ +LOCAL int uart_get_databits(int bits) +{ + switch (bits) + { + case 5: return CS5; + case 6: return CS6; + case 7: return CS7; + case 8: + default: return CS8; + } +} + +/** + * @brief 配置串口参数 + */ +LOCAL int uart_configure(int fd, stru_comm_instance *inst) +{ + struct termios tty; + + memset(&tty, 0, sizeof(tty)); + + if (tcgetattr(fd, &tty) != 0) + { + return -1; + } + + /* 输入标志：关闭软件流控 */ + tty.c_iflag &= ~(IGNBRK | BRKINT | PARMRK | ISTRIP | INLCR | IGNCR | ICRNL | IXON); + tty.c_iflag |= IGNPAR; + + /* 输出标志：原始模式 */ + tty.c_oflag &= ~OPOST; + + /* 控制标志 */ + tty.c_cflag &= ~(CSIZE | PARENB | CSTOPB | CRTSCTS); + tty.c_cflag |= CREAD | CLOCAL; + + /* 数据位 */ + tty.c_cflag |= uart_get_databits(inst->uart_data_bits); + + /* 停止位 */ + if (inst->uart_stop_bits == 2) + { + tty.c_cflag |= CSTOPB; + } + + /* 校验位 */ + switch (inst->uart_parity) + { + case 'E': + case 'e': + tty.c_cflag |= PARENB; + tty.c_iflag |= INPCK; + break; + + case 'O': + case 'o': + tty.c_cflag |= (PARENB | PARODD); + tty.c_iflag |= INPCK; + break; + + case 'N': + case 'n': + default: + /* 无校验 */ + break; + } + + /* 本地模式：原始模式 */ + tty.c_lflag &= ~(ICANON | ECHO | ECHOE | ECHONL | ISIG | IEXTEN); + + /* 超时：0.1 秒 */ + tty.c_cc[VMIN] = 0; + tty.c_cc[VTIME] = 1; + + /* 波特率 */ + speed_t speed = uart_get_baudrate(inst->uart_baudrate); + + cfsetispeed(&tty, speed); + cfsetospeed(&tty, speed); + + if (tcsetattr(fd, TCSANOW, &tty) != 0) + { + return -1; + } + + /* 清空缓冲区 */ + tcflush(fd, TCIOFLUSH); + + return 0; +} + +/** + * @brief 打开串口设备 + */ +int uart_open(stru_comm_instance *inst) +{ + int fd = open(inst->uart_device, O_RDWR | O_NOCTTY | O_NONBLOCK); + + if (fd == -1) + { + if (inst->debug) + { + MY_LOG_E("comm[%d] UART open %s failed: %s", + inst->id, inst->uart_device, strerror(errno)); + } + + return COMM_ERR_SOCKET; + } + + if (uart_configure(fd, inst) != 0) + { + if (inst->debug) + { + MY_LOG_E("comm[%d] UART configure failed: %s", + inst->id, strerror(errno)); + } + close(fd); + + return COMM_ERR_INTERNAL; + } + + if (epoll_add_fd(fd, inst->id, 0) != 0) + { + close(fd); + + return COMM_ERR_INTERNAL; + } + + inst->local_fd = fd; + + if (inst->debug) + { + MY_LOG_I("comm[%d] UART opened %s baud=%d databits=%d stopbits=%d parity=%c", + inst->id, inst->uart_device, inst->uart_baudrate, + inst->uart_data_bits, inst->uart_stop_bits, inst->uart_parity); + } + + return COMM_OK; +} + +/** + * @brief 关闭串口 + */ +void uart_close(stru_comm_instance *inst) +{ + if (inst->local_fd != -1) + { + epoll_del_fd(inst->local_fd); + close(inst->local_fd); + inst->local_fd = -1; + } +} + +/** + * @brief 处理 UART 可读事件 + */ +void handle_uart_read(stru_comm_instance *inst, int fd) +{ + unsigned char buf[COMM_RECV_BUF_SIZE]; + + while (1) + { + ssize_t n = read(fd, buf, sizeof(buf)); + + if (n > 0) + { + if (inst->recv_cb) + { + inst->recv_cb(inst->id, fd, buf, (int)n); + } + } + else if (n == 0) + { + break; + } + else + { + if (errno == EAGAIN || errno == EWOULDBLOCK) + { + break; + } + + if (errno != EINTR) + { + if (inst->debug) + { + MY_LOG_E("comm[%d] UART fd=%d read error: %s", + inst->id, fd, strerror(errno)); + } + } + break; + } + } +} diff --git a/src/public/libcomm/src/comm_udp.cpp b/src/public/libcomm/src/comm_udp.cpp new file mode 100644 index 0000000..e4c6bca --- /dev/null +++ b/src/public/libcomm/src/comm_udp.cpp @@ -0,0 +1,133 @@ +/** + * @file comm_udp.cpp + * @brief 通讯抽象层 — UDP 通讯 + * @details UDP 服务器/客户端：socket → bind → recvfrom/sendto + */ + +#include "comm_internal.h" + +#include +#include + +/** + * @brief 处理 UDP 可读事件 + */ +void handle_udp_read(stru_comm_instance *inst, int fd) +{ + unsigned char buf[COMM_RECV_BUF_SIZE]; + + while (1) + { + struct sockaddr_in from_addr; + socklen_t from_len = sizeof(from_addr); + + ssize_t n = recvfrom(fd, buf, sizeof(buf), 0, + (struct sockaddr *)&from_addr, &from_len); + + if (n > 0) + { + if (inst->recv_cb) + { + inst->recv_cb(inst->id, fd, buf, (int)n); + } + } + else + { + if (errno == EAGAIN || errno == EWOULDBLOCK) + { + break; + } + + if (errno != EINTR) + { + if (inst->debug) + { + MY_LOG_E("comm[%d] fd=%d recvfrom error: %s", + inst->id, fd, strerror(errno)); + } + } + break; + } + } +} + +/** + * @brief UDP bind + */ +int udp_bind(stru_comm_instance *inst) +{ + int fd; + struct sockaddr_in addr; + + fd = socket(AF_INET, SOCK_DGRAM, 0); + if (fd == -1) + { + return COMM_ERR_SOCKET; + } + + set_nonblock(fd); + + memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_port = htons(inst->local_port); + + if (inst->local_ip[0] != '\0') + { + inet_pton(AF_INET, inst->local_ip, &addr.sin_addr); + } + else + { + addr.sin_addr.s_addr = INADDR_ANY; + } + + if (bind(fd, (struct sockaddr *)&addr, sizeof(addr)) == -1) + { + close(fd); + + return COMM_ERR_BIND; + } + + if (epoll_add_fd(fd, inst->id, 0) != 0) + { + close(fd); + + return COMM_ERR_INTERNAL; + } + + inst->local_fd = fd; + + if (inst->debug) + { + MY_LOG_I("comm[%d] UDP bound to %s:%d", + inst->id, inst->local_ip, inst->local_port); + } + + return COMM_OK; +} + +/** + * @brief UDP 发送（sendto 到远端地址） + */ +int udp_sendto(stru_comm_instance *inst, int fd, const unsigned char *data, int len) +{ + struct sockaddr_in addr; + + memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_port = htons(inst->remote_port); + inet_pton(AF_INET, inst->remote_ip, &addr.sin_addr); + + ssize_t sent = sendto(fd, data, len, 0, (struct sockaddr *)&addr, sizeof(addr)); + + if (sent == -1) + { + if (inst->debug) + { + MY_LOG_E("comm[%d] fd=%d sendto error: %s", inst->id, fd, strerror(errno)); + } + + return COMM_ERR_SEND; + } + + return (int)sent; +}