diff --git a/Base/c_Types.h b/Base/c_Types.h index 6b323b1..72c1af2 100644 --- a/Base/c_Types.h +++ b/Base/c_Types.h @@ -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 /* ------------------------------------------------------------------------------------------------------------------ */ /* */ diff --git a/Foundation/c_TcpClient.c b/Foundation/c_TcpClient.c new file mode 100644 index 0000000..bce8440 --- /dev/null +++ b/Foundation/c_TcpClient.c @@ -0,0 +1,138 @@ +#include +#include + +/* ------------------------------------------------------------------------------------------------------------------ */ +/* */ + +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); +} + diff --git a/Foundation/c_TcpClient.h b/Foundation/c_TcpClient.h new file mode 100644 index 0000000..3cc2774 --- /dev/null +++ b/Foundation/c_TcpClient.h @@ -0,0 +1,161 @@ +#ifndef INCLUDED_C_TCPCLIENT_H +#define INCLUDED_C_TCPCLIENT_H + +#ifndef INCLUDED_C_SOCKET_H +#include +#endif /*INCLUDED_C_SOCKET_H*/ + + +#ifndef INCLUDED_C_ALLOCATOR_H +#include +#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*/ diff --git a/Foundation/c_TcpClient.t.c b/Foundation/c_TcpClient.t.c new file mode 100644 index 0000000..ff2a460 --- /dev/null +++ b/Foundation/c_TcpClient.t.c @@ -0,0 +1,30 @@ +#include "c_Test.h" +#include "c_TcpClient.h" +#include + +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; +} diff --git a/Foundation/c_TcpClientPollReactor.c b/Foundation/c_TcpClientPollReactor.c new file mode 100644 index 0000000..f2ea323 --- /dev/null +++ b/Foundation/c_TcpClientPollReactor.c @@ -0,0 +1,125 @@ +#include + +#if !defined(_WIN32) && !defined(_WIN64) + #include +#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; +} \ No newline at end of file diff --git a/Foundation/c_TcpClientPollReactor.h b/Foundation/c_TcpClientPollReactor.h new file mode 100644 index 0000000..711502a --- /dev/null +++ b/Foundation/c_TcpClientPollReactor.h @@ -0,0 +1,14 @@ +#ifndef INCLUDED_C_TCPCLIENTPOLLREACTOR_H +#define INCLUDED_C_TCPCLIENTPOLLREACTOR_H + +#ifndef INCLUDED_C_TCPCLIENT_H +#include +#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*/ diff --git a/Foundation/c_TcpClientPoolReactor_TimeClient.t.c b/Foundation/c_TcpClientPoolReactor_TimeClient.t.c new file mode 100644 index 0000000..1bd38a9 --- /dev/null +++ b/Foundation/c_TcpClientPoolReactor_TimeClient.t.c @@ -0,0 +1,70 @@ +#include "c_TcpClient.h" +#include "c_TcpClientPollReactor.h" +#include "c_SocketUtil.h" +#include +#include + +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; +} diff --git a/Foundation/c_TcpClient_TimeClient.t.c b/Foundation/c_TcpClient_TimeClient.t.c new file mode 100644 index 0000000..a273846 --- /dev/null +++ b/Foundation/c_TcpClient_TimeClient.t.c @@ -0,0 +1,74 @@ +#include +#include +#include +#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; +} diff --git a/Foundation/c_TcpEpollReactor.c b/Foundation/c_TcpEpollReactor.c new file mode 100644 index 0000000..2347a89 --- /dev/null +++ b/Foundation/c_TcpEpollReactor.c @@ -0,0 +1,174 @@ +#include +#include "c_TcpServer.h" + +#if defined(__linux__) +#include +#include +#include + +#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__ */ + diff --git a/Foundation/c_TcpEpollReactor.h b/Foundation/c_TcpEpollReactor.h new file mode 100644 index 0000000..2501f98 --- /dev/null +++ b/Foundation/c_TcpEpollReactor.h @@ -0,0 +1,13 @@ +#ifndef INCLUDED_C_TCPEPOLLREACTOR_H +#define INCLUDED_C_TCPEPOLLREACTOR_H + +#ifndef INCLUDED_C_TCPSERVER_H +#include +#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*/ diff --git a/Foundation/c_TcpServer.h b/Foundation/c_TcpServer.h index 7691835..3266885 100644 --- a/Foundation/c_TcpServer.h +++ b/Foundation/c_TcpServer.h @@ -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 */ };