Update

- Refactor curl support, add curl websockets
- Remove aki_socket config structure
- Add the ability to change aki_timer repeat
- Various fixes

Signed-off-by: Andrew Opalach <andrew@akon.city>

29 files changed, 771 insertions, 449 deletions

diff --git a/.gitignore b/.gitignore
index a4bfced..526a0f2 100644
--- a/.gitignore
+++ b/.gitignore
@@ -3,6 +3,10 @@ compile_commands.json
 .cache/
 .clangd/
 subprojects/c89atomic.wrap
+subprojects/curl/
+subprojects/curl.wrap
+subprojects/jansson/
+subprojects/jansson.wrap
 subprojects/libalabaster
 subprojects/libev-4.33/
 subprojects/packagecache/
diff --git a/include/aki/common.h b/include/aki/common.h
index 64a41c8..67ac037 100644
--- a/include/aki/common.h
+++ b/include/aki/common.h
@@ -2,6 +2,7 @@
 
 #include "../../src/common.h"
 
+#include "../../src/loop.h"
 #include "../../src/util/buffer.h"
 #include "../../src/util/packet.h"
 #include "../../src/util/timer/timer.h"
diff --git a/include/aki/http.h b/include/aki/http.h
index 1f75987..9a38452 100644
--- a/include/aki/http.h
+++ b/include/aki/http.h
@@ -1,5 +1,5 @@
 #pragma once
 
 #ifdef AKIYO_HAS_CURL
-#include "../../src/http.h"
+#include "../../src/curl/http.h"
 #endif
diff --git a/include/aki/websocket.h b/include/aki/websocket.h
new file mode 100644
index 0000000..db1ed43
--- /dev/null
+++ b/include/aki/websocket.h
@@ -0,0 +1,5 @@
+#pragma once
+
+#ifdef AKIYO_HAS_CURL
+#include "../../src/curl/websocket.h"
+#endif
diff --git a/meson.build b/meson.build
index 34a2686..150bda0 100644
--- a/meson.build
+++ b/meson.build
@@ -2,6 +2,9 @@ project('libakiyo', 'c', version: '0.1',
   default_options: ['warning_level=2', 'c_std=c99'])
 
 compiler = meson.get_compiler('c')
+cmake = import('cmake')
+
+is_debug = get_option('buildtype').startswith('debug')
 
 akiyo_src = [
   'src/common.c',
@@ -36,14 +39,42 @@ libev = subproject('libev').get_variable('ev')
 akiyo_src += ['src/ev_embed_compat.c']
 akiyo_deps += [libev]
 
+#curl_opts = cmake.subproject_options()
+#curl_opts.add_cmake_defines({ 'CMAKE_BUILD_TYPE': is_debug ? 'Debug' : 'Release' })
+#curl_opts.add_cmake_defines({ 'BUILD_CURL_EXE': false })
+#curl_opts.add_cmake_defines({ 'BUILD_SHARED_LIBS': false })
+#curl_opts.add_cmake_defines({ 'BUILD_STATIC_LIBS': true })
+#curl_opts.add_cmake_defines({ 'CURL_USE_LIBSSH2': false })
+#curl_opts.add_cmake_defines({ 'ENABLE_WEBSOCKETS': true })
+#libcurl = cmake.subproject('curl', options: curl_opts).dependency('libcurl_static')
+#libcurl = compiler.find_library('curl',
+#  dirs: [join_paths(meson.project_source_root(), 'subprojects/curl/build/lib')])
 libcurl = dependency('libcurl', required: false)
 akiyo_has_curl = libcurl.found()
 if akiyo_has_curl
-  akiyo_src += ['src/http.c']
+  akiyo_src += [
+    'src/curl/curl.c',
+    'src/curl/http.c',
+    'src/curl/websocket.c'
+  ]
   akiyo_deps += [libcurl]
+  #openssl = dependency('openssl')
+  #libidn2 = dependency('libidn2')
+  #zlib = dependency('zlib')
+  #akiyo_deps += [libcurl, openssl, libidn2, zlib]
   akiyo_args += ['-DAKIYO_HAS_CURL']
 endif
 
+jansson = dependency('jansson', required: false)
+if not jansson.found()
+  jansson_opts = cmake.subproject_options()
+  jansson_opts.add_cmake_defines({ 'CMAKE_BUILD_TYPE': is_debug ? 'Debug' : 'Release' })
+  jansson_opts.add_cmake_defines({ 'JANSSON_EXAMPLES': false })
+  jansson_opts.add_cmake_defines({ 'JANSSON_BUILD_DOCS': false })
+  jansson = cmake.subproject('jansson', options: jansson_opts).dependency('jansson')
+endif
+akiyo_deps += [jansson]
+
 if host_machine.system() != 'windows'
   akiyo_src += [
     'src/util/timer/timer_linux.c',
diff --git a/scripts/line_count.sh b/scripts/line_count.sh
new file mode 100755
index 0000000..aadb675
--- /dev/null
+++ b/scripts/line_count.sh
@@ -0,0 +1,2 @@
+#! /usr/bin/env sh
+find ./src ./include -type f | xargs wc -l
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;
 }
diff --git a/test/test.c b/test/test.c
index bf1654b..0b2f680 100644
--- a/test/test.c
+++ b/test/test.c
@@ -15,11 +15,12 @@ aki_thread_result AKI_THREADCALL test_thread(void *userdata)
 
 static struct aki_event_loop loop;
 
-static void line_callback(void *userdata, str *line)
+static u8 line_callback(void *userdata, str *line)
 {
     (void)userdata;
     al_printf("%.*s\n", AL_STR_PRINTF(line));
-    if (al_str_eq(line, al_str_c("quit"))) aki_event_loop_break(&loop);
+    if (al_str_eq(line, al_str_c("quit"))) return AKI_LINE_PROCESSOR_STOP;
+    return AKI_LINE_PROCESSOR_CONTINUE;
 }
 
 static void timer_callback(void *userdata, struct aki_timer *timer)
@@ -34,9 +35,11 @@ s32 main(void)
 
     aki_event_loop_init(&loop);
 
+    /*
     struct aki_timer timer;
     aki_timer_init(&timer, &loop, 1.0, timer_callback, NULL);
     aki_timer_again(&timer);
+    */
 
     /*
     struct aki_thread thread;
@@ -45,14 +48,12 @@ s32 main(void)
     */
 
 #ifndef _WIN32
-    /*
     struct aki_line_processor pro;
     pro.callback = line_callback;
     pro.userdata = NULL;
     aki_line_processor_init(&pro, al_str_c("\n"));
     aki_line_processor_open_fd(&pro, STDIN_FILENO);
     aki_line_processor_run(&pro, &loop);
-    */
 #endif
 
     aki_event_loop_run(&loop);