jianf/chat_rebot-connect-with-onebot-standard-
Template
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

修复部分bug

Browse Source
This commit is contained in:
jianf
2025-11-21 18:02:37 +08:00
parent 26dac05e5b
commit 08589dfe79
7 changed files with 153 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
c/network/http/http_rel.c
View File

@ -3,7 +3,7 @@
#include <string.h>
#include <unistd.h>
#include "network/mongoose/mongoose.h"
#include "tools/log/log.h"
#include "http_rel.h"
/* 接收状态辅助结构 */
struct recv_state {
@ -12,10 +12,11 @@ struct recv_state {
int error; // 错误标志
};
/* mongoose 事件处理函数 */
static void http_recv_handler(struct mg_connection *c, int ev, void *ev_data, void *fn_data)
static void http_recv_handler(struct mg_connection *c, int ev, void *ev_data)
{
struct recv_state *state = (struct recv_state *)fn_data;
// 从连接对象获取用户数据
struct recv_state *state = (struct recv_state *)c->fn_data;
switch (ev) {
case MG_EV_HTTP_MSG: {
@ -48,7 +49,7 @@ char *recv_http_request(int cfd)
mg_mgr_init(&mgr);
/* 将已连接的 socket 包装成 mongoose 连接 */
c = mg_wrapfd(&mgr, cfd, http_recv_handler, &state); // ← 修复此处
c = mg_wrapfd(&mgr, cfd, http_recv_handler, &state);
/* 设置 5 秒超时 */
int64_t end_time = mg_millis() + 5000;
@ -66,7 +67,12 @@ char *recv_http_request(int cfd)
/* 出错时释放内存 */
if (state.error) {
free(state.request);
free(state.request); // 安全释放（free(NULL) 是安全的）
return NULL;
}
/* 确保返回的请求不为空 */
if (!state.request) {
return NULL;
}
@ -83,3 +89,92 @@ const char *http_get_body(const char *buf)
if (*body == '\0') return NULL;
return body;
}
/**
* @brief 初始化HTTP监听socket，所有错误通过logmanager记录
* @param port 监听端口
* @param logger 日志管理器实例指针
* @return 成功返回监听fd，失败返回-1并记录日志
*/
int init_http_network(int port, log_manager *logger)
{
logs *log;
int fd;
/* 1. 创建socket */
fd = socket(AF_INET, SOCK_STREAM, 0);
if (fd == -1) {
log = malloc(sizeof(logs));
// cppcheck-suppress uninitdata
snprintf(log->log, sizeof(log->log),
"[FATAL] socket() failed: %s", strerror(errno));
logger->in_log(log, logger);
return -1;
}
/* 2. 设置SO_REUSEADDR，避免TIME_WAIT状态导致bind失败 */
int opt = 1;
if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) == -1) {
log = malloc(sizeof(logs));
snprintf(log->log, sizeof(log->log),
"[ERROR] setsockopt(SO_REUSEADDR) on fd=%d failed: %s",
fd, strerror(errno));
logger->in_log(log, logger);
close(fd);
return -1;
}
/* 3. 设置为非阻塞模式（配合epoll使用） */
int flags = fcntl(fd, F_GETFL, 0);
if (flags == -1) {
log = malloc(sizeof(logs));
snprintf(log->log, sizeof(log->log),
"[ERROR] fcntl(F_GETFL) on fd=%d failed: %s", fd, strerror(errno));
logger->in_log(log, logger);
close(fd);
return -1;
}
if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) {
log = malloc(sizeof(logs));
snprintf(log->log, sizeof(log->log),
"[ERROR] fcntl(O_NONBLOCK) on fd=%d failed: %s", fd, strerror(errno));
logger->in_log(log, logger);
close(fd);
return -1;
}
/* 4. 绑定到指定端口 */
struct sockaddr_in addr = {0};
addr.sin_family = AF_INET;
addr.sin_port = htons(port);
addr.sin_addr.s_addr = htonl(INADDR_ANY); // 监听所有网卡
if (bind(fd, (struct sockaddr *)&addr, sizeof(addr)) == -1) {
log = malloc(sizeof(logs));
snprintf(log->log, sizeof(log->log),
"[FATAL] bind(port %d) failed: %s (fd=%d)",
port, strerror(errno), fd);
logger->in_log(log, logger);
close(fd);
return -1;
}
/* 5. 开始监听 */
if (listen(fd, 10) == -1) {
log = malloc(sizeof(logs));
snprintf(log->log, sizeof(log->log),
"[FATAL] listen(fd=%d, backlog=10) failed: %s",
fd, strerror(errno));
logger->in_log(log, logger);
close(fd);
return -1;
}
/* 6. 成功日志 */
log = malloc(sizeof(logs));
snprintf(log->log, sizeof(log->log),
"[HTTP] Successfully listening on port %d (fd=%d)", port, fd);
logger->in_log(log, logger);
return fd;
}

