summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/curl/curl.c134
-rw-r--r--src/curl/curl.h23
-rw-r--r--src/curl/http.c226
-rw-r--r--src/curl/http.h (renamed from src/http.h)52
-rw-r--r--src/curl/websocket.c81
-rw-r--r--src/curl/websocket.h17
-rw-r--r--src/http.c256
-rw-r--r--src/line_processor.c48
-rw-r--r--src/line_processor.h18
-rw-r--r--src/packet_stream.c10
-rw-r--r--src/packet_stream.h2
-rw-r--r--src/rpc.c1
-rw-r--r--src/socket/socket.h22
-rw-r--r--src/socket/socket_linux.c181
-rw-r--r--src/socket/socket_windows.c28
-rw-r--r--src/timer.c26
-rw-r--r--src/timer.h7
-rw-r--r--src/util/buffer.c19
-rw-r--r--src/util/buffer.h5
-rw-r--r--src/util/error.h2
-rw-r--r--src/util/file/file_linux.c2
-rw-r--r--src/util/packet.c4
22 files changed, 721 insertions, 443 deletions
diff --git a/src/curl/curl.c b/src/curl/curl.c
new file mode 100644
index 0000000..3f4cb9e
--- /dev/null
+++ b/src/curl/curl.c
@@ -0,0 +1,134 @@
+#include <al/log.h>
+
+#include "curl.h"
+
+static void curl_socket_action_callback(struct ev_loop *loop, ev_io *w, s32 revents)
+{
+ (void)loop;
+ struct aki_curl *curl = (struct aki_curl *)w->data;
+ s32 action = ((revents & EV_READ) ? CURL_CSELECT_IN : 0) | ((revents & EV_WRITE) ? CURL_CSELECT_OUT : 0);
+ s32 running;
+ CURLMcode mc = curl_multi_socket_action(curl->multi_handle, curl->sock, action, &running);
+ if (mc != CURLM_OK) {
+ al_log_error("curl", "curl_multi_socket_action() failed (%s).", curl_multi_strerror(mc));
+ return;
+ }
+ curl->handle_events(curl);
+}
+
+static void set_sock(struct aki_curl *curl, s32 what)
+{
+ s32 action = ((what & CURL_POLL_IN) ? EV_READ : 0) | ((what & CURL_POLL_OUT) ? EV_WRITE : 0);
+ if (curl->event.active) ev_io_stop(curl->loop->ev, &curl->event);
+ ev_io_init(&curl->event, curl_socket_action_callback, curl->sock, action);
+ ev_io_start(curl->loop->ev, &curl->event);
+}
+
+static s32 sock_callback(CURL *e, curl_socket_t s, s32 what, void *cbp, void *sockp)
+{
+ (void)e;
+ struct aki_curl *curl = (struct aki_curl *)cbp;
+ if (what == CURL_POLL_REMOVE) {
+ ev_io_stop(curl->loop->ev, &curl->event);
+ curl_multi_assign(curl->multi_handle, curl->sock, NULL);
+ curl->sock = 0;
+ } else {
+ if (!sockp) {
+ curl->sock = s;
+ curl_multi_assign(curl->multi_handle, curl->sock, curl);
+ set_sock(curl, what);
+ } else {
+ set_sock(curl, what);
+ }
+ }
+ return 0;
+}
+
+static void timeout_callback(struct ev_loop *loop, ev_timer *w, s32 revents)
+{
+ (void)loop;
+ struct aki_curl *curl = (struct aki_curl *)w->data;
+ (void)revents;
+ ev_timer_stop(curl->loop->ev, w);
+ curl->timer_started = false;
+ s32 running;
+ curl_multi_socket_action(curl->multi_handle, CURL_SOCKET_TIMEOUT, 0, &running);
+ curl->handle_events(curl);
+}
+
+static s32 timer_callback(CURLM *multi, s64 timeout_ms, void *userp)
+{
+ (void)multi;
+ struct aki_curl *curl = (struct aki_curl *)userp;
+ if (timeout_ms == -1) {
+ ev_timer_stop(curl->loop->ev, &curl->timer);
+ return 0;
+ }
+ if (curl->timer_started) {
+ ev_timer_stop(curl->loop->ev, &curl->timer);
+ }
+ curl->timer.data = curl;
+ ev_timer_init(&curl->timer, timeout_callback, timeout_ms / 1000.0, 0.0);
+ curl->timer_started = true;
+ ev_timer_start(curl->loop->ev, &curl->timer);
+ return 0;
+}
+
+bool aki_curl_init(struct aki_curl *curl)
+{
+ curl->handle = curl_easy_init();
+ if (!curl->handle) goto err;
+#ifdef _DEBUG_
+ //curl_easy_setopt(curl->handle, CURLOPT_VERBOSE, 1L);
+#endif
+ curl_easy_setopt(curl->handle, CURLOPT_NOPROGRESS, 1L);
+ curl->multi_handle = curl_multi_init();
+ if (!curl->multi_handle) goto err;
+ curl_multi_setopt(curl->multi_handle, CURLMOPT_SOCKETFUNCTION, sock_callback);
+ curl_multi_setopt(curl->multi_handle, CURLMOPT_SOCKETDATA, curl);
+ curl_multi_setopt(curl->multi_handle, CURLMOPT_TIMERFUNCTION, timer_callback);
+ curl_multi_setopt(curl->multi_handle, CURLMOPT_TIMERDATA, curl);
+ curl->timer_started = false;
+ curl->loop = NULL;
+ curl->event.data = curl;
+ curl->event.active = 0;
+ return true;
+err:
+ aki_curl_close(curl);
+ return false;
+}
+
+void aki_curl_set_url(struct aki_curl *curl, str *url)
+{
+ char *c_str = al_str_to_c_str(url);
+ curl_easy_setopt(curl->handle, CURLOPT_URL, c_str);
+ al_free(c_str);
+}
+
+bool aki_curl_add_handle(struct aki_curl *curl)
+{
+ // CURLM_RECURSIVE_API_CALL (8) also means we can't even call this
+ // while a callback is running on another thread.
+ CURLMcode mc = curl_multi_add_handle(curl->multi_handle, curl->handle);
+ if (mc != CURLM_OK) {
+ al_log_error("curl", "curl_multi_add_handle() failed (%s).", curl_multi_strerror(mc));
+ return false;
+ }
+ return true;
+}
+
+bool aki_curl_remove_handle(struct aki_curl *curl)
+{
+ CURLMcode mc = curl_multi_remove_handle(curl->multi_handle, curl->handle);
+ if (mc != CURLM_OK) {
+ al_log_error("curl", "curl_multi_remove_handle() failed (%s).", curl_multi_strerror(mc));
+ return false;
+ }
+ return true;
+}
+
+void aki_curl_close(struct aki_curl *curl)
+{
+ if (curl->handle) curl_easy_cleanup(curl->handle);
+ if (curl->multi_handle) curl_multi_cleanup(curl->multi_handle);
+}
diff --git a/src/curl/curl.h b/src/curl/curl.h
new file mode 100644
index 0000000..1850dcf
--- /dev/null
+++ b/src/curl/curl.h
@@ -0,0 +1,23 @@
+#pragma once
+
+#include <curl/curl.h>
+#include <al/str.h>
+
+#include "../loop.h"
+
+struct aki_curl {
+ ev_io event;
+ ev_timer timer;
+ bool timer_started;
+ struct aki_event_loop *loop;
+ CURL *handle;
+ CURLM *multi_handle;
+ curl_socket_t sock;
+ void (*handle_events)(struct aki_curl *);
+};
+
+bool aki_curl_init(struct aki_curl *curl);
+void aki_curl_set_url(struct aki_curl *curl, str *url);
+bool aki_curl_add_handle(struct aki_curl *curl);
+bool aki_curl_remove_handle(struct aki_curl *curl);
+void aki_curl_close(struct aki_curl *curl);
diff --git a/src/curl/http.c b/src/curl/http.c
new file mode 100644
index 0000000..ecdec2d
--- /dev/null
+++ b/src/curl/http.c
@@ -0,0 +1,226 @@
+#include <al/log.h>
+
+#include "http.h"
+
+static void check_status_codes(struct aki_curl *curl)
+{
+ struct aki_http *http = (struct aki_http *)curl;
+ if (http->response_code <= 0) {
+ curl_easy_getinfo(curl->handle, CURLINFO_RESPONSE_CODE, &http->response_code);
+ if (http->response_code > 0) {
+ http->callback(http->userdata, AKI_HTTP_RESPONSE_CODE, NULL, http->response_code);
+ }
+ }
+ if (http->response_code > 0 && http->content_length < 0) {
+ curl_easy_getinfo(curl->handle, CURLINFO_CONTENT_LENGTH_DOWNLOAD_T, &http->content_length);
+ if (http->content_length >= 0) {
+ http->callback(http->userdata, AKI_HTTP_CONTENT_LENGTH, NULL, (s64)http->content_length);
+ }
+ }
+}
+
+static void http_handle_events(struct aki_curl *curl)
+{
+ s32 pending;
+ CURLMsg *msg;
+ struct aki_http *http = (struct aki_http *)curl;
+ while ((msg = curl_multi_info_read(curl->multi_handle, &pending))) {
+ switch (msg->msg) {
+ case CURLMSG_DONE:
+ al_assert(msg->easy_handle == curl->handle);
+ CURLcode code = msg->data.result;
+ if (code != CURLE_OK) {
+ al_log_error("http", "Curl error: %s.", curl_easy_strerror(code));
+ return;
+ }
+ if (!aki_curl_remove_handle(curl)) return;
+ check_status_codes(curl);
+ if (http->response_code == 200) {
+ http->callback(http->userdata, AKI_HTTP_FINISHED, NULL, http->response_code);
+ } else if (http->response_code / 100 == 3) {
+ char *redirect_url = NULL;
+ curl_easy_getinfo(curl->handle, CURLINFO_REDIRECT_URL, &redirect_url);
+ if (!redirect_url) {
+ http->callback(http->userdata, AKI_HTTP_ERROR, NULL, http->response_code);
+ return;
+ }
+ aki_curl_set_url(curl, al_str_c(redirect_url));
+ http->response_code = -1;
+ http->content_length = -1;
+ if (!aki_curl_add_handle(curl)) return;
+ http->callback(http->userdata, AKI_HTTP_REDIRECT, NULL, http->response_code);
+ } else {
+ http->callback(http->userdata, AKI_HTTP_ERROR, NULL, http->response_code);
+ }
+ return;
+ case CURLMSG_LAST:
+ al_log_debug("http", "Unhandled CURLMSG_LAST.");
+ break;
+ case CURLMSG_NONE:
+ al_log_debug("http", "Unhandled CURLMSG_NONE.");
+ break;
+ }
+ }
+}
+
+bool aki_http_init(struct aki_http *http)
+{
+ http->headers = NULL;
+ http->response_code = -1;
+ http->content_length = -1;
+ return aki_curl_init(&http->curl);
+}
+
+void aki_http_set_url(struct aki_http *http, str *url)
+{
+ aki_curl_set_url(&http->curl, url);
+}
+
+void aki_http_set_user_agent(struct aki_http *http, str *user_agent)
+{
+ char *c_str = al_str_to_c_str(user_agent);
+ curl_easy_setopt(http->curl.handle, CURLOPT_USERAGENT, c_str);
+ al_free(c_str);
+}
+
+void aki_http_add_header(struct aki_http *http, str *header)
+{
+ char *c_str = al_str_to_c_str(header);
+ http->headers = curl_slist_append(http->headers, c_str);
+ al_free(c_str);
+}
+
+void aki_http_set_range(struct aki_http *http, ssize_t start, ssize_t end)
+{
+ char range[64];
+ if (start < 0) al_sprintf(range, "-%zd", end);
+ else if (end < 0) al_sprintf(range, "%zd-", start);
+ else al_sprintf(range, "%zd-%zd", start, end);
+ curl_easy_setopt(http->curl.handle, CURLOPT_RANGE, range);
+}
+
+void aki_http_close(struct aki_http *http)
+{
+ aki_curl_close(&http->curl);
+}
+
+static size_t stream_write_callback(char *buffer, size_t size, size_t nmemb, void *userdata)
+{
+ struct aki_http *http = (struct aki_http *)userdata;
+ check_status_codes(&http->curl);
+ return http->callback(http->userdata, AKI_HTTP_WRITE, (u8 *)buffer, size * nmemb);
+}
+
+static size_t stream_read_callback(char *buffer, size_t size, size_t nmemb, void *userdata)
+{
+ struct aki_http *http = (struct aki_http *)userdata;
+ return http->callback(http->userdata, AKI_HTTP_READ, (u8 *)buffer, size * nmemb);
+}
+
+static size_t request_callback(void *userdata, u8 op, u8 *buf, s64 int0)
+{
+ struct aki_http_request *request = (struct aki_http_request *)userdata;
+ switch (op) {
+ case AKI_HTTP_FINISHED: {
+ request->callback(request->userdata, request, true);
+ break;
+ }
+ case AKI_HTTP_ERROR: {
+ request->callback(request->userdata, request, false);
+ break;
+ }
+ case AKI_HTTP_RESPONSE_CODE: {
+ break;
+ }
+ case AKI_HTTP_CONTENT_LENGTH: {
+ aki_buffer_ensure_space(&request->response, int0);
+ break;
+ }
+ case AKI_HTTP_WRITE: {
+ aki_buffer_append(&request->response, buf, int0);
+ break;
+ }
+ case AKI_HTTP_READ: {
+ al_assert(request->pointer >= 0);
+ size_t size = aki_buffer_get_size(&request->payload);
+ if (request->pointer + int0 > (off_t)size) {
+ int0 = size - request->pointer;
+ }
+ aki_buffer_read(&request->payload, buf, request->pointer, int0);
+ request->pointer += int0;
+ break;
+ }
+ }
+ return int0;
+}
+
+static bool init_request_internal(struct aki_http *http, u8 method)
+{
+ struct aki_curl *curl = &http->curl;
+ switch (method) {
+ case AKI_HTTP_HEAD:
+ curl_easy_setopt(curl->handle, CURLOPT_NOBODY, 1L);
+ break;
+ case AKI_HTTP_GET:
+ curl_easy_setopt(curl->handle, CURLOPT_HTTPGET, 1L);
+ break;
+ case AKI_HTTP_POST:
+ curl_easy_setopt(curl->handle, CURLOPT_POST, 1L);
+ break;
+ }
+ curl_easy_setopt(curl->handle, CURLOPT_WRITEFUNCTION, stream_write_callback);
+ curl_easy_setopt(curl->handle, CURLOPT_WRITEDATA, http);
+ curl_easy_setopt(curl->handle, CURLOPT_READFUNCTION, stream_read_callback);
+ curl_easy_setopt(curl->handle, CURLOPT_READDATA, http);
+ if (http->headers) {
+ curl_easy_setopt(curl->handle, CURLOPT_HTTPHEADER, http->headers);
+ }
+ curl->handle_events = http_handle_events;
+ return aki_curl_add_handle(curl);
+}
+
+bool aki_http_request_stream(struct aki_http *http, u8 method, struct aki_event_loop *loop,
+ size_t (*callback)(void *, u8, u8 *, s64), void *userdata)
+{
+ struct aki_curl *curl = &http->curl;
+ al_assert(curl->loop == NULL);
+ al_assert(method != AKI_HTTP_POST); // Not implemented.
+ curl->loop = loop;
+ http->callback = callback;
+ http->userdata = userdata;
+ return init_request_internal(http, method);
+}
+
+bool aki_http_request_init(struct aki_http_request *request)
+{
+ request->pointer = 0;
+ aki_buffer_init(&request->payload);
+ aki_buffer_init(&request->response);
+ return aki_http_init(&request->http);
+}
+
+bool aki_http_request(struct aki_http_request *request, u8 method, struct aki_event_loop *loop,
+ void (*callback)(void *, struct aki_http_request *, bool), void *userdata)
+{
+ struct aki_curl *curl = &request->http.curl;
+ al_assert(curl->loop == NULL);
+ curl->loop = loop;
+ request->http.callback = request_callback;
+ request->http.userdata = request;
+ size_t payload = aki_buffer_get_size(&request->payload);
+ if (payload > 0) {
+ curl_easy_setopt(curl->handle, CURLOPT_POSTFIELDSIZE, (s64)payload);
+ } else {
+ al_assert(method != AKI_HTTP_POST);
+ }
+ request->callback = callback;
+ request->userdata = userdata;
+ return init_request_internal(&request->http, method);
+}
+
+void aki_http_request_close(struct aki_http_request *request)
+{
+ aki_buffer_free(&request->payload);
+ aki_buffer_free(&request->response);
+ aki_http_close(&request->http);
+}
diff --git a/src/http.h b/src/curl/http.h
index 802a7f2..28942b9 100644
--- a/src/http.h
+++ b/src/curl/http.h
@@ -1,31 +1,17 @@
#pragma once
-#include <curl/curl.h>
#include <al/str.h>
-#include "util/buffer.h"
+#include "../util/buffer.h"
-#include "loop.h"
+#include "../loop.h"
+
+#include "curl.h"
enum {
AKI_HTTP_HEAD = 0,
AKI_HTTP_GET,
- AKI_HTTP_POST,
-};
-
-struct aki_http {
- ev_io event;
- ev_timer timer;
- bool timer_started;
- struct aki_event_loop *loop;
- CURL *handle;
- CURLM *multi_handle;
- char *url;
- struct curl_slist *headers;
- s64 response_code;
- curl_off_t content_length;
- curl_socket_t sock;
- void (*handle_events)(struct aki_http *);
+ AKI_HTTP_POST
};
enum {
@@ -38,19 +24,33 @@ enum {
AKI_HTTP_ERROR
};
-struct aki_http_stream {
- struct aki_http http;
+struct aki_http {
+ struct aki_curl curl;
+ s64 response_code;
+ curl_off_t content_length;
+ struct curl_slist *headers;
size_t (*callback)(void *, u8, u8 *, s64);
void *userdata;
};
+struct aki_http_request {
+ struct aki_http http;
+ off_t pointer;
+ struct aki_buffer payload;
+ struct aki_buffer response;
+ void (*callback)(void *, struct aki_http_request *, bool);
+ void *userdata;
+};
+
bool aki_http_init(struct aki_http *http);
-void aki_http_add_header(struct aki_http *http, str *header);
void aki_http_set_url(struct aki_http *http, str *url);
+void aki_http_set_user_agent(struct aki_http *http, str *user_agent);
+void aki_http_add_header(struct aki_http *http, str *header);
void aki_http_set_range(struct aki_http *http, ssize_t start, ssize_t end);
void aki_http_close(struct aki_http *http);
-
-bool aki_http_stream_init(struct aki_http_stream *stream);
-bool aki_http_stream_request(struct aki_http_stream *stream, u8 method, struct aki_event_loop *loop,
+bool aki_http_request_stream(struct aki_http *http, u8 method, struct aki_event_loop *loop,
size_t (*callback)(void *, u8, u8 *, s64), void *userdata);
-void aki_http_stream_close(struct aki_http_stream *stream);
+bool aki_http_request_init(struct aki_http_request *request);
+bool aki_http_request(struct aki_http_request *request, u8 method, struct aki_event_loop *loop,
+ void (*callback)(void *, struct aki_http_request *, bool), void *userdata);
+void aki_http_request_close(struct aki_http_request *request);
diff --git a/src/curl/websocket.c b/src/curl/websocket.c
new file mode 100644
index 0000000..7e1d9b9
--- /dev/null
+++ b/src/curl/websocket.c
@@ -0,0 +1,81 @@
+#include <al/log.h>
+#include <aki/socket.h>
+
+#include "websocket.h"
+
+static void websocket_handle_events(struct aki_curl *curl)
+{
+ s32 pending;
+ CURLMsg *msg;
+ while ((msg = curl_multi_info_read(curl->multi_handle, &pending))) {
+ switch (msg->msg) {
+ case CURLMSG_DONE:
+ aki_curl_remove_handle(curl);
+ aki_curl_close(curl);
+ return;
+ case CURLMSG_LAST:
+ break;
+ case CURLMSG_NONE:
+ break;
+ }
+ }
+}
+
+static size_t stream_write_callback(char *buffer, size_t size, size_t nmemb, void *userdata)
+{
+ (void)size;
+ (void)nmemb;
+ struct aki_websocket *ws = (struct aki_websocket *)userdata;
+ const struct curl_ws_frame *m = curl_ws_meta(ws->curl.handle);
+ size_t frame_size = m->offset + m->len + m->bytesleft;
+ al_log_debug("ws", "frame_size: %zu, bytesleft: %zu", frame_size, m->bytesleft);
+ aki_buffer_ensure_space(&ws->frame, frame_size);
+ al_memcpy(aki_buffer_get_ptr(&ws->frame, m->offset), buffer, m->len);
+ if (m->bytesleft == 0) {
+ ws->callback(ws->userdata, m, aki_buffer_get_ptr(&ws->frame, 0), frame_size);
+ }
+ return m->len;
+}
+
+bool aki_websocket_init(struct aki_websocket *ws)
+{
+ aki_buffer_init(&ws->frame);
+ return aki_curl_init(&ws->curl);
+}
+
+void aki_websocket_set_url(struct aki_websocket *ws, str *url)
+{
+ aki_curl_set_url(&ws->curl, url);
+}
+
+bool aki_websocket_connect(struct aki_websocket *ws, struct aki_event_loop *loop,
+ void (*callback)(void *, const struct curl_ws_frame *, u8 *, size_t), void *userdata)
+{
+ al_assert(ws->curl.loop == NULL);
+ ws->curl.loop = loop;
+ ws->callback = callback;
+ ws->userdata = userdata;
+ curl_easy_setopt(ws->curl.handle, CURLOPT_WRITEFUNCTION, stream_write_callback);
+ curl_easy_setopt(ws->curl.handle, CURLOPT_WRITEDATA, ws);
+ ws->curl.handle_events = websocket_handle_events;
+ return aki_curl_add_handle(&ws->curl);
+}
+
+void aki_websocket_disconnect(struct aki_websocket *ws, u16 status)
+{
+ size_t sent;
+ status = aki_htons(status);
+ curl_ws_send(ws->curl.handle, &status, sizeof(u16), &sent, 0, CURLWS_CLOSE);
+ al_assert(sent == sizeof(u16));
+}
+
+//aki_websocket_close();
+//aki_buffer_free(&ws->frame);
+
+bool aki_websocket_send_frame(struct aki_websocket *ws, u8 *data, size_t size)
+{
+ size_t sent;
+ curl_ws_send(ws->curl.handle, data, size, &sent, 0, CURLWS_TEXT);
+ al_assert(sent == size);
+ return true;
+}
diff --git a/src/curl/websocket.h b/src/curl/websocket.h
new file mode 100644
index 0000000..3dc740a
--- /dev/null
+++ b/src/curl/websocket.h
@@ -0,0 +1,17 @@
+#pragma once
+
+#include "http.h"
+
+struct aki_websocket {
+ struct aki_curl curl;
+ struct aki_buffer frame;
+ void (*callback)(void *, const struct curl_ws_frame *, u8 *, size_t);
+ void *userdata;
+};
+
+bool aki_websocket_init(struct aki_websocket *ws);
+void aki_websocket_set_url(struct aki_websocket *ws, str *url);
+bool aki_websocket_connect(struct aki_websocket *ws, struct aki_event_loop *loop,
+ void (*callback)(void *, const struct curl_ws_frame *, u8 *, size_t), void *userdata);
+void aki_websocket_disconnect(struct aki_websocket *ws, u16 status);
+bool aki_websocket_send_frame(struct aki_websocket *ws, u8 *data, size_t size);
diff --git a/src/http.c b/src/http.c
deleted file mode 100644
index 0036af2..0000000
--- a/src/http.c
+++ /dev/null
@@ -1,256 +0,0 @@
-#include <al/log.h>
-
-#include "http.h"
-
-#define USER_AGENT "Mozilla/5.0 (X11; Linux x86_64; rv:96.0) Gecko/20100101 Firefox/96.0"
-
-static void curl_socket_action_callback(struct ev_loop *loop, ev_io *w, s32 revents)
-{
- (void)loop;
- struct aki_http *http = (struct aki_http *)w->data;
-
- s32 action = ((revents & EV_READ) ? CURL_CSELECT_IN : 0) | ((revents & EV_WRITE) ? CURL_CSELECT_OUT : 0);
-
- s32 running;
- CURLMcode mc = curl_multi_socket_action(http->multi_handle, http->sock, action, &running);
- if (mc != CURLM_OK) {
- al_log_error("http", "curl_multi_socket_action() failed (%s).", curl_multi_strerror(mc));
- return;
- }
-
- http->handle_events(http);
-}
-
-static void set_sock(struct aki_http *http, s32 what)
-{
- http->event.data = http;
-
- s32 action = ((what & CURL_POLL_IN) ? EV_READ : 0) | ((what & CURL_POLL_OUT) ? EV_WRITE : 0);
-
- if (http->event.active) ev_io_stop(http->loop->ev, &http->event);
-
- ev_io_init(&http->event, curl_socket_action_callback, http->sock, action);
- ev_io_start(http->loop->ev, &http->event);
-}
-
-static s32 sock_callback(CURL *e, curl_socket_t s, s32 what, void *cbp, void *sockp)
-{
- (void)e;
- struct aki_http *http = (struct aki_http *)cbp;
- if (what == CURL_POLL_REMOVE) {
- http->sock = 0;
- ev_io_stop(http->loop->ev, &http->event);
- curl_multi_assign(http->multi_handle, http->sock, NULL);
- } else {
- if (!sockp) {
- http->sock = s;
- set_sock(http, what);
- curl_multi_assign(http->multi_handle, http->sock, http);
- } else {
- set_sock(http, what);
- }
- }
- return 0;
-}
-
-static void timeout_callback(struct ev_loop *loop, ev_timer *w, s32 revents)
-{
- (void)loop;
- struct aki_http *http = (struct aki_http *)w->data;
- (void)revents;
- ev_timer_stop(http->loop->ev, w);
- http->timer_started = false;
- s32 running;
- curl_multi_socket_action(http->multi_handle, CURL_SOCKET_TIMEOUT, 0, &running);
- http->handle_events(http);
-}
-
-static s32 timer_callback(CURLM *multi, s64 timeout_ms, void *userp)
-{
- (void)multi;
- struct aki_http *http = (struct aki_http *)userp;
- if (timeout_ms == -1) {
- ev_timer_stop(http->loop->ev, &http->timer);
- return 0;
- }
- if (http->timer_started) {
- ev_timer_stop(http->loop->ev, &http->timer);
- }
- http->timer.data = http;
- ev_timer_init(&http->timer, timeout_callback, timeout_ms / 1000.0, 0.0);
- http->timer_started = true;
- ev_timer_start(http->loop->ev, &http->timer);
- return 0;
-}
-
-bool aki_http_init(struct aki_http *http)
-{
- http->handle = curl_easy_init();
- if (!http->handle) goto err;
- //curl_easy_setopt(http->handle, CURLOPT_VERBOSE, 1L);
- curl_easy_setopt(http->handle, CURLOPT_NOPROGRESS, 1L);
- curl_easy_setopt(http->handle, CURLOPT_USERAGENT, USER_AGENT);
- http->multi_handle = curl_multi_init();
- if (!http->multi_handle) goto err;
- curl_multi_setopt(http->multi_handle, CURLMOPT_SOCKETFUNCTION, sock_callback);
- curl_multi_setopt(http->multi_handle, CURLMOPT_SOCKETDATA, http);
- curl_multi_setopt(http->multi_handle, CURLMOPT_TIMERFUNCTION, timer_callback);
- curl_multi_setopt(http->multi_handle, CURLMOPT_TIMERDATA, http);
- http->url = NULL;
- http->headers = NULL;
- http->timer_started = false;
- http->response_code = -1;
- return true;
-err:
- if (http->handle) curl_easy_cleanup(http->handle);
- if (http->multi_handle) curl_multi_cleanup(http->multi_handle);
- return false;
-}
-
-void aki_http_add_header(struct aki_http *http, str *header)
-{
- char *c_str = al_str_to_c_str(header);
- http->headers = curl_slist_append(http->headers, c_str);
- al_free(c_str);
-}
-
-void aki_http_set_url(struct aki_http *http, str *url)
-{
- if (http->url) al_free(http->url);
- http->url = al_str_to_c_str(url);
- curl_easy_setopt(http->handle, CURLOPT_URL, http->url);
-}
-
-void aki_http_set_range(struct aki_http *http, ssize_t start, ssize_t end)
-{
- char range[64];
- if (start < 0) sprintf(range, "-%zd", end);
- else if (end < 0) sprintf(range, "%zd-", start);
- else sprintf(range, "%zd-%zd", start, end);
- curl_easy_setopt(http->handle, CURLOPT_RANGE, range);
-}
-
-void aki_http_close(struct aki_http *http)
-{
- curl_easy_cleanup(http->handle);
- curl_multi_cleanup(http->multi_handle);
- al_free(http->url);
-}
-
-static void stream_handle_events(struct aki_http *http)
-{
- CURLMsg *msg;
- s32 pending;
- struct aki_http_stream *stream = (struct aki_http_stream *)http;
- while ((msg = curl_multi_info_read(http->multi_handle, &pending))) {
- switch (msg->msg) {
- case CURLMSG_DONE:
- al_assert(msg->easy_handle == http->handle);
- if (msg->data.result != CURLE_OK) {
- al_log_error("http", "Error handling curl_multi event (%s).", curl_easy_strerror(msg->data.result));
- return;
- }
- CURLMcode mc = curl_multi_remove_handle(http->multi_handle, msg->easy_handle);
- if (mc != CURLM_OK) {
- al_log_error("http", "curl_multi_remove_handle() failed (%s).", curl_multi_strerror(mc));
- return;
- }
- if (http->response_code < 0) {
- curl_easy_getinfo(http->handle, CURLINFO_RESPONSE_CODE, &http->response_code);
- }
- if (http->response_code == 200) {
- stream->callback(stream->userdata, AKI_HTTP_FINISHED, NULL, http->response_code);
- } else if (http->response_code / 100 == 3) {
- char *redirect_url = NULL;
- curl_easy_getinfo(http->handle, CURLINFO_REDIRECT_URL, &redirect_url);
- if (!redirect_url) {
- stream->callback(stream->userdata, AKI_HTTP_ERROR, NULL, http->response_code);
- return;
- }
- aki_http_set_url(http, al_str_c(redirect_url));
- mc = curl_multi_add_handle(http->multi_handle, http->handle);
- if (mc != CURLM_OK) {
- al_log_error("http", "curl_multi_add_handle() failed (%s).", curl_multi_strerror(mc));
- return;
- }
- stream->callback(stream->userdata, AKI_HTTP_REDIRECT, NULL, http->response_code);
- } else {
- stream->callback(stream->userdata, AKI_HTTP_ERROR, NULL, http->response_code);
- }
- return;
- case CURLMSG_LAST:
- al_log_warn("http", "Unhandled CURLMSG_LAST.");
- break;
- case CURLMSG_NONE:
- al_log_warn("http", "Unhandled CURLMSG_NONE.");
- break;
- }
- }
-}
-
-bool aki_http_stream_init(struct aki_http_stream *stream)
-{
- al_memset(stream, 0, sizeof(struct aki_http_stream));
- if (!aki_http_init(&stream->http)) return false;
- stream->http.handle_events = stream_handle_events;
- return true;
-}
-
-static size_t stream_write_callback(char *buffer, size_t size, size_t nmemb, void *userdata)
-{
- struct aki_http_stream *stream = (struct aki_http_stream *)userdata;
- if (stream->http.response_code < 0) {
- curl_easy_getinfo(stream->http.handle, CURLINFO_RESPONSE_CODE, &stream->http.response_code);
- if (stream->http.response_code > 0) {
- stream->callback(stream->userdata, AKI_HTTP_RESPONSE_CODE, NULL, stream->http.response_code);
- curl_easy_getinfo(stream->http.handle, CURLINFO_CONTENT_LENGTH_DOWNLOAD_T, &stream->http.content_length);
- if (stream->http.content_length >= 0) {
- stream->callback(stream->userdata, AKI_HTTP_CONTENT_LENGTH, NULL, (s64)stream->http.content_length);
- }
- }
- }
- return stream->callback(stream->userdata, AKI_HTTP_WRITE, (u8 *)buffer, size * nmemb);
-}
-
-static size_t stream_read_callback(char *buffer, size_t size, size_t nmemb, void *userdata)
-{
- struct aki_http_stream *stream = (struct aki_http_stream *)userdata;
- return stream->callback(stream->userdata, AKI_HTTP_READ, (u8 *)buffer, size * nmemb);
-}
-
-bool aki_http_stream_request(struct aki_http_stream *stream, u8 method, struct aki_event_loop *loop,
- size_t (*callback)(void *, u8, u8 *, s64), void *userdata)
-{
- al_assert(stream->http.url != NULL);
- al_assert(stream->http.loop == NULL);
- stream->http.loop = loop;
- stream->callback = callback;
- stream->userdata = userdata;
- switch (method) {
- case AKI_HTTP_HEAD:
- curl_easy_setopt(stream->http.handle, CURLOPT_NOBODY, 1L);
- break;
- case AKI_HTTP_GET:
- curl_easy_setopt(stream->http.handle, CURLOPT_HTTPGET, 1L);
- break;
- case AKI_HTTP_POST:
- al_assert(false); // Not implemented.
- }
- curl_easy_setopt(stream->http.handle, CURLOPT_WRITEFUNCTION, stream_write_callback);
- curl_easy_setopt(stream->http.handle, CURLOPT_WRITEDATA, stream);
- curl_easy_setopt(stream->http.handle, CURLOPT_READFUNCTION, stream_read_callback);
- curl_easy_setopt(stream->http.handle, CURLOPT_READDATA, stream);
- // CURLM_RECURSIVE_API_CALL (8) also means we can't even call this
- // while a callback is running on another thread.
- CURLMcode mc = curl_multi_add_handle(stream->http.multi_handle, stream->http.handle);
- if (mc != CURLM_OK) {
- al_log_error("http", "curl_multi_add_handle() failed (%s).", curl_multi_strerror(mc));
- return false;
- }
- return true;
-}
-
-void aki_http_stream_close(struct aki_http_stream *stream)
-{
- aki_http_close(&stream->http);
-}
diff --git a/src/line_processor.c b/src/line_processor.c
index 444142a..758f99c 100644
--- a/src/line_processor.c
+++ b/src/line_processor.c
@@ -8,22 +8,13 @@
#define CHUNK_SIZE 512
#define END al_str_c("X")
-struct line_processor_client {
- ev_io event;
- struct aki_socket s;
- size_t index;
- struct aki_buffer buf;
- size_t mark;
- struct aki_line_processor *pro;
-};
-
static void read_callback(struct ev_loop *loop, ev_io *w, s32 revents)
{
(void)loop;
- struct line_processor_client *client = (struct line_processor_client *)w->data;
+ struct aki_line_processor_client *client = (struct aki_line_processor_client *)w->data;
(void)revents;
- aki_buffer_ensure_size(&client->buf, client->index + CHUNK_SIZE);
+ aki_buffer_ensure_space(&client->buf, client->index + CHUNK_SIZE);
u8 *ptr = aki_buffer_get_ptr(&client->buf, client->index);
ssize_t ret;
if (client->pro->type == AKI_LINE_PROCESSOR_SOCKET) {
@@ -50,7 +41,13 @@ static void read_callback(struct ev_loop *loop, ev_io *w, s32 revents)
al_free(client);
return;
}
- client->pro->callback(client->pro->userdata, s);
+ switch (client->pro->callback(client->pro->userdata, s)) {
+ case AKI_LINE_PROCESSOR_CONTINUE:
+ break;
+ case AKI_LINE_PROCESSOR_STOP:
+ aki_line_processor_stop(client->pro);
+ return;
+ }
size_t skip = i + offset + 1;
client->index = client->index - skip;
al_memmove(ptr, &ptr[skip], client->index);
@@ -66,18 +63,19 @@ static void socket_connection_callback(struct ev_loop *loop, ev_io *w, s32 reven
(void)loop;
struct aki_line_processor *pro = (struct aki_line_processor *)w->data;
(void)revents;
- struct line_processor_client *client = al_alloc_object(struct line_processor_client);
- client->s.config.blocking = false;
- client->s.config.no_delay = false;
+ struct aki_line_processor_client *client = al_alloc_object(struct aki_line_processor_client);
if (!aki_socket_accept(pro->s, &client->s)) {
al_free(client);
return;
}
+ aki_socket_set_blocking(&client->s, false);
+ aki_socket_set_no_delay(&client->s, 1);
client->index = 0;
aki_buffer_init(&client->buf);
client->mark = 0;
client->pro = pro;
client->event.data = client;
+ al_array_push(pro->clients, client);
ev_io_init(&client->event, read_callback, aki_socket_get_fd(&client->s), EV_READ);
ev_io_start(pro->loop->ev, &client->event);
}
@@ -85,6 +83,7 @@ static void socket_connection_callback(struct ev_loop *loop, ev_io *w, s32 reven
void aki_line_processor_init(struct aki_line_processor *pro, str *delim)
{
al_str_clone(&pro->delim, delim);
+ al_array_init(pro->clients);
}
void aki_line_processor_open_socket(struct aki_line_processor *pro, struct aki_socket *socket)
@@ -109,15 +108,32 @@ void aki_line_processor_run(struct aki_line_processor *pro, struct aki_event_loo
ev_io_start(pro->loop->ev, &pro->event);
break;
case AKI_LINE_PROCESSOR_FD: { // Not supported on windows.
- struct line_processor_client *client = al_alloc_object(struct line_processor_client);
+ struct aki_line_processor_client *client = al_alloc_object(struct aki_line_processor_client);
client->index = 0;
aki_buffer_init(&client->buf);
client->mark = 0;
client->pro = pro;
client->event.data = client;
+ al_array_push(pro->clients, client);
ev_io_init(&client->event, read_callback, pro->fd, EV_READ);
ev_io_start(pro->loop->ev, &client->event);
break;
}
}
}
+
+void aki_line_processor_stop(struct aki_line_processor *pro)
+{
+ switch (pro->type) {
+ case AKI_LINE_PROCESSOR_SOCKET:
+ ev_io_stop(pro->loop->ev, &pro->event);
+ break;
+ case AKI_LINE_PROCESSOR_FD:
+ break;
+ }
+ struct aki_line_processor_client *client;
+ al_array_foreach(pro->clients, i, client) {
+ ev_io_stop(pro->loop->ev, &client->event);
+ }
+ al_array_free(pro->clients);
+}
diff --git a/src/line_processor.h b/src/line_processor.h
index 7ce3ca3..cc391ca 100644
--- a/src/line_processor.h
+++ b/src/line_processor.h
@@ -17,6 +17,20 @@ enum {
AKI_LINE_PROCESSOR_FD
};
+enum {
+ AKI_LINE_PROCESSOR_CONTINUE = 0,
+ AKI_LINE_PROCESSOR_STOP
+};
+
+struct aki_line_processor_client {
+ ev_io event;
+ struct aki_socket s;
+ size_t index;
+ struct aki_buffer buf;
+ size_t mark;
+ struct aki_line_processor *pro;
+};
+
struct aki_line_processor {
struct aki_event_loop *loop;
u8 type;
@@ -24,7 +38,8 @@ struct aki_line_processor {
struct aki_socket *s;
s32 fd;
str delim;
- void (*callback)(void *, str *line);
+ array(struct aki_line_processor_client *) clients;
+ u8 (*callback)(void *, str *line);
void *userdata;
};
@@ -32,3 +47,4 @@ void aki_line_processor_init(struct aki_line_processor *pro, str *delim);
void aki_line_processor_open_socket(struct aki_line_processor *pro, struct aki_socket *socket);
void aki_line_processor_open_fd(struct aki_line_processor *pro, s32 fd);
void aki_line_processor_run(struct aki_line_processor *pro, struct aki_event_loop *loop);
+void aki_line_processor_stop(struct aki_line_processor *pro);
diff --git a/src/packet_stream.c b/src/packet_stream.c
index 8da342f..5c415aa 100644
--- a/src/packet_stream.c
+++ b/src/packet_stream.c
@@ -23,15 +23,14 @@ bool aki_packet_stream_init(struct aki_packet_stream *stream, u8 type,
stream->flushed_callback = NULL;
stream->userdata = userdata;
stream->s.type = type;
- stream->s.config.blocking = false;
- stream->s.config.no_delay = false;
if (!aki_socket_init(&stream->s)) return false;
+ aki_socket_set_blocking(&stream->s, false);
init_internal(stream);
stream->connected = false;
return true;
}
-void aki_packet_stream_set_no_delay(struct aki_packet_stream *stream, bool no_delay)
+void aki_packet_stream_set_no_delay(struct aki_packet_stream *stream, s32 no_delay)
{
aki_socket_set_no_delay(&stream->s, no_delay);
}
@@ -75,7 +74,7 @@ static void stream_read_callback(struct ev_loop *loop, ev_io *w, s32 revents)
if (!stream->in.have_size && stream->in.index >= AKI_PACKET_HEADER_LENGTH) {
stream->in.have_size = true;
size = aki_packet_size(stream->in.packet);
- aki_buffer_ensure_size(&stream->in.packet->buffer, size);
+ aki_buffer_ensure_space(&stream->in.packet->buffer, size);
}
if (stream->in.have_size && stream->in.index >= size) {
struct aki_packet *packet = stream->in.packet;
@@ -138,12 +137,11 @@ static void socket_connection_callback(struct ev_loop *loop, ev_io *w, s32 reven
struct aki_packet_stream *server = (struct aki_packet_stream *)w->data;
(void)revents;
struct aki_packet_stream *client = al_alloc_object(struct aki_packet_stream);
- client->s.config.blocking = false;
- client->s.config.no_delay = false;
if (!aki_socket_accept(&server->s, &client->s)) {
al_free(client);
return;
}
+ aki_socket_set_blocking(&client->s, false);
client->connected = true;
client->loop = server->loop;
init_internal(client);
diff --git a/src/packet_stream.h b/src/packet_stream.h
index 027fac0..4e83639 100644
--- a/src/packet_stream.h
+++ b/src/packet_stream.h
@@ -39,7 +39,7 @@ bool aki_packet_stream_init(struct aki_packet_stream *stream, u8 type,
void (*connection_closed_callback)(void *, struct aki_packet_stream *),
void (*packet_callback)(void *, struct aki_packet_stream *, struct aki_packet *),
void (*packet_sent_callback)(void *, struct aki_packet *), void *userdata);
-void aki_packet_stream_set_no_delay(struct aki_packet_stream *stream, bool no_delay);
+void aki_packet_stream_set_no_delay(struct aki_packet_stream *stream, s32 no_delay);
void aki_packet_stream_listen(struct aki_packet_stream *stream,
struct aki_event_loop *loop, str *addr, s32 port);
void aki_packet_stream_connect(struct aki_packet_stream *stream,
diff --git a/src/rpc.c b/src/rpc.c
index edf73c0..90790e2 100644
--- a/src/rpc.c
+++ b/src/rpc.c
@@ -33,6 +33,7 @@ static void packet_callback(void *userdata, struct aki_packet_stream *s, struct
} else {
aki_packet_free(rpacket);
}
+ break;
}
}
}
diff --git a/src/socket/socket.h b/src/socket/socket.h
index ea307e7..4971c1c 100644
--- a/src/socket/socket.h
+++ b/src/socket/socket.h
@@ -21,10 +21,6 @@ enum {
struct aki_socket {
u8 type;
- struct {
- bool blocking;
- bool no_delay;
- } config;
#ifndef _WIN32
s32 sock;
struct sockaddr_in addr_in;
@@ -37,20 +33,25 @@ struct aki_socket {
#ifndef _WIN32
#define aki_read read
-#else
-#define aki_read _read
-#endif
-
-#ifndef _WIN32
#define aki_write write
+#define aki_htons htons
+#define aki_htonl htonl
+#define aki_ntohs ntohs
#else
+#define aki_read _read
#define aki_write _write
+#define aki_htons(...) 0
+#define aki_htonl(...) 0
+#define aki_ntohs(...) 0
#endif
bool aki_socket_init(struct aki_socket *s);
void aki_socket_set_blocking(struct aki_socket *s, bool blocking);
-void aki_socket_set_no_delay(struct aki_socket *s, bool no_delay);
+void aki_socket_set_no_delay(struct aki_socket *s, s32 no_delay);
+
+void aki_socket_set_send_buf(struct aki_socket *s, u32 sndbuf);
+u32 aki_socket_get_send_buf(struct aki_socket *s);
bool aki_socket_listen(struct aki_socket *s, str *addr, s32 port);
bool aki_socket_accept(struct aki_socket *s, struct aki_socket *c);
@@ -61,6 +62,7 @@ s32 aki_socket_get_fd(struct aki_socket *s);
ssize_t aki_socket_read(struct aki_socket *s, void *buf, size_t size);
ssize_t aki_socket_write(struct aki_socket *s, void *buf, size_t size);
ssize_t aki_socket_sendto(struct aki_socket *s, void *buf, size_t size);
+ssize_t aki_socket_recvfrom(struct aki_socket *s, void *buf, size_t size);
void aki_socket_shutdown(struct aki_socket *s);
void aki_socket_close(struct aki_socket *s);
diff --git a/src/socket/socket_linux.c b/src/socket/socket_linux.c
index b466dea..7bc9ea9 100644
--- a/src/socket/socket_linux.c
+++ b/src/socket/socket_linux.c
@@ -9,48 +9,6 @@
#include "socket.h"
-static void socket_set_blocking(struct aki_socket *s, bool blocking)
-{
- s32 flags = fcntl(s->sock, F_GETFL, 0);
- al_assert(flags != -1);
- flags = blocking ? (flags & ~O_NONBLOCK) : (flags | O_NONBLOCK);
- s32 ret = fcntl(s->sock, F_SETFL, flags);
- al_assert(ret == 0);
-}
-
-static void socket_set_no_delay(struct aki_socket *s, s32 no_delay)
-{
- if (s->type == AKI_SOCKET_TCP) {
- setsockopt(s->sock, IPPROTO_TCP, TCP_NODELAY, &no_delay, sizeof(no_delay));
- }
-}
-
-static void socket_set_send_buf(struct aki_socket *s, u32 sndbuf)
-{
- u32 send_queue_size = sndbuf;
- setsockopt(s->sock, SOL_SOCKET, SO_SNDBUF, &send_queue_size, sizeof(send_queue_size));
-
- socklen_t optlen = sizeof(send_queue_size);
- getsockopt(s->sock, SOL_SOCKET, SO_SNDBUF, &send_queue_size, &optlen);
-
- if (send_queue_size != sndbuf * 2) {
- al_log_warn("socket", "Failed to set SO_SNDBUF.");
- }
-}
-
-static void socket_set_recv_buf(struct aki_socket *s, u32 rcvbuf)
-{
- u32 receive_queue_size = rcvbuf;
- setsockopt(s->sock, SOL_SOCKET, SO_RCVBUF, &receive_queue_size, sizeof(receive_queue_size));
-
- socklen_t optlen = sizeof(receive_queue_size);
- getsockopt(s->sock, SOL_SOCKET, SO_RCVBUF, &receive_queue_size, &optlen);
-
- if (receive_queue_size != rcvbuf * 2) {
- al_log_warn("socket", "Failed to set SO_RCVBUF.");
- }
-}
-
bool aki_socket_init(struct aki_socket *s)
{
switch (s->type) {
@@ -70,63 +28,52 @@ bool aki_socket_init(struct aki_socket *s)
return false;
}
- aki_socket_set_blocking(s, s->config.blocking);
- aki_socket_set_no_delay(s, s->config.no_delay);
-
return true;
}
void aki_socket_set_blocking(struct aki_socket *s, bool blocking)
{
- s->config.blocking = blocking;
- socket_set_blocking(s, blocking);
+ s32 flags = fcntl(s->sock, F_GETFL, 0);
+ al_assert(flags != -1);
+ flags = blocking ? (flags & ~O_NONBLOCK) : (flags | O_NONBLOCK);
+ s32 ret = fcntl(s->sock, F_SETFL, flags);
+ al_assert(ret == 0);
}
-void aki_socket_set_no_delay(struct aki_socket *s, bool no_delay)
+void aki_socket_set_no_delay(struct aki_socket *s, s32 no_delay)
{
- s->config.no_delay = no_delay;
- socket_set_no_delay(s, no_delay);
+ if (s->type != AKI_SOCKET_TCP) return;
+ setsockopt(s->sock, IPPROTO_TCP, TCP_NODELAY, &no_delay, sizeof(no_delay));
}
-bool aki_socket_accept(struct aki_socket *s, struct aki_socket *c)
+u32 aki_socket_get_send_buf(struct aki_socket *s)
{
- c->type = s->type;
-
- struct sockaddr_storage addr;
- socklen_t len = sizeof(addr);
- if((c->sock = accept(s->sock, (struct sockaddr *)&addr, &len)) == -1) {
- return false;
- }
+ u32 send_queue_size;
+ socklen_t optlen = sizeof(send_queue_size);
+ getsockopt(s->sock, SOL_SOCKET, SO_SNDBUF, &send_queue_size, &optlen);
+ return send_queue_size / 2;
+}
- aki_socket_set_blocking(c, c->config.blocking);
- aki_socket_set_no_delay(c, c->config.no_delay);
+void aki_socket_set_send_buf(struct aki_socket *s, u32 sndbuf)
+{
+ u32 send_queue_size = sndbuf;
+ setsockopt(s->sock, SOL_SOCKET, SO_SNDBUF, &send_queue_size, sizeof(send_queue_size));
+ al_assert(aki_socket_get_send_buf(s) == send_queue_size);
+}
- switch (s->type) {
- case AKI_SOCKET_TCP: {
- s32 port = 0;
- char ipstr[INET6_ADDRSTRLEN] = {0};
- if (addr.ss_family == AF_INET) {
- struct sockaddr_in *s = (struct sockaddr_in *)&addr;
- port = ntohs(s->sin_port);
- inet_ntop(AF_INET, &s->sin_addr, ipstr, sizeof(ipstr));
- } else if (addr.ss_family == AF_INET6) {
- struct sockaddr_in6 *s = (struct sockaddr_in6 *)&addr;
- port = ntohs(s->sin6_port);
- inet_ntop(AF_INET6, &s->sin6_addr, ipstr, sizeof(ipstr));
- }
- al_log_info("socket", "Connection from %s:%d.", ipstr, port);
- break;
- }
- case AKI_SOCKET_UNIX: {
- al_log_info("socket", "Connection from a UNIX socket.");
- break;
- }
- case AKI_SOCKET_UDP: {
- al_assert(false);
- }
- }
+u32 aki_socket_get_recv_buf(struct aki_socket *s)
+{
+ u32 receive_queue_size;
+ socklen_t optlen = sizeof(receive_queue_size);
+ getsockopt(s->sock, SOL_SOCKET, SO_RCVBUF, &receive_queue_size, &optlen);
+ return receive_queue_size / 2;
+}
- return true;
+void aki_socket_set_recv_buf(struct aki_socket *s, u32 rcvbuf)
+{
+ u32 receive_queue_size = rcvbuf;
+ setsockopt(s->sock, SOL_SOCKET, SO_RCVBUF, &receive_queue_size, sizeof(receive_queue_size));
+ al_assert(aki_socket_get_recv_buf(s) == receive_queue_size);
}
bool aki_socket_listen(struct aki_socket *s, str *addr, s32 port)
@@ -154,10 +101,9 @@ bool aki_socket_listen(struct aki_socket *s, str *addr, s32 port)
addrlen = sizeof(s->addr_un);
break;
}
- case AKI_SOCKET_UDP: {
+ case AKI_SOCKET_UDP:
al_assert(false);
}
- }
if (bind(s->sock, saddr, addrlen) < 0) {
al_log_error("socket", "Failed to bind (%.*s:%d).", AL_STR_PRINTF(addr), port);
@@ -176,10 +122,47 @@ bool aki_socket_listen(struct aki_socket *s, str *addr, s32 port)
return true;
}
-bool aki_socket_connect(struct aki_socket *s, str *addr, s32 port)
+bool aki_socket_accept(struct aki_socket *s, struct aki_socket *c)
{
+ c->type = s->type;
+
+ struct sockaddr_storage addr;
+ socklen_t len = sizeof(addr);
+ if ((c->sock = accept(s->sock, (struct sockaddr *)&addr, &len)) == -1) {
+ return false;
+ }
+
switch (s->type) {
case AKI_SOCKET_TCP: {
+ s32 port = 0;
+ char ipstr[INET6_ADDRSTRLEN] = {0};
+ if (addr.ss_family == AF_INET) {
+ struct sockaddr_in *s = (struct sockaddr_in *)&addr;
+ port = ntohs(s->sin_port);
+ inet_ntop(AF_INET, &s->sin_addr, ipstr, sizeof(ipstr));
+ } else if (addr.ss_family == AF_INET6) {
+ struct sockaddr_in6 *s = (struct sockaddr_in6 *)&addr;
+ port = ntohs(s->sin6_port);
+ inet_ntop(AF_INET6, &s->sin6_addr, ipstr, sizeof(ipstr));
+ }
+ al_log_info("socket", "Connection from %s:%d.", ipstr, port);
+ break;
+ }
+ case AKI_SOCKET_UNIX:
+ al_log_info("socket", "Connection from a UNIX socket.");
+ break;
+ case AKI_SOCKET_UDP:
+ al_assert(false);
+ }
+
+ return true;
+}
+
+bool aki_socket_connect(struct aki_socket *s, str *addr, s32 port)
+{
+ switch (s->type) {
+ case AKI_SOCKET_TCP:
+ case AKI_SOCKET_UDP: {
char *c_str = al_str_to_c_str(addr);
struct hostent *hptr = gethostbyname(c_str);
al_free(c_str);
@@ -198,13 +181,20 @@ bool aki_socket_connect(struct aki_socket *s, str *addr, s32 port)
s->addr_in.sin_addr.s_addr = ((struct in_addr *)hptr->h_addr_list[0])->s_addr;
s->addr_in.sin_port = htons(port);
- s32 ret = connect(s->sock, (struct sockaddr *)&s->addr_in, sizeof(s->addr_in));
+ break;
+ }
+ default:
+ break;
+ }
+ switch (s->type) {
+ case AKI_SOCKET_TCP:
+ case AKI_SOCKET_UDP: {
+ s32 ret = connect(s->sock, (struct sockaddr *)&s->addr_in, sizeof(s->addr_in));
if (ret != 0 && errno != 115) {
al_log_error("socket", "Failed to connect: %s (%d).", aki_strerror(errno), errno);
return false;
}
-
break;
}
case AKI_SOCKET_UNIX: {
@@ -217,10 +207,9 @@ bool aki_socket_connect(struct aki_socket *s, str *addr, s32 port)
}
break;
}
- case AKI_SOCKET_UDP: {
+ default:
al_assert(false);
}
- }
signal(SIGPIPE, SIG_IGN);
@@ -239,7 +228,13 @@ ssize_t aki_socket_write(struct aki_socket *s, void *buf, size_t size)
ssize_t aki_socket_sendto(struct aki_socket *s, void *buf, size_t size)
{
- return sendto(s->sock, buf, size, 0, (const struct sockaddr *)&s->addr_un, sizeof(s->addr_un));
+ return sendto(s->sock, buf, size, 0, (const struct sockaddr *)&s->addr_in, sizeof(s->addr_in));
+}
+
+ssize_t aki_socket_recvfrom(struct aki_socket *s, void *buf, size_t size)
+{
+ socklen_t addrlen = sizeof(s->addr_in);
+ return recvfrom(s->sock, buf, size, 0, (struct sockaddr *)&s->addr_in, &addrlen);
}
s32 aki_socket_get_fd(struct aki_socket *s)
diff --git a/src/socket/socket_windows.c b/src/socket/socket_windows.c
index 6e72aea..b83ae52 100644
--- a/src/socket/socket_windows.c
+++ b/src/socket/socket_windows.c
@@ -2,18 +2,6 @@
#include "socket.h"
-static void socket_set_blocking(struct aki_socket *s, bool blocking)
-{
- u_long mode = (blocking) ? 0 : 1;
- ioctlsocket(s->sock, FIONBIO, &mode);
-}
-
-static void socket_set_no_delay(struct aki_socket *s, s32 no_delay)
-{
- (void)s;
- (void)no_delay;
-}
-
bool aki_socket_init(struct aki_socket *s)
{
switch (s->type) {
@@ -33,22 +21,19 @@ bool aki_socket_init(struct aki_socket *s)
return false;
}
- aki_socket_set_blocking(s, s->config.blocking);
- aki_socket_set_no_delay(s, s->config.no_delay);
-
return true;
}
void aki_socket_set_blocking(struct aki_socket *s, bool blocking)
{
- s->config.blocking = blocking;
- socket_set_blocking(s, blocking);
+ u_long mode = (blocking) ? 0 : 1;
+ ioctlsocket(s->sock, FIONBIO, &mode);
}
-void aki_socket_set_no_delay(struct aki_socket *s, bool no_delay)
+void aki_socket_set_no_delay(struct aki_socket *s, s32 no_delay)
{
- s->config.no_delay = no_delay;
- socket_set_no_delay(s, no_delay);
+ (void)s;
+ (void)no_delay;
}
bool aki_socket_listen(struct aki_socket *s, str *addr, s32 port)
@@ -76,9 +61,6 @@ bool aki_socket_accept(struct aki_socket *s, struct aki_socket *c)
return false;
}
- aki_socket_set_blocking(c, c->config.blocking);
- aki_socket_set_no_delay(c, c->config.no_delay);
-
return true;
}
diff --git a/src/timer.c b/src/timer.c
index e730ad5..52cdcca 100644
--- a/src/timer.c
+++ b/src/timer.c
@@ -9,17 +9,37 @@ static void timer_callback(struct ev_loop *loop, ev_timer *w, s32 revents)
}
void aki_timer_init(struct aki_timer *timer, struct aki_event_loop *loop,
- aki_os_tstamp repeat, void (*callback)(void *, struct aki_timer *), void *userdata)
+ void (*callback)(void *, struct aki_timer *), void *userdata)
{
timer->loop = loop;
timer->callback = callback;
timer->userdata = userdata;
timer->timer.data = timer;
- timer->timer.repeat = repeat;
+ timer->disabled = false;
ev_init(&timer->timer, timer_callback);
}
+void aki_timer_set_repeat(struct aki_timer *timer, aki_os_tstamp repeat)
+{
+ timer->timer.repeat = repeat;
+}
+
void aki_timer_again(struct aki_timer *timer)
{
- ev_timer_again(timer->loop->ev, &timer->timer);
+ if (!timer->disabled) ev_timer_again(timer->loop->ev, &timer->timer);
+}
+
+void aki_timer_disable(struct aki_timer *timer)
+{
+ timer->disabled = true;
+}
+
+void aki_timer_enable(struct aki_timer *timer)
+{
+ timer->disabled = false;
+}
+
+void aki_timer_stop(struct aki_timer *timer)
+{
+ ev_timer_stop(timer->loop->ev, &timer->timer);
}
diff --git a/src/timer.h b/src/timer.h
index 03b2f43..73080a4 100644
--- a/src/timer.h
+++ b/src/timer.h
@@ -8,10 +8,15 @@
struct aki_timer {
ev_timer timer;
struct aki_event_loop *loop;
+ bool disabled;
void (*callback)(void *, struct aki_timer *);
void *userdata;
};
void aki_timer_init(struct aki_timer *timer, struct aki_event_loop *loop,
- aki_os_tstamp repeat, void (*callback)(void *, struct aki_timer *), void *userdata);
+ void (*callback)(void *, struct aki_timer *), void *userdata);
+void aki_timer_set_repeat(struct aki_timer *timer, aki_os_tstamp repeat);
void aki_timer_again(struct aki_timer *timer);
+void aki_timer_disable(struct aki_timer *timer);
+void aki_timer_enable(struct aki_timer *timer);
+void aki_timer_stop(struct aki_timer *timer);
diff --git a/src/util/buffer.c b/src/util/buffer.c
index 26e242c..4f184b9 100644
--- a/src/util/buffer.c
+++ b/src/util/buffer.c
@@ -2,7 +2,7 @@
#define INITIAL_SIZE 128
-void aki_buffer_ensure_size(struct aki_buffer *buf, size_t size)
+void aki_buffer_ensure_space(struct aki_buffer *buf, size_t size)
{
if (buf->alloc >= size) return;
if (size < INITIAL_SIZE) buf->alloc = INITIAL_SIZE;
@@ -17,14 +17,29 @@ void aki_buffer_init(struct aki_buffer *buf)
buf->alloc = 0;
}
+size_t aki_buffer_get_size(struct aki_buffer *buf)
+{
+ return buf->size;
+}
+
+void aki_buffer_set_size(struct aki_buffer *buf, size_t size)
+{
+ buf->size = size;
+}
+
void aki_buffer_write(struct aki_buffer *buf, void *data, size_t index, size_t size)
{
size_t reach = index + size;
- aki_buffer_ensure_size(buf, reach);
+ aki_buffer_ensure_space(buf, reach);
if (reach > buf->size) buf->size = reach;
al_memcpy(&buf->data[index], data, size);
}
+void aki_buffer_append(struct aki_buffer *buf, void *data, size_t size)
+{
+ aki_buffer_write(buf, data, buf->size, size);
+}
+
u8 *aki_buffer_get_ptr(struct aki_buffer *buf, size_t index)
{
return &buf->data[index];
diff --git a/src/util/buffer.h b/src/util/buffer.h
index 3c40b3a..78d32d5 100644
--- a/src/util/buffer.h
+++ b/src/util/buffer.h
@@ -10,8 +10,11 @@ struct aki_buffer {
};
void aki_buffer_init(struct aki_buffer *buf);
-void aki_buffer_ensure_size(struct aki_buffer *buf, size_t size);
+void aki_buffer_ensure_space(struct aki_buffer *buf, size_t size);
+size_t aki_buffer_get_size(struct aki_buffer *buf);
+void aki_buffer_set_size(struct aki_buffer *buf, size_t size);
void aki_buffer_write(struct aki_buffer *buf, void *data, size_t index, size_t size);
+void aki_buffer_append(struct aki_buffer *buf, void *data, size_t size);
u8 *aki_buffer_get_ptr(struct aki_buffer *buf, size_t index);
void aki_buffer_read(struct aki_buffer *buf, void *ptr, size_t index, size_t size);
void aki_buffer_free(struct aki_buffer *buf);
diff --git a/src/util/error.h b/src/util/error.h
index d014faa..c7edd82 100644
--- a/src/util/error.h
+++ b/src/util/error.h
@@ -13,7 +13,7 @@ extern __declspec(thread) char errorbuf[ERROR_STR_MAXLEN];
static inline char *aki_strerror(s32 errnum)
{
#ifndef _WIN32
- strerror_r(errnum, errorbuf, ERROR_STR_MAXLEN);
+ (void)!strerror_r(errnum, errorbuf, ERROR_STR_MAXLEN);
#else
strerror_s(errorbuf, ERROR_STR_MAXLEN, errnum);
#endif
diff --git a/src/util/file/file_linux.c b/src/util/file/file_linux.c
index 97e9688..708bd2d 100644
--- a/src/util/file/file_linux.c
+++ b/src/util/file/file_linux.c
@@ -158,7 +158,7 @@ bool aki_dir_exists(str *path)
bool aki_dir_create(str *path)
{
char *c_str = al_str_to_c_str(path);
- s32 ret = mkdir(c_str, 0700);
+ s32 ret = mkdir(c_str, 0755);
al_free(c_str);
if (ret == -1) {
al_log_error("file", "mkdir(%.*s) failed (%s).", AL_STR_PRINTF(path), aki_strerror(errno));
diff --git a/src/util/packet.c b/src/util/packet.c
index c58d656..9e5a8d7 100644
--- a/src/util/packet.c
+++ b/src/util/packet.c
@@ -11,7 +11,7 @@ struct aki_packet *aki_packet_create(void)
struct aki_packet *aki_packet_clone(struct aki_packet *packet)
{
struct aki_packet *c = aki_packet_create();
- aki_buffer_ensure_size(&c->buffer, packet->buffer.size);
+ aki_buffer_ensure_space(&c->buffer, packet->buffer.size);
al_memcpy(aki_buffer_get_ptr(&c->buffer, 0), aki_buffer_get_ptr(&packet->buffer, 0), packet->buffer.size);
c->windex = packet->windex;
c->rindex = packet->rindex;
@@ -20,7 +20,7 @@ struct aki_packet *aki_packet_clone(struct aki_packet *packet)
void aki_packet_reset(struct aki_packet *packet)
{
- aki_buffer_ensure_size(&packet->buffer, AKI_PACKET_HEADER_LENGTH);
+ aki_buffer_ensure_space(&packet->buffer, AKI_PACKET_HEADER_LENGTH);
packet->rindex = AKI_PACKET_HEADER_LENGTH;
packet->windex = AKI_PACKET_HEADER_LENGTH;
}
generated by cgit v1.2.3-101-g0448 (git 2.54.0) at 2026-07-25 18:12:01 +0000