Update

- Add aki_poll and aki_fs_event
- Various aki_file improvments
- Add grow mode to aki_packet_pool
- aki_packet_stream_pause/resume -> cork
- Misc fixes and improvments

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

30 files changed, 365 insertions, 107 deletions

diff --git a/.gitignore b/.gitignore
index 526a0f2..9a297b3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -4,9 +4,8 @@ compile_commands.json
 .clangd/
 subprojects/c89atomic.wrap
 subprojects/curl/
-subprojects/curl.wrap
 subprojects/jansson/
-subprojects/jansson.wrap
 subprojects/libalabaster
 subprojects/libev-4.33/
+subprojects/liblfds.wrap
 subprojects/packagecache/
diff --git a/include/aki/common.h b/include/aki/common.h
index 67ac037..153530d 100644
--- a/include/aki/common.h
+++ b/include/aki/common.h
@@ -1,8 +1,6 @@
 #pragma once
 
 #include "../../src/common.h"
-
-#include "../../src/loop.h"
 #include "../../src/util/buffer.h"
-#include "../../src/util/packet.h"
+#include "../../src/util/error.h"
 #include "../../src/util/timer/timer.h"
diff --git a/include/aki/event_loop.h b/include/aki/event_loop.h
new file mode 100644
index 0000000..d532d92
--- /dev/null
+++ b/include/aki/event_loop.h
@@ -0,0 +1,5 @@
+#pragma once
+
+#include "../../src/loop.h"
+#include "../../src/poll.h"
+#include "../../src/util/packet.h"
diff --git a/include/aki/fs_event.h b/include/aki/fs_event.h
new file mode 100644
index 0000000..76db88a
--- /dev/null
+++ b/include/aki/fs_event.h
@@ -0,0 +1,3 @@
+#pragma once
+
+#include "../src/fs_event.h"
diff --git a/meson.build b/meson.build
index 150bda0..fb717e5 100644
--- a/meson.build
+++ b/meson.build
@@ -8,14 +8,6 @@ is_debug = get_option('buildtype').startswith('debug')
 
 akiyo_src = [
   'src/common.c',
-  'src/loop.c',
-  'src/timer.c',
-  'src/signal.c',
-  'src/packet_pool.c',
-  'src/packet_cache.c',
-  'src/packet_stream.c',
-  'src/rpc.c',
-  'src/line_processor.c',
   'src/winwrap.c',
   'src/util/error.c',
   'src/util/buffer.c',
@@ -24,45 +16,67 @@ akiyo_src = [
 ]
 akiyo_inc = [include_directories('include')]
 akiyo_deps = []
-akiyo_args = []
+akiyo_args = ['-Wno-strict-aliasing']
 
 alabaster = subproject('libalabaster')
 akiyo_deps += [alabaster.get_variable('alabaster')]
 
+math = compiler.find_library('m', required: false)
+if math.found()
+  akiyo_deps += [math]
+endif
+
 threads = dependency('threads', required: false)
 if threads.found()
   akiyo_deps += [threads]
 endif
 
-# libev gets embedded in ev_embed_compat.c.
-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
+if get_option('event-loop').enabled()
   akiyo_src += [
-    'src/curl/curl.c',
-    'src/curl/http.c',
-    'src/curl/websocket.c'
+    'src/loop.c',
+    'src/poll.c',
+    'src/fs_event.c',
+    'src/timer.c',
+    'src/signal.c',
+    'src/packet_pool.c',
+    'src/packet_cache.c',
+    'src/packet_stream.c',
+    'src/rpc.c',
+    'src/line_processor.c'
   ]
-  akiyo_deps += [libcurl]
-  #openssl = dependency('openssl')
-  #libidn2 = dependency('libidn2')
-  #zlib = dependency('zlib')
-  #akiyo_deps += [libcurl, openssl, libidn2, zlib]
-  akiyo_args += ['-DAKIYO_HAS_CURL']
+
+  # libev gets embedded in ev_embed_compat.c.
+  libev = subproject('libev').get_variable('ev')
+  akiyo_src += ['src/ev_embed_compat.c']
+  akiyo_deps += [libev]
+
+  libcurl = dependency('libcurl', required: false)
+  #libcurl = compiler.find_library('curl',
+  #  dirs: [join_paths(meson.project_source_root(), 'subprojects/curl/build/lib')])
+  if not libcurl.found()
+    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')
+  endif
+  akiyo_has_curl = libcurl.found()
+  if akiyo_has_curl
+    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
 endif
 
 jansson = dependency('jansson', required: false)
@@ -103,6 +117,6 @@ akiyo = declare_dependency(include_directories: akiyo_inc,
   dependencies: akiyo_deps, sources: akiyo_src,
   compile_args: akiyo_args)
 
-if get_option('build-tests') and not meson.is_subproject()
-  executable('tests', ['test/test.c'], dependencies: [akiyo])
+if get_option('tests') and not meson.is_subproject()
+  executable('tests', ['test/test.c'], dependencies: akiyo)
 endif
diff --git a/meson_options.txt b/meson_options.txt
index 891bc02..8d53466 100644
--- a/meson_options.txt
+++ b/meson_options.txt
@@ -1 +1,2 @@
-option('build-tests', type: 'boolean', value: true)
+option('event-loop', type: 'feature', value: 'enabled')
+option('tests', type: 'boolean', value: true)
diff --git a/src/common.c b/src/common.c
index b9ee5c2..3c76cbd 100644
--- a/src/common.c
+++ b/src/common.c
@@ -1,5 +1,6 @@
 #include <al/lib.h>
 #include <al/random.h>
+#include <locale.h>
 
 #ifndef _WIN32
 #include <unistd.h>
@@ -10,8 +11,10 @@
 #include "common.h"
 #include "winwrap.h"
 
-void aki_common_init(void)
+bool aki_common_init(void)
 {
+    if (!setlocale(LC_ALL, "")) return false;
+
     al_malloc_init();
     al_rand_init();
 
@@ -24,6 +27,8 @@ void aki_common_init(void)
     GetSystemInfo(&info);
     al_set_page_size(info.dwPageSize);
 #endif
+
+    return true;
 }
 
 void aki_common_close(void)
diff --git a/src/common.h b/src/common.h
index da4b3e3..c70cc24 100644
--- a/src/common.h
+++ b/src/common.h
@@ -1,4 +1,6 @@
 #pragma once
 
-void aki_common_init(void);
+#include <al/types.h>
+
+bool aki_common_init(void);
 void aki_common_close(void);
diff --git a/src/fs_event.c b/src/fs_event.c
new file mode 100644
index 0000000..8edce13
--- /dev/null
+++ b/src/fs_event.c
@@ -0,0 +1,56 @@
+#include <al/log.h>
+
+#include "util/error.h"
+
+#include "fs_event.h"
+
+static void event_callback(struct ev_loop *loop, ev_io *w, s32 revents)
+{
+    (void)loop;
+    struct aki_fs_event *fs = (struct aki_fs_event *)w->data;
+    (void)revents;
+    struct inotify_event event = { 0 };
+    ssize_t ret = aki_read(fs->fd, &event, sizeof(struct inotify_event));
+    if (ret == sizeof(struct inotify_event)) {
+        al_assert(event.wd == fs->wd);
+        al_assert(event.len == 0);
+        fs->callback(fs->userdata, &event);
+    }
+}
+
+void aki_fs_event_init(struct aki_fs_event *fs, str *path, u32 mask,
+    void (*callback)(void *, struct inotify_event *), void *userdata)
+{
+    fs->fd = inotify_init();
+    aki_fd_set_blocking(fs->fd, true);
+    fs->mask = mask;
+    fs->callback = callback;
+    fs->userdata = userdata;
+    fs->path = al_str_to_c_str(path);
+    fs->event.data = fs;
+    ev_io_init(&fs->event, event_callback, fs->fd, EV_READ);
+}
+
+bool aki_fs_event_start(struct aki_fs_event *fs, struct aki_event_loop *loop)
+{
+    fs->loop = loop;
+    fs->wd = inotify_add_watch(fs->fd, fs->path, fs->mask);
+    if (fs->wd == -1) {
+        al_log_error("fs_event", "inotify_add_watch(%s) failed (%s)", fs->path, aki_strerror(errno));
+        return false;
+    }
+    ev_io_start(fs->loop->ev, &fs->event);
+    return true;
+}
+
+void aki_fs_event_stop(struct aki_fs_event *fs)
+{
+    inotify_rm_watch(fs->fd, fs->wd);
+    ev_io_stop(fs->loop->ev, &fs->event);
+}
+
+void aki_fs_event_free(struct aki_fs_event *fs)
+{
+    close(fs->fd);
+    al_free(fs->path);
+}
diff --git a/src/fs_event.h b/src/fs_event.h
new file mode 100644
index 0000000..d45d7af
--- /dev/null
+++ b/src/fs_event.h
@@ -0,0 +1,25 @@
+#pragma once
+
+#include <al/str.h>
+#include <sys/inotify.h>
+
+#include "socket/socket.h"
+
+#include "loop.h"
+
+struct aki_fs_event {
+    s32 fd;
+    u32 mask;
+    char *path;
+    s32 wd;
+    ev_io event;
+    struct aki_event_loop *loop;
+    void (*callback)(void *, struct inotify_event *);
+    void *userdata;
+};
+
+void aki_fs_event_init(struct aki_fs_event *fs, str *path, u32 mask,
+    void (*callback)(void *, struct inotify_event *), void *userdata);
+bool aki_fs_event_start(struct aki_fs_event *fs, struct aki_event_loop *loop);
+void aki_fs_event_stop(struct aki_fs_event *fs);
+void aki_fs_event_free(struct aki_fs_event *fs);
diff --git a/src/line_processor.c b/src/line_processor.c
index de86dab..158c5cb 100644
--- a/src/line_processor.c
+++ b/src/line_processor.c
@@ -150,4 +150,5 @@ void aki_line_processor_stop(struct aki_line_processor *pro)
         }
     }
     al_array_free(pro->clients);