1
c/network/http/http_rel.h
View File

@ -3,5 +3,6 @@
const char *http_get_body(const char *buf);
char *recv_http_request(int cfd);
int init_http_network(int port, log_manager *logger);
#endif

108
c/network/network.c
View File

@ -7,6 +7,7 @@
#include "tools/log/log.h"
#include "tools/quit/quit.h"
#include "erroprocess/erroprocess.h"
#include <semaphore.h>
#include <unistd.h>
#include <stdio.h>
@ -18,6 +19,7 @@
#include <sys/socket.h>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <sys/mman.h>
static void safe_strcpy(char *dst, size_t dst_size, const char *src)
{
@ -77,94 +79,6 @@ int rbt_parse_json(const char *json_text, rbt_msg *out)
return 0; // 成功
}
/**
* @brief 初始化HTTP监听socket，所有错误通过logmanager记录
* @param port 监听端口
* @param logger 日志管理器实例指针
* @return 成功返回监听fd，失败返回-1并记录日志
*/
int init_http_network(int port, log_manager *logger)
{
logs *log;
int fd;
/* 1. 创建socket */
fd = socket(AF_INET, SOCK_STREAM, 0);
if (fd == -1) {
log = malloc(sizeof(logs));
// cppcheck-suppress uninitdata
snprintf(log->log, sizeof(log->log),
"[FATAL] socket() failed: %s", strerror(errno));
logger->in_log(log, logger);
return -1;
}
/* 2. 设置SO_REUSEADDR，避免TIME_WAIT状态导致bind失败 */
int opt = 1;
if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) == -1) {
log = malloc(sizeof(logs));
snprintf(log->log, sizeof(log->log),
"[ERROR] setsockopt(SO_REUSEADDR) on fd=%d failed: %s",
fd, strerror(errno));
logger->in_log(log, logger);
close(fd);
return -1;
}
/* 3. 设置为非阻塞模式（配合epoll使用） */
int flags = fcntl(fd, F_GETFL, 0);
if (flags == -1) {
log = malloc(sizeof(logs));
snprintf(log->log, sizeof(log->log),
"[ERROR] fcntl(F_GETFL) on fd=%d failed: %s", fd, strerror(errno));
logger->in_log(log, logger);
close(fd);
return -1;
}
if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) {
log = malloc(sizeof(logs));
snprintf(log->log, sizeof(log->log),
"[ERROR] fcntl(O_NONBLOCK) on fd=%d failed: %s", fd, strerror(errno));
logger->in_log(log, logger);
close(fd);
return -1;
}
/* 4. 绑定到指定端口 */
struct sockaddr_in addr = {0};
addr.sin_family = AF_INET;
addr.sin_port = htons(port);
addr.sin_addr.s_addr = htonl(INADDR_ANY); // 监听所有网卡
if (bind(fd, (struct sockaddr *)&addr, sizeof(addr)) == -1) {
log = malloc(sizeof(logs));
snprintf(log->log, sizeof(log->log),
"[FATAL] bind(port %d) failed: %s (fd=%d)",
port, strerror(errno), fd);
logger->in_log(log, logger);
close(fd);
return -1;
}
/* 5. 开始监听 */
if (listen(fd, 10) == -1) {
log = malloc(sizeof(logs));
snprintf(log->log, sizeof(log->log),
"[FATAL] listen(fd=%d, backlog=10) failed: %s",
fd, strerror(errno));
logger->in_log(log, logger);
close(fd);
return -1;
}
/* 6. 成功日志 */
log = malloc(sizeof(logs));
snprintf(log->log, sizeof(log->log),
"[HTTP] Successfully listening on port %d (fd=%d)", port, fd);
logger->in_log(log, logger);
return fd;
}
ssize_t read_req(int fd, void *buf)
{
@ -175,7 +89,7 @@ ssize_t read_req(int fd, void *buf)
return (n > 0) ? n : -1;
}
int process_message(char *req, log_manager *logger) {
int process_message(char *req, log_manager *logger,rbt_msg *swap) {
if(req == NULL) return 0;
int fd;
@ -195,18 +109,15 @@ int process_message(char *req, log_manager *logger) {
const char *body = http_get_body(req_buf);
free(req_buf);
rbt_msg message;
if(rbt_parse_json(body, &message) == 0) {
make_swap((void*)&message);
if(rbt_parse_json(body,swap) == 0) {
logs *log = malloc(sizeof(logs));
// cppcheck-suppress uninitdata
snprintf(log->log, sizeof(log->log), "%s message %s processed ok\n",
message.nickname, message.raw_message);
swap->nickname,swap->raw_message);
make_swap(swap);
logger->in_log(log, logger);
}
// ****** 修复2：发送HTTP响应 ******
const char *response =
"HTTP/1.1 200 OK\r\n"
"Content-Type: text/plain\r\n"
@ -248,10 +159,10 @@ void *pth_module(void *args_p)
free(args_p);
char name[256] = {'\0'};
sprintf(name,"chatrebot%lu",pthread_self());
int swap = create_swap(name);
int swapfd = create_swap(name);
//创建共享内存
char swap_arg[64] = {'\0'};
sprintf(swap_arg,"%d",swap);
sprintf(swap_arg,"%d",swapfd);
pid_t id = fork();
if(id == 0)
{
@ -266,6 +177,7 @@ void *pth_module(void *args_p)
sprintf(pth_log->log,"PID:%lu launched python plugines\n",pthread_self());
logger->in_log(pth_log,logger);
rbt_msg *swap = (rbt_msg*)mmap(NULL, sizeof(rbt_msg), PROT_READ|PROT_WRITE, MAP_SHARED,swapfd, 0);
//拉起python插件管理器
for(;;){
//线程池中，单个线程模型
@ -283,7 +195,7 @@ void *pth_module(void *args_p)
pth_log = (logs*)malloc(sizeof(logs));
sprintf(pth_log->log,"processd message");
logger->in_log(pth_log,logger);
process_message(req,logger);
process_message(req,logger,swap);
atomic_fetch_add(&pmd->status, 1);
}
}

11
c/network/network.h
View File

@ -14,20 +14,21 @@ typedef struct pthread_module
int fifo_fd[2];
atomic_int status;
}pth_m;
//打包线程模型参数
typedef struct net_args
{
log_manager *log;
pth_m *pth;
}net_args;
typedef struct network_manager
typedef struct network_manager//网络管理器
{
pth_m pool[MAX_POOL];
void *(*run_network)(void*);//启动网络监听
int (*start_pool)(struct network_manager*);
int (*shutdown_pool)(struct network_manager*);
int (*iss_work)(struct network_manager*,char *);
pth_m pool[MAX_POOL];
int fifo_fd[2];
log_manager *logmanager;
indiector *err_indictor;
@ -39,10 +40,10 @@ typedef struct network_manager
typedef struct rebot_message
{
char raw_message[MAX_MESSAGE_BUF];
char nickname[64];
char gid[32];
char uid[32];
char nickname[64];
char raw_message[MAX_MESSAGE_BUF];
char message_type;
sem_t status;
int state;

33
c/network/swap.c
View File

@ -8,27 +8,46 @@
#include <linux/memfd.h>
#include <sys/syscall.h>
#include <sys/mman.h>
#include "network.h"
#include <fcntl.h>
int make_swap(void *message)
#include "network.h"
#include "swap.h"
int make_swap(rbt_msg *swap)
{
rbt_msg *msg = (rbt_msg*)message;
printf("gid=%s uid=%s nick=%s raw=%s type=%c\n",
msg->gid, msg->uid, msg->nickname,
msg->raw_message, msg->message_type);
swap->state = NEWMSG;
sem_post(&swap->status);
}
int create_swap(const char *name)
{
int fd = memfd_create(name,0);
//申请共享内存
ftruncate(fd, sizeof(rbt_msg));
//调整关闭策略
int flags = fcntl(fd, F_GETFD);
flags &= ~FD_CLOEXEC;
fcntl(fd, F_SETFD, flags);
//调整大小
rbt_msg *init_msg = (rbt_msg*)mmap(NULL, sizeof(rbt_msg), PROT_READ|PROT_WRITE, MAP_SHARED, fd, 0);
char buf[MAX_MESSAGE_BUF] = {'\0'};
//初始化
memcpy(init_msg->raw_message,buf,MAX_MESSAGE_BUF);
memcpy(init_msg->nickname,buf,64);
sem_init(&init_msg->status,1,1);
init_msg->raw_message[0] = '\0';
init_msg->state = FREE;
init_msg->uid[0] = '\0';
munmap((void*)init_msg,sizeof(rbt_msg));
//初始化
return fd;
}
int close_swap(int shmid,rbt_msg *swap)
{
swap->state = QUITPLG;//置退出态
sem_post(&swap->status);//发送信号量
close(shmid);//关闭共享内存
}

7
c/network/swap.h
View File

@ -1,7 +1,12 @@
#ifndef SWAP
#define SWAP
int make_swap(void *);
#define QUITPLG 0
#define NEWMSG 1
#define FREE 2
int make_swap(rbt_msg *swap);
int create_swap(const char *name);
int close_swap(int shmid,rbt_msg *swap);
#endif

2
c/run_pluginmanager/run_pluginmanager.c
View File

@ -4,6 +4,6 @@ int main(int argc,char **argv)
{
for(;;)
{
sleep(10);
}
}