Everything before initial synced list

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

5 files changed, 565 insertions, 0 deletions

diff --git a/tests/action.h b/tests/action.h
new file mode 100644
index 0000000..0a3d22d
--- /dev/null
+++ b/tests/action.h
@@ -0,0 +1,14 @@
+#pragma once
+
+#include <al/types.h>
+
+enum {
+    CAMU_TEST_ACTION_ADD = 0,
+    CAMU_TEST_ACTION_SKIP,
+    CAMU_TEST_ACTION_SEEK
+};
+
+struct camu_test_action {
+    u8 action;
+    f32 roll;
+};
diff --git a/tests/meson.build b/tests/meson.build
new file mode 100644
index 0000000..97b0f60
--- /dev/null
+++ b/tests/meson.build
@@ -0,0 +1,3 @@
+#tests_src = ['runner.c', 'ui.c']
+#tests_deps = [common_deps, dependency('notcurses'), server, desktop]
+#executable('runner', tests_src, dependencies: tests_deps)
diff --git a/tests/runner.c b/tests/runner.c
new file mode 100644
index 0000000..13bbaa1
--- /dev/null
+++ b/tests/runner.c
@@ -0,0 +1,288 @@
+#define DISABLE_TUI 0
+
+#include <al/log.h>
+#include <al/random.h>
+#include <aki/common.h>
+#include <aki/event_loop.h>
+#include <aki/timer.h>
+
+#include "../src/server/server.h"
+#include "../src/server/list.h"
+#include "../src/server/common.h"
+#include "../src/sink/desktop.h"
+#include "../src/cache/handlers/file.h"
+#include "../src/codec/ffmpeg/common.h"
+
+#include "ui.h"
+#include "action.h"
+
+struct runner {
+    struct aki_event_loop loop;
+    struct camu_test_ui ui;
+    struct aki_poll input_poll;
+    struct aki_timer timer;
+    struct {
+        struct camu_server server;
+    } server;
+    struct {
+        u32 count;
+        atomic_s32 sinks_to_add;
+        array(struct camu_desktop *) pending_sinks;
+        struct aki_signal signal;
+        array(struct camu_desktop *) sinks;
+        atomic_bool cleanup_needed;
+        array(struct camu_desktop *) cleanup;
+    } sink;
+    struct aki_mutex mutex;
+};
+
+static void list_add_file(struct camu_server *server, struct lia_list *list, str *path)
+{
+    struct cch_entry *entry = cch_handler_file_create(path);
+    if (entry) {
+        struct camu_server_resource *resource = al_alloc_object(struct camu_server_resource);
+        resource->node = lia_server_create_node(&server->data.server, entry);
+        al_array_push(server->data.resources, resource);
+        al_str_clone(&resource->unique_id, path);
+        resource->entry = entry;
+        lia_list_add(list, resource, camu_server_entry_callback);
+    }
+}
+
+static void exit_callback(void *userdata, struct camu_desktop *desktop)
+{
+    struct runner *rn = (struct runner *)userdata;
+    al_log_info("runner", "Sink exited.");
+    aki_mutex_lock(&rn->mutex);
+    al_array_push(rn->sink.cleanup, desktop);
+    al_atomic_bool_store(&rn->sink.cleanup_needed, true, AL_ATOMIC_RELAXED);
+    if (--rn->sink.count) {
+        aki_mutex_unlock(&rn->mutex);
+        return;
+    }
+    aki_mutex_unlock(&rn->mutex);
+    aki_signal_stop(&rn->sink.signal);
+    camu_server_close(&rn->server.server);
+#if !DISABLE_TUI
+    aki_timer_disable(&rn->timer);
+    aki_timer_stop(&rn->timer);
+    aki_poll_stop(&rn->input_poll);
+#endif
+}
+
+static void signal_callback(void *userdata)
+{
+    struct runner *rn = (struct runner *)userdata;
+    aki_mutex_lock(&rn->mutex);
+    struct camu_desktop *desktop;
+    al_array_foreach(rn->sink.pending_sinks, i, desktop) {
+        if (!camu_desktop_connect(desktop, AKI_SOCKET_UNIX, &rn->loop, CAMU_UNIX_PATH, CAMU_PORT)) {
+        }
+        camu_test_ui_add_sink(&rn->ui, &desktop->sink);
+    }
+    rn->sink.pending_sinks.size = 0;
+    aki_mutex_unlock(&rn->mutex);
+}
+
+static void add_sink(struct runner *rn)
+{
+    struct camu_desktop *desktop = al_alloc_object(struct camu_desktop);
+    if (!camu_desktop_init(desktop, "cmv_test")) {
+    }
+    desktop->exit_callback = exit_callback;
+    desktop->userdata = rn;
+    al_array_push(rn->sink.sinks, desktop);
+    aki_mutex_lock(&rn->mutex);
+    rn->sink.count++;
+    al_array_push(rn->sink.pending_sinks, desktop);
+    aki_mutex_unlock(&rn->mutex);
+    aki_signal_send(&rn->sink.signal);
+}
+
+#if !DISABLE_TUI
+static void input_poll_callback(void *userdata, s32 revents)
+{
+    struct runner *rn = (struct runner *)userdata;
+    (void)revents;
+    struct ncinput input;
+    u32 ret;
+    do {
+        al_memset(&input, 0, sizeof(struct ncinput));
+        ret = notcurses_get_nblock(rn->ui.nc, &input);
+        if (ret == (u32)-1 || ret == 0) break;
+        if (input.evtype != NCTYPE_RELEASE) {
+            if (input.id == 'n') al_atomic_s32_add(&rn->sink.sinks_to_add, 1, AL_ATOMIC_RELAXED);
+        }
+    } while (1);
+}
+
+static s32 log_callback(void *userdata, char *msg)
+{
+    struct runner *rn = (struct runner *)userdata;
+    camu_test_ui_push_message(&rn->ui, msg);
+    return al_strlen(msg);
+}
+
+static void av_log_callback(void *userdata, int level, const char *fmt, va_list args)
+{
+    (void)userdata;
+    if (level < AV_LOG_DEBUG) al_logv("debug", "ff_debug", fmt, args);
+}
+
+//static f32 chances[] = { 0.2f, 0.5f, 1.0f };
+
+static void timer_callback(void *userdata, struct aki_timer *timer)
+{
+    struct runner *rn = (struct runner *)userdata;
+    (void)timer;
+    /*
+    f32 roll = al_rand() / (f32)AL_RAND_MAX;
+    struct camu_user *user = al_array_at(rn->server.server.users, 0);
+    struct lia_list *list = al_array_at(user->lists, 0);
+    //s32 sequence = al_array_at(list->entries, list->current)->sequence;
+    struct camu_test_action action = { .roll = roll };
+    if (roll < chances[0]) {
+        action.action = CAMU_TEST_ACTION_SEEK;
+        s32 amount = (al_rand() - (RAND_MAX / 2)) % 3;
+        //lia_list_skip(list, sequence, amount);
+    } else if (roll < chances[1]) {
+        action.action = CAMU_TEST_ACTION_SKIP;
+        f64 percent = al_rand() / (f64)AL_RAND_MAX;
+        //lia_list_seek(list, sequence, percent);
+    } else if (roll < chances[2]) {
+        action.action = CAMU_TEST_ACTION_ADD;
+    }
+    camu_test_ui_push_action(&rn->ui, &action);
+    */
+    camu_test_ui_draw(&rn->ui);
+    aki_timer_again(&rn->timer);
+}
+#endif
+
+static aki_thread_result AKI_THREADCALL event_loop_thread(void *userdata)
+{
+    struct runner *rn = (struct runner *)userdata;
+    aki_event_loop_run(&rn->loop);
+    aki_event_loop_destroy(&rn->loop);
+    return 0;
+}
+
+static struct runner rn;
+
+#define failure() do { \
+    goto out; \
+} while (0)
+
+#ifndef _WIN32
+s32 main(s32 argc, char *argv[])
+#else
+s32 wmain(s32 argc, wchar_t **argv)
+#endif
+{
+    (void)argc;
+    (void)argv;
+    if (!aki_common_init() || !stl_global_init()) {
+        return EXIT_FAILURE;
+    }
+
+    s32 ret = EXIT_FAILURE;
+
+#if !DISABLE_TUI
+    al_set_print(log_callback, &rn);
+    camu_ff_set_log_callback(av_log_callback);
+    if (!camu_test_ui_init(&rn.ui)) failure();
+#endif
+
+    aki_event_loop_init(&rn.loop);
+
+    if (!camu_server_init(&rn.server.server, AKI_SOCKET_UNIX, &rn.loop)) failure();
+    if (!camu_server_listen(&rn.server.server, CAMU_UNIX_PATH, CAMU_PORT)) failure();
+
+    struct camu_list *list = al_array_at(rn.server.server.lists, 0);
+    list_add_file(&rn.server.server, &list->impl, al_str_c("/home/andrew/dl/mp4/Biggest glow up of any ye song ever [fgbri3k71ljc1].mp4"));
+    list_add_file(&rn.server.server, &list->impl, al_str_c("/home/andrew/dl/mp4/This is FRIED [svlczpgt2m3d1].mp4"));
+    list_add_file(&rn.server.server, &list->impl, al_str_c("/home/andrew/pics/anime/41970900_p0.jpg"));
+    list_add_file(&rn.server.server, &list->impl, al_str_c("/home/andrew/c/camu/data/test_resources/2F49561C2BEE122D0E6FBE2B43A8F6AD_video_dashinit.mp4"));
+
+#if !DISABLE_TUI
+    camu_test_ui_add_list(&rn.ui, &list->impl);
+
+    aki_poll_init(&rn.input_poll, input_poll_callback, &rn);
+    aki_poll_set(&rn.input_poll, notcurses_inputready_fd(rn.ui.nc), AKI_POLL_READ);
+    aki_poll_start(&rn.input_poll, &rn.loop);
+
+    aki_timer_init(&rn.timer, &rn.loop, timer_callback, &rn);
+    aki_timer_set_repeat(&rn.timer, AKI_TS_FROM_USEC(333333));
+    timer_callback(&rn, NULL);
+#endif
+
+    rn.sink.count = 0;
+    al_atomic_s32_store(&rn.sink.sinks_to_add, 0, AL_ATOMIC_RELAXED);
+    al_array_init(rn.sink.pending_sinks);
+    aki_signal_init(&rn.sink.signal, signal_callback, &rn);
+    aki_signal_start(&rn.sink.signal, &rn.loop);
+    al_array_init(rn.sink.sinks);
+    al_atomic_bool_store(&rn.sink.cleanup_needed, false, AL_ATOMIC_RELAXED);
+    al_array_init(rn.sink.cleanup);
+
+    aki_mutex_init(&rn.mutex);
+
+    struct aki_thread thread0;
+    aki_thread_create(&thread0, event_loop_thread, &rn);
+
+    s32 sinks_to_add;
+    bool any_tick;
+    struct camu_desktop *desktop;
+    do {
+        sinks_to_add = al_atomic_s32_load(&rn.sink.sinks_to_add, AL_ATOMIC_RELAXED);
+        for (s32 i = 0; i < sinks_to_add; i++) {
+            add_sink(&rn);
+            al_atomic_s32_sub(&rn.sink.sinks_to_add, 1, AL_ATOMIC_RELAXED);
+        }
+        if (rn.sink.count == 0) {
+            any_tick = true;
+            aki_thread_sleep(AKI_TS_FROM_USEC(1000));
+            continue;
+        }
+        // If any sink blocks this thread, things go awry and we
+        // will crash. I'm not sure why. In wayland this manifests
+        // as getting disconnected from the wayland display.
+        any_tick = false;
+        al_array_foreach(rn.sink.sinks, i, desktop) {
+            if (!camu_desktop_tick(desktop)) {
+                camu_desktop_stop(desktop);
+                al_array_remove_at_iter(rn.sink.sinks, i);
+            } else {
+                any_tick = true;
+            }
+        }
+        // This is a really bad race condition. It's quite likely that sink entries
+        // won't be done cleanup by the time we are calling desktop_free() here.
+        //if (al_atomic_bool_load(&rn.sink.cleanup_needed, AL_ATOMIC_RELAXED)) {
+        //    aki_mutex_lock(&rn.mutex);
+        //    al_array_foreach(rn.sink.cleanup, i, desktop) {
+        //        camu_desktop_free(desktop);
+        //    }
+        //    rn.sink.cleanup.size = 0;
+        //    al_atomic_bool_store(&rn.sink.cleanup_needed, false, AL_ATOMIC_RELAXED);
+        //    aki_mutex_unlock(&rn.mutex);
+        //}
+    } while (any_tick);
+
+    aki_thread_join(&thread0);
+    al_assert(rn.sink.sinks.size == 0);
+    al_array_foreach(rn.sink.cleanup, i, desktop) {
+        camu_desktop_free(desktop);
+    }
+    camu_server_free(&rn.server.server);
+
+    ret = EXIT_SUCCESS;
+out:
+#if !DISABLE_TUI
+    camu_test_ui_close(&rn.ui);
+#endif
+    stl_global_close();
+    aki_common_close();
+
+    return ret;
+}
diff --git a/tests/ui.c b/tests/ui.c
new file mode 100644
index 0000000..7998e93
--- /dev/null
+++ b/tests/ui.c
@@ -0,0 +1,220 @@
+#include "ui.h"
+
+static void layout_ui(struct camu_test_ui *ui)
+{
+    struct ncplane_options opts = { 0 };
+
+    if (ui->messages.p) ncplane_destroy(ui->messages.p);
+    opts.x = ui->term_cols / 2;
+    opts.y = 0;
+    opts.rows = ui->term_rows / 2;
+    opts.cols = ui->term_cols / 2;
+    ui->messages.p = ncplane_create(notcurses_stdplane(ui->nc), &opts);
+
+    if (ui->actions.p) ncplane_destroy(ui->actions.p);
+    opts.y = ui->term_rows / 2;
+    opts.rows = ui->term_rows - opts.rows;
+    ui->actions.p = ncplane_create(notcurses_stdplane(ui->nc), &opts);
+
+    if (ui->lists.p) ncplane_destroy(ui->lists.p);
+    opts.x = 0;
+    opts.y = 0;
+    opts.rows = ui->term_rows / 2;
+    ui->lists.p = ncplane_create(notcurses_stdplane(ui->nc), &opts);
+
+    if (ui->sinks.p) ncplane_destroy(ui->sinks.p);
+    opts.y = ui->term_rows / 2;
+    opts.rows = ui->term_rows - opts.rows;
+    ui->sinks.p = ncplane_create(notcurses_stdplane(ui->nc), &opts);
+}
+
+static int resize_cb(struct ncplane *p)
+{
+    struct camu_test_ui *ui = (struct camu_test_ui *)ncplane_userptr(p);
+    notcurses_stddim_yx(ui->nc, &ui->term_rows, &ui->term_cols);
+    layout_ui(ui);
+    camu_test_ui_draw(ui);
+    return 0;
+}
+
+bool camu_test_ui_init(struct camu_test_ui *ui)
+{
+    if (!(ui->nc = notcurses_core_init(NULL, stdin))) {
+        return false;
+    }
+
+    notcurses_stddim_yx(ui->nc, &ui->term_rows, &ui->term_cols);
+    ncplane_set_resizecb(notcurses_stdplane(ui->nc), resize_cb);
+    ncplane_set_userptr(notcurses_stdplane(ui->nc), ui);
+
+    ui->messages.p = NULL;
+    al_array_init(ui->messages.a);
+
+    ui->actions.p = NULL;
+    al_array_init(ui->actions.a);
+
+    ui->lists.p = NULL;
+    al_array_init(ui->lists.a);
+
+    ui->sinks.p = NULL;
+    al_array_init(ui->sinks.a);
+
+    ui->buffer = al_malloc(256);
+    ui->buffer[0] = '\0';
+
+    layout_ui(ui);
+
+    return true;
+}
+
+void camu_test_ui_push_message(struct camu_test_ui *ui, char *msg)
+{
+    al_array_push(ui->messages.a, msg);
+}
+
+void camu_test_ui_push_action(struct camu_test_ui *ui, struct camu_test_action *action)
+{
+    al_array_push(ui->actions.a, *action);
+}
+
+void camu_test_ui_add_list(struct camu_test_ui *ui, struct lia_list *list)
+{
+    al_array_push(ui->lists.a, list);
+}
+
+void camu_test_ui_add_sink(struct camu_test_ui *ui, struct camu_sink *sink)
+{
+    al_array_push(ui->sinks.a, sink);
+}
+
+static void draw_list(struct ncplane *p, struct lia_list *list)
+{
+    for (u32 i = 0; i < list->entries.size; i++) {
+        struct lia_list_entry *entry = al_array_at(list->entries, i);
+        str *unique_id;
+        entry->callback(entry, LIANA_ENTRY_ID, &unique_id);
+        u32 len = AL_MIN(unique_id->len, ncplane_dim_x(p) - 2);
+        for (u32 j = 0; j < len; j++) {
+            ncplane_putchar_yx(p, i + 1, j + 1, al_str_at(unique_id, j));
+        }
+    }
+}
+
+static void print_formatted_time(f64 seconds, s32 *text, char *buffer)
+{
+    s64 minute = (s64)(seconds / 60.0);
+    s64 hour = minute / 60L;
+    minute -= hour * 60L;
+    if (hour > 0) {
+        *text += al_sprintf(buffer + *text, "%.2ld:", hour);
+    }
+    *text += al_sprintf(buffer + *text, "%.2ld:%06.3f", minute, fmod(seconds, 60.0));
+}
+
+static void draw_sink(struct camu_test_ui *ui, struct ncplane *p, struct camu_sink *sink, u32 y)
+{
+    u64 c = 0;
+    ncchannels_set_fg_rgb8(&c, 255, 255, 255);
+    ncplane_cursor_move_yx(p, y, 1);
+    ncplane_rounded_box(p, NCSTYLE_NONE, c, y + 2, ncplane_dim_x(p) - 3, 0);
+    struct camu_sink_entry *entry = camu_sink_get_current(sink);
+    if (entry) {
+        f64 pts = camu_clock_get_pts(&entry->clock, 0.0);
+        f64 duration = entry->client.duration / 1000000.0;
+        s32 text = 0;
+        ui->buffer[0] = '\0';
+        text += al_sprintf(ui->buffer + text, "(");
+        print_formatted_time(pts, &text, ui->buffer);
+        text += al_sprintf(ui->buffer + text, "/");
+        print_formatted_time(duration, &text, ui->buffer);
+        text += al_sprintf(ui->buffer + text, ")");
+        f64 percent = (duration == 0.0 || pts > duration) ? 1.0 : pts / duration;
+        s32 parts = ((ncplane_dim_x(p) - 5) - text) * percent;
+        al_memset(ui->buffer + text, '-', parts);
+        ui->buffer[text + parts] = '\0';
+        ncplane_putstr_yx(p, y + 1, 2, ui->buffer);
+    }
+    camu_sink_return_current(sink);
+}
+
+void camu_test_ui_draw(struct camu_test_ui *ui)
+{
+    u64 c = 0;
+    ncchannels_set_fg_rgb8(&c, 255, 255, 255);
+
+    // Log messages
+    struct ncplane *p = ui->messages.p;
+    ncplane_erase(p);
+    ncplane_rounded_box(p, NCSTYLE_NONE, c, ncplane_dim_y(p) - 1, ncplane_dim_x(p) - 1, 0);
+    u32 size = ui->messages.a.size;
+    u32 height = ncplane_dim_y(p) - 2;
+    u32 index = size > height ? size - height : 0;
+    for (u32 i = index; i < size; i++) {
+        char *msg = al_array_at(ui->messages.a, i);
+        u32 msg_width = AL_MIN(ncplane_dim_x(p) - 2, al_strlen(msg));
+        for (u32 j = 0; j < msg_width; j++) {
+            ncplane_putchar_yx(p, (i - index) + 1, 1 + j, msg[j]);
+        }
+    }
+
+    // Actions
+    p = ui->actions.p;
+    ncplane_erase(p);
+    ncplane_rounded_box(p, NCSTYLE_NONE, c, ncplane_dim_y(p) - 1, ncplane_dim_x(p) - 1, 0);
+    size = ui->actions.a.size;
+    height = ncplane_dim_y(p) - 2;
+    index = size > height ? size - height : 0;
+    for (u32 i = index; i < size; i++) {
+        struct camu_test_action *action = &al_array_at(ui->actions.a, i);
+        ui->buffer[0] = '\0';
+        u32 text = al_sprintf(ui->buffer, "%f", action->roll);
+        ncplane_putstr_yx(p, (i - index) + 1, 1, ui->buffer);
+        switch (action->action) {
+        case CAMU_TEST_ACTION_ADD:
+            ncplane_putstr_yx(p, (i - index) + 1, text + 2, "ADD");
+            break;
+        case CAMU_TEST_ACTION_SKIP:
+            ncplane_putstr_yx(p, (i - index) + 1, text + 2, "SKIP");
+            break;
+        case CAMU_TEST_ACTION_SEEK:
+            ncplane_putstr_yx(p, (i - index) + 1, text + 2, "SEEK");
+            break;
+        }
+    }
+
+    // Lists
+    p = ui->lists.p;
+    ncplane_erase(p);
+    ncplane_rounded_box(p, NCSTYLE_NONE, c, ncplane_dim_y(p) - 1, ncplane_dim_x(p) - 1, 0);
+    struct lia_list *list;
+    al_array_foreach(ui->lists.a, i, list) {
+        draw_list(p, list);
+    }
+
+    // Sinks
+    p = ui->sinks.p;
+    ncplane_erase(p);
+    ncplane_rounded_box(p, NCSTYLE_NONE, c, ncplane_dim_y(p) - 1, ncplane_dim_x(p) - 1, 0);
+    u32 y = 1;
+    struct camu_sink *sink;
+    al_array_foreach(ui->sinks.a, i, sink) {
+        draw_sink(ui, p, sink, y);
+        y += 3;
+    }
+
+    notcurses_refresh(ui->nc, NULL, NULL);
+    notcurses_render(ui->nc);
+}
+
+void camu_test_ui_close(struct camu_test_ui *ui)
+{
+    char *msg;
+    al_array_foreach(ui->messages.a, i, msg) {
+        al_free(msg);
+    }
+    al_array_free(ui->messages.a);
+    al_array_free(ui->actions.a);
+    al_array_free(ui->lists.a);
+    al_array_free(ui->sinks.a);
+    notcurses_stop(ui->nc);
+}
diff --git a/tests/ui.h b/tests/ui.h
new file mode 100644
index 0000000..34c7ed1
--- /dev/null
+++ b/tests/ui.h
@@ -0,0 +1,40 @@
+#pragma once
+
+#include <al/array.h>
+#include <notcurses/notcurses.h>
+
+#include "../src/liana/list.h"
+#include "../src/libsink/sink.h"
+
+#include "action.h"
+
+struct camu_test_ui {
+    struct notcurses *nc;
+    u32 term_rows;
+    u32 term_cols;
+    struct {
+        struct ncplane *p;
+        array(char *) a;
+    } messages;
+    struct {
+        struct ncplane *p;
+        array(struct camu_test_action) a;
+    } actions;
+    struct {
+        struct ncplane *p;
+        array(struct lia_list *) a;
+    } lists;
+    struct {
+        struct ncplane *p;
+        array(struct camu_sink *) a;
+    } sinks;
+    char *buffer;
+};
+
+bool camu_test_ui_init(struct camu_test_ui *ui);
+void camu_test_ui_push_message(struct camu_test_ui *ui, char *msg);
+void camu_test_ui_push_action(struct camu_test_ui *ui, struct camu_test_action *action);
+void camu_test_ui_add_list(struct camu_test_ui *ui, struct lia_list *list);
+void camu_test_ui_add_sink(struct camu_test_ui *ui, struct camu_sink *sink);
+void camu_test_ui_draw(struct camu_test_ui *ui);
+void camu_test_ui_close(struct camu_test_ui *ui);