+    al_str_free(&pro->delim);
 }
diff --git a/src/packet_pool.c b/src/packet_pool.c
index 4c942b1..683efcb 100644
--- a/src/packet_pool.c
+++ b/src/packet_pool.c
@@ -79,12 +79,18 @@ void aki_packet_pool_init_ex(struct aki_packet_pool *pool, u32 size, struct aki_
     init_internal(pool, size, loop, mode, callback, userdata);
 }
 
+static const bool grow = false;
+
 struct aki_packet *aki_packet_pool_get(struct aki_packet_pool *pool)
 {
     al_assert(pool->mode == AKI_PACKET_POOL_MODE_POOL);
     aki_mutex_lock(&pool->mutex);
     if (!pool->disabled && !pool->empty.size) {
-        aki_cond_wait(&pool->cond, &pool->mutex);
+        if (grow) {
+            al_array_push(pool->empty, aki_packet_create());
+        } else {
+            aki_cond_wait(&pool->cond, &pool->mutex);
+        }
     }
     if (pool->disabled) {
         aki_mutex_unlock(&pool->mutex);
diff --git a/src/packet_stream.c b/src/packet_stream.c
index 5c415aa..0096f7c 100644
--- a/src/packet_stream.c
+++ b/src/packet_stream.c
@@ -202,16 +202,12 @@ void aki_packet_stream_connect(struct aki_packet_stream *stream,
     ev_timer_start(loop->ev, &stream->timer);
 }
 
-void aki_packet_stream_pause(struct aki_packet_stream *stream)
+void aki_packet_stream_cork(struct aki_packet_stream *stream, bool cork)
 {
-    if (stream->connected && stream->revent.active) {
+    if (!stream->connected) return;
+    if (cork && stream->revent.active) {
         ev_io_stop(stream->loop->ev, &stream->revent);
-    }
-}
-
-void aki_packet_stream_resume(struct aki_packet_stream *stream)
-{
-    if (stream->connected && !stream->revent.active) {
+    } else if (!cork && !stream->revent.active) {
         ev_io_start(stream->loop->ev, &stream->revent);
     }
 }
@@ -238,7 +234,7 @@ void aki_packet_stream_disconnect(struct aki_packet_stream *stream)
 {
     if (stream->connected) {
         aki_socket_shutdown(&stream->s);
-        aki_packet_stream_resume(stream);
+        aki_packet_stream_cork(stream, false);
     }
 }
 
diff --git a/src/packet_stream.h b/src/packet_stream.h
index 4e83639..964bb9b 100644
--- a/src/packet_stream.h
+++ b/src/packet_stream.h
@@ -44,8 +44,7 @@ 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,
     struct aki_event_loop *loop, str *addr, s32 port);
-void aki_packet_stream_pause(struct aki_packet_stream *stream);
-void aki_packet_stream_resume(struct aki_packet_stream *stream);
+void aki_packet_stream_cork(struct aki_packet_stream *stream, bool cork);
 void aki_packet_stream_send_packet(struct aki_packet_stream *stream, struct aki_packet *packet);
 void aki_packet_stream_flush(struct aki_packet_stream *stream, void (*flushed_callback)(void *));
 void aki_packet_stream_disconnect(struct aki_packet_stream *stream);
diff --git a/src/poll.c b/src/poll.c
new file mode 100644
index 0000000..ff04a63
--- /dev/null
+++ b/src/poll.c
@@ -0,0 +1,32 @@
+#include "poll.h"
+
+static void event_callback(struct ev_loop *loop, ev_io *w, s32 revents)
+{
+    (void)loop;
+    struct aki_poll *poll = (struct aki_poll *)w->data;
+    poll->callback(poll->userdata, revents);
+}
+
+void aki_poll_init(struct aki_poll *poll, void (*callback)(void *, s32 revents), void *userdata)
+{
+    poll->callback = callback;
+    poll->userdata = userdata;
+    poll->event.data = poll;
+    ev_init(&poll->event, event_callback);
+}
+
+void aki_poll_set(struct aki_poll *poll, s32 fd, s32 events)
+{
+    ev_io_set(&poll->event, fd, events);
+}
+
+void aki_poll_start(struct aki_poll *poll, struct aki_event_loop *loop)
+{
+    poll->loop = loop;
+    ev_io_start(poll->loop->ev, &poll->event);
+}
+
+void aki_poll_stop(struct aki_poll *poll)
+{
+    ev_io_stop(poll->loop->ev, &poll->event);
+}
diff --git a/src/poll.h b/src/poll.h
new file mode 100644
index 0000000..6a0e826
--- /dev/null
+++ b/src/poll.h
@@ -0,0 +1,18 @@
+#pragma once
+
+#include "loop.h"
+
+#define AKI_POLL_READ EV_READ
+#define AKI_POLL_WRITE EV_WRITE
+
+struct aki_poll {
+    ev_io event;
+    struct aki_event_loop *loop;
+    void (*callback)(void *, s32 revents);
+    void *userdata;
+};
+
+void aki_poll_init(struct aki_poll *poll, void (*callback)(void *, s32 revents), void *userdata);
+void aki_poll_start(struct aki_poll *poll, struct aki_event_loop *loop);
+void aki_poll_set(struct aki_poll *poll, s32 fd, s32 events);
+void aki_poll_stop(struct aki_poll *poll);
diff --git a/src/socket/socket.h b/src/socket/socket.h
index 4971c1c..c0cf796 100644
--- a/src/socket/socket.h
+++ b/src/socket/socket.h
@@ -1,11 +1,14 @@
 #pragma once
 
 #ifndef _WIN32
+#include <unistd.h>
 #include <sys/socket.h>
 #include <netinet/in.h>
 #include <arpa/inet.h>
 #include <netinet/in.h>
+#include <poll.h>
 #include <sys/un.h>
+#include <sys/eventfd.h>
 #else
 #include "../winwrap.h"
 #endif
@@ -34,15 +37,20 @@ struct aki_socket {
 #ifndef _WIN32
 #define aki_read read
 #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
+
+#define aki_htons htons
+#define aki_htonl htonl
+#define aki_ntohs ntohs
+
+#ifndef _WIN32 // Posix-only helpers.
+#define aki_pollfd pollfd
+#define aki_nfds nfds_t
+void aki_fd_set_blocking(s32 fd, bool blocking);
+s32 aki_poll_fds(struct aki_pollfd *fds, aki_nfds nfds, s64 timeout_ns);
 #endif
 
 bool aki_socket_init(struct aki_socket *s);
diff --git a/src/socket/socket_linux.c b/src/socket/socket_linux.c
index 7bc9ea9..3d141b7 100644
--- a/src/socket/socket_linux.c
+++ b/src/socket/socket_linux.c
@@ -9,6 +9,27 @@
 
 #include "socket.h"
 
+void aki_fd_set_blocking(s32 fd, bool blocking)
+{
+    s32 flags = fcntl(fd, F_GETFL, 0);
+    al_assert(flags != -1);
+    flags = blocking ? (flags & ~O_NONBLOCK) : (flags | O_NONBLOCK);
+    s32 ret = fcntl(fd, F_SETFL, flags);
+    al_assert(ret == 0);
+}
+
+#define AKI_TIME_S_TO_NS(s) ((s) * INT64_C(1000000000))
+
+// https://github.com/mpv-player/mpv/blob/e575ec4fc3654387c7358bd3640877ef32628d2c/osdep/poll_wrapper.c#L29
+s32 aki_poll_fds(struct aki_pollfd *fds, aki_nfds nfds, s64 timeout_ns)
+{
+    struct timespec ts;
+    ts.tv_sec  = timeout_ns / AKI_TIME_S_TO_NS(1);
+    ts.tv_nsec = timeout_ns % AKI_TIME_S_TO_NS(1);
+    struct timespec *tsp = timeout_ns >= 0 ? &ts : NULL;
+    return ppoll(fds, nfds, tsp, NULL);
+}
+
 bool aki_socket_init(struct aki_socket *s)
 {
     switch (s->type) {
@@ -33,11 +54,7 @@ bool aki_socket_init(struct aki_socket *s)
 
 void aki_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);
+    aki_fd_set_blocking(s->sock, blocking);
 }
 
 void aki_socket_set_no_delay(struct aki_socket *s, s32 no_delay)
@@ -251,4 +268,3 @@ void aki_socket_close(struct aki_socket *s)
 {
     close(s->sock);
 }
-
diff --git a/src/util/error.c b/src/util/error.c
index f055416..7e09d20 100644
--- a/src/util/error.c
+++ b/src/util/error.c
@@ -1,7 +1,3 @@
 #include "error.h"
 
-#ifndef _WIN32
-__thread char errorbuf[ERROR_STR_MAXLEN];
-#else
-__declspec(thread) char errorbuf[ERROR_STR_MAXLEN];
-#endif
+__al_thread char errorbuf[AKI_ERROR_STR_MAXLEN];
diff --git a/src/util/error.h b/src/util/error.h
index c7edd82..0849ddb 100644
--- a/src/util/error.h
+++ b/src/util/error.h
@@ -2,20 +2,16 @@
 
 #include <errno.h>
 
-#define ERROR_STR_MAXLEN 94
+#define AKI_ERROR_STR_MAXLEN 94
 
-#ifndef _WIN32
-extern __thread char errorbuf[ERROR_STR_MAXLEN];
-#else
-extern __declspec(thread) char errorbuf[ERROR_STR_MAXLEN];
-#endif
+extern __al_thread char errorbuf[AKI_ERROR_STR_MAXLEN];
 
 static inline char *aki_strerror(s32 errnum)
 {
 #ifndef _WIN32
-    (void)!strerror_r(errnum, errorbuf, ERROR_STR_MAXLEN);
+    (void)!strerror_r(errnum, errorbuf, AKI_ERROR_STR_MAXLEN);
 #else
-    strerror_s(errorbuf, ERROR_STR_MAXLEN, errnum);
+    strerror_s(errorbuf, AKI_ERROR_STR_MAXLEN, errnum);
 #endif
     return errorbuf;
 }
diff --git a/src/util/file/file.h b/src/util/file/file.h
index 465d76a..a6611bd 100644
--- a/src/util/file/file.h
+++ b/src/util/file/file.h
@@ -14,6 +14,13 @@
 #endif
 
 #include <al/str.h>
+#include <al/lib.h>
+
+enum {
+    AKI_FILE_READONLY = 1,
+    AKI_FILE_CREATE = 1 << 1,
+    AKI_FILE_LOCK = 1 << 2
+};
 
 enum {
     AKI_ENTRY_FILE = 0,
@@ -60,8 +67,9 @@ struct aki_dir_entry {
     str path;
 };
 
-bool aki_file_open(struct aki_file *file, str *path, bool create);
+bool aki_file_open(struct aki_file *file, str *path, s32 flags);
 bool aki_file_exists(str *path);
+bool aki_file_create(str *path);
 bool aki_file_truncate(struct aki_file *file, off_t size);
 off_t aki_file_seek(struct aki_file *file, off_t offset, s32 whence);
 size_t aki_file_get_filesize(struct aki_file *file);
@@ -69,6 +77,7 @@ bool aki_file_write(struct aki_file *file, void *buf, size_t size);
 bool aki_file_read(struct aki_file *file, void *buf, size_t size);
 s32 aki_file_read_as_str(struct aki_file *file, str *out);
 s32 aki_file_read_as_c_str(struct aki_file *file, char **out);
+s32 aki_file_replace(struct aki_file *file, str *buf);
 void *aki_file_mmap(struct aki_file *file);
 void *aki_file_mremap(struct aki_file *file, size_t size, void *old_map);
 void aki_file_munmap(struct aki_file *file, void *map);
diff --git a/src/util/file/file_linux.c b/src/util/file/file_linux.c
index 708bd2d..f916e0b 100644
--- a/src/util/file/file_linux.c
+++ b/src/util/file/file_linux.c
@@ -1,16 +1,33 @@
-#include <sys/mman.h>
 #include <al/log.h>
+#include <sys/mman.h>
 
 #include "../error.h"
 
 #include "file.h"
 
-bool aki_file_open(struct aki_file *file, str *path, bool create)
+static bool lock_file_internal(s32 fd)
+{
+    struct flock l = {
+        .l_type = F_WRLCK,
+        .l_whence = SEEK_SET,
+        .l_start = 0,
+        .l_len = 0
+    };
+    if (fcntl(fd, F_SETLKW, &l) == -1) {
+        al_log_error("file", "fcntl(F_SETLKW) failed (%s).", aki_strerror(errno));
+        close(fd);
+        return false;
+    }
+    return true;
+}
+
+bool aki_file_open(struct aki_file *file, str *path, s32 flags)
 {
+    al_assert(!((flags & AKI_FILE_READONLY) && (flags & AKI_FILE_LOCK)));
     char *c_str = al_str_to_c_str(path);
-    s32 flags = O_RDWR;
-    if (create) flags |= O_CREAT;
-    file->fd = open(c_str, flags, 0666);
+    s32 oflags = (flags & AKI_FILE_READONLY) ? O_RDONLY : O_RDWR;
+    if (flags & AKI_FILE_CREATE) oflags |= O_CREAT;
+    file->fd = open(c_str, oflags, 0666);
     al_free(c_str);
     if (file->fd == -1) {
         al_log_error("file", "open(%.*s) failed (%s).", AL_STR_PRINTF(path), aki_strerror(errno));
@@ -24,6 +41,7 @@ bool aki_file_open(struct aki_file *file, str *path, bool create)
         return false;
     }
     file->filesize = sb.st_size;
+    if (flags & AKI_FILE_LOCK) lock_file_internal(file->fd);
     return true;
 }
 
@@ -32,12 +50,19 @@ bool aki_file_exists(str *path)
     char *c_str = al_str_to_c_str(path);
     s32 ret = access(c_str, F_OK);
     al_free(c_str);
-    if (ret != 0) {
-        al_log_error("file", "access(%.*s, F_OK) failed (%s).", AL_STR_PRINTF(path), aki_strerror(errno));
-    }
     return ret == 0;
 }
 
+bool aki_file_create(str *path)
+{
+    struct aki_file file;
+    if (aki_file_open(&file, path, AKI_FILE_CREATE)) {
+        aki_file_close(&file);
+        return true;
+    }
+    return false;
+}
+
 off_t aki_file_seek(struct aki_file *file, off_t offset, s32 whence)
 {
     off_t res = lseek(file->fd, offset, whence);
@@ -98,7 +123,6 @@ void *aki_file_mmap(struct aki_file *file)
 
 void *aki_file_mremap(struct aki_file *file, size_t size, void *old_map)
 {
-    printf("old_size: %zu, new_size: %zu\n", file->filesize, size);
     void *map = mremap(old_map, file->filesize, size, MREMAP_MAYMOVE);
     if (map == MAP_FAILED) {
         al_log_error("file", "mremap(%p, %zu, %zu) failed (%s).", old_map,
@@ -152,6 +176,7 @@ bool aki_dir_exists(str *path)
         }
         return false;
     }
+    closedir(dir.dir);
     return true;
 }
 
diff --git a/src/util/file/file_stdio.c b/src/util/file/file_stdio.c
index 52d8761..fa1e4ff 100644
--- a/src/util/file/file_stdio.c
+++ b/src/util/file/file_stdio.c
@@ -4,10 +4,10 @@
 
 #include "file.h"
 
-bool aki_file_open(struct aki_file *file, str *path, bool create)
+bool aki_file_open(struct aki_file *file, str *path, s32 flags)
 {
     char *c_str = al_str_to_c_str(path);
-    const char *mode = create ? "wb+" : "rb+";
+    const char *mode = (flags & AKI_FILE_CREATE) ? "wb+" : "rb+";
 #ifndef _WIN32
     file->file = fopen(c_str, mode);
 #else
diff --git a/src/util/file/util.c b/src/util/file/util.c
index e893de7..fe7785f 100644
--- a/src/util/file/util.c
+++ b/src/util/file/util.c
@@ -23,3 +23,13 @@ s32 aki_file_read_as_c_str(struct aki_file *file, char **out)
     (*out)[size] = '\n';
     return size;
 }
+
+s32 aki_file_replace(struct aki_file *file, str *buf)
+{
+    if (aki_file_seek(file, 0, SEEK_SET) == (off_t)-1 ||
+        !aki_file_write(file, buf->data, buf->len) ||
+        !aki_file_truncate(file, buf->len)) {
+        return -1;
+    }
+    return buf->len;
+}
diff --git a/src/util/packet.c b/src/util/packet.c
index 9e5a8d7..2d8b440 100644
--- a/src/util/packet.c
+++ b/src/util/packet.c
@@ -52,12 +52,18 @@ DEFINE_PACKET_WRITE_FUNC(s64)
 DEFINE_PACKET_WRITE_FUNC(f32)
 DEFINE_PACKET_WRITE_FUNC(f64)
 
-void aki_packet_write_string(struct aki_packet *packet, str *s)
+void aki_packet_write_str(struct aki_packet *packet, str *s)
 {
     AKI_PACKET_WRITE_TYPE(packet, u32, s->len);
     AKI_PACKET_WRITE_DATA(packet, s->data, s->len);
 }
 
+void aki_packet_write_wstr(struct aki_packet *packet, wstr *w)
+{
+    AKI_PACKET_WRITE_TYPE(packet, u32, w->len);
+    AKI_PACKET_WRITE_DATA(packet, w->data, w->len * sizeof(wchar_t));
+}
+
 void aki_packet_write_buffer(struct aki_packet *packet, struct aki_buffer *buf)
 {
     AKI_PACKET_WRITE_TYPE(packet, size_t, buf->size);
@@ -83,13 +89,20 @@ DEFINE_PACKET_READ_FUNC(s64)
 DEFINE_PACKET_READ_FUNC(f32)
 DEFINE_PACKET_READ_FUNC(f64)
 
-void aki_packet_read_string(struct aki_packet *packet, str *s)
+void aki_packet_read_str(struct aki_packet *packet, str *s)
 {
     AKI_PACKET_READ_TYPE(packet, u32, s->len);
     AKI_PACKET_READ_DATA(packet, s->len, s->data);
     s->alloc = 0;
 }
 
+void aki_packet_read_wstr(struct aki_packet *packet, wstr *w)
+{
+    AKI_PACKET_READ_TYPE(packet, u32, w->len);
+    AKI_PACKET_READ_DATA(packet, w->len * sizeof(wchar_t), w->data);
+    w->alloc = 0;
+}
+
 void aki_packet_read_buffer(struct aki_packet *packet, struct aki_buffer *buf)
 {
     AKI_PACKET_READ_TYPE(packet, size_t, buf->size);
diff --git a/src/util/packet.h b/src/util/packet.h
index 14acf68..c6b4235 100644
--- a/src/util/packet.h
+++ b/src/util/packet.h
@@ -2,6 +2,7 @@
 
 #include <al/types.h>
 #include <al/str.h>
+#include <al/wstr.h>
 #include <al/array.h>
 
 #include "../util/buffer.h"
@@ -47,7 +48,8 @@ void aki_packet_write_u32(struct aki_packet *packet, u32 v);
 void aki_packet_write_s64(struct aki_packet *packet, s64 v);
 void aki_packet_write_u64(struct aki_packet *packet, u64 v);
 void aki_packet_write_f32(struct aki_packet *packet, f32 v);
-void aki_packet_write_string(struct aki_packet *packet, str *s);
+void aki_packet_write_str(struct aki_packet *packet, str *s);
+void aki_packet_write_wstr(struct aki_packet *packet, wstr *w);
 void aki_packet_write_buffer(struct aki_packet *packet, struct aki_buffer *buf);
 
 s8 aki_packet_read_s8(struct aki_packet *packet);
@@ -59,7 +61,8 @@ u32 aki_packet_read_u32(struct aki_packet *packet);
 s64 aki_packet_read_s64(struct aki_packet *packet);
 u64 aki_packet_read_u64(struct aki_packet *packet);
 f32 aki_packet_read_f32(struct aki_packet *packet);
-void aki_packet_read_string(struct aki_packet *packet, str *s);
+void aki_packet_read_str(struct aki_packet *packet, str *s);
+void aki_packet_read_wstr(struct aki_packet *packet, wstr *w);
 void aki_packet_read_buffer(struct aki_packet *packet, struct aki_buffer *buf);
 
 void aki_packet_free(struct aki_packet *packet);
diff --git a/subprojects/curl.wrap b/subprojects/curl.wrap
new file mode 100644
index 0000000..245b709
--- /dev/null
+++ b/subprojects/curl.wrap
@@ -0,0 +1,4 @@
+[wrap-git]
+url = https://github.com/curl/curl.git
+revision = curl-8_4_0
+depth = 1
diff --git a/subprojects/jansson.wrap b/subprojects/jansson.wrap
new file mode 100644
index 0000000..23114ba
--- /dev/null
+++ b/subprojects/jansson.wrap
@@ -0,0 +1,4 @@
+[wrap-git]
+url = https://github.com/akheron/jansson.git
+revision = v2.14
+depth = 1
diff --git a/subprojects/libev.wrap b/subprojects/libev.wrap
index 21430df..22fed20 100644
--- a/subprojects/libev.wrap
+++ b/subprojects/libev.wrap
@@ -6,7 +6,6 @@ source_filename = libev-4.33.tar.gz
 source_hash = 507eb7b8d1015fbec5b935f34ebed15bf346bed04a11ab82b8eee848c4205aea
 
 patch_directory = libev
-
 diff_files = libev/libev_aki_sleep.diff
 
 [provide]
diff --git a/test/test.c b/test/test.c
index 0b2f680..60799e6 100644
--- a/test/test.c
+++ b/test/test.c
@@ -1,7 +1,9 @@
 #include <aki/common.h>
 #include <aki/thread.h>
-#include <aki/line_processor.h>
-#include <aki/timer.h>
+#include <aki/event_loop.h>
+#include <aki/fs_event.h>
+//#include <aki/line_processor.h>
+//#include <aki/timer.h>
 
 aki_thread_result AKI_THREADCALL test_thread(void *userdata)
 {
@@ -15,6 +17,7 @@ aki_thread_result AKI_THREADCALL test_thread(void *userdata)
 
 static struct aki_event_loop loop;
 
+/*
 static u8 line_callback(void *userdata, str *line)
 {
     (void)userdata;
@@ -28,6 +31,12 @@ static void timer_callback(void *userdata, struct aki_timer *timer)
     printf("Yo\n");
     aki_timer_again(timer);
 }
+*/
+
+static void fs_event_callback(void *userdata, struct inotify_event *event)
+{
+    al_printf("%i\n", event->mask);
+}
 
 s32 main(void)
 {
@@ -35,18 +44,22 @@ s32 main(void)
 
     aki_event_loop_init(&loop);
 
+    struct aki_fs_event fs_event;
+    aki_fs_event_init(&fs_event, al_str_c("/home/andrew/.config/taro/DP-3/selection"), IN_CLOSE_WRITE,
+        fs_event_callback, NULL);
+    aki_fs_event_start(&fs_event, &loop);
+
     /*
     struct aki_timer timer;
     aki_timer_init(&timer, &loop, 1.0, timer_callback, NULL);
     aki_timer_again(&timer);
     */
 
-    /*
-    struct aki_thread thread;
-    aki_thread_create(&thread, test_thread, NULL);
-    aki_thread_join(&thread);
-    */
+    //struct aki_thread thread;
+    //aki_thread_create(&thread, test_thread, NULL);
+    //aki_thread_join(&thread);
 
+/*
 #ifndef _WIN32
     struct aki_line_processor pro;
     pro.callback = line_callback;
@@ -55,8 +68,10 @@ s32 main(void)
     aki_line_processor_open_fd(&pro, STDIN_FILENO);
     aki_line_processor_run(&pro, &loop);
 #endif
+*/
 
     aki_event_loop_run(&loop);
+    aki_event_loop_destroy(&loop);
 
     aki_common_close();