path: root/src/liana/list.c

#define AL_LOG_SECTION "list"
//#define AL_LOG_ENABLE_TRACE
#include <al/log.h>
#include <al/lib.h>
#include <al/random.h>
#include <nnwt/time.h>

#include "list.h"
#include "list_cmp.h"

enum {
    ADD_SINK = 0,
    REMOVE_SINK,
    ADD,
    SKIPTO,
    SKIP,
    TOGGLE_PAUSE,
    SEEK,
    END,
    REVERSE,
    SORT,
    SHUFFLE,
    UNSET,
    CLEAR
};

static inline u32 get_incremental_id(struct lia_list *list)
{
    list->increment = al_u32_inc_wrap(list->increment);
    al_assert(list->increment > 0); // Wrapping not currently handled.
    return list->increment;
}

void lia_list_init(struct lia_list *list, str *name)
{
    al_str_clone(&list->name, name);
    list->current = -1;
    list->idle = true;
    list->closed = false;
    list->increment = 0;
    al_array_init(list->entries);
    al_array_init(list->sinks);
    al_array_init(list->command_queue);
    list->cmd = NULL;
}

// These small functions may seem excessive but their purpose is an attempt
// to reduce noise in parts that are harder to understand.
static inline void list_signal_meta(struct lia_list *list, struct lia_list_entry *entry, u8 meta)
{
    list->callback(list->userdata, LIANA_LIST_META, entry, (u8[]){ meta });
}

static inline void list_add_entry(struct lia_list *list, struct lia_list_entry *entry)
{
    al_array_push(list->entries, entry);
    list_signal_meta(list, entry, LIANA_META_ADDED_ENTRY);
}

static inline void entry_unload(struct lia_list *list, struct lia_list_entry *entry)
{
    list->callback(list->userdata, LIANA_UNLOAD_ENTRY, entry, NULL);
}

static inline void entry_free(struct lia_list_entry *entry)
{
    al_str_free(&entry->brief);
    al_free(entry);
}

static inline void entry_ref(struct lia_list *list, struct lia_list_entry *entry)
{
    list->callback(list->userdata, LIANA_REF_ENTRY, entry, NULL);
}

static inline void entry_unref(struct lia_list *list, struct lia_list_entry *entry)
{
    list->callback(list->userdata, LIANA_UNREF_ENTRY, entry, NULL);
}

static void unref_all_entries(struct lia_list *list, s32 trigger)
{
    struct lia_list_entry *entry;
    s32 sequence;
    al_array_foreach(list->entries, i, entry) {
        sequence = (s32)i;
        if (sequence != list->current && sequence != trigger) {
            entry_unref(list, entry);
        }
    }
}

// @TODO: When implementing list_remove(), consider this case.
//  remove() sequence 0 causes a skip to sequence 1 and sequence 1 fails to load.
//  Do we properly move to list->current = -1 and an idle list?
static bool entry_load_and_get_duration(struct lia_list *list, struct lia_list_entry *entry, s32 sequence, bool *error)
{
    u8 last_load = entry->load;
    list->callback(list->userdata, LIANA_LOAD_ENTRY, entry, &entry->load);
    if (entry->load == LIANA_ENTRY_ERRORED) {
        // If sequence is <0 that must mean entry is not yet added to list->entries.
        if (sequence >= 0) {
            al_array_remove_at(list->entries, (u32)sequence);
            if (list->current > sequence) {
                struct lia_list_sink *sink;
                al_array_foreach(list->sinks, i, sink) {
                    al_assert(sink->set == list->current);
                    sink->set--;
                }
                list->current--;
            }
            struct lia_list_cmd *cmd = list->cmd;
            al_assert(cmd);
            al_assert(cmd->sequence != sequence);
            if (cmd->sequence > sequence) {
                cmd->sequence--;
            }
            al_array_foreach(list->command_queue, i, cmd) {
                if (cmd->sequence > sequence) {
                    cmd->sequence--;
                }
            }
        }
        *error = true;
        list_signal_meta(list, entry, LIANA_META_ENTRY_ERRORED);
        al_assert(!al_array_contains(list->entries, entry));
        entry_unload(list, entry);
        entry_free(entry);
        return false;
    }
    *error = false;
    if (entry->load == LIANA_ENTRY_LOADED) {
        if (last_load != LIANA_ENTRY_LOADED) { // Newly loaded entry.
            unref_all_entries(list, sequence);
        }
        list->callback(list->userdata, LIANA_GET_ENTRY_DURATION, entry, &entry->duration);
        return true;
    }
    return false;
}

