ypc
/
RTU_ALL_AI
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(libcomm): 通讯抽象层模块 — TCP/UDP/UART + epoll 异步事件循环 

- 纯 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() 事件循环控制
This commit is contained in:
ypc 2026-07-08 13:35:23 +08:00
parent 72e5e37c2d
commit 9d4c2902c3
7 changed files with 1616 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
release/src/public/libcomm/makefile Normal file
View File

@ -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

187
src/public/libcomm/inc/myComm.h Normal file
View File

@ -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_ */

722
src/public/libcomm/src/comm_core.cpp Normal file
View File

@ -0,0 +1,722 @@
/**
* @file comm_core.cpp
* @brief 通讯抽象层 — 核心模块
* @details 全局变量定义、epoll 工具函数、事件循环主循环、
* comm_create/connect/disconnect/destroy/run/stop 实现。
*/
#include "comm_internal.h"
#include <sys/eventfd.h>
/* === 全局变量定义 === */
std::map<int, stru_comm_instance *> g_instances;
std::map<int, stru_comm_fd_info> g_fd_map;
std::mutex g_mutex;
std::atomic<bool> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> 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;
}

98
src/public/libcomm/src/comm_internal.h Normal file
View File

@ -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 <map>
#include <vector>
#include <mutex>
#include <thread>
#include <atomic>
#include <cstring>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <sys/socket.h>
#include <sys/epoll.h>
/* === 常量 === */
#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<int> client_fds;
} stru_comm_instance;
/* === 全局状态（定义在 comm_core.cpp） === */
extern std::map<int, stru_comm_instance *> g_instances;
extern std::map<int, stru_comm_fd_info> g_fd_map;
extern std::mutex g_mutex;
extern std::atomic<bool> 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

217
src/public/libcomm/src/comm_tcp.cpp Normal file
View File

@ -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 <arpa/inet.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
/**
* @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;
}

228
src/public/libcomm/src/comm_uart.cpp Normal file
View File

@ -0,0 +1,228 @@
/**
* @file comm_uart.cpp
* @brief 通讯抽象层 — UART 串口通讯
* @details 串口设备打开、波特率/数据位/停止位/校验位配置、
* 非阻塞读写、epoll 集成。
*/
#include "comm_internal.h"
#include <termios.h>
/**
* @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;
}
}
}

133
src/public/libcomm/src/comm_udp.cpp Normal file
View File

@ -0,0 +1,133 @@
/**
* @file comm_udp.cpp
* @brief 通讯抽象层 — UDP 通讯
* @details UDP 服务器/客户端：socket → bind → recvfrom/sendto
*/
#include "comm_internal.h"
#include <arpa/inet.h>
#include <netinet/in.h>
/**
* @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;
}