SORT(*libmsg_center*.a:*.o*) (.data .data* .sdata .sdata*)
SORT(*ohos*.a:*.o*) (.data .data* .sdata .sdata*)
SORT(*liblwIP*.a:*.o*) (.data .data* .sdata .sdata*)
+ SORT(*libslip*.a:*.o*) (.data .data* .sdata .sdata*)
SORT(*xts_test*.a:*.o*) (.data .data* .sdata .sdata*)
*libnativelauncher*.a:*.o* (.data .data* .sdata .sdata*)
SORT(*application/wearable*.a:*.o*)(.data .data* .sdata .sdata*)
SORT(*libmsg_center*.a:*.o*) (.bss .bss* .sbss .sbss* COMMON)
SORT(*ohos*.a:*.o*) (.bss .bss* .sbss .sbss* COMMON)
SORT(*liblwIP*.a:*.o*) (.bss .bss* .sbss .sbss* COMMON)
+ SORT(*libslip*.a:*.o*) (.bss .bss* .sbss .sbss* COMMON)
SORT(*xts_test*.a:*.o*) (.bss .bss* .sbss .sbss* COMMON)
*libnativelauncher*.a:*.o* (.bss .bss* .sbss .sbss* COMMON)
SORT(*application/wearable*.a:*.o*) (.bss .bss* .sbss .sbss* COMMON)
-\r
-#ifndef __BK_WEBSOCKET_H__\r
-#define __BK_WEBSOCKET_H__\r
-// #include "aic_osa.h"\r
-// #include <bits/types.h>\r
-// WebSocket 消息类型枚举\r
-\r
-#ifdef __cplusplus\r
-#if __cplusplus\r
-extern "C" {\r
-#endif /* __cplusplus */\r
-#endif /* __cplusplus */\r
-\r
-typedef enum {\r
- WS_CONT = 0x00,\r
- WS_TEXT = 0x01,\r
- WS_BINARY = 0x02,\r
- WS_CLOSE = 0x8,\r
- WS_PING = 0x9,\r
- WS_PONG = 0xa\r
-} WebSocketMessageType;\r
-\r
-/* WebSocket数据包回调函数类型 */\r
-typedef void (*on_ws_packet_callback)(uint8_t* data, size_t datalen, WebSocketMessageType type);\r
-\r
-/* WebSocket组包缓冲区配置 */\r
-#define WS_FRAME_BUFFER_SIZE 256\r
-#define WS_PAYLOAD_BUFFER_SIZE 128\r
-\r
-/* WebSocket组包状态 */\r
-typedef enum {\r
- WS_PARSE_HEADER, /* 解析帧头 */\r
- WS_PARSE_EXTENDED, /* 解析扩展长度 */\r
- WS_PARSE_MASK, /* 解析掩码 */\r
- WS_PARSE_PAYLOAD /* 解析载荷数据 */\r
-} ws_parse_state_t;\r
-\r
-/* WebSocket组包缓冲区结构 */\r
-typedef struct {\r
- uint8_t frame_buffer[WS_FRAME_BUFFER_SIZE]; /* 帧缓冲区 */\r
- uint32_t frame_buffer_len; /* 帧缓冲区当前长度 */\r
- \r
- uint8_t payload_buffer[WS_PAYLOAD_BUFFER_SIZE]; /* 载荷缓冲区 */\r
- uint32_t payload_buffer_len; /* 载荷缓冲区当前长度 */\r
- \r
- ws_parse_state_t parse_state; /* 解析状态 */\r
- uint8_t fin; /* FIN标志 */\r
- WebSocketMessageType opcode; /* 操作码 */\r
- uint8_t mask; /* 掩码标志 */\r
- uint64_t payload_len; /* 载荷长度 */\r
- uint32_t payload_received; /* 已接收载荷长度 */\r
- uint8_t masking_key[4]; /* 掩码密钥 */\r
- \r
- on_ws_packet_callback packet_callback; /* 数据包回调函数 */\r
-} ws_packet_parser_t;\r
-\r
-/* WebSocket消息队列系统 */\r
-#define WS_QUEUE_SIZE 4\r
-#define WS_MESSAGE_MAX_SIZE 512\r
-\r
-typedef struct {\r
- uint8_t data[WS_MESSAGE_MAX_SIZE];\r
- uint16_t length;\r
- WebSocketMessageType type;\r
- uint32_t timestamp;\r
- bool in_use;\r
-} ws_message_item_t;\r
-\r
-typedef struct {\r
- ws_message_item_t messages[WS_QUEUE_SIZE];\r
- uint8_t head;\r
- uint8_t tail;\r
- uint8_t count;\r
- osMutexId_t mutex;\r
- osSemaphoreId_t semaphore;\r
- bool initialized;\r
-} ws_message_queue_t;\r
-\r
-\r
-// WebSocket 客户端结构体\r
-typedef struct {\r
- void* tcp_handle;\r
- ws_message_queue_t *msg_queue;\r
- char host[128];\r
- char path[128];\r
- char query[128];\r
- int port;\r
- \r
- /* 组包解析器 */\r
- ws_packet_parser_t parser;\r
-} WebSocketClient;\r
-\r
-\r
-// 初始化 WebSocket 客户端\r
-WebSocketClient *init_websocket_client(const char *url, const char *headers,uint8_t use_dns);\r
-\r
-// 设置WebSocket数据包回调函数\r
-void ws_set_packet_callback(WebSocketClient *client, on_ws_packet_callback callback);\r
-\r
-// 处理接收到的原始数据\r
-int ws_process_received_data(WebSocketClient *client, const uint8_t *data, size_t len);\r
-\r
-// 关闭 WebSocket 客户端\r
-void lwip_close_websocket_client(WebSocketClient *client) ;\r
-\r
-// 接收 WebSocket 消息\r
-int receive_websocket_message(WebSocketClient *client, unsigned char *buf,int len,WebSocketMessageType *code);\r
-int send_websocket_message(WebSocketClient *client, uint8_t *message, size_t message_len, WebSocketMessageType type) ;\r
-\r
-\r
-#ifdef __cplusplus\r
-#if __cplusplus\r
-}\r
-#endif /* __cplusplus */\r
-#endif /* __cplusplus */\r
-\r
-// 发送 WebSocket 消息\r
-// int send_websocket_message(WebSocketClient *client, uint8_t *message, size_t message_len, WebSocketMessageType type) ;\r
-\r
-\r
-#endif\r
+
+#ifndef __BK_WEBSOCKET_H__
+#define __BK_WEBSOCKET_H__
+// #include "aic_osa.h"
+// #include <bits/types.h>
+// WebSocket 消息类型枚举
+
+#ifdef __cplusplus
+#if __cplusplus
+extern "C" {
+#endif /* __cplusplus */
+#endif /* __cplusplus */
+
+typedef enum {
+ WS_CONT = 0x00,
+ WS_TEXT = 0x01,
+ WS_BINARY = 0x02,
+ WS_CLOSE = 0x8,
+ WS_PING = 0x9,
+ WS_PONG = 0xa
+} WebSocketMessageType;
+
+/* WebSocket数据包回调函数类型 */
+typedef void (*on_ws_packet_callback)(uint8_t* data, size_t datalen, WebSocketMessageType type);
+
+/* WebSocket组包缓冲区配置 */
+#define WS_FRAME_BUFFER_SIZE 256
+#define WS_PAYLOAD_BUFFER_SIZE 128
+
+/* WebSocket组包状态 */
+typedef enum {
+ WS_PARSE_HEADER, /* 解析帧头 */
+ WS_PARSE_EXTENDED, /* 解析扩展长度 */
+ WS_PARSE_MASK, /* 解析掩码 */
+ WS_PARSE_PAYLOAD /* 解析载荷数据 */
+} ws_parse_state_t;
+
+/* WebSocket组包缓冲区结构 */
+typedef struct {
+ uint8_t frame_buffer[WS_FRAME_BUFFER_SIZE]; /* 帧缓冲区 */
+ uint32_t frame_buffer_len; /* 帧缓冲区当前长度 */
+
+ uint8_t payload_buffer[WS_PAYLOAD_BUFFER_SIZE]; /* 载荷缓冲区 */
+ uint32_t payload_buffer_len; /* 载荷缓冲区当前长度 */
+
+ ws_parse_state_t parse_state; /* 解析状态 */
+ uint8_t fin; /* FIN标志 */
+ WebSocketMessageType opcode; /* 操作码 */
+ uint8_t mask; /* 掩码标志 */
+ uint64_t payload_len; /* 载荷长度 */
+ uint32_t payload_received; /* 已接收载荷长度 */
+ uint8_t masking_key[4]; /* 掩码密钥 */
+
+ on_ws_packet_callback packet_callback; /* 数据包回调函数 */
+} ws_packet_parser_t;
+
+/* WebSocket消息队列系统 */
+#define WS_QUEUE_SIZE 4
+#define WS_MESSAGE_MAX_SIZE 512
+
+typedef struct {
+ uint8_t data[WS_MESSAGE_MAX_SIZE];
+ uint16_t length;
+ WebSocketMessageType type;
+ uint32_t timestamp;
+ bool in_use;
+} ws_message_item_t;
+
+typedef struct {
+ ws_message_item_t messages[WS_QUEUE_SIZE];
+ uint8_t head;
+ uint8_t tail;
+ uint8_t count;
+ osMutexId_t mutex;
+ osSemaphoreId_t semaphore;
+ bool initialized;
+} ws_message_queue_t;
+
+
+// WebSocket 客户端结构体
+typedef struct {
+ void* tcp_handle;
+ ws_message_queue_t *msg_queue;
+ char host[128];
+ char path[128];
+ char query[128];
+ int port;
+
+ /* 组包解析器 */
+ ws_packet_parser_t parser;
+} WebSocketClient;
+
+
+// 初始化 WebSocket 客户端
+WebSocketClient *init_websocket_client(const char *url, const char *headers,uint8_t use_dns);
+
+// 设置WebSocket数据包回调函数
+void ws_set_packet_callback(WebSocketClient *client, on_ws_packet_callback callback);
+
+// 处理接收到的原始数据
+int ws_process_received_data(WebSocketClient *client, const uint8_t *data, size_t len);
+
+// 关闭 WebSocket 客户端
+void lwip_close_websocket_client(WebSocketClient *client) ;
+
+// 接收 WebSocket 消息
+int receive_websocket_message(WebSocketClient *client, unsigned char *buf,int len,WebSocketMessageType *code);
+int send_websocket_message(WebSocketClient *client, uint8_t *message, size_t message_len, WebSocketMessageType type) ;
+
+
+#ifdef __cplusplus
+#if __cplusplus
+}
+#endif /* __cplusplus */
+#endif /* __cplusplus */
+
+// 发送 WebSocket 消息
+// int send_websocket_message(WebSocketClient *client, uint8_t *message, size_t message_len, WebSocketMessageType type) ;
+
+
+#endif
{
WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"asr_server_deinit");
- if(send_websocket_message(asr_client,(uint8_t *)"WS_CLOSE",strlen("WS_CLOSE"),WS_CLOSE))
- {
- WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"WS_CLOSE");
+ // 检查TCP管理器是否仍然可用再发送关闭消息
+ extern bool tcp_manager_is_initialized(void);
+ if (asr_client && tcp_manager_is_initialized()) {
+ if(send_websocket_message(asr_client,(uint8_t *)"WS_CLOSE",strlen("WS_CLOSE"),WS_CLOSE))
+ {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"WS_CLOSE");
+ }
+ } else {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"跳过发送WS_CLOSE: TCP管理器不可用");
}
}
{
WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"chat_server_deinit");
- if(send_websocket_message(chat_client,(uint8_t *)"WS_CLOSE",strlen("WS_CLOSE"),WS_CLOSE))
- {
- WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"WS_CLOSE");
+ // 检查TCP管理器是否仍然可用再发送关闭消息
+ extern bool tcp_manager_is_initialized(void);
+ if (chat_client && tcp_manager_is_initialized()) {
+ if(send_websocket_message(chat_client,(uint8_t *)"WS_CLOSE",strlen("WS_CLOSE"),WS_CLOSE))
+ {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"WS_CLOSE");
+ }
+ } else {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"跳过发送WS_CLOSE: TCP管理器不可用");
}
}
{
WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"tts_server_deinit");
- if(send_websocket_message(tts_client,(uint8_t *)"WS_CLOSE",strlen("WS_CLOSE"),WS_CLOSE))
- {
- WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"WS_CLOSE");
+
+ // 检查TCP管理器是否仍然可用再发送关闭消息
+ extern bool tcp_manager_is_initialized(void);
+ if (tts_client && tcp_manager_is_initialized()) {
+ if(send_websocket_message(tts_client,(uint8_t *)"WS_CLOSE",strlen("WS_CLOSE"),WS_CLOSE))
+ {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"WS_CLOSE");
+ }
+ } else {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"跳过发送WS_CLOSE: TCP管理器不可用");
}
}
void ai_cleanup_resources(void) {
AI_LOCK();
+ // 首先关闭所有WebSocket连接
+ extern void lwip_close_websocket_client(WebSocketClient *client);
+ if (chat_client) {
+ lwip_close_websocket_client(chat_client);
+ chat_client = NULL;
+ }
+ if (asr_client) {
+ lwip_close_websocket_client(asr_client);
+ asr_client = NULL;
+ }
+ if (tts_client) {
+ lwip_close_websocket_client(tts_client);
+ tts_client = NULL;
+ }
+
// 清理WebSocket消息队列
ws_queues_deinit_all();
AI_UNLOCK();
- // 清理TCP管理器
+ // æ\9c\80å\90\8eæ¸\85ç\90\86TCP管ç\90\86å\99¨
tcp_simple_manager_deinit();
WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "AI resources cleaned up");
printf("tcp_send_data_directly failed: %d", tcp_result);
ret = -1;
}
-
- int retlen = tcp_recv_data_directly(client->tcp_handle, (uint8_t *)response, response_len);
-
- if(retlen<0)
- {
- printf("tcp_recv_data_directly failed");
- ret = -1;
- }
else
{
- ret = retlen;
+ // 发送成功，尝试接收响应
+ int retlen = tcp_recv_data_directly(client->tcp_handle, (uint8_t *)response, response_len);
+
+ if(retlen<0)
+ {
+ printf("tcp_recv_data_directly failed");
+ ret = -1;
+ }
+ else
+ {
+ ret = retlen;
+ }
}
free(request);
int tcp_get_debug_level(void)
{
return g_tcp_debug_level;
+}
+
+/**
+ * @brief 检查TCP管理器是否已初始化
+ */
+bool tcp_manager_is_initialized(void)
+{
+ return g_manager_initialized;
}
\ No newline at end of file
/* 连接参数定义 */
#define TCP_CONNECT_TIMEOUT_MS 10000
#define TCP_DNS_TIMEOUT_MS 8000
-#define TCP_SEND_QUEUE_SIZE 4
-#define TCP_SEND_BUFFER_SIZE 128
+#define TCP_SEND_QUEUE_SIZE 16
+#define TCP_SEND_BUFFER_SIZE 1024
#define TCP_CONNECT_RETRY_COUNT 3
#define TCP_CONNECT_RETRY_DELAY_MS 1000
*/
int tcp_get_debug_level(void);
+/**
+ * @brief 检查TCP管理器是否已初始化
+ */
+bool tcp_manager_is_initialized(void);
+
#ifdef __cplusplus
}
#endif
+++ /dev/null
-/**
- * @file tcp_simple_manager.c
- * @brief TCP简单管理器实现
- */
-
-#include "tcp_simple_manager.h"
-#include "lwip/sockets.h"
-#include "lwip/netdb.h"
-#include "lwip/sys.h"
-#include "lwip/err.h"
-#include <string.h>
-#include <stdlib.h>
-#include <unistd.h>
-#include <sys/select.h>
-#include <fcntl.h>
-#include <errno.h>
-#include <netinet/in.h>
-#include <arpa/inet.h>
-
-/* lwIP socket函数重定义，确保使用lwIP接口 */
-#define socket lwip_socket
-#define connect lwip_connect
-#define close lwip_close
-#define send lwip_send
-#define recv lwip_recv
-#define bind lwip_bind
-#define listen lwip_listen
-#define accept lwip_accept
-#define getsockopt lwip_getsockopt
-#define setsockopt lwip_setsockopt
-
-/* 内部状态标志 */
-static bool g_manager_initialized = false;
-static bool g_manager_running = false;
-
-/* 线程相关 */
-static sys_thread_t g_io_thread;
-static sys_sem_t g_shutdown_sem;
-
-/* 队列相关 */
-static sys_mbox_t g_cmd_queue;
-static sys_mbox_t g_recv_queue;
-
-/* 连接池 */
-static tcp_connection_info_t g_connections[TCP_MANAGER_MAX_CONNECTIONS];
-static sys_mutex_t g_connections_mutex;
-
-/* 统计信息 */
-static tcp_manager_stats_t g_stats;
-static sys_mutex_t g_stats_mutex;
-
-/* 当前线程ID检查 - 用于tcp_send_data_directly的安全检查 */
-static sys_thread_t g_io_thread_id;
-
-/* 简单的线程ID获取函数 - 适配lwIP系统 */
-static sys_thread_t sys_thread_self(void) {
- /* 对于嵌入式系统，这里可以返回当前任务ID */
- /* 暂时使用简化实现，实际环境中应该调用系统API */
- extern UINT32 LOS_CurTaskIDGet(void);
- return (sys_thread_t)LOS_CurTaskIDGet();
-}
-
-/* 简单的消息队列计数函数 - 暂时返回0 */
-static int sys_mbox_num_msgs(sys_mbox_t *mbox) {
- /* 简化实现，实际环境中应该查询队列状态 */
- (void)mbox;
- return 0;
-}
-
-/* 内部函数声明 */
-static void tcp_manager_io_thread(void* arg);
-static tcp_connection_info_t* tcp_manager_find_free_connection(void);
-static tcp_connection_info_t* tcp_manager_find_connection_by_handle(tcp_handle_t handle);
-static tcp_connection_info_t* tcp_manager_find_connection_by_pcb(struct tcp_pcb* pcb);
-static void tcp_manager_cleanup_connection(tcp_connection_info_t* conn);
-static err_t tcp_manager_lwip_connected_cb(void* arg, struct tcp_pcb* pcb, err_t err);
-static err_t tcp_manager_lwip_recv_cb(void* arg, struct tcp_pcb* pcb, struct pbuf* p, err_t err);
-static err_t tcp_manager_lwip_sent_cb(void* arg, struct tcp_pcb* pcb, u16_t len);
-static void tcp_manager_lwip_err_cb(void* arg, err_t err);
-static void tcp_manager_process_command(tcp_command_t* cmd);
-static void tcp_manager_process_recv_data(tcp_recv_data_t* recv_data);
-static tcp_manager_result_t tcp_manager_send_internal(tcp_connection_info_t* conn, uint8_t* data, uint16_t datalen);
-static tcp_manager_result_t tcp_manager_close_internal(tcp_connection_info_t* conn);
-static void tcp_manager_check_connection_health(void);
-static void tcp_manager_handle_socket_read(tcp_connection_info_t* conn);
-static void tcp_manager_handle_socket_error(tcp_connection_info_t* conn);
-static void tcp_manager_process_connect_command(tcp_command_t* cmd);
-
-/* 工具函数 */
-static void tcp_manager_update_stats_locked(void (*update_func)(void)) {
- sys_mutex_lock(&g_stats_mutex);
- update_func();
- sys_mutex_unlock(&g_stats_mutex);
-}
-
-static void stats_inc_errors(void) {
- g_stats.errors++;
-}
-
-static void stats_inc_bytes_sent_value(uint16_t bytes) {
- g_stats.bytes_sent += bytes;
-}
-
-static void stats_inc_bytes_received_value(uint16_t bytes) {
- g_stats.bytes_received += bytes;
-}
-
-/* 全局变量用于传递参数给统计函数 */
-static uint16_t g_temp_bytes_count = 0;
-
-static void stats_inc_bytes_sent(void) {
- g_stats.bytes_sent += g_temp_bytes_count;
-}
-
-static void stats_inc_bytes_received(void) {
- g_stats.bytes_received += g_temp_bytes_count;
-}
-
-/**
- * @brief 初始化TCP管理器
- */
-tcp_manager_result_t tcp_simple_manager_init(void)
-{
- if (g_manager_initialized) {
- return TCP_MANAGER_OK;
- }
-
- /* 清零统计信息 */
- memset(&g_stats, 0, sizeof(g_stats));
-
- /* 初始化连接池 */
- memset(g_connections, 0, sizeof(g_connections));
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- g_connections[i].handle = (tcp_handle_t)(uintptr_t)(i + 1); // 避免NULL handle
- g_connections[i].sockfd = -1; // 初始化为无效socket
- g_connections[i].in_use = false;
- g_connections[i].state = TCP_STATE_DISCONNECTED;
- }
-
- /* 创建互斥锁 */
- if (sys_mutex_new(&g_connections_mutex) != ERR_OK) {
- return TCP_MANAGER_ERROR;
- }
-
- if (sys_mutex_new(&g_stats_mutex) != ERR_OK) {
- sys_mutex_free(&g_connections_mutex);
- return TCP_MANAGER_ERROR;
- }
-
- /* 创建信号量 */
- if (sys_sem_new(&g_shutdown_sem, 0) != ERR_OK) {
- sys_mutex_free(&g_connections_mutex);
- sys_mutex_free(&g_stats_mutex);
- return TCP_MANAGER_ERROR;
- }
-
- /* 创建队列 */
- if (sys_mbox_new(&g_cmd_queue, TCP_MANAGER_CMD_QUEUE_SIZE) != ERR_OK) {
- sys_mutex_free(&g_connections_mutex);
- sys_mutex_free(&g_stats_mutex);
- sys_sem_free(&g_shutdown_sem);
- return TCP_MANAGER_ERROR;
- }
-
- if (sys_mbox_new(&g_recv_queue, TCP_MANAGER_RECV_QUEUE_SIZE) != ERR_OK) {
- sys_mutex_free(&g_connections_mutex);
- sys_mutex_free(&g_stats_mutex);
- sys_sem_free(&g_shutdown_sem);
- sys_mbox_free(&g_cmd_queue);
- return TCP_MANAGER_ERROR;
- }
-
- /* 启动IO线程 */
- g_manager_running = true;
- g_io_thread = sys_thread_new("tcp_manager_io", tcp_manager_io_thread, NULL,
- TCP_MANAGER_THREAD_STACK_SIZE, TCP_MANAGER_THREAD_PRIORITY);
- if (g_io_thread == (sys_thread_t)-1) {
- g_manager_running = false;
- sys_mutex_free(&g_connections_mutex);
- sys_mutex_free(&g_stats_mutex);
- sys_sem_free(&g_shutdown_sem);
- sys_mbox_free(&g_cmd_queue);
- sys_mbox_free(&g_recv_queue);
- return TCP_MANAGER_THREAD_ERROR;
- }
-
- g_manager_initialized = true;
- return TCP_MANAGER_OK;
-}
-
-/**
- * @brief 反初始化TCP管理器
- */
-tcp_manager_result_t tcp_simple_manager_deinit(void)
-{
- if (!g_manager_initialized) {
- return TCP_MANAGER_NOT_INITIALIZED;
- }
-
- /* 发送关闭命令 */
- tcp_command_t cmd = {0};
- cmd.type = TCP_CMD_SHUTDOWN;
-
- if (sys_mbox_trypost(&g_cmd_queue, &cmd) != ERR_OK) {
- /* 如果队列满了，强制关闭 */
- g_manager_running = false;
- }
-
- /* 等待线程结束 */
- sys_sem_wait(&g_shutdown_sem);
-
- /* 清理资源 */
- sys_mbox_free(&g_cmd_queue);
- sys_mbox_free(&g_recv_queue);
- sys_sem_free(&g_shutdown_sem);
- sys_mutex_free(&g_connections_mutex);
- sys_mutex_free(&g_stats_mutex);
-
- g_manager_initialized = false;
- return TCP_MANAGER_OK;
-}
-
-/**
- * @brief 连接到指定地址
- */
-tcp_handle_t tcp_open_url(const char* host, uint16_t port,
- tcp_on_connect_cb connect_cb,
- tcp_on_disconnect_cb disconnect_cb,
- tcp_on_data_received_cb data_cb)
-{
- if (!g_manager_initialized || !host) {
- return NULL;
- }
-
- /* 查找空闲连接 */
- sys_mutex_lock(&g_connections_mutex);
- tcp_connection_info_t* conn = tcp_manager_find_free_connection();
- if (!conn) {
- sys_mutex_unlock(&g_connections_mutex);
- return NULL;
- }
-
- /* 初始化连接信息 */
- conn->in_use = true;
- conn->state = TCP_STATE_CONNECTING;
- conn->remote_port = port;
- conn->connect_cb = connect_cb;
- conn->disconnect_cb = disconnect_cb;
- conn->data_cb = data_cb;
- conn->last_activity = sys_now();
-
- tcp_handle_t handle = conn->handle;
- sys_mutex_unlock(&g_connections_mutex);
-
- /* 创建连接命令 */
- tcp_command_t* cmd = (tcp_command_t*)malloc(sizeof(tcp_command_t));
- if (!cmd) {
- /* 清理连接 */
- sys_mutex_lock(&g_connections_mutex);
- conn->in_use = false;
- conn->state = TCP_STATE_DISCONNECTED;
- sys_mutex_unlock(&g_connections_mutex);
- return NULL;
- }
-
- cmd->type = TCP_CMD_CONNECT;
- cmd->handle = handle;
- cmd->params.connect.host = strdup(host);
- cmd->params.connect.port = port;
- cmd->params.connect.connect_cb = connect_cb;
- cmd->params.connect.disconnect_cb = disconnect_cb;
- cmd->params.connect.data_cb = data_cb;
-
- if (sys_mbox_trypost(&g_cmd_queue, cmd) != ERR_OK) {
- /* 队列满了 */
- free(cmd->params.connect.host);
- free(cmd);
- sys_mutex_lock(&g_connections_mutex);
- conn->in_use = false;
- conn->state = TCP_STATE_DISCONNECTED;
- sys_mutex_unlock(&g_connections_mutex);
- return NULL;
- }
-
- return handle;
-}
-
-/**
- * @brief 发送数据
- */
-tcp_manager_result_t tcp_send_data(tcp_handle_t handle, uint8_t* data, uint16_t datalen)
-{
- if (!g_manager_initialized || !handle || !data || datalen == 0) {
- return TCP_MANAGER_ERROR;
- }
-
- /* 创建发送命令 */
- tcp_command_t* cmd = (tcp_command_t*)malloc(sizeof(tcp_command_t));
- if (!cmd) {
- return TCP_MANAGER_MEMORY_ERROR;
- }
-
- uint8_t* data_copy = (uint8_t*)malloc(datalen);
- if (!data_copy) {
- free(cmd);
- return TCP_MANAGER_MEMORY_ERROR;
- }
-
- memcpy(data_copy, data, datalen);
-
- cmd->type = TCP_CMD_SEND;
- cmd->handle = handle;
- cmd->params.send.data = data_copy;
- cmd->params.send.length = datalen;
- cmd->params.send.need_free = true;
-
- if (sys_mbox_trypost(&g_cmd_queue, cmd) != ERR_OK) {
- free(data_copy);
- free(cmd);
- return TCP_MANAGER_QUEUE_FULL;
- }
-
- return TCP_MANAGER_OK;
-}
-
-/**
- * @brief 直接发送数据 (仅在回调中使用)
- */
-tcp_manager_result_t tcp_send_data_directly(tcp_handle_t handle, uint8_t* data, uint16_t datalen)
-{
- if (!g_manager_initialized || !handle || !data || datalen == 0) {
- return TCP_MANAGER_ERROR;
- }
-
- /* 检查是否在IO线程中调用 */
- if (sys_thread_self() != g_io_thread_id) {
- /* 不在IO线程中，使用普通发送 */
- return tcp_send_data(handle, data, datalen);
- }
-
- /* 在IO线程中，可以直接发送 */
- sys_mutex_lock(&g_connections_mutex);
- tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(handle);
- if (!conn || conn->state != TCP_STATE_CONNECTED) {
- sys_mutex_unlock(&g_connections_mutex);
- return TCP_MANAGER_INVALID_HANDLE;
- }
-
- tcp_manager_result_t result = tcp_manager_send_internal(conn, data, datalen);
- sys_mutex_unlock(&g_connections_mutex);
-
- return result;
-}
-
-/**
- * @brief 关闭TCP连接
- */
-tcp_manager_result_t tcp_manager_close(tcp_handle_t handle)
-{
- if (!g_manager_initialized || !handle) {
- return TCP_MANAGER_ERROR;
- }
-
- /* 创建关闭命令 */
- tcp_command_t* cmd = (tcp_command_t*)malloc(sizeof(tcp_command_t));
- if (!cmd) {
- return TCP_MANAGER_MEMORY_ERROR;
- }
-
- cmd->type = TCP_CMD_CLOSE;
- cmd->handle = handle;
-
- if (sys_mbox_trypost(&g_cmd_queue, cmd) != ERR_OK) {
- free(cmd);
- return TCP_MANAGER_QUEUE_FULL;
- }
-
- return TCP_MANAGER_OK;
-}
-
-/**
- * @brief 获取连接状态
- */
-tcp_connection_state_t tcp_get_connection_state(tcp_handle_t handle)
-{
- if (!g_manager_initialized || !handle) {
- return TCP_STATE_ERROR;
- }
-
- sys_mutex_lock(&g_connections_mutex);
- tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(handle);
- tcp_connection_state_t state = conn ? conn->state : TCP_STATE_ERROR;
- sys_mutex_unlock(&g_connections_mutex);
-
- return state;
-}
-
-/**
- * @brief 获取管理器统计信息
- */
-tcp_manager_result_t tcp_get_manager_stats(tcp_manager_stats_t* stats)
-{
- if (!stats) {
- return TCP_MANAGER_ERROR;
- }
-
- sys_mutex_lock(&g_stats_mutex);
- memcpy(stats, &g_stats, sizeof(tcp_manager_stats_t));
-
- /* 更新实时统计 */
- stats->recv_queue_count = sys_mbox_valid(&g_recv_queue) ?
- sys_mbox_num_msgs(&g_recv_queue) : 0;
- stats->cmd_queue_count = sys_mbox_valid(&g_cmd_queue) ?
- sys_mbox_num_msgs(&g_cmd_queue) : 0;
-
- sys_mutex_lock(&g_connections_mutex);
- stats->active_connections = 0;
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- if (g_connections[i].in_use && g_connections[i].state == TCP_STATE_CONNECTED) {
- stats->active_connections++;
- }
- }
- sys_mutex_unlock(&g_connections_mutex);
-
- sys_mutex_unlock(&g_stats_mutex);
- return TCP_MANAGER_OK;
-}
-
-/**
- * @brief 直接关闭TCP连接 (仅在回调中使用)
- */
-tcp_manager_result_t tcp_manager_close_directly(tcp_handle_t handle)
-{
- if (!g_manager_initialized || !handle) {
- return TCP_MANAGER_ERROR;
- }
-
- /* 检查是否在IO线程中调用 */
- if (sys_thread_self() != g_io_thread_id) {
- /* 不在IO线程中，使用普通关闭 */
- return tcp_manager_close(handle);
- }
-
- /* 在IO线程中，可以直接关闭 */
- sys_mutex_lock(&g_connections_mutex);
- tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(handle);
- if (!conn) {
- sys_mutex_unlock(&g_connections_mutex);
- return TCP_MANAGER_INVALID_HANDLE;
- }
-
- tcp_manager_result_t result = tcp_manager_close_internal(conn);
- sys_mutex_unlock(&g_connections_mutex);
-
- return result;
-}
-
-/**
- * @brief 设置连接超时时间
- */
-tcp_manager_result_t tcp_set_timeout(tcp_handle_t handle, uint32_t timeout_ms)
-{
- /* 简化实现，可以在连接信息中添加超时字段 */
- (void)handle;
- (void)timeout_ms;
- return TCP_MANAGER_OK;
-}
-
-/* ========== 内部函数实现 ========== */
-
-/**
- * @brief IO线程主函数 - 使用select机制监控socket
- */
-static void tcp_manager_io_thread(void* arg)
-{
- (void)arg;
-
- g_io_thread_id = sys_thread_self();
- fd_set read_fds, write_fds, error_fds;
- int max_fd = 0;
- struct timeval timeout;
-
- while (g_manager_running) {
- /* 重置文件描述符集合 */
- FD_ZERO(&read_fds);
- FD_ZERO(&write_fds);
- FD_ZERO(&error_fds);
- max_fd = 0;
-
- /* 添加所有活跃连接的socket到监控集合 */
- sys_mutex_lock(&g_connections_mutex);
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- if (g_connections[i].in_use && g_connections[i].sockfd >= 0) {
- FD_SET(g_connections[i].sockfd, &read_fds);
- FD_SET(g_connections[i].sockfd, &error_fds);
- if (g_connections[i].state == TCP_STATE_CONNECTING) {
- FD_SET(g_connections[i].sockfd, &write_fds);
- }
- if (g_connections[i].sockfd > max_fd) {
- max_fd = g_connections[i].sockfd;
- }
- }
- }
- sys_mutex_unlock(&g_connections_mutex);
-
- /* 设置超时时间 */
- timeout.tv_sec = 0;
- timeout.tv_usec = 50000; /* 50ms */
-
- /* 使用select进行多路复用监控 */
- int select_result = select(max_fd + 1, &read_fds, &write_fds, &error_fds, &timeout);
-
- if (select_result > 0) {
- /* 有socket活动，处理事件 */
- sys_mutex_lock(&g_connections_mutex);
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- if (g_connections[i].in_use && g_connections[i].sockfd >= 0) {
- int sockfd = g_connections[i].sockfd;
-
- /* 检查连接完成事件 */
- if (g_connections[i].state == TCP_STATE_CONNECTING && FD_ISSET(sockfd, &write_fds)) {
- int error = 0;
- socklen_t len = sizeof(error);
- if (getsockopt(sockfd, SOL_SOCKET, SO_ERROR, &error, &len) == 0 && error == 0) {
- /* 连接成功 */
- g_connections[i].state = TCP_STATE_CONNECTED;
- g_connections[i].last_activity = sys_now();
- if (g_connections[i].connect_cb) {
- g_connections[i].connect_cb(g_connections[i].handle, true);
- }
- } else {
- /* 连接失败 */
- g_connections[i].state = TCP_STATE_ERROR;
- if (g_connections[i].connect_cb) {
- g_connections[i].connect_cb(g_connections[i].handle, false);
- }
- tcp_manager_cleanup_connection(&g_connections[i]);
- }
- }
-
- /* 检查读事件 */
- if (FD_ISSET(sockfd, &read_fds)) {
- tcp_manager_handle_socket_read(&g_connections[i]);
- }
-
- /* 检查错误事件 */
- if (FD_ISSET(sockfd, &error_fds)) {
- tcp_manager_handle_socket_error(&g_connections[i]);
- }
- }
- }
- sys_mutex_unlock(&g_connections_mutex);
- } else if (select_result < 0 && errno != EINTR) {
- /* select错误 */
- tcp_manager_update_stats_locked(stats_inc_errors);
- }
-
- /* 处理命令队列 - 非阻塞处理 */
- void* msg;
- while (sys_mbox_tryfetch(&g_cmd_queue, &msg) == ERR_OK) {
- tcp_command_t* cmd = (tcp_command_t*)msg;
- tcp_manager_process_command(cmd);
-
- if (cmd->type == TCP_CMD_SHUTDOWN) {
- g_manager_running = false;
- goto cleanup;
- }
- }
-
- /* 处理接收队列 - 非阻塞处理 */
- while (sys_mbox_tryfetch(&g_recv_queue, &msg) == ERR_OK) {
- tcp_recv_data_t* recv_data = (tcp_recv_data_t*)msg;
- tcp_manager_process_recv_data(recv_data);
- free(recv_data);
- }
-
- /* 检查连接健康状态和超时 */
- tcp_manager_check_connection_health();
- }
-
-cleanup:
- /* 清理所有连接 */
- sys_mutex_lock(&g_connections_mutex);
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- if (g_connections[i].in_use) {
- tcp_manager_cleanup_connection(&g_connections[i]);
- }
- }
- sys_mutex_unlock(&g_connections_mutex);
-
- /* 通知主线程可以退出 */
- sys_sem_signal(&g_shutdown_sem);
-}
-
-/**
- * @brief 查找空闲连接
- */
-static tcp_connection_info_t* tcp_manager_find_free_connection(void)
-{
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- if (!g_connections[i].in_use) {
- return &g_connections[i];
- }
- }
- return NULL;
-}
-
-/**
- * @brief 根据句柄查找连接
- */
-static tcp_connection_info_t* tcp_manager_find_connection_by_handle(tcp_handle_t handle)
-{
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- if (g_connections[i].in_use && g_connections[i].handle == handle) {
- return &g_connections[i];
- }
- }
- return NULL;
-}
-
-/**
- * @brief 根据PCB查找连接
- */
-static tcp_connection_info_t* tcp_manager_find_connection_by_pcb(struct tcp_pcb* pcb)
-{
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- if (g_connections[i].in_use && g_connections[i].pcb == pcb) {
- return &g_connections[i];
- }
- }
- return NULL;
-}
-
-/**
- * @brief 清理连接
- */
-static void tcp_manager_cleanup_connection(tcp_connection_info_t* conn)
-{
- if (!conn) return;
-
- if (conn->sockfd >= 0) {
- close(conn->sockfd);
- conn->sockfd = -1;
- }
-
- conn->in_use = false;
- conn->state = TCP_STATE_DISCONNECTED;
- conn->connect_cb = NULL;
- conn->disconnect_cb = NULL;
- conn->data_cb = NULL;
-}
-
-/* ========== 移除了lwIP内部回调函数，使用标准socket接口 ========== */
-
-/* ========== 命令处理函数 ========== */
-
-static void tcp_manager_process_command(tcp_command_t* cmd)
-{
- switch (cmd->type) {
- case TCP_CMD_CONNECT: {
- tcp_manager_process_connect_command(cmd);
- break;
- }
-
- case TCP_CMD_SEND: {
- sys_mutex_lock(&g_connections_mutex);
- tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(cmd->handle);
- if (conn && conn->state == TCP_STATE_CONNECTED && conn->sockfd >= 0) {
- tcp_manager_send_internal(conn, cmd->params.send.data, cmd->params.send.length);
- }
- sys_mutex_unlock(&g_connections_mutex);
-
- /* 释放命令资源 */
- if (cmd->params.send.need_free) {
- free(cmd->params.send.data);
- }
- free(cmd);
- break;
- }
-
- case TCP_CMD_CLOSE: {
- sys_mutex_lock(&g_connections_mutex);
- tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(cmd->handle);
- if (conn) {
- conn->state = TCP_STATE_CLOSING;
- if (conn->disconnect_cb) {
- conn->disconnect_cb(conn->handle);
- }
- tcp_manager_cleanup_connection(conn);
- }
- sys_mutex_unlock(&g_connections_mutex);
-
- free(cmd);
- break;
- }
-
- case TCP_CMD_SHUTDOWN:
- /* 不释放cmd，因为它在栈上 */
- break;
-
- default:
- free(cmd);
- break;
- }
-}
-
-static void tcp_manager_process_recv_data(tcp_recv_data_t* recv_data)
-{
- sys_mutex_lock(&g_connections_mutex);
- tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(recv_data->handle);
- if (conn && conn->data_cb) {
- tcp_on_data_received_cb callback = conn->data_cb;
- sys_mutex_unlock(&g_connections_mutex);
-
- /* 调用用户回调 */
- callback(recv_data->handle, recv_data->data, recv_data->length);
- } else {
- sys_mutex_unlock(&g_connections_mutex);
- }
-}
-
-static tcp_manager_result_t tcp_manager_send_internal(tcp_connection_info_t* conn, uint8_t* data, uint16_t datalen)
-{
- if (!conn || conn->sockfd < 0 || !data || datalen == 0) {
- return TCP_MANAGER_ERROR;
- }
-
- ssize_t sent = send(conn->sockfd, data, datalen, 0);
- if (sent > 0) {
- conn->last_activity = sys_now();
- g_temp_bytes_count = (uint16_t)sent;
- tcp_manager_update_stats_locked(stats_inc_bytes_sent);
- return TCP_MANAGER_OK;
- }
-
- tcp_manager_update_stats_locked(stats_inc_errors);
- return TCP_MANAGER_ERROR;
-}
-
-/**
- * @brief 处理连接命令
- */
-static void tcp_manager_process_connect_command(tcp_command_t* cmd)
-{
- sys_mutex_lock(&g_connections_mutex);
- tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(cmd->handle);
- if (!conn) {
- sys_mutex_unlock(&g_connections_mutex);
- goto cleanup;
- }
-
- /* 创建socket */
- int sockfd = socket(AF_INET, SOCK_STREAM, 0);
- if (sockfd < 0) {
- conn->state = TCP_STATE_ERROR;
- if (conn->connect_cb) {
- conn->connect_cb(conn->handle, false);
- }
- tcp_manager_cleanup_connection(conn);
- tcp_manager_update_stats_locked(stats_inc_errors);
- sys_mutex_unlock(&g_connections_mutex);
- goto cleanup;
- }
-
- /* 设置非阻塞模式 */
- int flags = fcntl(sockfd, F_GETFL, 0);
- if (flags >= 0) {
- fcntl(sockfd, F_SETFL, flags | O_NONBLOCK);
- }
-
- /* 设置socket选项 */
- int opt = 1;
- setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
-
- conn->sockfd = sockfd;
- conn->state = TCP_STATE_CONNECTING;
- strcpy(conn->remote_ip, cmd->params.connect.host);
- conn->remote_port = cmd->params.connect.port;
-
- /* 解析地址并连接 */
- struct sockaddr_in server_addr;
- memset(&server_addr, 0, sizeof(server_addr));
- server_addr.sin_family = AF_INET;
- server_addr.sin_port = htons(cmd->params.connect.port);
-
- /* 尝试解析IP地址 */
- if (inet_aton(cmd->params.connect.host, &server_addr.sin_addr) == 0) {
- /* 不是IP地址，尝试域名解析 */
- struct hostent *he = gethostbyname(cmd->params.connect.host);
- if (he && he->h_addr_list[0]) {
- memcpy(&server_addr.sin_addr, he->h_addr_list[0], he->h_length);
- } else {
- /* 解析失败 */
- conn->state = TCP_STATE_ERROR;
- if (conn->connect_cb) {
- conn->connect_cb(conn->handle, false);
- }
- tcp_manager_cleanup_connection(conn);
- tcp_manager_update_stats_locked(stats_inc_errors);
- sys_mutex_unlock(&g_connections_mutex);
- goto cleanup;
- }
- }
-
- /* 非阻塞连接 */
- int result = connect(sockfd, (struct sockaddr*)&server_addr, sizeof(server_addr));
- if (result == 0) {
- /* 立即连接成功 */
- conn->state = TCP_STATE_CONNECTED;
- conn->last_activity = sys_now();
- if (conn->connect_cb) {
- conn->connect_cb(conn->handle, true);
- }
- } else if (errno == EINPROGRESS) {
- /* 连接正在进行中，等待select检测 */
- /* 状态已经设置为TCP_STATE_CONNECTING */
- } else {
- /* 连接失败 */
- conn->state = TCP_STATE_ERROR;
- if (conn->connect_cb) {
- conn->connect_cb(conn->handle, false);
- }
- tcp_manager_cleanup_connection(conn);
- tcp_manager_update_stats_locked(stats_inc_errors);
- }
-
- sys_mutex_unlock(&g_connections_mutex);
-
-cleanup:
- /* 释放命令资源 */
- free(cmd->params.connect.host);
- free(cmd);
-}
-
-/**
- * @brief 处理socket读事件
- */
-static void tcp_manager_handle_socket_read(tcp_connection_info_t* conn)
-{
- if (!conn || conn->sockfd < 0) return;
-
- uint8_t buffer[TCP_MANAGER_RECV_BUFFER_SIZE];
- ssize_t bytes_read = recv(conn->sockfd, buffer, sizeof(buffer), MSG_DONTWAIT);
-
- if (bytes_read > 0) {
- /* 分配接收数据结构 */
- tcp_recv_data_t* recv_data = (tcp_recv_data_t*)malloc(sizeof(tcp_recv_data_t));
- if (!recv_data) {
- tcp_manager_update_stats_locked(stats_inc_errors);
- return;
- }
-
- recv_data->handle = conn->handle;
- recv_data->length = (uint16_t)bytes_read;
- recv_data->timestamp = sys_now();
- memcpy(recv_data->data, buffer, bytes_read);
-
- /* 更新统计 */
- g_temp_bytes_count = recv_data->length;
- tcp_manager_update_stats_locked(stats_inc_bytes_received);
-
- /* 投递到接收队列 */
- if (sys_mbox_trypost(&g_recv_queue, recv_data) != ERR_OK) {
- free(recv_data);
- tcp_manager_update_stats_locked(stats_inc_errors);
- }
-
- conn->last_activity = sys_now();
- } else if (bytes_read == 0) {
- /* 连接被对端关闭 */
- conn->state = TCP_STATE_DISCONNECTED;
- if (conn->disconnect_cb) {
- conn->disconnect_cb(conn->handle);
- }
- tcp_manager_cleanup_connection(conn);
- } else if (errno != EAGAIN && errno != EWOULDBLOCK) {
- /* 读取错误 */
- tcp_manager_handle_socket_error(conn);
- }
-}
-
-/**
- * @brief 处理socket错误事件
- */
-static void tcp_manager_handle_socket_error(tcp_connection_info_t* conn)
-{
- if (!conn) return;
-
- conn->state = TCP_STATE_ERROR;
- if (conn->disconnect_cb) {
- conn->disconnect_cb(conn->handle);
- }
-
- tcp_manager_cleanup_connection(conn);
- tcp_manager_update_stats_locked(stats_inc_errors);
-}
-
-/**
- * @brief 检查连接健康状态
- */
-static void tcp_manager_check_connection_health(void)
-{
- uint32_t current_time = sys_now();
-
- sys_mutex_lock(&g_connections_mutex);
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- if (g_connections[i].in_use) {
- /* 检查连接超时 (30秒无活动) */
- if (current_time - g_connections[i].last_activity > 30000) {
- if (g_connections[i].state == TCP_STATE_CONNECTED) {
- /* 发送心跳包或标记为超时 */
- g_connections[i].state = TCP_STATE_ERROR;
- if (g_connections[i].disconnect_cb) {
- g_connections[i].disconnect_cb(g_connections[i].handle);
- }
- tcp_manager_cleanup_connection(&g_connections[i]);
- }
- }
- }
- }
- sys_mutex_unlock(&g_connections_mutex);
-}
-
-/**
- * @brief 内部关闭连接实现
- */
-static tcp_manager_result_t tcp_manager_close_internal(tcp_connection_info_t* conn)
-{
- if (!conn) return TCP_MANAGER_ERROR;
-
- conn->state = TCP_STATE_CLOSING;
-
- /* 如果有断开回调，先调用 */
- if (conn->disconnect_cb) {
- conn->disconnect_cb(conn->handle);
- }
-
- /* 清理连接资源 */
- tcp_manager_cleanup_connection(conn);
-
- return TCP_MANAGER_OK;
-}
\ No newline at end of file
+++ /dev/null
-/**
- * @file tcp_simple_manager.c
- * @brief TCP简单管理器实现
- */
-
-#include "tcp_simple_manager.h"
-#include "lwip/sockets.h"
-#include "lwip/netdb.h"
-#include "lwip/sys.h"
-#include "lwip/err.h"
-#include <string.h>
-#include <stdlib.h>
-#include <unistd.h>
-#include <sys/select.h>
-#include <fcntl.h>
-#include <errno.h>
-#include <netinet/in.h>
-#include <arpa/inet.h>
-
-/* lwIP socket函数重定义，确保使用lwIP接口 */
-#define socket lwip_socket
-#define connect lwip_connect
-#define close lwip_close
-#define send lwip_send
-#define recv lwip_recv
-#define bind lwip_bind
-#define listen lwip_listen
-#define accept lwip_accept
-#define getsockopt lwip_getsockopt
-#define setsockopt lwip_setsockopt
-
-/* 内部状态标志 */
-static bool g_manager_initialized = false;
-static bool g_manager_running = false;
-
-/* 线程相关 */
-static sys_thread_t g_io_thread;
-static sys_sem_t g_shutdown_sem;
-
-/* 队列相关 */
-static sys_mbox_t g_cmd_queue;
-static sys_mbox_t g_recv_queue;
-
-/* 连接池 */
-static tcp_connection_info_t g_connections[TCP_MANAGER_MAX_CONNECTIONS];
-static sys_mutex_t g_connections_mutex;
-
-/* 统计信息 */
-static tcp_manager_stats_t g_stats;
-static sys_mutex_t g_stats_mutex;
-
-/* 当前线程ID检查 - 用于tcp_send_data_directly的安全检查 */
-static sys_thread_t g_io_thread_id;
-
-/* 简单的线程ID获取函数 - 适配lwIP系统 */
-static sys_thread_t sys_thread_self(void) {
- /* 对于嵌入式系统，这里可以返回当前任务ID */
- /* 暂时使用简化实现，实际环境中应该调用系统API */
- extern UINT32 LOS_CurTaskIDGet(void);
- return (sys_thread_t)LOS_CurTaskIDGet();
-}
-
-/* 简单的消息队列计数函数 - 暂时返回0 */
-static int sys_mbox_num_msgs(sys_mbox_t *mbox) {
- /* 简化实现，实际环境中应该查询队列状态 */
- (void)mbox;
- return 0;
-}
-
-/* 内部函数声明 */
-static void tcp_manager_io_thread(void* arg);
-static tcp_connection_info_t* tcp_manager_find_free_connection(void);
-static tcp_connection_info_t* tcp_manager_find_connection_by_handle(tcp_handle_t handle);
-static tcp_connection_info_t* tcp_manager_find_connection_by_pcb(struct tcp_pcb* pcb);
-static void tcp_manager_cleanup_connection(tcp_connection_info_t* conn);
-static err_t tcp_manager_lwip_connected_cb(void* arg, struct tcp_pcb* pcb, err_t err);
-static err_t tcp_manager_lwip_recv_cb(void* arg, struct tcp_pcb* pcb, struct pbuf* p, err_t err);
-static err_t tcp_manager_lwip_sent_cb(void* arg, struct tcp_pcb* pcb, u16_t len);
-static void tcp_manager_lwip_err_cb(void* arg, err_t err);
-static void tcp_manager_process_command(tcp_command_t* cmd);
-static void tcp_manager_process_recv_data(tcp_recv_data_t* recv_data);
-static tcp_manager_result_t tcp_manager_send_internal(tcp_connection_info_t* conn, uint8_t* data, uint16_t datalen);
-static tcp_manager_result_t tcp_manager_close_internal(tcp_connection_info_t* conn);
-static void tcp_manager_check_connection_health(void);
-static void tcp_manager_handle_socket_read(tcp_connection_info_t* conn);
-static void tcp_manager_handle_socket_error(tcp_connection_info_t* conn);
-static void tcp_manager_process_connect_command(tcp_command_t* cmd);
-
-/* 工具函数 */
-static void tcp_manager_update_stats_locked(void (*update_func)(void)) {
- sys_mutex_lock(&g_stats_mutex);
- update_func();
- sys_mutex_unlock(&g_stats_mutex);
-}
-
-static void stats_inc_errors(void) {
- g_stats.errors++;
-}
-
-static void stats_inc_bytes_sent_value(uint16_t bytes) {
- g_stats.bytes_sent += bytes;
-}
-
-static void stats_inc_bytes_received_value(uint16_t bytes) {
- g_stats.bytes_received += bytes;
-}
-
-/* 全局变量用于传递参数给统计函数 */
-static uint16_t g_temp_bytes_count = 0;
-
-static void stats_inc_bytes_sent(void) {
- g_stats.bytes_sent += g_temp_bytes_count;
-}
-
-static void stats_inc_bytes_received(void) {
- g_stats.bytes_received += g_temp_bytes_count;
-}
-
-/**
- * @brief 初始化TCP管理器
- */
-tcp_manager_result_t tcp_simple_manager_init(void)
-{
- if (g_manager_initialized) {
- return TCP_MANAGER_OK;
- }
-
- /* 清零统计信息 */
- memset(&g_stats, 0, sizeof(g_stats));
-
- /* 初始化连接池 */
- memset(g_connections, 0, sizeof(g_connections));
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- g_connections[i].handle = (tcp_handle_t)(uintptr_t)(i + 1); // 避免NULL handle
- g_connections[i].sockfd = -1; // 初始化为无效socket
- g_connections[i].in_use = false;
- g_connections[i].state = TCP_STATE_DISCONNECTED;
- }
-
- /* 创建互斥锁 */
- if (sys_mutex_new(&g_connections_mutex) != ERR_OK) {
- return TCP_MANAGER_ERROR;
- }
-
- if (sys_mutex_new(&g_stats_mutex) != ERR_OK) {
- sys_mutex_free(&g_connections_mutex);
- return TCP_MANAGER_ERROR;
- }
-
- /* 创建信号量 */
- if (sys_sem_new(&g_shutdown_sem, 0) != ERR_OK) {
- sys_mutex_free(&g_connections_mutex);
- sys_mutex_free(&g_stats_mutex);
- return TCP_MANAGER_ERROR;
- }
-
- /* 创建队列 */
- if (sys_mbox_new(&g_cmd_queue, TCP_MANAGER_CMD_QUEUE_SIZE) != ERR_OK) {
- sys_mutex_free(&g_connections_mutex);
- sys_mutex_free(&g_stats_mutex);
- sys_sem_free(&g_shutdown_sem);
- return TCP_MANAGER_ERROR;
- }
-
- if (sys_mbox_new(&g_recv_queue, TCP_MANAGER_RECV_QUEUE_SIZE) != ERR_OK) {
- sys_mutex_free(&g_connections_mutex);
- sys_mutex_free(&g_stats_mutex);
- sys_sem_free(&g_shutdown_sem);
- sys_mbox_free(&g_cmd_queue);
- return TCP_MANAGER_ERROR;
- }
-
- /* 启动IO线程 */
- g_manager_running = true;
- g_io_thread = sys_thread_new("tcp_manager_io", tcp_manager_io_thread, NULL,
- TCP_MANAGER_THREAD_STACK_SIZE, TCP_MANAGER_THREAD_PRIORITY);
- if (g_io_thread == (sys_thread_t)-1) {
- g_manager_running = false;
- sys_mutex_free(&g_connections_mutex);
- sys_mutex_free(&g_stats_mutex);
- sys_sem_free(&g_shutdown_sem);
- sys_mbox_free(&g_cmd_queue);
- sys_mbox_free(&g_recv_queue);
- return TCP_MANAGER_THREAD_ERROR;
- }
-
- g_manager_initialized = true;
- return TCP_MANAGER_OK;
-}
-
-/**
- * @brief 反初始化TCP管理器
- */
-tcp_manager_result_t tcp_simple_manager_deinit(void)
-{
- if (!g_manager_initialized) {
- return TCP_MANAGER_NOT_INITIALIZED;
- }
-
- /* 发送关闭命令 */
- tcp_command_t cmd = {0};
- cmd.type = TCP_CMD_SHUTDOWN;
-
- if (sys_mbox_trypost(&g_cmd_queue, &cmd) != ERR_OK) {
- /* 如果队列满了，强制关闭 */
- g_manager_running = false;
- }
-
- /* 等待线程结束 */
- sys_sem_wait(&g_shutdown_sem);
-
- /* 清理资源 */
- sys_mbox_free(&g_cmd_queue);
- sys_mbox_free(&g_recv_queue);
- sys_sem_free(&g_shutdown_sem);
- sys_mutex_free(&g_connections_mutex);
- sys_mutex_free(&g_stats_mutex);
-
- g_manager_initialized = false;
- return TCP_MANAGER_OK;
-}
-
-/**
- * @brief 连接到指定地址
- */
-tcp_handle_t tcp_open_url(const char* host, uint16_t port,
- tcp_on_connect_cb connect_cb,
- tcp_on_disconnect_cb disconnect_cb,
- tcp_on_data_received_cb data_cb)
-{
- if (!g_manager_initialized || !host) {
- return NULL;
- }
-
- /* 查找空闲连接 */
- sys_mutex_lock(&g_connections_mutex);
- tcp_connection_info_t* conn = tcp_manager_find_free_connection();
- if (!conn) {
- sys_mutex_unlock(&g_connections_mutex);
- return NULL;
- }
-
- /* 初始化连接信息 */
- conn->in_use = true;
- conn->state = TCP_STATE_CONNECTING;
- conn->remote_port = port;
- conn->connect_cb = connect_cb;
- conn->disconnect_cb = disconnect_cb;
- conn->data_cb = data_cb;
- conn->last_activity = sys_now();
-
- tcp_handle_t handle = conn->handle;
- sys_mutex_unlock(&g_connections_mutex);
-
- /* 创建连接命令 */
- tcp_command_t* cmd = (tcp_command_t*)malloc(sizeof(tcp_command_t));
- if (!cmd) {
- /* 清理连接 */
- sys_mutex_lock(&g_connections_mutex);
- conn->in_use = false;
- conn->state = TCP_STATE_DISCONNECTED;
- sys_mutex_unlock(&g_connections_mutex);
- return NULL;
- }
-
- cmd->type = TCP_CMD_CONNECT;
- cmd->handle = handle;
- cmd->params.connect.host = strdup(host);
- cmd->params.connect.port = port;
- cmd->params.connect.connect_cb = connect_cb;
- cmd->params.connect.disconnect_cb = disconnect_cb;
- cmd->params.connect.data_cb = data_cb;
-
- if (sys_mbox_trypost(&g_cmd_queue, cmd) != ERR_OK) {
- /* 队列满了 */
- free(cmd->params.connect.host);
- free(cmd);
- sys_mutex_lock(&g_connections_mutex);
- conn->in_use = false;
- conn->state = TCP_STATE_DISCONNECTED;
- sys_mutex_unlock(&g_connections_mutex);
- return NULL;
- }
-
- return handle;
-}
-
-/**
- * @brief 发送数据
- */
-tcp_manager_result_t tcp_send_data(tcp_handle_t handle, uint8_t* data, uint16_t datalen)
-{
- if (!g_manager_initialized || !handle || !data || datalen == 0) {
- return TCP_MANAGER_ERROR;
- }
-
- /* 创建发送命令 */
- tcp_command_t* cmd = (tcp_command_t*)malloc(sizeof(tcp_command_t));
- if (!cmd) {
- return TCP_MANAGER_MEMORY_ERROR;
- }
-
- uint8_t* data_copy = (uint8_t*)malloc(datalen);
- if (!data_copy) {
- free(cmd);
- return TCP_MANAGER_MEMORY_ERROR;
- }
-
- memcpy(data_copy, data, datalen);
-
- cmd->type = TCP_CMD_SEND;
- cmd->handle = handle;
- cmd->params.send.data = data_copy;
- cmd->params.send.length = datalen;
- cmd->params.send.need_free = true;
-
- if (sys_mbox_trypost(&g_cmd_queue, cmd) != ERR_OK) {
- free(data_copy);
- free(cmd);
- return TCP_MANAGER_QUEUE_FULL;
- }
-
- return TCP_MANAGER_OK;
-}
-
-/**
- * @brief 直接发送数据 (仅在回调中使用)
- */
-tcp_manager_result_t tcp_send_data_directly(tcp_handle_t handle, uint8_t* data, uint16_t datalen)
-{
- if (!g_manager_initialized || !handle || !data || datalen == 0) {
- return TCP_MANAGER_ERROR;
- }
-
- /* 检查是否在IO线程中调用 */
- if (sys_thread_self() != g_io_thread_id) {
- /* 不在IO线程中，使用普通发送 */
- return tcp_send_data(handle, data, datalen);
- }
-
- /* 在IO线程中，可以直接发送 */
- sys_mutex_lock(&g_connections_mutex);
- tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(handle);
- if (!conn || conn->state != TCP_STATE_CONNECTED) {
- sys_mutex_unlock(&g_connections_mutex);
- return TCP_MANAGER_INVALID_HANDLE;
- }
-
- tcp_manager_result_t result = tcp_manager_send_internal(conn, data, datalen);
- sys_mutex_unlock(&g_connections_mutex);
-
- return result;
-}
-
-/**
- * @brief 关闭TCP连接
- */
-tcp_manager_result_t tcp_manager_close(tcp_handle_t handle)
-{
- if (!g_manager_initialized || !handle) {
- return TCP_MANAGER_ERROR;
- }
-
- /* 创建关闭命令 */
- tcp_command_t* cmd = (tcp_command_t*)malloc(sizeof(tcp_command_t));
- if (!cmd) {
- return TCP_MANAGER_MEMORY_ERROR;
- }
-
- cmd->type = TCP_CMD_CLOSE;
- cmd->handle = handle;
-
- if (sys_mbox_trypost(&g_cmd_queue, cmd) != ERR_OK) {
- free(cmd);
- return TCP_MANAGER_QUEUE_FULL;
- }
-
- return TCP_MANAGER_OK;
-}
-
-/**
- * @brief 获取连接状态
- */
-tcp_connection_state_t tcp_get_connection_state(tcp_handle_t handle)
-{
- if (!g_manager_initialized || !handle) {
- return TCP_STATE_ERROR;
- }
-
- sys_mutex_lock(&g_connections_mutex);
- tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(handle);
- tcp_connection_state_t state = conn ? conn->state : TCP_STATE_ERROR;
- sys_mutex_unlock(&g_connections_mutex);
-
- return state;
-}
-
-/**
- * @brief 获取管理器统计信息
- */
-tcp_manager_result_t tcp_get_manager_stats(tcp_manager_stats_t* stats)
-{
- if (!stats) {
- return TCP_MANAGER_ERROR;
- }
-
- sys_mutex_lock(&g_stats_mutex);
- memcpy(stats, &g_stats, sizeof(tcp_manager_stats_t));
-
- /* 更新实时统计 */
- stats->recv_queue_count = sys_mbox_valid(&g_recv_queue) ?
- sys_mbox_num_msgs(&g_recv_queue) : 0;
- stats->cmd_queue_count = sys_mbox_valid(&g_cmd_queue) ?
- sys_mbox_num_msgs(&g_cmd_queue) : 0;
-
- sys_mutex_lock(&g_connections_mutex);
- stats->active_connections = 0;
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- if (g_connections[i].in_use && g_connections[i].state == TCP_STATE_CONNECTED) {
- stats->active_connections++;
- }
- }
- sys_mutex_unlock(&g_connections_mutex);
-
- sys_mutex_unlock(&g_stats_mutex);
- return TCP_MANAGER_OK;
-}
-
-/**
- * @brief 直接关闭TCP连接 (仅在回调中使用)
- */
-tcp_manager_result_t tcp_manager_close_directly(tcp_handle_t handle)
-{
- if (!g_manager_initialized || !handle) {
- return TCP_MANAGER_ERROR;
- }
-
- /* 检查是否在IO线程中调用 */
- if (sys_thread_self() != g_io_thread_id) {
- /* 不在IO线程中，使用普通关闭 */
- return tcp_manager_close(handle);
- }
-
- /* 在IO线程中，可以直接关闭 */
- sys_mutex_lock(&g_connections_mutex);
- tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(handle);
- if (!conn) {
- sys_mutex_unlock(&g_connections_mutex);
- return TCP_MANAGER_INVALID_HANDLE;
- }
-
- tcp_manager_result_t result = tcp_manager_close_internal(conn);
- sys_mutex_unlock(&g_connections_mutex);
-
- return result;
-}
-
-/**
- * @brief 设置连接超时时间
- */
-tcp_manager_result_t tcp_set_timeout(tcp_handle_t handle, uint32_t timeout_ms)
-{
- /* 简化实现，可以在连接信息中添加超时字段 */
- (void)handle;
- (void)timeout_ms;
- return TCP_MANAGER_OK;
-}
-
-/* ========== 内部函数实现 ========== */
-
-/**
- * @brief IO线程主函数 - 使用select机制监控socket
- */
-static void tcp_manager_io_thread(void* arg)
-{
- (void)arg;
-
- g_io_thread_id = sys_thread_self();
- fd_set read_fds, write_fds, error_fds;
- int max_fd = 0;
- struct timeval timeout;
-
- while (g_manager_running) {
- /* 重置文件描述符集合 */
- FD_ZERO(&read_fds);
- FD_ZERO(&write_fds);
- FD_ZERO(&error_fds);
- max_fd = 0;
-
- /* 添加所有活跃连接的socket到监控集合 */
- sys_mutex_lock(&g_connections_mutex);
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- if (g_connections[i].in_use && g_connections[i].sockfd >= 0) {
- FD_SET(g_connections[i].sockfd, &read_fds);
- FD_SET(g_connections[i].sockfd, &error_fds);
- if (g_connections[i].state == TCP_STATE_CONNECTING) {
- FD_SET(g_connections[i].sockfd, &write_fds);
- }
- if (g_connections[i].sockfd > max_fd) {
- max_fd = g_connections[i].sockfd;
- }
- }
- }
- sys_mutex_unlock(&g_connections_mutex);
-
- /* 设置超时时间 */
- timeout.tv_sec = 0;
- timeout.tv_usec = 50000; /* 50ms */
-
- /* 使用select进行多路复用监控 */
- int select_result = select(max_fd + 1, &read_fds, &write_fds, &error_fds, &timeout);
-
- if (select_result > 0) {
- /* 有socket活动，处理事件 */
- sys_mutex_lock(&g_connections_mutex);
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- if (g_connections[i].in_use && g_connections[i].sockfd >= 0) {
- int sockfd = g_connections[i].sockfd;
-
- /* 检查连接完成事件 */
- if (g_connections[i].state == TCP_STATE_CONNECTING && FD_ISSET(sockfd, &write_fds)) {
- int error = 0;
- socklen_t len = sizeof(error);
- if (getsockopt(sockfd, SOL_SOCKET, SO_ERROR, &error, &len) == 0 && error == 0) {
- /* 连接成功 */
- g_connections[i].state = TCP_STATE_CONNECTED;
- g_connections[i].last_activity = sys_now();
- if (g_connections[i].connect_cb) {
- g_connections[i].connect_cb(g_connections[i].handle, true);
- }
- } else {
- /* 连接失败 */
- g_connections[i].state = TCP_STATE_ERROR;
- if (g_connections[i].connect_cb) {
- g_connections[i].connect_cb(g_connections[i].handle, false);
- }
- tcp_manager_cleanup_connection(&g_connections[i]);
- }
- }
-
- /* 检查读事件 */
- if (FD_ISSET(sockfd, &read_fds)) {
- tcp_manager_handle_socket_read(&g_connections[i]);
- }
-
- /* 检查错误事件 */
- if (FD_ISSET(sockfd, &error_fds)) {
- tcp_manager_handle_socket_error(&g_connections[i]);
- }
- }
- }
- sys_mutex_unlock(&g_connections_mutex);
- } else if (select_result < 0 && errno != EINTR) {
- /* select错误 */
- tcp_manager_update_stats_locked(stats_inc_errors);
- }
-
- /* 处理命令队列 - 非阻塞处理 */
- void* msg;
- while (sys_mbox_tryfetch(&g_cmd_queue, &msg) == ERR_OK) {
- tcp_command_t* cmd = (tcp_command_t*)msg;
- tcp_manager_process_command(cmd);
-
- if (cmd->type == TCP_CMD_SHUTDOWN) {
- g_manager_running = false;
- goto cleanup;
- }
- }
-
- /* 处理接收队列 - 非阻塞处理 */
- while (sys_mbox_tryfetch(&g_recv_queue, &msg) == ERR_OK) {
- tcp_recv_data_t* recv_data = (tcp_recv_data_t*)msg;
- tcp_manager_process_recv_data(recv_data);
- free(recv_data);
- }
-
- /* 检查连接健康状态和超时 */
- tcp_manager_check_connection_health();
- }
-
-cleanup:
- /* 清理所有连接 */
- sys_mutex_lock(&g_connections_mutex);
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- if (g_connections[i].in_use) {
- tcp_manager_cleanup_connection(&g_connections[i]);
- }
- }
- sys_mutex_unlock(&g_connections_mutex);
-
- /* 通知主线程可以退出 */
- sys_sem_signal(&g_shutdown_sem);
-}
-
-/**
- * @brief 查找空闲连接
- */
-static tcp_connection_info_t* tcp_manager_find_free_connection(void)
-{
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- if (!g_connections[i].in_use) {
- return &g_connections[i];
- }
- }
- return NULL;
-}
-
-/**
- * @brief 根据句柄查找连接
- */
-static tcp_connection_info_t* tcp_manager_find_connection_by_handle(tcp_handle_t handle)
-{
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- if (g_connections[i].in_use && g_connections[i].handle == handle) {
- return &g_connections[i];
- }
- }
- return NULL;
-}
-
-/**
- * @brief 根据PCB查找连接
- */
-static tcp_connection_info_t* tcp_manager_find_connection_by_pcb(struct tcp_pcb* pcb)
-{
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- if (g_connections[i].in_use && g_connections[i].pcb == pcb) {
- return &g_connections[i];
- }
- }
- return NULL;
-}
-
-/**
- * @brief 清理连接
- */
-static void tcp_manager_cleanup_connection(tcp_connection_info_t* conn)
-{
- if (!conn) return;
-
- if (conn->sockfd >= 0) {
- close(conn->sockfd);
- conn->sockfd = -1;
- }
-
- conn->in_use = false;
- conn->state = TCP_STATE_DISCONNECTED;
- conn->connect_cb = NULL;
- conn->disconnect_cb = NULL;
- conn->data_cb = NULL;
-}
-
-/* ========== 移除了lwIP内部回调函数，使用标准socket接口 ========== */
-
-/* ========== 命令处理函数 ========== */
-
-static void tcp_manager_process_command(tcp_command_t* cmd)
-{
- switch (cmd->type) {
- case TCP_CMD_CONNECT: {
- tcp_manager_process_connect_command(cmd);
- break;
- }
-
- case TCP_CMD_SEND: {
- sys_mutex_lock(&g_connections_mutex);
- tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(cmd->handle);
- if (conn && conn->state == TCP_STATE_CONNECTED && conn->sockfd >= 0) {
- tcp_manager_send_internal(conn, cmd->params.send.data, cmd->params.send.length);
- }
- sys_mutex_unlock(&g_connections_mutex);
-
- /* 释放命令资源 */
- if (cmd->params.send.need_free) {
- free(cmd->params.send.data);
- }
- free(cmd);
- break;
- }
-
- case TCP_CMD_CLOSE: {
- sys_mutex_lock(&g_connections_mutex);
- tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(cmd->handle);
- if (conn) {
- conn->state = TCP_STATE_CLOSING;
- if (conn->disconnect_cb) {
- conn->disconnect_cb(conn->handle);
- }
- tcp_manager_cleanup_connection(conn);
- }
- sys_mutex_unlock(&g_connections_mutex);
-
- free(cmd);
- break;
- }
-
- case TCP_CMD_SHUTDOWN:
- /* 不释放cmd，因为它在栈上 */
- break;
-
- default:
- free(cmd);
- break;
- }
-}
-
-static void tcp_manager_process_recv_data(tcp_recv_data_t* recv_data)
-{
- sys_mutex_lock(&g_connections_mutex);
- tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(recv_data->handle);
- if (conn && conn->data_cb) {
- tcp_on_data_received_cb callback = conn->data_cb;
- sys_mutex_unlock(&g_connections_mutex);
-
- /* 调用用户回调 */
- callback(recv_data->handle, recv_data->data, recv_data->length);
- } else {
- sys_mutex_unlock(&g_connections_mutex);
- }
-}
-
-static tcp_manager_result_t tcp_manager_send_internal(tcp_connection_info_t* conn, uint8_t* data, uint16_t datalen)
-{
- if (!conn || conn->sockfd < 0 || !data || datalen == 0) {
- return TCP_MANAGER_ERROR;
- }
-
- ssize_t sent = send(conn->sockfd, data, datalen, 0);
- if (sent > 0) {
- conn->last_activity = sys_now();
- g_temp_bytes_count = (uint16_t)sent;
- tcp_manager_update_stats_locked(stats_inc_bytes_sent);
- return TCP_MANAGER_OK;
- }
-
- tcp_manager_update_stats_locked(stats_inc_errors);
- return TCP_MANAGER_ERROR;
-}
-
-/**
- * @brief 处理连接命令
- */
-static void tcp_manager_process_connect_command(tcp_command_t* cmd)
-{
- sys_mutex_lock(&g_connections_mutex);
- tcp_connection_info_t* conn = tcp_manager_find_connection_by_handle(cmd->handle);
- if (!conn) {
- sys_mutex_unlock(&g_connections_mutex);
- goto cleanup;
- }
-
- /* 创建socket */
- int sockfd = socket(AF_INET, SOCK_STREAM, 0);
- if (sockfd < 0) {
- conn->state = TCP_STATE_ERROR;
- if (conn->connect_cb) {
- conn->connect_cb(conn->handle, false);
- }
- tcp_manager_cleanup_connection(conn);
- tcp_manager_update_stats_locked(stats_inc_errors);
- sys_mutex_unlock(&g_connections_mutex);
- goto cleanup;
- }
-
- /* 设置非阻塞模式 */
- int flags = fcntl(sockfd, F_GETFL, 0);
- if (flags >= 0) {
- fcntl(sockfd, F_SETFL, flags | O_NONBLOCK);
- }
-
- /* 设置socket选项 */
- int opt = 1;
- setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
-
- conn->sockfd = sockfd;
- conn->state = TCP_STATE_CONNECTING;
- strcpy(conn->remote_ip, cmd->params.connect.host);
- conn->remote_port = cmd->params.connect.port;
-
- /* 解析地址并连接 */
- struct sockaddr_in server_addr;
- memset(&server_addr, 0, sizeof(server_addr));
- server_addr.sin_family = AF_INET;
- server_addr.sin_port = htons(cmd->params.connect.port);
-
- /* 尝试解析IP地址 */
- if (inet_aton(cmd->params.connect.host, &server_addr.sin_addr) == 0) {
- /* 不是IP地址，尝试域名解析 */
- struct hostent *he = gethostbyname(cmd->params.connect.host);
- if (he && he->h_addr_list[0]) {
- memcpy(&server_addr.sin_addr, he->h_addr_list[0], he->h_length);
- } else {
- /* 解析失败 */
- conn->state = TCP_STATE_ERROR;
- if (conn->connect_cb) {
- conn->connect_cb(conn->handle, false);
- }
- tcp_manager_cleanup_connection(conn);
- tcp_manager_update_stats_locked(stats_inc_errors);
- sys_mutex_unlock(&g_connections_mutex);
- goto cleanup;
- }
- }
-
- /* 非阻塞连接 */
- int result = connect(sockfd, (struct sockaddr*)&server_addr, sizeof(server_addr));
- if (result == 0) {
- /* 立即连接成功 */
- conn->state = TCP_STATE_CONNECTED;
- conn->last_activity = sys_now();
- if (conn->connect_cb) {
- conn->connect_cb(conn->handle, true);
- }
- } else if (errno == EINPROGRESS) {
- /* 连接正在进行中，等待select检测 */
- /* 状态已经设置为TCP_STATE_CONNECTING */
- } else {
- /* 连接失败 */
- conn->state = TCP_STATE_ERROR;
- if (conn->connect_cb) {
- conn->connect_cb(conn->handle, false);
- }
- tcp_manager_cleanup_connection(conn);
- tcp_manager_update_stats_locked(stats_inc_errors);
- }
-
- sys_mutex_unlock(&g_connections_mutex);
-
-cleanup:
- /* 释放命令资源 */
- free(cmd->params.connect.host);
- free(cmd);
-}
-
-/**
- * @brief 处理socket读事件
- */
-static void tcp_manager_handle_socket_read(tcp_connection_info_t* conn)
-{
- if (!conn || conn->sockfd < 0) return;
-
- uint8_t buffer[TCP_MANAGER_RECV_BUFFER_SIZE];
- ssize_t bytes_read = recv(conn->sockfd, buffer, sizeof(buffer), MSG_DONTWAIT);
-
- if (bytes_read > 0) {
- /* 分配接收数据结构 */
- tcp_recv_data_t* recv_data = (tcp_recv_data_t*)malloc(sizeof(tcp_recv_data_t));
- if (!recv_data) {
- tcp_manager_update_stats_locked(stats_inc_errors);
- return;
- }
-
- recv_data->handle = conn->handle;
- recv_data->length = (uint16_t)bytes_read;
- recv_data->timestamp = sys_now();
- memcpy(recv_data->data, buffer, bytes_read);
-
- /* 更新统计 */
- g_temp_bytes_count = recv_data->length;
- tcp_manager_update_stats_locked(stats_inc_bytes_received);
-
- /* 投递到接收队列 */
- if (sys_mbox_trypost(&g_recv_queue, recv_data) != ERR_OK) {
- free(recv_data);
- tcp_manager_update_stats_locked(stats_inc_errors);
- }
-
- conn->last_activity = sys_now();
- } else if (bytes_read == 0) {
- /* 连接被对端关闭 */
- conn->state = TCP_STATE_DISCONNECTED;
- if (conn->disconnect_cb) {
- conn->disconnect_cb(conn->handle);
- }
- tcp_manager_cleanup_connection(conn);
- } else if (errno != EAGAIN && errno != EWOULDBLOCK) {
- /* 读取错误 */
- tcp_manager_handle_socket_error(conn);
- }
-}
-
-/**
- * @brief 处理socket错误事件
- */
-static void tcp_manager_handle_socket_error(tcp_connection_info_t* conn)
-{
- if (!conn) return;
-
- conn->state = TCP_STATE_ERROR;
- if (conn->disconnect_cb) {
- conn->disconnect_cb(conn->handle);
- }
-
- tcp_manager_cleanup_connection(conn);
- tcp_manager_update_stats_locked(stats_inc_errors);
-}
-
-/**
- * @brief 检查连接健康状态
- */
-static void tcp_manager_check_connection_health(void)
-{
- uint32_t current_time = sys_now();
-
- sys_mutex_lock(&g_connections_mutex);
- for (int i = 0; i < TCP_MANAGER_MAX_CONNECTIONS; i++) {
- if (g_connections[i].in_use) {
- /* 检查连接超时 (30秒无活动) */
- if (current_time - g_connections[i].last_activity > 30000) {
- if (g_connections[i].state == TCP_STATE_CONNECTED) {
- /* 发送心跳包或标记为超时 */
- g_connections[i].state = TCP_STATE_ERROR;
- if (g_connections[i].disconnect_cb) {
- g_connections[i].disconnect_cb(g_connections[i].handle);
- }
- tcp_manager_cleanup_connection(&g_connections[i]);
- }
- }
- }
- }
- sys_mutex_unlock(&g_connections_mutex);
-}
-
-/**
- * @brief 内部关闭连接实现
- */
-static tcp_manager_result_t tcp_manager_close_internal(tcp_connection_info_t* conn)
-{
- if (!conn) return TCP_MANAGER_ERROR;
-
- conn->state = TCP_STATE_CLOSING;
-
- /* 如果有断开回调，先调用 */
- if (conn->disconnect_cb) {
- conn->disconnect_cb(conn->handle);
- }
-
- /* 清理连接资源 */
- tcp_manager_cleanup_connection(conn);
-
- return TCP_MANAGER_OK;
-}
\ No newline at end of file
projects / mirrors / FD230.git / commitdiff