static void unload_all_entires(struct lia_list *list)
{
    struct lia_list_cmd *cmd = list->cmd;
    if (cmd && cmd->entry) {
        al_assert(cmd->op == ADD);
        entry_unload(list, cmd->entry);
        list->cmd = NULL;
    }
    al_array_foreach_rev(list->command_queue, i, cmd) {
        if (cmd->entry) {
            al_assert(cmd->op == ADD);
            entry_unload(list, cmd->entry);
            al_array_remove_at(list->command_queue, i);
        }
    }
    struct lia_list_entry *entry;
    al_array_foreach(list->entries, i, entry) {
        entry_unload(list, entry);
    }
}

static inline void sink_set_entry(struct lia_list_sink *sink, struct lia_list_entry *entry, s32 sequence, struct lia_timing *time)
{
    sink->callback(sink->userdata, LIANA_SINK_SET, entry, sequence, time);
}

static inline void sink_seek_entry(struct lia_list_sink *sink, struct lia_list_entry *entry, s32 sequence, struct lia_timing *time)
{
    sink->callback(sink->userdata, LIANA_SINK_SEEK, entry, sequence, time);
}

static inline void sink_pause_entry(struct lia_list_sink *sink, struct lia_list_entry *entry, s32 sequence, struct lia_timing *time)
{
    sink->callback(sink->userdata, LIANA_SINK_PAUSE, entry, sequence, time);
}

// @TODO: This should probably give a sequence or ID.
static inline void sink_unset_entry(struct lia_list_sink *sink)
{
    sink->callback(sink->userdata, LIANA_SINK_UNSET, NULL, -1, NULL);
}

static bool list_set_current(struct lia_list *list, struct lia_list_entry *entry, s32 sequence)
{
    list->idle = false;
    entry_ref(list, entry);
    al_assert(list->current != sequence);
    list->current = sequence;
    list_signal_meta(list, entry, LIANA_META_CURRENT_CHANGED);
    return true;
}

static void pump_queue(struct lia_list *list);

static bool handle_add_sink(struct lia_list *list, struct lia_list_sink *sink)
{
    // The list being idle is not equivalent to current = -1.
    if (list->current >= 0) {
        struct lia_list_entry *current = al_array_at(list->entries, list->current);
        u8 pause;
        u64 at = LIANA_TIMESTAMP_INVALID;
        u64 pos = current->offset;
        if (current->paused_at != LIANA_TIMESTAMP_INVALID) {
            pause = LIANA_PAUSE_NONE;
        } else {
            at = nn_get_timestamp() + LIANA_BASE_DELAY;
            if (at > current->start && at - current->start > LIANA_BASE_DELAY) {
                pos += at - current->start;
            } else {
                at = current->start;
            }
            // This sink could have an entry set from a connection we no longer
            // know about. In that case skipping to this entry with a pause_and_swap_to()
            // would be better. If this sink is empty that's still okay because
            // PAUSE_BOTH is required to handle that case sink-side.
            pause = LIANA_PAUSE_BOTH;
        }
        struct lia_timing time = {
            .at = at,
            .pos = pos,
            .pause = pause
        };
        sink->set = list->current;
        sink_set_entry(sink, current, list->current, &time);
    }
    al_array_push(list->sinks, sink);
    return true;
}


