Browse Source

chat headless: cross-platform sockets via socket_compat (Windows build)

topo_upd
Evgeny 2 months ago
parent
commit
b7348f3237
  1. 54
      src/chat/chat_headless_control.c

54
src/chat/chat_headless_control.c

@ -22,12 +22,8 @@
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
#include <stdio.h> #include <stdio.h>
#include <errno.h> #include "../../lib/platform_compat.h"
#include <unistd.h> #include "../../lib/socket_compat.h"
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <fcntl.h>
#define MAX_CLIENTS 16 #define MAX_CLIENTS 16
#define RECV_BUF_SIZE 16384 #define RECV_BUF_SIZE 16384
@ -104,7 +100,7 @@ static const char* json_get_cmd_end(const char* cmd_start) {
/* ── Send helpers ── */ /* ── Send helpers ── */
static void cli_send(struct headless_client* cli, const char* data, size_t len) { static void cli_send(struct headless_client* cli, const char* data, size_t len) {
if (!cli || cli->closing || cli->fd < 0 || !data || len == 0) return; if (!cli || cli->closing || cli->fd == SOCKET_INVALID || !data || len == 0) return;
if (cli->send_len > 0) { if (cli->send_len > 0) {
size_t space = sizeof(cli->send_buf) - (size_t)cli->send_len; size_t space = sizeof(cli->send_buf) - (size_t)cli->send_len;
if (len + 1 > space) return; if (len + 1 > space) return;
@ -112,8 +108,8 @@ static void cli_send(struct headless_client* cli, const char* data, size_t len)
cli->send_len += (int)len; cli->send_len += (int)len;
return; return;
} }
ssize_t r = write(cli->fd, data, len); ssize_t r = send(cli->fd, data, (int)len, 0);
if (r < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) { if (r < 0 && (socket_get_error() == ERR_AGAIN || socket_get_error() == ERR_WOULDBLOCK)) {
if (len <= sizeof(cli->send_buf)) { memcpy(cli->send_buf, data, len); cli->send_len = (int)len; cli->send_offset = 0; } if (len <= sizeof(cli->send_buf)) { memcpy(cli->send_buf, data, len); cli->send_len = (int)len; cli->send_offset = 0; }
if (g_hc.ua) uasync_set_socket_write(g_hc.ua, cli->socket_id, 1); if (g_hc.ua) uasync_set_socket_write(g_hc.ua, cli->socket_id, 1);
return; return;
@ -182,8 +178,8 @@ static void client_write_callback(socket_t fd, void* arg) {
struct headless_client* cli = (struct headless_client*)arg; struct headless_client* cli = (struct headless_client*)arg;
if (!cli || cli->closing || fd != cli->fd) return; if (!cli || cli->closing || fd != cli->fd) return;
if (cli->send_len <= 0) { if (g_hc.ua) uasync_set_socket_write(g_hc.ua, cli->socket_id, 0); return; } if (cli->send_len <= 0) { if (g_hc.ua) uasync_set_socket_write(g_hc.ua, cli->socket_id, 0); return; }
ssize_t r = write(cli->fd, cli->send_buf + cli->send_offset, (size_t)(cli->send_len - cli->send_offset)); ssize_t r = send(cli->fd, cli->send_buf + cli->send_offset, (int)(cli->send_len - cli->send_offset), 0);
if (r < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) return; if (r < 0 && (socket_get_error() == ERR_AGAIN || socket_get_error() == ERR_WOULDBLOCK)) return;
if (r <= 0) { cli->send_len = 0; cli->send_offset = 0; if (g_hc.ua) uasync_set_socket_write(g_hc.ua, cli->socket_id, 0); return; } if (r <= 0) { cli->send_len = 0; cli->send_offset = 0; if (g_hc.ua) uasync_set_socket_write(g_hc.ua, cli->socket_id, 0); return; }
cli->send_offset += (int)r; cli->send_offset += (int)r;
if (cli->send_offset >= cli->send_len) { cli->send_len = 0; cli->send_offset = 0; if (g_hc.ua) uasync_set_socket_write(g_hc.ua, cli->socket_id, 0); } if (cli->send_offset >= cli->send_len) { cli->send_len = 0; cli->send_offset = 0; if (g_hc.ua) uasync_set_socket_write(g_hc.ua, cli->socket_id, 0); }
@ -195,7 +191,7 @@ static void hc_close_client(struct headless_client* cli) {
if (!cli || cli->closing) return; if (!cli || cli->closing) return;
cli->closing = 1; cli->closing = 1;
if (g_hc.ua && cli->socket_id) uasync_remove_socket_t(g_hc.ua, cli->fd); if (g_hc.ua && cli->socket_id) uasync_remove_socket_t(g_hc.ua, cli->fd);
if (cli->fd >= 0) { close(cli->fd); cli->fd = -1; } if (cli->fd != SOCKET_INVALID) { socket_close_wrapper(cli->fd); cli->fd = SOCKET_INVALID; }
} }
/* ── Read callback ── */ /* ── Read callback ── */
@ -204,7 +200,7 @@ static void client_read_callback(socket_t fd, void* arg) {
struct headless_client* cli = (struct headless_client*)arg; struct headless_client* cli = (struct headless_client*)arg;
if (!cli || cli->closing) return; if (!cli || cli->closing) return;
char tmp[4096]; char tmp[4096];
ssize_t r = read(fd, tmp, sizeof(tmp) - 1); ssize_t r = recv(fd, tmp, (int)sizeof(tmp) - 1, 0);
if (r <= 0) { hc_close_client(cli); return; } if (r <= 0) { hc_close_client(cli); return; }
for (ssize_t i = 0; i < r; i++) { for (ssize_t i = 0; i < r; i++) {
if (tmp[i] == '\n') { if (tmp[i] == '\n') {
@ -455,18 +451,18 @@ static void accept_callback(socket_t fd, void* arg) {
(void)arg; (void)arg;
struct sockaddr_storage addr; socklen_t alen = sizeof(addr); struct sockaddr_storage addr; socklen_t alen = sizeof(addr);
socket_t cfd = accept(fd, (struct sockaddr*)&addr, &alen); socket_t cfd = accept(fd, (struct sockaddr*)&addr, &alen);
if (cfd < 0) { if (errno != EAGAIN && errno != EWOULDBLOCK) DEBUG_ERROR((int)DEBUG_CATEGORY_HEADLESS, "headless: accept failed: %s", strerror(errno)); return; } if (cfd == SOCKET_INVALID) { int e = socket_get_error(); if (e != ERR_AGAIN && e != ERR_WOULDBLOCK) DEBUG_ERROR((int)DEBUG_CATEGORY_HEADLESS, "headless: accept failed: %s", socket_strerror(e)); return; }
int flags = fcntl(cfd, F_GETFL, 0); if (flags >= 0) fcntl(cfd, F_SETFL, flags | O_NONBLOCK); socket_set_nonblocking(cfd);
if (g_hc.client_count >= MAX_CLIENTS) { close(cfd); DEBUG_WARN((int)DEBUG_CATEGORY_HEADLESS, "headless: max clients reached"); return; } if (g_hc.client_count >= MAX_CLIENTS) { socket_close_wrapper(cfd); DEBUG_WARN((int)DEBUG_CATEGORY_HEADLESS, "headless: max clients reached"); return; }
struct headless_client* cli = u_calloc(1, sizeof(*cli)); struct headless_client* cli = u_calloc(1, sizeof(*cli));
if (!cli) { close(cfd); DEBUG_ERROR((int)DEBUG_CATEGORY_HEADLESS, "headless: failed to allocate client"); return; } if (!cli) { socket_close_wrapper(cfd); DEBUG_ERROR((int)DEBUG_CATEGORY_HEADLESS, "headless: failed to allocate client"); return; }
cli->fd = cfd; cli->fd = cfd;
cli->socket_id = uasync_add_socket_t(g_hc.ua, cfd, client_read_callback, client_write_callback, NULL, cli); cli->socket_id = uasync_add_socket_t(g_hc.ua, cfd, client_read_callback, client_write_callback, NULL, cli);
if (!cli->socket_id) { u_free(cli); close(cfd); DEBUG_ERROR((int)DEBUG_CATEGORY_HEADLESS, "headless: failed to register client fd"); return; } if (!cli->socket_id) { u_free(cli); socket_close_wrapper(cfd); DEBUG_ERROR((int)DEBUG_CATEGORY_HEADLESS, "headless: failed to register client fd"); return; }
cli->next = g_hc.clients; g_hc.clients = cli; g_hc.client_count++; cli->next = g_hc.clients; g_hc.clients = cli; g_hc.client_count++;
DEBUG_INFO((int)DEBUG_CATEGORY_HEADLESS, "headless: client connected fd=%d total=%d", (int)cfd, g_hc.client_count); DEBUG_INFO((int)DEBUG_CATEGORY_HEADLESS, "headless: client connected fd=%d total=%d", (int)cfd, g_hc.client_count);
@ -481,7 +477,7 @@ static void hc_cleanup(void* arg) {
if ((*prev)->closing) { if ((*prev)->closing) {
struct headless_client* dead = *prev; *prev = dead->next; struct headless_client* dead = *prev; *prev = dead->next;
if (dead->socket_id && g_hc.ua) uasync_remove_socket_t(g_hc.ua, dead->fd); if (dead->socket_id && g_hc.ua) uasync_remove_socket_t(g_hc.ua, dead->fd);
if (dead->fd >= 0) close(dead->fd); if (dead->fd != SOCKET_INVALID) socket_close_wrapper(dead->fd);
u_free(dead); g_hc.client_count--; u_free(dead); g_hc.client_count--;
} else prev = &(*prev)->next; } else prev = &(*prev)->next;
} }
@ -499,23 +495,23 @@ int chat_headless_control_init(struct UASYNC* ua, struct UTUN_INSTANCE* inst,
g_hc.ua = ua; g_hc.inst = inst; g_hc.ua = ua; g_hc.inst = inst;
g_hc.listen_fd = socket(AF_INET, SOCK_STREAM, 0); g_hc.listen_fd = socket(AF_INET, SOCK_STREAM, 0);
if (g_hc.listen_fd < 0) { DEBUG_ERROR((int)DEBUG_CATEGORY_HEADLESS, "headless: socket() failed: %s", strerror(errno)); return -1; } if (g_hc.listen_fd == SOCKET_INVALID) { DEBUG_ERROR((int)DEBUG_CATEGORY_HEADLESS, "headless: socket() failed: %s", socket_strerror(socket_get_error())); return -1; }
int reuse = 1; setsockopt(g_hc.listen_fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)); socket_set_reuseaddr(g_hc.listen_fd, 1);
int flags = fcntl(g_hc.listen_fd, F_GETFL, 0); if (flags >= 0) fcntl(g_hc.listen_fd, F_SETFL, flags | O_NONBLOCK); socket_set_nonblocking(g_hc.listen_fd);
struct sockaddr_in sin; memset(&sin, 0, sizeof(sin)); struct sockaddr_in sin; memset(&sin, 0, sizeof(sin));
sin.sin_family = AF_INET; sin.sin_port = htons((uint16_t)port); sin.sin_family = AF_INET; sin.sin_port = htons((uint16_t)port);
if (inet_pton(AF_INET, bind_ip, &sin.sin_addr) != 1) { close(g_hc.listen_fd); g_hc.listen_fd = -1; return -1; } if (inet_pton(AF_INET, bind_ip, &sin.sin_addr) != 1) { socket_close_wrapper(g_hc.listen_fd); g_hc.listen_fd = SOCKET_INVALID; return -1; }
if (bind(g_hc.listen_fd, (struct sockaddr*)&sin, sizeof(sin)) < 0) { if (bind(g_hc.listen_fd, (struct sockaddr*)&sin, sizeof(sin)) < 0) {
DEBUG_ERROR((int)DEBUG_CATEGORY_HEADLESS, "headless: bind(%s:%d) failed: %s", bind_ip, port, strerror(errno)); DEBUG_ERROR((int)DEBUG_CATEGORY_HEADLESS, "headless: bind(%s:%d) failed: %s", bind_ip, port, socket_strerror(socket_get_error()));
close(g_hc.listen_fd); g_hc.listen_fd = -1; return -1; socket_close_wrapper(g_hc.listen_fd); g_hc.listen_fd = SOCKET_INVALID; return -1;
} }
if (listen(g_hc.listen_fd, 5) < 0) { close(g_hc.listen_fd); g_hc.listen_fd = -1; return -1; } if (listen(g_hc.listen_fd, 5) < 0) { socket_close_wrapper(g_hc.listen_fd); g_hc.listen_fd = SOCKET_INVALID; return -1; }
g_hc.listen_sock_id = uasync_add_socket_t(ua, g_hc.listen_fd, accept_callback, NULL, NULL, NULL); g_hc.listen_sock_id = uasync_add_socket_t(ua, g_hc.listen_fd, accept_callback, NULL, NULL, NULL);
if (!g_hc.listen_sock_id) { close(g_hc.listen_fd); g_hc.listen_fd = -1; return -1; } if (!g_hc.listen_sock_id) { socket_close_wrapper(g_hc.listen_fd); g_hc.listen_fd = SOCKET_INVALID; return -1; }
chat_event_set_handler(hc_on_chat_event); chat_event_set_handler(hc_on_chat_event);
uasync_set_timeout(ua, 5000, NULL, hc_cleanup, "hc_cleanup"); uasync_set_timeout(ua, 5000, NULL, hc_cleanup, "hc_cleanup");
@ -530,9 +526,9 @@ void chat_headless_control_destroy(void) {
g_hc.running = 0; g_hc.running = 0;
chat_event_set_handler(NULL); chat_event_set_handler(NULL);
if (g_hc.listen_sock_id && g_hc.ua) uasync_remove_socket_t(g_hc.ua, g_hc.listen_fd); if (g_hc.listen_sock_id && g_hc.ua) uasync_remove_socket_t(g_hc.ua, g_hc.listen_fd);
if (g_hc.listen_fd >= 0) { close(g_hc.listen_fd); g_hc.listen_fd = -1; } if (g_hc.listen_fd != SOCKET_INVALID) { socket_close_wrapper(g_hc.listen_fd); g_hc.listen_fd = SOCKET_INVALID; }
struct headless_client* c = g_hc.clients; struct headless_client* c = g_hc.clients;
while (c) { struct headless_client* n = c->next; if (c->fd >= 0) close(c->fd); u_free(c); c = n; } while (c) { struct headless_client* n = c->next; if (c->fd != SOCKET_INVALID) socket_close_wrapper(c->fd); u_free(c); c = n; }
g_hc.clients = NULL; g_hc.client_count = 0; g_hc.clients = NULL; g_hc.client_count = 0;
DEBUG_INFO((int)DEBUG_CATEGORY_HEADLESS, "headless: destroyed"); DEBUG_INFO((int)DEBUG_CATEGORY_HEADLESS, "headless: destroyed");
} }

Loading…
Cancel
Save