pchen/cKit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

TcpClient/Reactor 框架

Browse Source
This commit is contained in:
pchen
2026-09-08 12:38:28 +08:00
parent ac4045a979
commit aae773a7fa
11 changed files with 801 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Base/c_Types.h
+1
View File
@@ -123,6 +123,7 @@ typedef int c_err_t;
#define C_SUCCESS C_ERR_OK
#define C_ERR_NOTFOUND C_ERR_FAIL
#define C_ERR_OUTOFBOUND C_ERR_FAIL
#define C_ERR_EOF C_ERR_OUTOFBOUND
/* ------------------------------------------------------------------------------------------------------------------ */
/* */
Foundation/c_TcpClient.c
+138
View File
@@ -0,0 +1,138 @@
#include <c_TcpClient.h>
#include <c_AddrInfo.h>
/* ------------------------------------------------------------------------------------------------------------------ */
/* */
c_err_t c_TcpClient_Connect(c_TcpClient_t* self, int family, const char* server, const char* port, void* args, c_Allocator_t* allocator) {
if (!self || (!server && !port)) return C_ERR_PARAM;
self->allocator = allocator ? *allocator : c_DefaultAllocator;
self->sock = C_INVALID_SOCKET;
self->is_linked = C_FALSE;
/* 1. Resolve network address dynamically using your custom c_AddrInfo_t wrapper */
c_AddrInfo_t addr_resolver;
c_err_t err = c_AddrInfo_Init(&addr_resolver, family, SOCK_STREAM, 0, server, port);
if (err != C_ERR_OK) {
c_Socket_Destroy();
return C_ERR_FAIL;
}
c_AddrInfo_t curr_node = addr_resolver;
c_bool_t connected = C_FALSE;
/* 2. Traverse the resolved address link chain until a connection succeeds */
while (curr_node.addrinfo_p != NULL) {
/* Open the system socket using descriptors from addrinfo */
self->sock = socket(
c_AddrInfo_GetFamily(&curr_node),
c_AddrInfo_GetSockType(&curr_node),
c_AddrInfo_GetProtocol(&curr_node)
);
if (c_Socket_IsValid(self->sock)) {
/* Execute synchronous network connect transaction handshake */
int rc = connect(self->sock, curr_node.addrinfo_p->ai_addr, (int)curr_node.addrinfo_p->ai_addrlen);
if (rc != C_SOCKET_ERROR) {
connected = C_TRUE;
break; /* Route successfully established, break chain safely */
}
/* Connection failed on this node: close socket using custom macro helper */
c_Socket_Close(self->sock);
}
/* Shift forward to parse alternative network address frames */
c_AddrInfo_GetNext(&curr_node, &curr_node);
}
c_AddrInfo_Destroy(&addr_resolver);
if (!connected) {
if (self->fnOnError) {
self->fnOnError(self, C_TCPCLIENT_ERR_ON_CONNECT, args);
}
return C_ERR_FAIL; /* Host unreachable or timed out */
}
self->is_linked = C_TRUE;
if (self->fnOnConnect) {
if (!self->fnOnConnect(self, args)) {
self->is_linked = C_FALSE;
}
}
return C_SUCCESS;
}
c_err_t c_TcpClient_Send(c_TcpClient_t* self, const void* buffer, c_size_t bytes_to_send, c_size_t* out_bytes_sent) {
if (!self || !self->is_linked || !c_Socket_IsValid(self->sock) || !buffer) return C_ERR_PARAM;
if (bytes_to_send == 0) {
if (out_bytes_sent) *out_bytes_sent = 0;
return C_SUCCESS;
}
c_size_t total_sent = 0;
const char* ptr = (const char*)buffer;
/* Loop until the complete raw buffer payload block is flushed out to the wire */
while (total_sent < bytes_to_send) {
c_size_t remaining = bytes_to_send - total_sent;
#if defined(_WIN32) || defined(_WIN64)
int n = send(self->sock, ptr + total_sent, (int)remaining, 0);
#else
#ifdef MSG_NOSIGNAL
ssize_t n = send(self->sock, ptr + total_sent, (size_t)remaining, MSG_NOSIGNAL); /* Guard against SIGPIPE */
#else
ssize_t n = send(self->sock, ptr + total_sent, (size_t)remaining, 0);
#endif
#endif
if (n == C_SOCKET_ERROR) {
if (out_bytes_sent) *out_bytes_sent = total_sent;
return C_ERR_FAIL; /* Network pipeline transmission error */
}
total_sent += (c_size_t)n;
}
if (out_bytes_sent) *out_bytes_sent = total_sent;
return C_SUCCESS;
}
c_err_t c_TcpClient_Recv(c_TcpClient_t* self, void* buffer, c_size_t max_bytes_to_read, c_size_t* out_bytes_read) {
if (!self || !self->is_linked || !c_Socket_IsValid(self->sock) || !buffer || max_bytes_to_read == 0) return C_ERR_PARAM;
#if defined(_WIN32) || defined(_WIN64)
int n = recv(self->sock, (char*)buffer, (int)max_bytes_to_read, 0);
#else
ssize_t n = recv(self->sock, buffer, (size_t)max_bytes_to_read, 0);
#endif
if (n == 0) {
if (out_bytes_read) *out_bytes_read = 0;
return C_ERR_EOF; /* Connection clean cut closed by peer node */
}
if (n == C_SOCKET_ERROR) {
if (out_bytes_read) *out_bytes_read = 0;
return C_ERR_FAIL; /* Network pipeline failure */
}
if (out_bytes_read) *out_bytes_read = (c_size_t)n;
return C_SUCCESS;
}
void c_TcpClient_Destroy(c_TcpClient_t* self) {
if (!self) return;
self->is_linked = C_FALSE;
if (c_Socket_IsValid(self->sock)) {
c_Socket_Close(self->sock);
}
if (self->reactor) c_TcpClientReactor_Destroy(self->reactor);
}
c_err_t c_TcpClient_PollEvents(c_TcpClient_t* self, long timeout_ms, void* args) {
if (!self || !self->reactor || self->is_linked==C_FALSE) return C_ERR_PARAM;
return c_TcpClientReactor_Poll(self->reactor, timeout_ms, self, args);
}
Foundation/c_TcpClient.h
+161
View File
@@ -0,0 +1,161 @@
#ifndef INCLUDED_C_TCPCLIENT_H
#define INCLUDED_C_TCPCLIENT_H
#ifndef INCLUDED_C_SOCKET_H
#include <c_Socket.h>
#endif /*INCLUDED_C_SOCKET_H*/
#ifndef INCLUDED_C_ALLOCATOR_H
#include <c_Allocator.h>
#endif /*INCLUDED_C_ALLOCATOR_H*/
/* ------------------------------------------------------------------------------------------------------------------ */
/* */
#define C_TCPCLIENT_ERR_ON_POLL 1001
#define C_TCPCLIENT_ERR_ON_ACCEPT 1002
#define C_TCPCLIENT_ERR_ON_POLLERR 1003
#define C_TCPCLIENT_ERR_ON_RECV 1004
#define C_TCPCLIENT_ERR_ON_CONNECT 1005
#define C_TCPCLIENT_INITIALIZE {0}
/* ------------------------------------------------------------------------------------------------------------------ */
/* */
typedef struct c_TcpClient_t c_TcpClient_t;
typedef bool (*c_TcpClient_OnConnect_t)(c_TcpClient_t* client, void* args);
typedef bool (*c_TcpClient_OnData_t)(c_TcpClient_t* client, void* args);
typedef void (*c_TcpClient_OnDisconnect_t)(c_TcpClient_t* client, void* args);
typedef void (*c_TcpClient_OnError_t)(c_TcpClient_t* client, c_err_t error_code, void* args);
typedef struct c_TcpClientReactor_t c_TcpClientReactor_t;
typedef struct {
c_err_t (*add)(c_TcpClientReactor_t* self, c_socket_t sock, uint32_t events);
c_err_t (*remove)(c_TcpClientReactor_t* self, c_socket_t sock);
c_err_t (*poll)(c_TcpClientReactor_t* self, long timeout_ms, c_TcpClient_t* client, void* args);
void (*destroy)(c_TcpClientReactor_t* self);
} c_TcpClientReactorVtbl_t;
struct c_TcpClientReactor_t {
const c_TcpClientReactorVtbl_t* vtbl;
};
struct c_TcpClient_t {
c_socket_t sock;
c_bool_t is_linked;
c_TcpClientReactor_t* reactor;
c_TcpClient_OnConnect_t fnOnConnect;
c_TcpClient_OnData_t fnOnData;
c_TcpClient_OnDisconnect_t fnOnDisconnect;
c_TcpClient_OnError_t fnOnError;
c_Allocator_t allocator;
};
/* ------------------------------------------------------------------------------------------------------------------ */
/* */
C_STATIC_FORCE_INLINE
c_err_t c_TcpClientReactor_Add(c_TcpClientReactor_t* self, c_socket_t sock, uint32_t events) {
if (!self || !self->vtbl || !self->vtbl->add || !c_Socket_IsValid(sock)) return C_ERR_PARAM;
return self->vtbl->add(self, sock, events);
}
C_STATIC_FORCE_INLINE
c_err_t c_TcpClientReactor_Remove(c_TcpClientReactor_t* self, c_socket_t sock) {
if (!self || !self->vtbl || !self->vtbl->remove || !c_Socket_IsValid(sock)) return C_ERR_PARAM;
return self->vtbl->remove(self, sock);
}
C_STATIC_FORCE_INLINE
c_err_t c_TcpClientReactor_Poll(c_TcpClientReactor_t* self, long timeout_ms, c_TcpClient_t* client, void* args) {
if (!self || !self->vtbl || !self->vtbl->poll || !client) return C_ERR_PARAM;
return self->vtbl->poll(self, timeout_ms, client, args);
}
C_STATIC_FORCE_INLINE
void c_TcpClientReactor_Destroy(c_TcpClientReactor_t* self) {
if (!self || !self->vtbl || !self->vtbl->destroy) return;
self->vtbl->destroy(self);
}
/* ------------------------------------------------------------------------------------------------------------------ */
/* */
/**
* @brief Establishes a synchronous stream connection to a remote server host.
* @param family Address family (e.g., AF_INET, AF_INET6, or AF_UNSPEC)
* @param server Host destination string (e.g., "127.0.0.1" or "domain.com")
* @param port Service port string identifier (e.g., "8080")
* @param args User defined arguments
* @param allocator Explicit allocator context used to configure socket instances
*/
c_err_t c_TcpClient_Connect(c_TcpClient_t* self, int family, const char* server, const char* port,
void* args,
c_Allocator_t* allocator);
/**
* @brief Sends a raw data block across the socket, looping until the full buffer is dispatched.
* @param buffer Pointer to the source data block payload to transmit.
* @param bytes_to_send Exact size threshold of the payload block in bytes.
* @param out_bytes_sent Receives the actual number of bytes written out to the wire.
*/
c_err_t c_TcpClient_Send(c_TcpClient_t* self, const void* buffer, c_size_t bytes_to_send, c_size_t* out_bytes_sent);
/**
* @brief Reads a data block up to the requested size from the socket channel.
* @param buffer Pointer to the destination buffer to hold incoming data.
* @param max_bytes_to_read Upper ceiling threshold size of bytes to pull.
* @param out_bytes_read Receives the actual number of bytes fetched from the wire.
*/
c_err_t c_TcpClient_Recv(c_TcpClient_t* self, void* buffer, c_size_t max_bytes_to_read, c_size_t* out_bytes_read);
/**
* @brief Safely shuts down pipelines, closes active sockets, and recycles memory.
*/
void c_TcpClient_Destroy(c_TcpClient_t* self);
/**
* @brief Executes a non-recursive, single-pass asynchronous polling event dispatch iteration for the client.
*/
c_err_t c_TcpClient_PollEvents(c_TcpClient_t* self, long timeout_ms, void* args);
/* ------------------------------------------------------------------------------------------------------------------ */
/* */
C_STATIC_FORCE_INLINE
void c_TcpClient_SetReactor(c_TcpClient_t* self, c_TcpClientReactor_t* reactor) {
if (!self || !reactor) return;
self->reactor = reactor;
}
C_STATIC_FORCE_INLINE
void c_TcpClient_SetOnConnect(c_TcpClient_t* self, c_TcpClient_OnConnect_t fnOnConnect) {
if (!self ) return;
self->fnOnConnect = fnOnConnect;
}
C_STATIC_FORCE_INLINE
void c_TcpClient_SetOnData(c_TcpClient_t* self, c_TcpClient_OnData_t fnOnData) {
if (!self ) return;
self->fnOnData = fnOnData;
}
C_STATIC_FORCE_INLINE
void c_TcpClient_SetOnDisconnect(c_TcpClient_t* self, c_TcpClient_OnDisconnect_t fnOnDisconnect) {
if (!self ) return;
self->fnOnDisconnect = fnOnDisconnect;
}
C_STATIC_FORCE_INLINE
void c_TcpClient_SetOnError(c_TcpClient_t* self, c_TcpClient_OnError_t fnOnError) {
if (!self ) return;
self->fnOnError = fnOnError;
}
#endif /*INCLUDED_C_TCPCLIENT_H*/
Foundation/c_TcpClient.t.c
+30
View File
@@ -0,0 +1,30 @@
#include "c_Test.h"
#include "c_TcpClient.h"
#include <string.h>
TEST_CASE(test_raw_stream_free_client_lifecycle) {
c_Socket_Init();
c_TcpClient_t client=C_TCPCLIENT_INITIALIZE;
/* Connect operation should gracefully fail on an unbound dead port */
c_err_t err = c_TcpClient_Connect(&client, AF_INET, "127.0.0.1", "9999", 0, NULL);
ASSERT_INT_EQ(C_ERR_FAIL, err);
ASSERT_FALSE(client.is_linked);
ASSERT_FALSE(c_Socket_IsValid(client.sock));
/* Raw Send operations must catch parameter unassigned states safely */
c_size_t sent = 0;
err = c_TcpClient_Send(&client, "DATA", 4, &sent);
ASSERT_INT_EQ(C_ERR_PARAM, err);
/* Execute explicit structure cleanup safety tests */
c_TcpClient_Destroy(&client);
c_Socket_Destroy();
}
int main(void) {
TEST_START(TcpClient_RawStreamFree_Suite);
RUN_TEST(test_raw_stream_free_client_lifecycle);
TEST_REPORT();
RETURN_TEST_STATUS;
}
Foundation/c_TcpClientPollReactor.c
+125
View File
@@ -0,0 +1,125 @@
#include <c_TcpClientPollReactor.h>
#if !defined(_WIN32) && !defined(_WIN64)
#include <poll.h>
#endif
typedef struct {
c_TcpClientReactor_t base;
struct pollfd* fds;
c_size_t count;
c_size_t capacity;
c_Allocator_t allocator;
} c_ClientPollImpl_t;
static c_err_t ClientPoll_Add(c_TcpClientReactor_t* self, c_socket_t sock, uint32_t events) {
c_ClientPollImpl_t* impl = (c_ClientPollImpl_t*)self;
if (impl->count >= impl->capacity) {
c_size_t old_cap = impl->capacity;
c_size_t new_cap = old_cap == 0 ? 4 : old_cap * 2;
struct pollfd* new_fds = (struct pollfd*)c_Allocator_Realloc(&impl->allocator, impl->fds, old_cap * sizeof(struct pollfd), new_cap * sizeof(struct pollfd));
if (!new_fds) return C_ERR_NOMEM;
impl->fds = new_fds;
impl->capacity = new_cap;
}
impl->fds[impl->count].fd = (int)sock;
impl->fds[impl->count].events = (short)(events ? POLLIN : 0);
impl->fds[impl->count].revents = 0;
impl->count++;
return C_SUCCESS;
}
static c_err_t ClientPoll_Remove(c_TcpClientReactor_t* self, c_socket_t sock) {
c_ClientPollImpl_t* impl = (c_ClientPollImpl_t*)self;
for (c_size_t i = 0; i < impl->count; ++i) {
if (impl->fds[i].fd == (int)sock) {
impl->fds[i] = impl->fds[impl->count - 1]; /* O(1) tail-swap */
impl->count--;
return C_SUCCESS;
}
}
return C_ERR_NOTFOUND;
}
static c_err_t ClientPoll_Poll(c_TcpClientReactor_t* self, long timeout_ms, c_TcpClient_t* client, void* args) {
c_ClientPollImpl_t* impl = (c_ClientPollImpl_t*)self;
if (!client->is_linked || !c_Socket_IsValid(client->sock)) return C_SUCCESS;
if (impl->count == 0) {
ClientPoll_Add(self, client->sock, 1);
}
#if defined(_WIN32) || defined(_WIN64)
int ret = WSAPoll(impl->fds, (ULONG)impl->count, (INT)timeout_ms);
#else
int ret = poll(impl->fds, (nfds_t)impl->count, (int)timeout_ms);
#endif
if (ret < 0) {
if (client->fnOnError) {
client->fnOnError(client, C_TCPCLIENT_ERR_ON_POLL, args);
}
return C_SUCCESS;
}
if (ret == 0) return C_SUCCESS;
struct pollfd* client_fd = &impl->fds[0];
if (client_fd->revents & (POLLIN | POLLHUP | POLLERR)) {
c_bool_t keep_alive = C_TRUE;
if (client_fd->revents & POLLERR) {
if (client->fnOnError) {
client->fnOnError(client, C_TCPCLIENT_ERR_ON_POLLERR, args);
}
keep_alive = C_FALSE;
}
if (keep_alive) {
char peek_buf;
#if defined(_WIN32) || defined(_WIN64)
int peek_res = recv(client->sock, &peek_buf, 1, MSG_PEEK);
#else
ssize_t peek_res = recv(client->sock, &peek_buf, 1, MSG_PEEK);
#endif
if (peek_res == 0 || peek_res == C_SOCKET_ERROR) {
keep_alive = C_FALSE;
} else if (client->fnOnData) {
keep_alive = client->fnOnData(client, args);
}
}
if (!keep_alive || (client_fd->revents & POLLHUP)) {
ClientPoll_Remove(self, client->sock);
if (client->fnOnDisconnect) {
client->fnOnDisconnect(client, args);
}
c_Socket_Close(client->sock);
client->is_linked = false;
return C_ERR_EOF;
}
}
return C_SUCCESS;
}
static void ClientPoll_Destroy(c_TcpClientReactor_t* self) {
c_ClientPollImpl_t* impl = (c_ClientPollImpl_t*)self;
c_Allocator_t alloc = impl->allocator;
if (impl->fds) c_Allocator_Free(&alloc, impl->fds);
c_Allocator_Free(&alloc, impl);
}
static const c_TcpClientReactorVtbl_t g_ClientPollVtbl = { ClientPoll_Add, ClientPoll_Remove, ClientPoll_Poll, ClientPoll_Destroy };
c_err_t c_TcpClientPollReactor_Create(c_TcpClientReactor_t** out_reactor, c_Allocator_t* allocator) {
if (!out_reactor) return C_ERR_PARAM;
c_Allocator_t alloc = allocator ? *allocator : c_DefaultAllocator;
c_ClientPollImpl_t* impl = (c_ClientPollImpl_t*)c_Allocator_Calloc(&alloc, 1, sizeof(c_ClientPollImpl_t));
if (!impl) return C_ERR_NOMEM;
impl->base.vtbl = &g_ClientPollVtbl;
impl->fds = NULL;
impl->count = 0;
impl->capacity = 0;
impl->allocator = alloc;
*out_reactor = &impl->base;
return C_SUCCESS;
}
Foundation/c_TcpClientPollReactor.h
+14
View File
@@ -0,0 +1,14 @@
#ifndef INCLUDED_C_TCPCLIENTPOLLREACTOR_H
#define INCLUDED_C_TCPCLIENTPOLLREACTOR_H
#ifndef INCLUDED_C_TCPCLIENT_H
#include <c_TcpClient.h>
#endif /*INCLUDED_C_TCPCLIENT_H*/
/* ------------------------------------------------------------------------------------------------------------------ */
/* */
c_err_t c_TcpClientPollReactor_Create(c_TcpClientReactor_t** out_reactor, c_Allocator_t* allocator);
#endif /*INCLUDED_C_TCPCLIENTPOLLREACTOR_H*/
Foundation/c_TcpClientPoolReactor_TimeClient.t.c
+70
View File
@@ -0,0 +1,70 @@
#include "c_TcpClient.h"
#include "c_TcpClientPollReactor.h"
#include "c_SocketUtil.h"
#include <stdlib.h>
#include <stdio.h>
static bool OnConnect(c_TcpClient_t* client, void* args) {
char ip[128];
char port[32];
c_SocketUtil_GetNameInfoForSocket(client->sock, ip, sizeof(ip), port, sizeof(port));
printf("[CONN] %s:%s\n", ip, port);
c_TcpClient_Send(client, "Hello", 5, 0);
return true;
}
static bool OnData(c_TcpClient_t* client, void* args) {
char ip[128];
char port[32];
c_SocketUtil_GetNameInfoForSocket(client->sock, ip, sizeof(ip), port, sizeof(port));
printf("[DATA] %s:%s\n", ip, port);
char buffer[1024]={0};
c_size_t recv_bytes = 0;
c_TcpClient_Recv(client, buffer, sizeof(buffer), &recv_bytes);
printf("%.*s\n", (int)recv_bytes, buffer);
return false;
}
static void OnDisconnect(c_TcpClient_t* client, void* args) {
char ip[128];
char port[32];
c_SocketUtil_GetNameInfoForSocket(client->sock, ip, sizeof(ip), port, sizeof(port));
printf("[CLOSE] %s:%s\n", ip, port);
}
static void OnError(c_TcpClient_t* client, c_err_t err, void* args) {
if (c_Socket_IsValid(client->sock)) {
char ip[128];
char port[32];
c_SocketUtil_GetNameInfoForSocket(client->sock, ip, sizeof(ip), port, sizeof(port));
printf("[ERROR] %s:%s code:%d\n", ip, port, err);
}else {
printf("[ERROR] code:%d\n", err);
}
}
int main(int argc, char** argv){
c_Socket_Init();
c_TcpClient_t client=C_TCPCLIENT_INITIALIZE;
c_TcpClient_SetOnConnect(&client, OnConnect);
c_TcpClient_SetOnData(&client, OnData);
c_TcpClient_SetOnDisconnect(&client, OnDisconnect);
c_TcpClient_SetOnError(&client, OnError);
c_err_t err = c_TcpClient_Connect(&client, AF_INET, "127.0.0.1", "1313", 0, NULL);
c_TcpClientReactor_t* reactor = NULL;
c_TcpClientPollReactor_Create(&reactor, NULL);
c_TcpClient_SetReactor(&client, reactor);
while (client.is_linked) {
c_TcpClient_PollEvents(&client, 1000, NULL);
}
c_TcpClient_Destroy(&client);
c_Socket_Destroy();
return 0;
}
Foundation/c_TcpClient_TimeClient.t.c
+74
View File
@@ -0,0 +1,74 @@
#include <c_TcpClient.h>
#include <stdlib.h>
#include <stdio.h>
#include "c_Test.h"
int main(int argc, char** argv){
c_Socket_Init();
const char* server_host = "127.0.0.1";
const char* server_port = "1313"; /* Your time server daytime service port */
c_TcpClient_t client;
printf("[CLIENT] Connecting to Time Server at %s:%s...\n", server_host, server_port);
/* 1. Establish a raw stream connection using your explicit family-agnostic constructor */
c_err_t err = c_TcpClient_Connect(&client, AF_INET, server_host, server_port, NULL);
if (err != C_SUCCESS) {
fprintf(stderr, "[ERROR] Unable to connect to remote Time Server host destination.\n");
return -1;
}
printf("[CLIENT] Connection successful! Awaiting network timestamp data package...\n");
char recv_buffer[128];
memset(recv_buffer, 0, sizeof(recv_buffer));
c_size_t total_bytes_read = 0;
char send_buffer[1024];
c_size_t send_bytes = 0;
/*
* 2. Execute a raw block read operation.
* Loop to read stream chunks until the server sends the full packet and shuts down (returns C_ERR_EOF).
*/
while (true) {
c_size_t chunk_bytes = 0;
char* current_ptr = recv_buffer + total_bytes_read;
c_size_t available_space = sizeof(recv_buffer) - total_bytes_read - 1;
if (available_space == 0) break; /* Safety boundary ceiling check */
if (send_bytes == 0) {
c_size_t sz = snprintf(send_buffer, sizeof(send_buffer), "GET_TIME");
c_TcpClient_Send(&client, send_buffer, sz, &send_bytes);
}else {
// c_size_t sz = snprintf(send_buffer, sizeof(send_buffer), "STOP");
// c_TcpClient_Send(&client, send_buffer, sz, &send_bytes);
break;
}
err = c_TcpClient_Recv(&client, current_ptr, available_space, &chunk_bytes);
if (err == C_ERR_EOF) {
break; /* Server closed the connection gracefully, transaction complete */
}
if (err != C_SUCCESS) {
fprintf(stderr, "[ERROR] Network stream read transaction failure caught.\n");
c_TcpClient_Destroy(&client);
return -1;
}
total_bytes_read += chunk_bytes;
}
/* 3. Null-terminate the raw buffer safely before string logging operations */
recv_buffer[total_bytes_read] = '\0';
printf("\n " COLOR_GREEN "[SUCCESS] Network Time String Intercepted:" COLOR_RESET "\n");
printf(" >>> %s", recv_buffer);
/* 4. Safely close physical socket descriptors and reclaim global system assets */
c_TcpClient_Destroy(&client);
printf("\n[CLIENT] Client connection resources cleanly recycled.\n");
c_Socket_Destroy();
return 0;
}
Foundation/c_TcpEpollReactor.c
+174
View File
@@ -0,0 +1,174 @@
#include <c_TcpEpollReactor.h>
#include "c_TcpServer.h"
#if defined(__linux__)
#include <sys/epoll.h>
#include <unistd.h>
#include <string.h>
#define MAX_EPOLL_EVENTS 1024
typedef struct {
c_TcpReactor_t base;
int epoll_fd; /* Raw Linux epoll kernel handle instance descriptor */
struct epoll_event events[MAX_EPOLL_EVENTS]; /* Pre-allocated active kernel alerts buffer */
c_Allocator_t allocator; /* Deep copy of the user-provided allocator */
} c_EPollReactorImpl_t;
static c_err_t EPoll_Add(c_TcpReactor_t* self, c_socket_t sock, uint32_t events) {
c_EPollReactorImpl_t* impl = (c_EPollReactorImpl_t*)self;
struct epoll_event ev;
memset(&ev, 0, sizeof(ev));
/* Map incoming events to POLLIN / read data availability triggers */
ev.events = (events ? EPOLLIN : 0) | EPOLLERR | EPOLLHUP;
ev.data.fd = (int)sock;
if (epoll_ctl(impl->epoll_fd, EPOLL_CTL_ADD, (int)sock, &ev) < 0) {
return C_ERR_FAIL;
}
return C_SUCCESS;
}
static c_err_t EPoll_Remove(c_TcpReactor_t* self, c_socket_t sock) {
c_EPollReactorImpl_t* impl = (c_EPollReactorImpl_t*)self;
/* Passing NULL for the epoll_event structure is valid for deletions since Linux 2.6.9 */
if (epoll_ctl(impl->epoll_fd, EPOLL_CTL_DEL, (int)sock, NULL) < 0) {
return C_ERR_NOTFOUND;
}
return C_SUCCESS;
}
static c_err_t EPoll_Poll(c_TcpReactor_t* self, int timeout_ms, c_TcpServer_t* server, void* args) {
c_EPollReactorImpl_t* impl = (c_EPollReactorImpl_t*)self;
if (!server->is_running) return C_SUCCESS;
/* Intercept readiness frames directly out of the epoll kernel tree instance */
int ret = epoll_wait(impl->epoll_fd, impl->events, MAX_EPOLL_EVENTS, timeout_ms);
if (ret < 0) {
if (server->fnOnError) {
server->fnOnError(server, server->listen_sock, C_TCPSERVER_ERR_ON_POLL, args);
}
return C_SUCCESS;
}
if (ret == 0) return C_SUCCESS; /* Timeout pass, exit cleanly */
/*
* Iterative Forward Processing Pass:
* Unlike arrays, epoll_wait packs active ready elements sequentially into indices 0 to ret-1.
* This provides a strict O(1) loop traversal path that is entirely immune to index mutations.
*/
for (int i = 0; i < ret; ++i) {
struct epoll_event* current_ev = &impl->events[i];
c_socket_t active_sock = (c_socket_t)current_ev->data.fd;
/* Channel A: Master server listener handles incoming connection handshakes */
if (active_sock == server->listen_sock) {
if (current_ev->events & EPOLLIN) {
c_SockAddr_t peer_addr;
socklen_t addr_len = sizeof(peer_addr);
c_socket_t client = accept(server->listen_sock, (struct sockaddr*)&peer_addr, &addr_len);
if (c_Socket_IsValid(client)) {
bool keep = true;
if (server->fnOnConnect) {
keep = server->fnOnConnect(server, client, &peer_addr, args);
}
if (keep) {
EPoll_Add(self, client, 1); /* Inject new client into the red-black tree watchlist */
} else {
c_Socket_Close(client);
}
} else if (server->fnOnError) {
server->fnOnError(server, server->listen_sock, C_TCPSERVER_ERR_ON_ACCEPT, args);
}
}
if (current_ev->events & (EPOLLERR | EPOLLHUP)) {
if (server->fnOnError) {
server->fnOnError(server, server->listen_sock, C_TCPSERVER_ERR_ON_POLLERR, args);
}
}
}
/* Channel B: Connected active client descriptors processing data readiness loops */
else {
c_bool_t keep_alive = C_TRUE;
/* Intercept hardware-level operational error signals first */
if (current_ev->events & EPOLLERR) {
if (server->fnOnError) {
server->fnOnError(server, active_sock, C_TCPSERVER_ERR_ON_POLLERR, args);
}
keep_alive = C_FALSE;
}
if (keep_alive && (current_ev->events & EPOLLIN)) {
/* Identify clean socket EOF drops via zero-byte peeks */
char peek_buf;
ssize_t peek_res = recv(active_sock, &peek_buf, 1, MSG_PEEK);
if (peek_res == 0 || peek_res < 0) {
keep_alive = C_FALSE; /* Connection explicitly severed by remote peer node */
} else if (server->fnOnRequest) {
keep_alive = server->fnOnRequest(server, active_sock, args);
}
}
/* Clean up active client descriptor tracking states if loop signaling drops or errors out */
if (!keep_alive || (current_ev->events & EPOLLHUP)) {
/*
* Remove explicitly from epoll tracking first.
* Closing the socket handle automatically removes it from epoll,
* but manual synchronization ensures clean deterministic state tracking.
*/
EPoll_Remove(self, active_sock);
if (server->fnOnDisconnect) {
server->fnOnDisconnect(server, active_sock, args);
}
c_Socket_Close(active_sock);
}
}
}
return C_SUCCESS;
}
static void EPoll_Destroy(c_TcpReactor_t* self) {
c_EPollReactorImpl_t* impl = (c_EPollReactorImpl_t*)self;
c_Allocator_t alloc = impl->allocator;
if (impl->epoll_fd >= 0) {
close(impl->epoll_fd);
}
c_Allocator_Free(&alloc, impl);
}
static const c_TcpReactorVtbl_t g_EPollReactorVtbl = { EPoll_Add, EPoll_Remove, EPoll_Poll, EPoll_Destroy };
/* ------------------------------------------------------------------------------------------------------------------ */
/* */
c_err_t c_TcpEPollReactor_Create(c_TcpReactor_t** out_reactor, c_Allocator_t* allocator) {
if (!out_reactor) return C_ERR_PARAM;
c_Allocator_t alloc = allocator ? *allocator : c_DefaultAllocator;
c_EPollReactorImpl_t* impl = (c_EPollReactorImpl_t*)c_Allocator_Calloc(&alloc, 1, sizeof(c_EPollReactorImpl_t));
if (!impl) return C_ERR_NOMEM;
/* Create the epoll instance handle (Size parameter must be > 0; ignored since Linux 2.6.8) */
impl->epoll_fd = epoll_create(1);
if (impl->epoll_fd < 0) {
c_Allocator_Free(&alloc, impl);
return C_ERR_FAIL;
}
impl->base.vtbl = &g_EPollReactorVtbl;
impl->allocator = alloc;
*out_reactor = &impl->base;
return C_SUCCESS;
}
#endif /* __linux__ */
Foundation/c_TcpEpollReactor.h
+13
View File
@@ -0,0 +1,13 @@
#ifndef INCLUDED_C_TCPEPOLLREACTOR_H
#define INCLUDED_C_TCPEPOLLREACTOR_H
#ifndef INCLUDED_C_TCPSERVER_H
#include <c_TcpServer.h>
#endif /*INCLUDED_C_TCPSERVER_H*/
#if defined(__linux__)
c_err_t c_TcpEPollReactor_Create(c_TcpReactor_t** out_reactor, c_Allocator_t* allocator);
#endif
#endif /*INCLUDED_C_TCPEPOLLREACTOR_H*/
Foundation/c_TcpServer.h
+1 -1
View File
@@ -85,12 +85,12 @@ struct c_TcpReactor_t {
struct c_TcpServer_t {
c_socket_t listen_sock; /* Master server listening socket descriptor */
c_bool_t is_running; /* Server master thread loop state flag */
c_TcpReactor_t* reactor; /* Polymorphic multiplexing backend driver link */
c_TcpServer_OnConnect_t fnOnConnect; /* Connection accept callback hook */
c_TcpServer_OnRequest_t fnOnRequest; /* Data incoming available callback hook */
c_TcpServer_OnDisconnect_t fnOnDisconnect; /* Disconnect termination event hook */
c_TcpServer_OnError_t fnOnError; /* Network operational error event hook */
c_bool_t is_running; /* Server master thread loop state flag */
c_Allocator_t allocator; /* Deep copy of the user-provided allocator */
};