static void handle_remove_sink(struct lia_list *list, void *userdata)
{
    struct lia_list_cmd *cmd = list->cmd;
    // handle_remove_sink() runs immediately, so if there's an ADD_SINK
    // queued for this sink, remove it. It should only ever be possible to
    // have one queued ADD_SINK for each sink.
    if (cmd && cmd->op == ADD_SINK && cmd->sink->userdata == userdata) {
        al_free(cmd->sink);
        al_free(cmd);
        list->cmd = NULL;
        return;
    }
    al_array_foreach(list->command_queue, i, cmd) {
        if (cmd->op == ADD_SINK && cmd->sink->userdata == userdata) {
            al_free(cmd->sink);
            al_free(cmd);
            al_array_remove_at(list->command_queue, i);
            return;
        }
    }
    struct lia_list_sink *sink;
    al_array_foreach(list->sinks, i, sink) {
        if (sink->userdata == userdata) {
            al_array_remove_at(list->sinks, i);
            al_free(sink);
            break;
        }
    }
    if (list->closed && !list->sinks.count) {
        unload_all_entires(list);
    }
}

static struct lia_list_entry *get_entry_from_sequence(struct lia_list *list, s32 sequence)
{
    s32 size = (s32)list->entries.count;
    if (sequence < 0 || sequence >= size) {
        return NULL;
    }
    return al_array_at(list->entries, sequence);
}

static s32 get_sequence_from_entry_id(struct lia_list *list, u32 id)
{
    // Not returning the entry pointer here is a meaningful simplification.
    struct lia_list_entry *entry;
    al_array_foreach(list->entries, i, entry) {
        if (entry->id == id) return i;
    }
    return -1;
}

static u8 skipto_entry(struct lia_list_entry *current, struct lia_list_entry *target, u64 at)
{
    al_assert(at != LIANA_TIMESTAMP_INVALID);

    if (target->held) { // Resume target if it's held.
        al_assert(target->paused_at != LIANA_TIMESTAMP_INVALID);
        target->paused_at = LIANA_TIMESTAMP_INVALID;
        target->start = at;
        target->held = false;
    } else if (target->start == LIANA_TIMESTAMP_INVALID && target->paused_at == LIANA_TIMESTAMP_INVALID) {
        // Target was never started before.
        target->start = at;
    } else {
        // Target is static or paused. This is strict but should hold true.
        al_assert(target->duration == 0 || target->paused_at != LIANA_TIMESTAMP_INVALID);
    }

    u8 pause;

    // If current->duration = LIANA_TIMESTAMP_INVALID, handling of a static entry happens on the sink.
    if (current->duration == 0 || current->paused_at != LIANA_TIMESTAMP_INVALID) {
        if (target->duration == 0 || target->paused_at != LIANA_TIMESTAMP_INVALID) {
            // Swap entries and ignore their clocks.
            pause = LIANA_PAUSE_NONE;
        } else {
            // Swap entries and resume target.
            al_assert(target->start != LIANA_TIMESTAMP_INVALID);
            pause = LIANA_PAUSE_RESUME;
        }
    } else { // Current is playing.
        if (target->duration == 0 || target->paused_at != LIANA_TIMESTAMP_INVALID) {
            // Pause-swap current, don't touch target's clock.
            pause = LIANA_PAUSE_PAUSE;
        } else {
            // Pause-swap current and resume target.
            al_assert(target->start != LIANA_TIMESTAMP_INVALID);
            pause = LIANA_PAUSE_BOTH;
        }
    }

    // Pause current to be resumed if it becomes the target of a skip (hold).
    if (current->duration != 0 && current->paused_at == LIANA_TIMESTAMP_INVALID) {
        al_assert(current->start != LIANA_TIMESTAMP_INVALID);
        current->paused_at = at;
        if (current->paused_at < current->start) {
            current->paused_at = current->start;
        }
        current->offset += current->paused_at - current->start;
        current->start = LIANA_TIMESTAMP_INVALID;
        current->held = true;
    }

    return pause;
}

static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index)
{
    if (sequence == LIANA_SEQUENCE_ANY) {
        sequence = list->current;
    }
    if (sequence < 0) return true; // list->current = -1
    if (sequence == index) return true;
    if (sequence != list->current) {
        log_warn("Discarding out of date skip().");
        return true;
    }

    struct lia_list_entry *current = get_entry_from_sequence(list, sequence);
    struct lia_list_entry *target = get_entry_from_sequence(list, index);
    al_assert(current && !current->held && current != target);
    if (!target) return true;
    bool error;
    if (!entry_load_and_get_duration(list, target, index, &error)) {
        if (error) {
            // On an error, sequence will address the same entry but index might not.
            // entry_load_and_get_duration() may also adjust the current cmd's sequence,
            // so use cmd->sequence here.
            struct lia_list_cmd *cmd = list->cmd;
            return handle_skipto(list, cmd->sequence, index);
        }
        return false;
    }

    u64 at = nn_get_timestamp() + LIANA_BASE_PING;
    u64 pos = target->offset;
    u8 pause = skipto_entry(current, target, at);

    log_trace("skipto(#%u-#%u): pause: %s, held: %s.", current->id, target->id,
        lia_pause_op_name(pause), BOOLSTR(current->held));

    struct lia_timing time = {
        .at = at,
        .pos = pos,
        .pause = pause
    };

    struct lia_list_sink *sink;
    al_array_foreach(list->sinks, i, sink) {
        al_assert(sink->set != index);
        sink->set = index;
        sink_set_entry(sink, target, index, &time);
    }

    return list_set_current(list, target, index);
}

static bool handle_skip(struct lia_list *list, s32 sequence, s32 n)
{
    if (sequence == LIANA_SEQUENCE_ANY) sequence = list->current;
    if (sequence < 0) return true; // list->current = -1
    return handle_skipto(list, sequence, sequence + n);
}

static bool handle_add(struct lia_list *list, struct lia_list_entry *entry)
{
    if (list->idle) {
        bool error;
        if (!entry_load_and_get_duration(list, entry, -1, &error)) {
            // We gave a sequence of -1 so, on error, don't add to list->entries.
            return error;
        }
    }
    list_add_entry(list, entry);
    if (list->idle) {
        if (list->current == -1) { // Start the list.
            entry->start = nn_get_timestamp() + LIANA_BASE_DELAY;
            struct lia_timing time = {
                .at = entry->start,
                .pos = entry->offset,
                .pause = LIANA_PAUSE_RESUME
            };
            struct lia_list_sink *sink;
            al_array_foreach(list->sinks, i, sink) {
                al_assert(sink->set == -1);
                sink->set = 0;
                sink_set_entry(sink, entry, 0, &time);
            }
            return list_set_current(list, entry, 0);
        } else { // Skip to the added entry.
            // This is done via SKIPTO for consistency. Ended entries must still be held.
            struct lia_list_cmd *cmd = list->cmd;
            cmd->op = SKIPTO;
            cmd->sequence = list->current;
            cmd->arg0.i = list->current + 1;
            return handle_skipto(list, cmd->sequence, cmd->arg0.i);
        }
    }
    al_assert(list->current != -1);
    return true;
}

static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts)
{
    if (sequence == LIANA_SEQUENCE_ANY) {
        sequence = list->current;
    }
    if (sequence < 0) return;
    if (sequence != list->current) {
        log_warn("Discarding out of date toggle_pause().");
        return;
    }

    struct lia_list_entry *entry = get_entry_from_sequence(list, sequence);
    if (!entry || entry->ended || entry->duration == 0) return;
    al_assert(!entry->held);

    u64 now = nn_get_timestamp();
    u8 pause = (entry->paused_at == LIANA_TIMESTAMP_INVALID) ? LIANA_PAUSE_PAUSE : LIANA_PAUSE_RESUME;
    u64 at;

    switch (pause) {
    case LIANA_PAUSE_PAUSE: {
        al_assert(entry->start != LIANA_TIMESTAMP_INVALID);
        entry->paused_at = now + LIANA_PAUSE_DELAY;
        if (entry->paused_at < entry->start) {
            entry->paused_at = entry->start;
            // @TODO: Delay would need to apply to newly added sinks and skipto().
            //entry->delay = entry->start - entry->paused_at;
        } else {
            entry->offset += entry->paused_at - entry->start;
        }
        entry->start = LIANA_TIMESTAMP_INVALID;
        at = entry->paused_at;
        break;
    }
    case LIANA_PAUSE_RESUME: {
        al_assert(entry->start == LIANA_TIMESTAMP_INVALID);
        if (entry->delay != LIANA_TIMESTAMP_INVALID) {
            entry->start = now + MAX(entry->delay, LIANA_PAUSE_DELAY);
            entry->delay = LIANA_TIMESTAMP_INVALID;
        } else {
            entry->start = now + LIANA_PAUSE_DELAY;
        }
        entry->paused_at = LIANA_TIMESTAMP_INVALID;
        at = entry->start;
        break;
    }
    }

    log_trace("toggle_pause(#%u): pts: %f, pause: %hhu.", entry->id, pts, pause);

    struct lia_timing time = {
        .at = at,
        .pos = entry->offset,
        .pause = pause
    };

    struct lia_list_sink *sink;
    al_array_foreach(list->sinks, i, sink) {
        if (sequence == list->current && sink->set != sequence) {
            sink->set = sequence;
            sink_set_entry(sink, entry, sequence, &time);
        } else {
            sink_pause_entry(sink, entry, sequence, &time);
        }
    }

    list_signal_meta(list, entry, LIANA_META_ENTRY_PAUSED);
}

static void handle_seek(struct lia_list *list, s32 sequence, u32 id, u64 pos)
{
    if (sequence == LIANA_SEQUENCE_ANY) {
        sequence = list->current;
    } else {
        sequence = get_sequence_from_entry_id(list, id);
    }
    if (sequence < 0) return;

    struct lia_list_entry *entry = get_entry_from_sequence(list, sequence);
    al_assert(entry);

    if (entry->duration == 0 || entry->duration == LIANA_TIMESTAMP_INVALID) {
        log_warn("Skipping seek on entry with unknown or 0 duration.");
        return;
    }

    pos = CLAMP(pos, (u64)0, entry->duration);
    u64 at = nn_get_timestamp() + LIANA_BASE_DELAY;
    u8 pause;
    if (entry->paused_at == LIANA_TIMESTAMP_INVALID) {
        entry->start = at;
        pause = LIANA_PAUSE_RESUME;
    } else {
        pause = LIANA_PAUSE_NONE;
    }

    entry->ended = false;
    entry->offset = pos;
    entry->reset_token = get_incremental_id(list);

    log_trace("seek(#%u): pos: %f, pause: %hhu.", entry->id, pos / 1000000.0, pause);

    list->idle = false;

    struct lia_timing time = {
        .at = at,
        .pos = pos,
        .pause = pause
    };

    struct lia_list_sink *sink;
    al_array_foreach(list->sinks, i, sink) {
        if (sequence == list->current && sink->set != sequence) {
            sink->set = sequence;
            sink_set_entry(sink, entry, sequence, &time);
        } else {
            sink_seek_entry(sink, entry, sequence, &time);
        }
    }

    list_signal_meta(list, entry, LIANA_META_ENTRY_SEEKED);
}

static bool handle_end(struct lia_list *list, u32 id, u32 reset_token)
{
    s32 sequence = get_sequence_from_entry_id(list, id);
    if (sequence < 0) return true;

    struct lia_list_entry *entry = get_entry_from_sequence(list, sequence);

    if (reset_token != entry->reset_token) {
        log_warn("Got end() with out of order or incorrect reset id, ignoring.");
        return true;
    }

    if (entry->ended) {
        log_warn("Got end() from an already ended resource, ignoring.");
        return true;
    }

    log_trace("end(#%u).", entry->id);

    entry->ended = true;
    entry->offset = entry->duration;

#ifdef LIANA_LIST_SCUFFED_LOOP
    struct lia_list_cmd *cmd = list->cmd;
    cmd->op = SEEK;
    cmd->sequence = sequence;
    cmd->arg0.u = id;
    cmd->arg1.l = 0;
    handle_seek(list, cmd->sequence, cmd->arg0.u, cmd->arg1.l);
    return true;
#endif

    s32 size = (s32)list->entries.count;
    if (sequence == list->current) {
        s32 next = sequence + 1;
        if (next < size) {
            struct lia_list_cmd *cmd = list->cmd;
            cmd->op = SKIPTO;
            cmd->sequence = sequence;
            cmd->arg0.i = next;
            return handle_skipto(list, cmd->sequence, cmd->arg0.i);
        } else {
            list->idle = true;
        }
    }

    return true;
}

static bool adjust_for_order_change(struct lia_list *list, struct lia_list_entry *previous)
{
    al_assert(list->current >= 0);
    list_signal_meta(list, previous, LIANA_META_ORDER_PROBABLY_CHANGED);
    struct lia_list_cmd *cmd = list->cmd;
    struct lia_list_entry *entry = NULL;
    al_array_foreach(list->entries, i, entry) {
        if (entry->opaque == previous->opaque) {
            if (i == (u32)list->current) {
                return true;
            }
            cmd->op = SKIPTO;
            cmd->sequence = i;
            cmd->arg0.i = list->current;
            struct lia_list_sink *sink;
            al_array_foreach(list->sinks, j, sink) {
                al_assert(sink->set == list->current);
                sink->set = i;
            }
            list->current = i;
            break;
        }
    }
    al_assert(cmd->op == SKIPTO);
    // This SKIPTO being in-place is a consideration for the sequence of any
    // commands that were queued after this one.
    return handle_skipto(list, cmd->sequence, cmd->arg0.i);
}

static bool handle_reverse(struct lia_list *list)
{
    if (list->current < 0) return true;
    struct lia_list_entry *previous = al_array_at(list->entries, list->current);
    u32 size = list->entries.count;
    for (u32 i = 0; i < size; i++) {
        u32 tail = size - (i + 1);
        if (tail <= i) break;
        SWAP(al_array_at(list->entries, i), al_array_at(list->entries, tail));
    }
    log_trace("reverse()");
    return adjust_for_order_change(list, previous);
}

static bool handle_sort(struct lia_list *list)
{
    if (list->current < 0) return true;
    struct lia_list_entry *previous = al_array_at(list->entries, list->current);
    al_array_sort(list->entries, struct lia_list_entry *, camu_db_compare);
    log_trace("sort()");
    return adjust_for_order_change(list, previous);
}

static bool handle_shuffle(struct lia_list *list)
{
    u32 size = list->entries.count;
    if (list->current < 0 || size < 2) return true;
    struct lia_list_entry *previous = al_array_at(list->entries, list->current);
    /* https://en.wikipedia.org/wiki/Fisher%E2%80%93Yates_shuffle
     for i from 0 to n−2 do
         j ← random integer such that i ≤ j ≤ n-1
         exchange a[i] and a[j]
    */
    if (size == 2) {
        if (al_random_int(0, 1) == 0) {
            SWAP(al_array_at(list->entries, 0), al_array_at(list->entries, 1));
        }
    } else {
        for (u32 i = 0; i < size - 2; i++) {
            u32 j = al_random_int(i, size - 1);
            SWAP(al_array_at(list->entries, i), al_array_at(list->entries, j));
        }
    }
    log_trace("shuffle()");
    return adjust_for_order_change(list, previous);
}

static void unset_current(struct lia_list *list)
{
    struct lia_list_sink *sink;
    al_array_foreach(list->sinks, i, sink) {
        if (sink->set >= 0) {
            sink_unset_entry(sink);
        }
        sink->set = -1;
    }
    list->idle = true;
}

static void handle_unset(struct lia_list *list)
{
    unset_current(list);
}

static void handle_clear(struct lia_list *list)
{
    unset_current(list);
    list->current = -1;
    unload_all_entires(list);
    struct lia_list_entry *entry;
    al_array_foreach(list->entries, i, entry) {
        entry_free(entry);
    }
    list->entries.count = 0;
}

static void run_queue(struct lia_list *list)
{
    if (!list->cmd) {
        if (list->command_queue.count) {
            al_array_pop_at(list->command_queue, 0, list->cmd);
        } else {
            return;
        }
    }
    struct lia_list_cmd *cmd = list->cmd;
    switch (cmd->op) {
    case ADD_SINK:
        if (!handle_add_sink(list, cmd->sink)) {
            return;
        }
        break;
    case REMOVE_SINK:
        handle_remove_sink(list, cmd->userdata);
        break;
    case ADD:
        if (!handle_add(list, cmd->entry)) {
            return;
        }
        break;
    // Return on SKIPTO/SKIP: Target entry not loaded.
    case SKIPTO:
        if (!handle_skipto(list, cmd->sequence, cmd->arg0.i)) {
            return;
        }
        break;
    case SKIP:
        if (!handle_skip(list, cmd->sequence, cmd->arg0.i)) {
            return;
        }
        break;
    case TOGGLE_PAUSE:
        handle_toggle_pause(list, cmd->sequence, cmd->arg1.f);
        break;
    case SEEK:
        handle_seek(list, cmd->sequence, cmd->arg0.u, cmd->arg1.l);
        break;
    case END:
        if (!handle_end(list, cmd->arg0.u, cmd->arg1.u)) {
            // Converted to a SKIP and target entry not loaded.
            return;
        }
        break;
    // Return on order change: Converted to SKIPTO and new current not loaded.
    case REVERSE:
        if (!handle_reverse(list)) {
            return;
        }
        break;
    case SORT:
        if (!handle_sort(list)) {
            return;
        }
        break;
    case SHUFFLE:
        if (!handle_shuffle(list)) {
            return;
        }
        break;
    case UNSET:
        handle_unset(list);
        break;
    case CLEAR:
        handle_clear(list);
        break;
    }
    al_free(cmd);
    list->cmd = NULL;
    pump_queue(list);
}

void pump_queue(struct lia_list *list)
{
    run_queue(list);
}

void lia_list_pump(struct lia_list *list)
{
    pump_queue(list);
}

void lia_list_add_sink(struct lia_list *list, void (*callback)(void *, u8, struct lia_list_entry *, s32, struct lia_timing *), void *userdata)
{
    struct lia_list_sink *sink = al_alloc_object(struct lia_list_sink);
    sink->set = -1;
    sink->callback = callback;
    sink->userdata = userdata;
    struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd);
    cmd->op = ADD_SINK;
    cmd->sink = sink;
    al_array_push(list->command_queue, cmd);
    pump_queue(list);
}

void lia_list_remove_sink(struct lia_list *list, void *userdata)
{
    // Don't queue remove sink because we can't let any currently queued
    // commands touch this sink.
    handle_remove_sink(list, userdata);
}

void lia_list_add(struct lia_list *list, str *brief, void *opaque, u64 duration, u8 load)
{
    struct lia_list_entry *entry = al_alloc_object(struct lia_list_entry);
    entry->opaque = opaque;
    al_str_clone(&entry->brief, brief);
    entry->id = get_incremental_id(list);
    entry->start = LIANA_TIMESTAMP_INVALID;
    entry->paused_at = LIANA_TIMESTAMP_INVALID;
    entry->offset = 0;
    entry->delay = LIANA_TIMESTAMP_INVALID;
    entry->duration = duration;
    entry->load = load;
    entry->held = false;
    entry->ended = false;
    entry->reset_token = get_incremental_id(list);
    entry->list = list;
    struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd);
    cmd->op = ADD;
    cmd->entry = entry;
    al_array_push(list->command_queue, cmd);
    pump_queue(list);
}

void lia_list_skipto(struct lia_list *list, s32 sequence, s32 index)
{
    if (sequence == index || index < 0) return;
    struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd);
    cmd->op = SKIPTO;
    cmd->sequence = sequence;
    cmd->arg0.i = index;
    al_array_push(list->command_queue, cmd);
    pump_queue(list);
}

void lia_list_skip(struct lia_list *list, s32 sequence, s32 n)
{
    if (n == 0) return;
    struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd);
    cmd->op = SKIP;
    cmd->sequence = sequence;
    cmd->arg0.i = n;
    al_array_push(list->command_queue, cmd);
    pump_queue(list);
}

void lia_list_toggle_pause(struct lia_list *list, s32 sequence, f64 pts)
{
    struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd);
    cmd->op = TOGGLE_PAUSE;
    cmd->sequence = sequence;
    cmd->arg1.f = pts;
    al_array_push(list->command_queue, cmd);
    pump_queue(list);
}

void lia_list_seek(struct lia_list *list, s32 sequence, u32 id, u64 pos)
{
    struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd);
    cmd->op = SEEK;
    cmd->sequence = sequence;
    cmd->arg0.u = id;
    cmd->arg1.l = pos;
    al_array_push(list->command_queue, cmd);
    pump_queue(list);
}

void lia_list_end(struct lia_list *list, u32 id, u32 reset_token)
{
    struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd);
    cmd->op = END;
    cmd->arg0.u = id;
    cmd->arg1.u = reset_token;
    al_array_push(list->command_queue, cmd);
    pump_queue(list);
}

void lia_list_reverse(struct lia_list *list)
{
    struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd);
    cmd->op = REVERSE;
    al_array_push(list->command_queue, cmd);
    pump_queue(list);
}

void lia_list_sort(struct lia_list *list)
{
    struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd);
    cmd->op = SORT;
    al_array_push(list->command_queue, cmd);
    pump_queue(list);
}

void lia_list_shuffle(struct lia_list *list)
{
    struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd);
    cmd->op = SHUFFLE;
    al_array_push(list->command_queue, cmd);
    pump_queue(list);
}

void lia_list_unset(struct lia_list *list)
{
    struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd);
    cmd->op = UNSET;
    al_array_push(list->command_queue, cmd);
    pump_queue(list);
}

void lia_list_clear(struct lia_list *list)
{
    struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd);
    cmd->op = CLEAR;
    al_array_push(list->command_queue, cmd);
    pump_queue(list);
}

void lia_list_close(struct lia_list *list)
{
    list->closed = true;
    if (!list->sinks.count) {
        unload_all_entires(list);
    }
}

void lia_list_free(struct lia_list *list)
{
    struct lia_list_cmd *cmd;
    al_array_foreach(list->command_queue, i, cmd) {
        al_free(cmd);
    }
    al_array_free(list->command_queue);
    if (list->cmd) al_free(list->cmd);
    struct lia_list_entry *entry;
    al_array_foreach(list->entries, i, entry) {
        entry_free(entry);
    }
    al_array_free(list->entries);
    struct lia_list_sink *sink;
    al_array_foreach(list->sinks, i, sink) {
        al_free(sink);
    }
    al_array_free(list->sinks);
    al_str_free(&list->name);
}