#define AL_LOG_SECTION "sink"
//#define AL_LOG_ENABLE_TRACE
#include <al/log.h>
#include <al/random.h>
#include <nnwt/multiplex.h>

#include "../server/common.h"
#include "../buffer/common.h"
#include "../liana/list.h"

#include "sink.h"
#include "common.h"

//#define CAMU_SINK_ONESHOT

// Requested state of the sinks outputs.
enum {
    SINK_PAUSED = 0,
    SINK_PLAYING
};

enum {
    // Created.
    BUFFER_INIT = 0,
    // Set but not configured.
    BUFFER_QUEUED,
    // Errored buffer was removed, treat it as empty from now on.
    BUFFER_DETACHED,
    // Ready to receive data.
    BUFFER_CONFIGURED,
    // The next call to add can add the buffer.
    BUFFER_SET_OR_BUFFERED,
    // Treat the buffer like it's added, even if it might not be.
    BUFFER_ADDED,
    // Effectively SET_OR_BUFFERED but not addable until after a reset.
    BUFFER_ENDED,
    // Same as BUFFER_ENDED but will be detached on a reset.
    BUFFER_ERRORED
};

// Command queue commands.
enum {
    // Sink operations.
    START = 0,
    STOP,
    ADD_BUFFER,
    REMOVE_BUFFER,
    CLEAR_BUFFERS,
    EJECT_ENTRY,
    CLOSE,
    // List actions.
    SKIP,
    TOGGLE_PAUSE,
    SEEK,
    RESEEK,
    SHUFFLE,
    END
};

// Number of entries to keep buffered at one time.
#define ENTRY_MAX_AGE 4
#define SINK_LRU_MAX UINT16_MAX
AL_STATIC_ASSERT(max_age_lt_lru, ENTRY_MAX_AGE, <, SINK_LRU_MAX);

// Store connection number in the upper 16 bits so IDs don't conflict after a server restart.
// If the sink disconnects but the server didn't restart, this will invalidate IDs that _do_ map
// to the same resource. That could be wasteful. It's also currently (ab)used by reseek().
#define LOCAL_ENTRY_ID(sink, id) (((u64)(sink)->connection_number) << 48 | (u64)(id)) // id is a u32.
#define REMOTE_ENTRY_ID(id) ((u32)(id & 0x7fffffff))
#define CONNECTION_NUMBER(id) ((id >> 48) & 0xffff)

// Only sink->target can ever be 0xb00b.
#define ENTRY_IS_VALID(entry) ((entry) && (entry) != (struct camu_sink_entry *)0xb00b)

// printf format for entries.
#define ENTRY_FMT "#%u(%p)"
#define ENTRY_ARG(entry) (ENTRY_IS_VALID(entry) ? REMOTE_ENTRY_ID((entry)->id) : 0), (entry)

#define AUDIO_STATE(entry) ((entry)->audio.state)
#define VIDEO_STATE(entry) ((entry)->video.state)

// If a buffer is still INIT or QUEUED after the entry is configured, it's "empty".
// An ERRORED buffer will be DETACHED after CLIENT_REMOVE_BUFFERS and is then considered empty.
// Empty is a state that cannot change while a buffer could be in use (push()/read()).
#define AUDIO_EMPTY(entry) (AUDIO_STATE(entry) <= BUFFER_DETACHED)
#define VIDEO_EMPTY(entry) (VIDEO_STATE(entry) <= BUFFER_DETACHED)

#define AUDIO_ENDED(entry) (AUDIO_STATE(entry) >= BUFFER_ENDED)
#define VIDEO_ENDED(entry) (VIDEO_STATE(entry) >= BUFFER_ENDED)

#define AUDIO_STREAM(entry) ((al_assert(!AUDIO_EMPTY(entry)), (entry)->audio.buf.stream))
#define VIDEO_STREAM(entry) ((al_assert(!VIDEO_EMPTY(entry)), (entry)->video.buf.stream))
#define VIDEO_IS_STATIC(entry) ((al_assert(!VIDEO_EMPTY(entry)), (entry)->video.buf.is_static))

#define ENTRY_EVAL_ENDED(entry) \
    ((AUDIO_ENDED(entry) && (VIDEO_ENDED(entry) || VIDEO_EMPTY(entry) || VIDEO_IS_STATIC(entry))) || \
     (AUDIO_EMPTY(entry) && VIDEO_ENDED(entry)))
// This assert would fail if using ENTRY_ENDED() in end_entry_and_advance_queue() before setting entry->ended.
#define ENTRY_ENDED(entry) (al_assert(entry->ended == ENTRY_EVAL_ENDED(entry)), entry->ended)

// IGNORED = ENDED or EMPTY.
#define AUDIO_ADDED_OR_IGNORED(entry) (AUDIO_STATE(entry) >= BUFFER_ADDED || AUDIO_EMPTY(entry))
#define VIDEO_ADDED_OR_IGNORED(entry) (VIDEO_STATE(entry) >= BUFFER_ADDED || VIDEO_EMPTY(entry))

#define AUDIO_ENDED_OR_EMPTY(entry) (AUDIO_ENDED(entry) || AUDIO_EMPTY(entry))
#define VIDEO_ENDED_OR_EMPTY(entry) (VIDEO_ENDED(entry) || VIDEO_EMPTY(entry))

#if defined CAMU_SCREEN_THREADED && defined CAMU_MIXER_THREADED
#define BLOCKING_SLEEP(sink, delay) ((void)sink, nn_thread_sleep(delay))
#else
#define BLOCKING_SLEEP(sink, delay) nn_event_loop_sleep(sink->loop, delay)
#endif

#define CMD(cmd, ...) ((struct camu_sink_cmd){ .op = cmd, __VA_ARGS__ })

// Functions that might happen on separate threads.
// CAMU_MIXER_THREADED:
//  audio_buffer_callback()
//  mixer_callback()
// CAMU_SCREEN_THREADED:
//  video_buffer_callback()
// CAMU_MIXER_THREADED or CAMU_SCREEN_THREADED:
//  clock_callback()
// client_callback()::LIANA_CLIENT_DATA
// client_callback()::LIANA_CLIENT_EOF/ERRORED

static inline bool entry_audio_buffer_held(struct camu_sink_entry *entry)
{
#ifdef CAMU_MIXER_THREADED
    return atomic_load(bool)(&entry->audio.buf.ref, AL_ATOMIC_RELAXED);
#else
    (void)entry;
    return false;
#endif
}

static inline bool entry_video_buffer_held(struct camu_sink_entry *entry)
{
#ifdef CAMU_SCREEN_THREADED
    return atomic_load(bool)(&entry->video.buf.ref, AL_ATOMIC_RELAXED);
#else
    (void)entry;
    return false;
#endif
}

static void queue_cmds(struct camu_sink *sink, u32 count, ...)
{
    camu_queue_lock(sink->queue);
    va_list cmds;
    va_start(cmds, count);
    for (u32 i = 0; i < count; i++) {
        camu_queue_push(sink->queue, va_arg(cmds, struct camu_sink_cmd));
    }
    camu_queue_unlock(sink->queue);
    nn_signal_send(&sink->queue_signal);
}

static void queue_cmd(struct camu_sink *sink, struct camu_sink_cmd cmd)
{
    queue_cmds(sink, 1, cmd);
}

static void refresh_video_output(struct camu_sink *sink)
{
#ifndef CAMU_SINK_NO_VIDEO
    sink->callback(sink->userdata, CAMU_SINK_REFRESH_VIDEO, 0, NULL);
#else
    (void)sink;
#endif
}

static inline void add_entry_audio_buffer(struct camu_sink_entry *entry)
{
    log_trace("add_entry_audio_buffer("ENTRY_FMT").", ENTRY_ARG(entry));
    struct camu_sink *sink = entry->sink;
#ifdef CAMU_MIXER_THREADED_START_STOP
    sink->callback(sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
#else
    queue_cmd(sink, CMD(ADD_BUFFER, .v.u = CAMU_SINK_AUDIO, .opaque = entry));
#endif
}

static inline void add_entry_video_buffer(struct camu_sink_entry *entry)
{
#ifndef CAMU_SINK_NO_VIDEO
    log_trace("add_entry_video_buffer("ENTRY_FMT").", ENTRY_ARG(entry));
    struct camu_sink *sink = entry->sink;
    sink->callback(sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
#else
    (void)entry;
#endif
}

static void remove_entry_audio_buffer(struct camu_sink_entry *entry)
{
    al_assert(!AUDIO_ENDED(entry));
    al_assert(AUDIO_STATE(entry) != BUFFER_INIT);
    log_trace("remove_entry_audio_buffer("ENTRY_FMT"), do_remove: %s.", ENTRY_ARG(entry),
        BOOLSTR(AUDIO_STATE(entry) == BUFFER_ADDED));
    switch (AUDIO_STATE(entry)) {
    case BUFFER_ADDED:
        AUDIO_STATE(entry) = BUFFER_SET_OR_BUFFERED;
        struct camu_sink *sink = entry->sink;
#ifdef CAMU_MIXER_THREADED_START_STOP
        sink->callback(sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
#else
        queue_cmd(sink, CMD(REMOVE_BUFFER, .v.u = CAMU_SINK_AUDIO, .opaque = entry));
#endif
        break;
    case BUFFER_SET_OR_BUFFERED:
        AUDIO_STATE(entry) = BUFFER_CONFIGURED;
        break;
    case BUFFER_QUEUED:
        AUDIO_STATE(entry) = BUFFER_INIT;
        break;
    }
}

static void remove_entry_video_buffer(struct camu_sink_entry *entry)
{
    // Don't assert !entry->ended here because of static video handling.
    al_assert(!VIDEO_ENDED(entry));
    al_assert(VIDEO_STATE(entry) != BUFFER_INIT);
    log_trace("remove_entry_video_buffer("ENTRY_FMT"), do_remove: %s.", ENTRY_ARG(entry),
        BOOLSTR(VIDEO_STATE(entry) == BUFFER_ADDED));
    switch (VIDEO_STATE(entry)) {
    case BUFFER_ADDED:
        VIDEO_STATE(entry) = BUFFER_SET_OR_BUFFERED;
#ifndef CAMU_SINK_NO_VIDEO
        struct camu_sink *sink = entry->sink;
        sink->callback(sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
#endif
        break;
    case BUFFER_SET_OR_BUFFERED:
        VIDEO_STATE(entry) = BUFFER_CONFIGURED;
        break;
    case BUFFER_QUEUED:
        // This can be hit when skipping through entries very fast.
        VIDEO_STATE(entry) = BUFFER_INIT;
        break;
    }
}

// It's possible for some of an entry's buffers to be ENDED while others are still ADDED.
// This means entry->ended and BUFFER_ENDED have two distinct considerations.
// entry->ended: Completely ignored and needs special handling in CLIENT_REMOVE_BUFFERS.
// BUFFER_ENDED: No-op'd in remove_entry_buffers(), add_or_queue_entry() and not added by do_add_entry(), otherwise unchanged.
//  Another note: remove_entry_buffers() and add_or_queue_entry() are only ever called by switch_to().
//   switch_to() -> add_or_queue_entry().
//   switch_to() -> maybe_add_to_previous() -> (possibly delayed)remove_entry_buffers().
static void remove_entry_buffers(struct camu_sink_entry *entry)
{
    log_trace("remove_entry_buffers("ENTRY_FMT"), audio_state: %hhu, video_state: %hhu.",
        ENTRY_ARG(entry), AUDIO_STATE(entry), VIDEO_STATE(entry));
    if (!AUDIO_ENDED(entry)) remove_entry_audio_buffer(entry);
    if (!VIDEO_ENDED(entry)) remove_entry_video_buffer(entry);
}

// Disconnecting a packet stream twice before a reconnect is an error.
static void maybe_disconnect_entry(struct camu_sink_entry *entry)
{
    if (!entry->disconnected) {
        // disconnect() could free entry.
        entry->disconnected = true;
        lia_client_disconnect(&entry->client);
    }
}

// sink->current could be NULL.
static inline struct camu_sink_entry *get_entry_for_command(struct camu_sink *sink)
{
    return ENTRY_IS_VALID(sink->target) ? sink->target : sink->current;
}

static inline s32 get_sequence_for_command(struct camu_sink *sink, struct camu_sink_entry *entry)
{
    // If local, using entry->sequence could only lead to feeling like your inputs were eaten.
    if (!sink->local && entry) {
        return entry->sequence;
    }
    // SEQUENCE_ANY resolves order on the server.
    return LIANA_SEQUENCE_ANY;
}

static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
{
    switch (cmd->op) {
    case START: {
        switch (cmd->v.u) {
        case CAMU_SINK_AUDIO:
            if (sink->audio.state == SINK_PAUSED) {
                sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_AUDIO, NULL);
                sink->audio.state = SINK_PLAYING;
            }
            break;
        case CAMU_SINK_VIDEO:
            if (sink->video.state == SINK_PAUSED) {
#ifndef CAMU_SINK_NO_VIDEO
                sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_VIDEO, NULL);
#endif
                sink->video.state = SINK_PLAYING;
            }
            break;
        }
        break;
    }
    case STOP: {
        switch (cmd->v.u) {
        case CAMU_SINK_AUDIO:
            if (sink->audio.state == SINK_PLAYING) {
                sink->callback(sink->userdata, CAMU_SINK_STOP, CAMU_SINK_AUDIO, NULL);
                sink->audio.state = SINK_PAUSED;
            }
            break;
        case CAMU_SINK_VIDEO:
            if (sink->video.state == SINK_PLAYING) {
#ifndef CAMU_SINK_NO_VIDEO
                sink->callback(sink->userdata, CAMU_SINK_STOP, CAMU_SINK_VIDEO, NULL);
#endif
                sink->video.state = SINK_PAUSED;
            }
            break;
        }
        break;
    }
    case ADD_BUFFER: {
        struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
        switch (cmd->v.u) {
        case CAMU_SINK_AUDIO:
            sink->callback(sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
            break;
        case CAMU_SINK_VIDEO:
#ifndef CAMU_SINK_NO_VIDEO
            sink->callback(sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
#endif
            break;
        }
        break;
    }
    case REMOVE_BUFFER: {
        struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
        switch (cmd->v.u) {
        case CAMU_SINK_AUDIO:
            sink->callback(sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
            break;
        case CAMU_SINK_VIDEO:
#ifndef CAMU_SINK_NO_VIDEO
            sink->callback(sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
#endif
            break;
        }
        break;
    }
    case CLEAR_BUFFERS: {
        switch (cmd->v.u) {
        case CAMU_SINK_AUDIO:
            sink->callback(sink->userdata, CAMU_SINK_CLEAR, CAMU_SINK_AUDIO, NULL);
            break;
        case CAMU_SINK_VIDEO:
#ifndef CAMU_SINK_NO_VIDEO
            sink->callback(sink->userdata, CAMU_SINK_CLEAR, CAMU_SINK_VIDEO, NULL);
#endif
            break;
        }
        break;
    }
    case EJECT_ENTRY: {
        struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
        maybe_disconnect_entry(entry);
        break;
    }
    case CLOSE: {
        nn_signal_stop(&sink->queue_signal);
        sink->callback(sink->userdata, CAMU_SINK_EXIT, 0, NULL);
        return;
    }
    case SKIP: {
        if (!sink->connected) return;
        struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
        struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
        nn_packet_write_str(packet, &sink->default_list);
        nn_packet_write_u8(packet, CAMU_LIST_SKIP);
        nn_packet_write_s32(packet, get_sequence_for_command(sink, entry));
        nn_packet_write_s32(packet, (s32)cmd->v.i);
        nn_rpc_connection_command(sink->conn, packet, NULL, NULL);
        break;
    }
    case TOGGLE_PAUSE: {
        if (!sink->connected) return;
        struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
        struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
        nn_packet_write_str(packet, &sink->default_list);
        nn_packet_write_u8(packet, CAMU_LIST_TOGGLE_PAUSE);
        nn_packet_write_s32(packet, get_sequence_for_command(sink, entry));
        nn_packet_write_f64(packet, cmd->v.f);
        nn_rpc_connection_command(sink->conn, packet, NULL, NULL);
        break;
    }
    case SEEK: {
        if (!sink->connected) return;
        struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
        struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
        nn_packet_write_str(packet, &sink->default_list);
        nn_packet_write_u8(packet, CAMU_LIST_SEEK);
        nn_packet_write_s32(packet, entry->sequence);
        nn_packet_write_u32(packet, REMOTE_ENTRY_ID(entry->id));
        nn_packet_write_u64(packet, cmd->v.u);
        nn_rpc_connection_command(sink->conn, packet, NULL, NULL);
        break;
    }
    case RESEEK: {
        // Crude way to trigger "re-add sink to list".
        if (sink->conn) nn_rpc_conn_disconnect(sink->conn);
        break;
    }
    case SHUFFLE: {
        if (!sink->connected) return;
        struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
        nn_packet_write_str(packet, &sink->default_list);
        nn_packet_write_u8(packet, CAMU_LIST_SHUFFLE);
        nn_rpc_connection_command(sink->conn, packet, NULL, NULL);
        break;
    }
    case END: {
        if (!sink->connected) return;
        struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
        struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
        nn_packet_write_str(packet, &sink->default_list);
        nn_packet_write_u8(packet, CAMU_LIST_END);
        nn_packet_write_u32(packet, REMOTE_ENTRY_ID(entry->id));
        nn_packet_write_u32(packet, (u32)cmd->v.u);
        nn_rpc_connection_command(sink->conn, packet, NULL, NULL);
        break;
    }
    }
}

static void queue_signal_callback(void *userdata)
{
    struct camu_sink *sink = (struct camu_sink *)userdata;
    u32 count;
    struct camu_sink_cmd cmd;
    for (;;) {
        camu_queue_try_pop(sink->queue, count, cmd);
        if (count == 0) break;
        handle_sink_cmd(sink, &cmd);
    }
}

static void mixer_callback(void *userdata, u8 op)
{
    struct camu_sink *sink = (struct camu_sink *)userdata;
    if (op == CAMU_MIXER_EMPTY) {
        log_info("Mixer empty.");
        // Regardless of if we are checking an entry's state, we have to sync
        // with do_add_entry() because the order of START/STOPs in the queue matters.
        nn_mutex_lock(&sink->lock);
        // This check is too loose. Still about as good as any naive implementation
        // of SINK_EMPTY, though.
        if (!(sink->current && AUDIO_STATE(sink->current) == BUFFER_ADDED)) {
            queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_AUDIO));
        }
        nn_mutex_unlock(&sink->lock);
    }
}

static s32 entry_lru_compare(const void *a, const void *b)
{
    struct camu_sink_entry *aa = *((struct camu_sink_entry **)a);
    struct camu_sink_entry *bb = *((struct camu_sink_entry **)b);
    if (aa->lru > bb->lru) return -1;
    else if (aa->lru < bb->lru) return 1;
    return 0;
}

static void maybe_cleanup_old_entries(struct camu_sink *sink)
{
    al_array_sort(sink->entries, struct camu_sink_entry *, entry_lru_compare);
    // We check size <= MAX_AGE in the loops because sink->lru is not
    // indicative of the amount of entries we have loaded.
    // The most obvious reason being it's incremented when moving back
    // and forth between two entries. As well as for buffer and queue operations.
    // We have to handle sink->lru wrapping in a step before the default case.
    // 0 65532 65533 65534 65535
    // 0 1 65533 65534 65535
    // 0 1 2 65534 65535
    // 0 1 2 3 65535
    // 0 1 2 3 4
    struct camu_sink_entry *entry;
    al_array_foreach_rev(sink->entries, i, entry) {
        if (sink->entries.count <= ENTRY_MAX_AGE) return;
        u16 entry_age = (SINK_LRU_MAX - entry->lru) + sink->lru;
        if (entry->lru > sink->lru && entry_age > ENTRY_MAX_AGE) {
            al_array_remove_at(sink->entries, i);
            maybe_disconnect_entry(entry);
        }
    }
    // Make sure we don't have to consider wrapping in the second loop.
    if (sink->lru < ENTRY_MAX_AGE) return;
    al_array_foreach_rev(sink->entries, i, entry) {
        al_assert(sink->lru >= entry->lru);
        if (sink->entries.count <= ENTRY_MAX_AGE) return;
        u16 entry_age = sink->lru - entry->lru;
        if (entry_age > ENTRY_MAX_AGE) {
            al_array_remove_at(sink->entries, i);
            maybe_disconnect_entry(entry);
        }
    }
}

static void maybe_run_previous(struct camu_sink *sink)
{
    log_trace("maybe_run_previous(), previous_count: %u.", sink->previous.count);
    struct camu_sink_entry *previous;
    al_array_foreach(sink->previous, i, previous) {
        remove_entry_buffers(previous);
        // Cleanup entries from old connections. Especially important to
        // keep reseek() from being overly wasteful.
        if (CONNECTION_NUMBER(previous->id) != sink->connection_number) {
            queue_cmd(sink, CMD(EJECT_ENTRY, .opaque = previous));
        }
    }
    sink->previous.count = 0;
}

// Due to the looseness of the previous queue, we may have to explicitly remove an
// entry if it becomes incorrect to attempt removing it's buffers. The most obvious example
// being at the point it's freed.
static void run_previous_if_contains(struct camu_sink *sink, struct camu_sink_entry *key)
{
    bool removed = false;
    struct camu_sink_entry *previous;
    al_array_foreach(sink->previous, i, previous) {
        if (previous == key) {
            maybe_run_previous(sink);
            removed = true;
            break;
        }
    }
    log_trace("run_previous_if_contains("ENTRY_FMT"), removed: %s.", ENTRY_ARG(key), BOOLSTR(removed));
}

static bool maybe_remove_from_previous(struct camu_sink *sink, struct camu_sink_entry *entry)
{
    bool removed = false;
    struct camu_sink_entry *previous;
    al_array_foreach(sink->previous, i, previous) {
        if (previous == entry) {
            al_array_remove_at(sink->previous, i);
            removed = true;
            break;
        }
    }
    log_trace("maybe_remove_from_previous("ENTRY_FMT"), removed: %s.", ENTRY_ARG(entry), BOOLSTR(removed));
    return removed;
}

// Every call to maybe_add_to_previous() must map to a remove_entry_buffers().
static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry *previous,
    struct camu_sink_entry *target)
{
    log_trace("maybe_add_to_previous("ENTRY_FMT"), target: "ENTRY_FMT".", ENTRY_ARG(previous), ENTRY_ARG(target));
    al_assert(previous != target);
    bool ignore_target = target == (struct camu_sink_entry *)0xb00b || ENTRY_ENDED(target);
    if (ignore_target || (AUDIO_STATE(previous) != BUFFER_ADDED && VIDEO_STATE(previous) != BUFFER_ADDED)) {
        remove_entry_buffers(previous);
        return;
    }
    al_array_push(sink->previous, previous);
}

static void after_add_entry(struct camu_sink_entry *entry, bool skip_audio, bool skip_video)
{
    struct camu_sink *sink = entry->sink;
    maybe_run_previous(sink);
    queue_cmds(entry->sink, 2,
        CMD((skip_video || entry->paused) ? STOP : START, .v.u = CAMU_SINK_VIDEO),
        CMD((skip_audio || entry->paused) ? STOP : START, .v.u = CAMU_SINK_AUDIO)
    );
    // Clear the screen if skipping from a video to an audio-only entry.
    if (VIDEO_ENDED_OR_EMPTY(entry)) refresh_video_output(sink);
}

// Call this after setting state to ADDED because this entry might be in previous.
static void do_add_entry(struct camu_sink_entry *entry)
{
    bool skip_audio = AUDIO_ENDED_OR_EMPTY(entry);
    // Static videos are unconditionally added in add_video_if_set_and_buffered().
    bool skip_video = VIDEO_ENDED_OR_EMPTY(entry) || VIDEO_IS_STATIC(entry);
    if (!skip_audio && !skip_video) {
        al_assert(VIDEO_STATE(entry) == AUDIO_STATE(entry));
    }
    // Whether to add audio or video first could be a consideration for responsiveness.
    if (!skip_video) add_entry_video_buffer(entry);
    if (!skip_audio) add_entry_audio_buffer(entry);
    after_add_entry(entry, skip_audio, skip_video);
}

static void add_audio_if_set_and_buffered(struct camu_sink_entry *entry)
{
    al_assert(!ENTRY_ENDED(entry));
    al_assert(AUDIO_STATE(entry) != BUFFER_INIT);
    al_assert(AUDIO_STATE(entry) != BUFFER_QUEUED);
    al_assert(AUDIO_STATE(entry) != BUFFER_ADDED);
    al_assert(!AUDIO_ENDED(entry));
    switch (AUDIO_STATE(entry)) {
    case BUFFER_CONFIGURED:
        AUDIO_STATE(entry) = BUFFER_SET_OR_BUFFERED;
        break;
    case BUFFER_SET_OR_BUFFERED:
        AUDIO_STATE(entry) = BUFFER_ADDED;
        if (VIDEO_ADDED_OR_IGNORED(entry) || VIDEO_IS_STATIC(entry)) {
            do_add_entry(entry);
        }
        break;
    }
}

static void add_video_if_set_and_buffered(struct camu_sink_entry *entry)
{
    // Static video buffers will be added/removed with ended set.
    if (ENTRY_ENDED(entry)) al_assert(VIDEO_IS_STATIC(entry));
    al_assert(VIDEO_STATE(entry) != BUFFER_INIT);
    al_assert(VIDEO_STATE(entry) != BUFFER_QUEUED);
    al_assert(VIDEO_STATE(entry) != BUFFER_ADDED);
    al_assert(!VIDEO_ENDED(entry));
    switch (VIDEO_STATE(entry)) {
    case BUFFER_CONFIGURED:
        VIDEO_STATE(entry) = BUFFER_SET_OR_BUFFERED;
        break;
    case BUFFER_SET_OR_BUFFERED:
        VIDEO_STATE(entry) = BUFFER_ADDED;
        if (VIDEO_IS_STATIC(entry)) {
            add_entry_video_buffer(entry);
            bool skip_audio = AUDIO_ENDED_OR_EMPTY(entry);
            if (skip_audio) {
                after_add_entry(entry, skip_audio, true);
            }
        } else if (AUDIO_ADDED_OR_IGNORED(entry)) {
            do_add_entry(entry);
        }
        break;
    }
}

static void add_or_queue_entry(struct camu_sink_entry *entry)
{
    log_trace("add_or_queue_entry("ENTRY_FMT"), audio_state: %hhu, video_state: %hhu.",
        ENTRY_ARG(entry), AUDIO_STATE(entry), VIDEO_STATE(entry));
    al_assert(AUDIO_STATE(entry) != BUFFER_QUEUED);
    al_assert(VIDEO_STATE(entry) != BUFFER_QUEUED);
    // Either buffer could be DETACHED.
    if (AUDIO_STATE(entry) == BUFFER_INIT) {
        AUDIO_STATE(entry) = BUFFER_QUEUED;
    } else if (!AUDIO_ENDED_OR_EMPTY(entry)) {
        add_audio_if_set_and_buffered(entry);
    }
    if (VIDEO_STATE(entry) == BUFFER_INIT) {
        VIDEO_STATE(entry) = BUFFER_QUEUED;
    } else if (!VIDEO_ENDED_OR_EMPTY(entry)) {
        add_video_if_set_and_buffered(entry);
    }
}

static void ensure_static_video_removed(struct camu_sink_entry *entry)
{
    if (!VIDEO_EMPTY(entry) && VIDEO_IS_STATIC(entry)) {
        remove_entry_video_buffer(entry);
        struct camu_sink *sink = entry->sink;
        while (entry_video_buffer_held(entry)) { BLOCKING_SLEEP(sink, NNWT_TS_FROM_USEC(2000)); }
    }
}

// This is the only function that sets sink->current.
static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
{
    struct camu_sink_entry *current = sink->current;

    log_trace("switch_to("ENTRY_FMT"), current: "ENTRY_FMT".", ENTRY_ARG(target), ENTRY_ARG(current));

    if (current) {
        struct camu_sink_entry *suspended = sink->suspended;
        al_assert(current != target);
        if (suspended) {
            al_assert(suspended == current);
            // It should be impossible for a buffer to be QUEUED while it's entry is suspended.
            // Even for a static video buffer because we wouldn't know if it was static yet.
            al_assert(AUDIO_STATE(suspended) != BUFFER_QUEUED);
            al_assert(VIDEO_STATE(suspended) != BUFFER_QUEUED);
            ensure_static_video_removed(suspended);
            al_assert(AUDIO_STATE(suspended) != BUFFER_ADDED);
            al_assert(VIDEO_STATE(suspended) != BUFFER_ADDED);
            sink->suspended = NULL;
            log_warn("Unset suspended entry as a substitute for remove.");
        } else {
            maybe_add_to_previous(sink, current, target);
        }
    }

    bool dangling_target = target == (struct camu_sink_entry *)0xb00b;
    if (dangling_target) log_trace("Ignoring dangling target.");
    bool stop_video = dangling_target;
    if (!dangling_target) {
        target->audio.ignore_paused = false;
        if (!sink->local && !target->paused) {
            camu_audio_buffer_resync(&target->audio.buf);
        }
        if (!maybe_remove_from_previous(sink, target)) {
            add_or_queue_entry(target);
        }
        al_assert(AUDIO_STATE(target) != BUFFER_INIT);
        al_assert(VIDEO_STATE(target) != BUFFER_INIT);
        // If target was just created, it's AUDIO/VIDEO_STATE() will be QUEUED.
        // Meaning, at this point, it's video buffer would be considered empty
        // as well as being too early to tell if it's static or not.
        stop_video = VIDEO_ENDED_OR_EMPTY(target) || VIDEO_IS_STATIC(target);
    }

    if (stop_video) {
        queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_VIDEO));
        if (dangling_target || VIDEO_ENDED_OR_EMPTY(target)) {
            refresh_video_output(sink);
        }
    }

    sink->current = dangling_target ? NULL : target;
}

static void pause_and_swap_to(struct camu_sink *sink, struct camu_sink_entry *target, u64 at)
{
    struct camu_sink_entry *current = sink->current;
    log_trace("pause_and_swap_to("ENTRY_FMT", %.2f), current: "ENTRY_FMT".",
        ENTRY_ARG(target), at / 1000000.0, ENTRY_ARG(current));
    al_assert(target != current);
    al_assert(!sink->target);
    sink->target = target;
    bool immediate = !current || ENTRY_ENDED(current);
    if (current) {
        current->audio.ignore_paused = true;
        immediate |= camu_clock_pause(&current->clock, at);
    }
    if (immediate) {
        switch_to(sink, sink->target);
        sink->target = NULL;
    }
}

static bool end_entry_and_advance_queue(struct camu_sink *sink, struct camu_sink_entry *entry)
{
    log_debug("Entry ("ENTRY_FMT") ended.", ENTRY_ARG(entry));
    run_previous_if_contains(sink, entry);
#ifdef CAMU_SINK_ONESHOT
    sink->callback(sink->userdata, CAMU_SINK_MOCK_CLOSE, 0, NULL);
    return false;
#endif
    // This entry's buffers cannot be added again until after a reset.
    entry->ended = true;
    al_assert(ENTRY_EVAL_ENDED(entry));
    queue_cmd(sink, CMD(END, .v.u = entry->reset_token, .opaque = entry));
#ifdef LIANA_LIST_SCUFFED_LOOP
    log_info("Looping.");
    return true;
#endif
    if (sink->target) {
        log_info("Buffers swapped on end() (Gapless if queued).");
        switch_to(sink, sink->target);
        sink->target = NULL;
        return true;
    }
    return false;
}

static void audio_buffer_callback(void *userdata, u8 op)
{
    struct camu_sink_entry *entry = (struct camu_sink_entry *)userdata;
    struct camu_sink *sink = entry->sink;
    switch (op) {
    case CAMU_BUFFER_BUFFERED:
        lia_vcr_set_buffered(entry->audio.track);
        nn_mutex_lock(&sink->lock);
        add_audio_if_set_and_buffered(entry);
        nn_mutex_unlock(&sink->lock);
        break;
    case CAMU_BUFFER_CORK:
        lia_vcr_cork(entry->audio.track);
        break;
    case CAMU_BUFFER_UNCORK:
        lia_vcr_uncork(entry->audio.track);
        break;
    case CAMU_BUFFER_PAUSED: // Comes from audio read() thread.
        nn_mutex_lock(&sink->lock);
        // entry->paused could plausibly be false here if the lock was held
        // by pause_command_callback() to resume. This can be simulated by
        // calling list_toggle_pause() twice in server/list_action_callback()
        // for each sink request. Spaced by an nn_event_loop_sleep(~15500us)
        // (no_video, MINIAUDIO_LOW_LATENCY mode).
        if (!entry->audio.ignore_paused && entry->paused) {
            log_info("Audio buffer paused.");
            queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_AUDIO));
        }
        nn_mutex_unlock(&sink->lock);
        break;
    case CAMU_BUFFER_EOF:
    case CAMU_BUFFER_ERRORED: {
        bool error = op == CAMU_BUFFER_ERRORED;
        if (error) {
            log_error("Audio buffer errored.");
        } else {
            log_debug("Audio EOF.");
        }
        nn_mutex_lock(&sink->lock);
        // EOF and ERRORED come from the outputs read() thread. So, having threaded
        // outputs means anything could have happened while waiting on the lock above.
        // For example, if we were locked in CLIENT_REMOVE_BUFFERS, state could have
        // dropped all the way to CONFIGURED before we acquired the lock here.
        // This should also maintain a consistent state in the more common case of switch_to()
        // right before a buffer EOF.
        if (AUDIO_STATE(entry) == BUFFER_ADDED) {
            remove_entry_audio_buffer(entry);
        }
        AUDIO_STATE(entry) = error ? BUFFER_ERRORED : BUFFER_ENDED;
        if (VIDEO_ENDED_OR_EMPTY(entry) || VIDEO_IS_STATIC(entry)) {
            end_entry_and_advance_queue(sink, entry);
        }
        nn_mutex_unlock(&sink->lock);
        break;
    }
    }
}

static void video_buffer_callback(void *userdata, u8 op)
{
    struct camu_sink_entry *entry = (struct camu_sink_entry *)userdata;
    struct camu_sink *sink = entry->sink;
    switch (op) {
    case CAMU_BUFFER_BUFFERED:
        lia_vcr_set_buffered(entry->video.track);
        nn_mutex_lock(&sink->lock);
        add_video_if_set_and_buffered(entry);
        nn_mutex_unlock(&sink->lock);
        break;
    case CAMU_BUFFER_CORK:
        lia_vcr_cork(entry->video.track);
        break;
    case CAMU_BUFFER_UNCORK:
        lia_vcr_uncork(entry->video.track);
        break;
    case CAMU_BUFFER_EOF:
    case CAMU_BUFFER_ERRORED: {
        bool error = op == CAMU_BUFFER_ERRORED;
        if (error) {
            log_error("Video buffer errored.");
        } else {
            log_debug("Video EOF.");
        }
        nn_mutex_lock(&sink->lock);
        if (!AUDIO_EMPTY(entry)) {
            camu_audio_buffer_set_no_video(&entry->audio.buf, true);
        }
        if (VIDEO_IS_STATIC(entry) && !error) {
            nn_mutex_unlock(&sink->lock);
            return;
        }
        // Video state could be ADDED, SET_OR_BUFFERED, or CONFIGURED.
        // See note about threaded outputs in audio_buffer_callback(EOF|ERRORED).
        if (VIDEO_STATE(entry) == BUFFER_ADDED) {
            remove_entry_video_buffer(entry);
        }
        VIDEO_STATE(entry) = error ? BUFFER_ERRORED : BUFFER_ENDED;
        bool swapped = false;
        if (AUDIO_ENDED_OR_EMPTY(entry)) {
            swapped = end_entry_and_advance_queue(sink, entry);
        }
        nn_mutex_unlock(&sink->lock);
        if (!swapped) {
            queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_VIDEO));
        }
        break;
    }
    }
}

static void clock_callback(void *userdata, u8 op)
{
    struct camu_sink_entry *entry = (struct camu_sink_entry *)userdata;
    struct camu_sink *sink = entry->sink;
    if (op == CAMU_CLOCK_PAUSED) {
        nn_mutex_lock(&sink->lock);
        log_trace("clock_paused("ENTRY_FMT"), target: "ENTRY_FMT".", ENTRY_ARG(entry), ENTRY_ARG(sink->target));
        if (entry == sink->current) {
            if (sink->target) {
                switch_to(sink, sink->target);
                sink->target = NULL;
            } else if (entry->paused) {
                queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_VIDEO));
            }
        }
        nn_mutex_unlock(&sink->lock);
    }
}

static void evaluate_and_set_buffer_params(struct camu_sink *sink, struct camu_sink_entry *entry)
{
    bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_STATIC(entry);
    f64 audio = camu_mixer_get_latency(sink->audio.mixer);
    u32 frames = 0;
    f64 video = 0.0;
    if (!ignore_video) {
        struct camu_renderer *renderer = sink->video.renderer;
        if (renderer) {
            f64 avg_frame_duration = entry->video.buf.avg_frame_duration;
            frames = renderer->get_latency(renderer);
            video = frames * avg_frame_duration;
        }
    }
    camu_audio_buffer_set_latency(&entry->audio.buf, audio);
    camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video);
    camu_video_buffer_set_latency(&entry->video.buf, video);
    log_info("video_latency: %fs (%u frames), audio_latency: %fs.", video, frames, audio);
    if (sink->local) {
        camu_audio_buffer_set_ignore_desync(&entry->audio.buf, ignore_video);
        // When the video buffer starts the clock, we have to consider the audio
        // buffer is treating the last period of silence sent while the clock was paused
        // as part of the stream. This poses an issue for sync because it requires more
        // than a period-length offset to not skip audio data at the start of the stream.
        // Combined with the fact that the timing of a request for the next period
        // doesn't have to be uniform. Something that is expected behavior but becomes
        // a consideration when thinking about "Has the clock started yet?".
        // @TODO: Could make a diagram of this.
        camu_clock_offset(&entry->clock, MAX(audio, video) * 1.25);
    }
}

static void run_queue_by_opaque(struct camu_sink *sink, void *opaque)
{
    camu_queue_lock(sink->queue);
    struct camu_sink_cmd *cmd;
    al_array_foreach_ptr(sink->queue.a, i, cmd) {
        if (cmd->opaque == opaque) {
            handle_sink_cmd(sink, cmd);
            al_array_remove_at_iter(sink->queue.a, i);
        }
    }
    camu_queue_unlock(sink->queue);
}

static void remove_from_queue_by_opaque(struct camu_sink *sink, void *opaque)
{
    camu_queue_lock(sink->queue);
    struct camu_sink_cmd *cmd;
    al_array_foreach_ptr_rev(sink->queue.a, i, cmd) {
        if (cmd->opaque == opaque) {
            al_array_remove_at(sink->queue.a, i);
        }
    }
    camu_queue_unlock(sink->queue);
}

static void client_callback(void *userdata, u8 op, struct camu_codec_stream *stream, void *opaque)
{
    struct camu_sink_entry *entry = (struct camu_sink_entry *)userdata;
    al_assert(ENTRY_IS_VALID(entry));
    struct camu_sink *sink = entry->sink;
    switch (op) {
    case LIANA_CLIENT_CONFIGURE: {
        // No data will be sent until all selected streams are configured.
        switch (stream->type) {
        case CAMU_STREAM_AUDIO:
            entry->audio.track = (struct lia_vcr_track *)opaque;
            if (!camu_audio_buffer_configure(&entry->audio.buf, stream, sink->audio.mixer)) {
                log_error("Audio buffer failed to configure.");
                maybe_disconnect_entry(entry);
                return;
            }
            nn_mutex_lock(&sink->lock);
            al_assert(AUDIO_EMPTY(entry));
            if (AUDIO_STATE(entry) == BUFFER_QUEUED) {
                AUDIO_STATE(entry) = BUFFER_SET_OR_BUFFERED;
            } else if (AUDIO_STATE(entry) == BUFFER_INIT) {
                AUDIO_STATE(entry) = BUFFER_CONFIGURED;
            }
            nn_mutex_unlock(&sink->lock);
            break;
        case CAMU_STREAM_VIDEO:
            entry->video.track = (struct lia_vcr_track *)opaque;
            if (!camu_video_buffer_configure(&entry->video.buf, stream, sink->video.renderer)) {
                log_error("Video buffer failed to configure.");
                maybe_disconnect_entry(entry);
                return;
            }
            nn_mutex_lock(&sink->lock);
            al_assert(VIDEO_EMPTY(entry));
            if (VIDEO_STATE(entry) == BUFFER_QUEUED) {
                VIDEO_STATE(entry) = BUFFER_SET_OR_BUFFERED;
            } else if (VIDEO_STATE(entry) == BUFFER_INIT) {
                VIDEO_STATE(entry) = BUFFER_CONFIGURED;
            }
            nn_mutex_unlock(&sink->lock);
            break;
        case CAMU_STREAM_SUBTITLE:
            if (!camu_video_buffer_configure_subtitles(&entry->video.buf, stream)) {
                log_warn("Video buffer couldn't configure subtitles.");
            }
            break;
        case CAMU_STREAM_ATTACHMENT: {
            struct camu_renderer *renderer = sink->video.renderer;
            if (renderer) renderer->add_font(renderer, stream);
            break;
        }
        }
        break;
    }
    case LIANA_CLIENT_CONFIGURE_COMPLETE: {
        // All present buffers were configured.
        nn_mutex_lock(&sink->lock);
        // This is mainly to assert DETACHED handling.
        al_assert(!VIDEO_ENDED(entry) && !AUDIO_ENDED(entry));
        evaluate_and_set_buffer_params(sink, entry);
        nn_mutex_unlock(&sink->lock);
        break;
    }
    case LIANA_CLIENT_DATA: {
        struct camu_codec_frame *frame = (struct camu_codec_frame *)opaque;
        // Even though !AUDIO/VIDEO_EMPTY() is a value that cannot change at this point, we
        // still have to lock because AUDIO/VIDEO_STATE() is not atomic.
        switch (stream->type) {
        case CAMU_STREAM_AUDIO:
            nn_locked_assert(!AUDIO_EMPTY(entry), &sink->lock);
            camu_audio_buffer_push(&entry->audio.buf, frame);
            break;
        case CAMU_STREAM_VIDEO:
            nn_locked_assert(!VIDEO_EMPTY(entry), &sink->lock);
            camu_video_buffer_push(&entry->video.buf, frame);
            break;
        default:
            camu_codec_frame_discard(frame);
            break;
        }
        break;
    }
    case LIANA_CLIENT_SUBTITLE: {
        switch (stream->type) {
        case CAMU_STREAM_SUBTITLE: {
            camu_video_buffer_push_subtitle(&entry->video.buf, (struct camu_codec_packet *)opaque);
            break;
        }
        }
        break;
    }
    case LIANA_CLIENT_REMOVE_BUFFERS: {
        struct lia_reconnect_info *rec = (struct lia_reconnect_info *)opaque;
        nn_mutex_lock(&sink->lock);

        log_trace("remove_buffers("ENTRY_FMT", %s, %s), entry == current: %s, static: %s.",
            ENTRY_ARG(entry), BOOLSTR(rec->reconnect), BOOLSTR(rec->unconfigured),
            BOOLSTR(entry == sink->current),
            BOOLSTR(!VIDEO_EMPTY(entry) && VIDEO_IS_STATIC(entry)));

        if (entry == sink->current) {
            if (rec->reconnect) {
                // Else, possible failed or aborted reconnect (Empty buffer could be INIT).
                al_assert(AUDIO_STATE(entry) != BUFFER_INIT);
                al_assert(VIDEO_STATE(entry) != BUFFER_INIT);
            }
            if (sink->target) {
                // This is necessary to avoid re-adding an entry with an in-between clock state.
                // See note in clock.c::camu_clock_seek().
                switch_to(sink, sink->target);
                sink->target = NULL;
            } else if (rec->reconnect) {
                // If this entry is still current on CLIENT_RECONNECTED, re-add it's buffers.
                sink->suspended = entry;
            }
        }

        // This entry might be in previous if it was added to previous then,
        //  1. it's being cleaned up after ENTRY_MAX_AGE - 1 entries were added but none buffered.
        //  2. it was seeked.
        run_previous_if_contains(sink, entry);
        // AUDIO/VIDEO_STATE() could be INIT at this point, even if entry = current.

        // We should treat CLIENT_REMOVE_BUFFERS as a function that removes an entry's buffers and
        // resets it's buffered state. For a non-empty entry that means remove_entry_audio/video_buffer()
        // twice and for an empty buffer, knock it down to BUFFER_INIT.
        if (AUDIO_STATE(entry) == BUFFER_QUEUED) AUDIO_STATE(entry) = BUFFER_INIT;
        if (VIDEO_STATE(entry) == BUFFER_QUEUED) VIDEO_STATE(entry) = BUFFER_INIT;

        // Don't consider unconfigured entries past this point.
        if (rec->unconfigured) {
            al_assert(AUDIO_EMPTY(entry) && VIDEO_EMPTY(entry));
            nn_mutex_unlock(&sink->lock);
            return;
        }

        // An empty buffer is guaranteed to not be held. An ended buffer is already
        // removed and will be further handled at the end of this case.
        bool skip_audio = AUDIO_EMPTY(entry);
        if (!skip_audio && !AUDIO_ENDED(entry)) {
            // Remove for re-add in CLIENT_RECONNECTED.
            remove_entry_audio_buffer(entry);
            // Remove again for another add in BUFFER_BUFFERED.
            remove_entry_audio_buffer(entry);
        }
        bool skip_video = VIDEO_EMPTY(entry);
        // Don't remove a static video buffer if we are reconnecting, unless it's errored.
        skip_video = skip_video || (rec->reconnect && VIDEO_IS_STATIC(entry) && VIDEO_STATE(entry) != BUFFER_ERRORED);
        if (!skip_video && !VIDEO_ENDED(entry)) {
            remove_entry_video_buffer(entry);
            remove_entry_video_buffer(entry);
        }
        // If MIXER_THREADED_START_STOP is not set, REMOVE_BUFFER is not thread-safe.
        // Meaning it must be run on the event loop. So, resolve any queued REMOVE_BUFFER
        // requests so we can safely block the loop.
        run_queue_by_opaque(sink, entry);

        // Unlock to wait.
        nn_mutex_unlock(&sink->lock);

        while ((!skip_audio && entry_audio_buffer_held(entry)) || (!skip_video && entry_video_buffer_held(entry))) {
            BLOCKING_SLEEP(sink, NNWT_TS_FROM_USEC(2000));
        }
        // At this point we can be sure that the entry's buffers are no longer in use.

        // Re-lock to check an entries ended state. As it could have been set at some point after unlocking to wait.
        nn_mutex_lock(&sink->lock);

        if (rec->reconnect) {
            // Detach errored buffers. If we detach both streams, the entry will be closed.
            if (AUDIO_STATE(entry) == BUFFER_ERRORED) {
                al_array_push(rec->detached, AUDIO_STREAM(entry));
            }
            if (VIDEO_STATE(entry) == BUFFER_ERRORED) {
                al_array_push(rec->detached, VIDEO_STREAM(entry));
            } else if (VIDEO_STATE(entry) == BUFFER_ADDED && VIDEO_IS_STATIC(entry)) {
                // Try to not request a duplicate frame. We can only be sure the frame wasn't
                // dropped by the VCR if it's ADDED.
                rec->mask &= ~(1 << VIDEO_STREAM(entry)->index);
            }
        }

        // Finalize the removal of the buffers by handling "ended" (ENDED or ERRORED) buffers.
        // ENDED: Set to CONFIGURED to emulate the two removes earlier in this case.
        // ERRORED: Set to DETACHED and they will now be considered empty.
        if (AUDIO_STATE(entry) == BUFFER_ENDED) {
            al_assert(!AUDIO_EMPTY(entry));
            AUDIO_STATE(entry) = BUFFER_CONFIGURED;
        } else if (AUDIO_STATE(entry) == BUFFER_ERRORED) {
            AUDIO_STATE(entry) = BUFFER_DETACHED;
        }
        if (VIDEO_STATE(entry) == BUFFER_ENDED) {
            al_assert(!VIDEO_EMPTY(entry));
            VIDEO_STATE(entry) = BUFFER_CONFIGURED;
        } else if (VIDEO_STATE(entry) == BUFFER_ERRORED) {
            VIDEO_STATE(entry) = BUFFER_DETACHED;
        }
        // Unset ended, consistent with the logic in list.
        entry->ended = false;

        nn_mutex_unlock(&sink->lock);

        break;
    }
    case LIANA_CLIENT_RECOVER_TO: {
        struct lia_timing *time = (struct lia_timing *)opaque;
        time->pos = camu_clock_get_last_pts(&entry->clock) * (u64)1000000;
        time->at = 0;
        break;
    }
    case LIANA_CLIENT_RESUME_AT: { // This is called after the client reconnects, before CLIENT_RECONNECTED.
        struct lia_timing *time = (struct lia_timing *)opaque;
        if (sink->local) time->at = 0;
        f64 pos = time->pos / 1000000.0;
        log_trace("resume_at("ENTRY_FMT"), pos: %f, paused_at: %f.", ENTRY_ARG(entry), pos, entry->clock.paused_at);
        nn_mutex_lock(&sink->lock);
        bool ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_STATIC(entry);
        if (!AUDIO_EMPTY(entry)) {
            camu_audio_buffer_reset(&entry->audio.buf);
            // no_video is set to true in video BUFFER_EOF as a fail-safe. Reset it here.
            camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video);
        }
        if (!ignore_video) {
            camu_video_buffer_reset(&entry->video.buf, pos);
        }
        camu_clock_seek(&entry->clock, pos, time->at);
        nn_mutex_unlock(&sink->lock);
        break;
    }
    case LIANA_CLIENT_RECONNECTED: {
        struct lia_reconnect_info *rec = (struct lia_reconnect_info *)opaque;
        nn_mutex_lock(&sink->lock);
        log_trace("reconnected("ENTRY_FMT"), suspended: "ENTRY_FMT", audio_state: %hhu, video_state: %hhu.",
            ENTRY_ARG(entry), ENTRY_ARG(sink->suspended), AUDIO_STATE(entry), VIDEO_STATE(entry));
        if (entry == sink->suspended) {
            al_assert(entry == sink->current);
            // Let this be the only other explicit BUFFER_ENDED check, or this will
            // become too complicated.
            al_assert(AUDIO_STATE(entry) != BUFFER_ENDED);
            al_assert(VIDEO_STATE(entry) != BUFFER_ENDED);
            // The value of unconfigured remains consistent from CLIENT_REMOVE_BUFFERS.
            if (rec->unconfigured) {
                al_assert(AUDIO_EMPTY(entry) && VIDEO_EMPTY(entry));
            }
            // An unconfigured entry would have still been queued if it was current.
            if (AUDIO_STATE(entry) == BUFFER_INIT) {
                AUDIO_STATE(entry) = BUFFER_QUEUED;
            } else if (!AUDIO_EMPTY(entry)) {
                add_audio_if_set_and_buffered(entry);
            }
            if (VIDEO_STATE(entry) == BUFFER_INIT) {
                VIDEO_STATE(entry) = BUFFER_QUEUED;
            } else if (!VIDEO_EMPTY(entry) && !VIDEO_IS_STATIC(entry)) {
                add_video_if_set_and_buffered(entry);
            }
            sink->suspended = NULL;
        }
        nn_mutex_unlock(&sink->lock);
        break;
    }
    case LIANA_CLIENT_EOF:
    case LIANA_CLIENT_ERRORED: {
        bool error = op == LIANA_CLIENT_ERRORED;
        switch (stream->type) {
        case CAMU_STREAM_AUDIO: {
            nn_locked_assert(!AUDIO_EMPTY(entry), &sink->lock);
            camu_audio_buffer_flush(&entry->audio.buf, error);
            break;
        }
        case CAMU_STREAM_VIDEO: {
            nn_locked_assert(!VIDEO_EMPTY(entry), &sink->lock);
            // Static video buffers are immediately flushed inside the buffer.
            if (!VIDEO_IS_STATIC(entry) || error) {
                camu_video_buffer_flush(&entry->video.buf, error);
            }
            break;
        }
        }
        break;
    }
    case LIANA_CLIENT_CLOSED: {
        struct lia_reconnect_info *rec = (struct lia_reconnect_info *)opaque;
        nn_mutex_lock(&sink->lock);
        // We can be assured that CLIENT_REMOVE_BUFFERS has been called on this entry.

        // If a client is closed after a failed reconnect, a static video buffer could still be added.
        if (rec->reconnect) ensure_static_video_removed(entry);
        al_assert(AUDIO_STATE(entry) != BUFFER_ADDED);
        al_assert(VIDEO_STATE(entry) != BUFFER_ADDED);

        bool removed = al_array_remove(sink->entries, entry);
        remove_from_queue_by_opaque(sink, entry);

        if (entry == sink->target) {
            sink->target = (struct camu_sink_entry *)0xb00b;
            log_warn("Attempting to handle a disconnected target.");
        } else if (entry == sink->current) {
            if (sink->suspended) {
                al_assert(sink->suspended == sink->current);
                sink->suspended = NULL;
            }
            if (sink->target) {
                switch_to(sink, sink->target);
                sink->target = NULL;
            } else {
                sink->current = NULL;
                if (removed) { // Don't stop video on exit.
                    queue_cmd(sink, CMD(STOP, .v.u = CAMU_SINK_VIDEO));
                }
            }
            // If current was never fully added we need to call this here.
            maybe_run_previous(sink);
        }

        nn_mutex_unlock(&sink->lock);

        lia_client_free(&entry->client);
        camu_audio_buffer_free(&entry->audio.buf);
        camu_video_buffer_free(&entry->video.buf);
        log_warn("Entry ("ENTRY_FMT") closed by %s.", ENTRY_ARG(entry), removed ? "force" : "cleanup");
        al_free(entry);

        break;
    }
    }
}

static struct camu_sink_entry *create_entry(struct camu_sink *sink, u64 id)
{
    struct camu_sink_entry *entry = al_alloc_object(struct camu_sink_entry);
    entry->sink = sink;
    entry->id = id;
    entry->disconnected = false;
    entry->ended = false;

    camu_clock_init(&entry->clock, clock_callback, entry);

    entry->client.callback = client_callback;
    entry->client.userdata = entry;
    entry->client.renderer = sink->video.renderer;
    entry->client.prefs = sink->prefs;

    AUDIO_STATE(entry) = BUFFER_INIT;
    entry->audio.ignore_paused = false;
    camu_audio_buffer_init(&entry->audio.buf, &entry->clock);
    entry->audio.buf.callback = audio_buffer_callback;
    entry->audio.buf.userdata = entry;

    VIDEO_STATE(entry) = BUFFER_INIT;
    camu_video_buffer_init(&entry->video.buf, &entry->clock);
    entry->video.buf.callback = video_buffer_callback;
    entry->video.buf.userdata = entry;
    // Two entries in order could share the same pointer if the first
    // entry was just freed. This has to be accounted for in the renderer
    // cache or it won't update on the first frame of the new entry.
    // This is most likely to happen when the sink reconnects to a server.
    union { f64 f; u64 u; } fv = { .u = id };
    entry->video.buf.seek_pts = fv.f;

    al_array_push(sink->entries, entry);

    return entry;
}

static struct camu_sink_entry *get_entry_from_id(struct camu_sink *sink, u64 id)
{
    struct camu_sink_entry *entry;
    al_array_foreach(sink->entries, i, entry) {
        if (entry->id == id) return entry;
    }
    return NULL;
}

static bool set_command_callback(void *userdata, struct nn_rpc_connection *conn,
    struct nn_packet *packet, struct nn_packet *rpacket)
{
    struct camu_sink *sink = (struct camu_sink *)userdata;
    (void)rpacket;

    u8 op = nn_packet_read_u8(packet);
    if (op == LIANA_SINK_UNSET) {
        nn_mutex_lock(&sink->lock);
        struct camu_sink_entry *current = sink->current;
        nn_mutex_unlock(&sink->lock);
        if (current) {
            maybe_disconnect_entry(current);
        }
        nn_mutex_lock(&sink->lock);
        goto out;
    }

    // Liana node info.
    str addr;
    nn_packet_read_str(packet, &addr);
    u16 port = nn_packet_read_u16(packet);
    u32 node_id = nn_packet_read_u32(packet);

    // List entry info.
    u64 id = LOCAL_ENTRY_ID(sink, nn_packet_read_u32(packet));
    s32 sequence = nn_packet_read_s32(packet);
    u64 at = nn_packet_read_u64(packet);
    u64 pos = nn_packet_read_u64(packet);
    u8 pause = nn_packet_read_u8(packet);
    u32 reset_token = nn_packet_read_u32(packet);

    struct camu_sink_entry *entry = get_entry_from_id(sink, id);
    bool create = !entry;
    if (create) entry = create_entry(sink, id);
    entry->sequence = sequence;
    entry->lru = sink->lru;
    sink->lru = al_u16_add_wrap(sink->lru, 1, SINK_LRU_MAX);
    entry->reset_token = reset_token;
    if (create) {
        entry->paused = pause == LIANA_PAUSE_NONE || pause == LIANA_PAUSE_PAUSE;
        camu_clock_set(&entry->clock, pos / 1000000.0);
        lia_client_connect(&entry->client, sink->loop, sink->type, &addr, port, node_id, pos);
    }

    // Don't lock before client_connect() or we could deadlock in CLIENT_CLOSED on a failed socket_connect().
    nn_mutex_lock(&sink->lock);

    struct camu_sink_entry *current = sink->current;

    if (op == LIANA_SINK_BUFFER) {
        log_trace("buffered("ENTRY_FMT"), created: %s.", ENTRY_ARG(entry), BOOLSTR(create));
        goto out;
    } else if (op == LIANA_SINK_BUFFER_AND_QUEUE) {
        log_trace("queued("ENTRY_FMT"), %s, created: %s.", ENTRY_ARG(entry), lia_pause_op_name(pause), BOOLSTR(create));
        goto out;
    }

    if (sink->local) at = 0;

    // For target to be set that must mean current is set, armed to pause, and not ended.
    struct camu_sink_entry *prev_target = sink->target;
    log_trace("set("ENTRY_FMT"), %s, created: %s, target: "ENTRY_FMT".", ENTRY_ARG(entry),
        lia_pause_op_name(pause), BOOLSTR(create), ENTRY_ARG(prev_target));
    switch (pause) {
    case LIANA_PAUSE_NONE:
        if (prev_target) {
            sink->target = NULL;
        } else {
            al_assert(entry != current);
        }
        if (entry != current) {
            switch_to(sink, entry);
        }
        break;
    case LIANA_PAUSE_RESUME:
        if (prev_target) {
            sink->target = NULL;
            if (entry == current) {
                // Switched back to current before pause_and_swap_to() completed.
                camu_audio_buffer_resync(&entry->audio.buf);
            }
        } else {
            al_assert(entry != current);
        }
        if (entry != current) {
            switch_to(sink, entry);
        }
        camu_clock_resume(&entry->clock, at);
        break;
    case LIANA_PAUSE_PAUSE:
        if (prev_target) {
            al_assert(current);
            al_assert(!ENTRY_ENDED(current));
            al_assert(prev_target != entry);
            if (entry == current) {
                // pause_and_swap_to() negated.
                sink->target = NULL;
            } else {
                sink->target = entry;
            }
            if (prev_target != (struct camu_sink_entry *)0xb00b) {
                camu_clock_pause(&prev_target->clock, at);
            }
        } else {
            al_assert(entry != current);
            pause_and_swap_to(sink, entry, at);
        }
        break;
    case LIANA_PAUSE_BOTH:
        if (prev_target) {
            al_assert(current);
            al_assert(!ENTRY_ENDED(current));
            al_assert(prev_target != entry);
            if (entry == current) {
                sink->target = NULL;
                camu_audio_buffer_resync(&entry->audio.buf);
            } else {
                sink->target = entry;
            }
            if (prev_target != (struct camu_sink_entry *)0xb00b) {
                camu_clock_pause(&prev_target->clock, at);
            }
        } else {
            al_assert(entry != current);
            pause_and_swap_to(sink, entry, at);
        }
        camu_clock_resume(&entry->clock, at);
        break;
    }

out:
    nn_mutex_unlock(&sink->lock);
    if (op != LIANA_SINK_BUFFER) {
        maybe_cleanup_old_entries(sink);
    }

    nn_packet_stream_return_packet(conn->stream, packet);

    return false;
}

static bool pause_command_callback(void *userdata, struct nn_rpc_connection *conn,
    struct nn_packet *packet, struct nn_packet *rpacket)
{
    struct camu_sink *sink = (struct camu_sink *)userdata;
    (void)rpacket;

    u64 id = LOCAL_ENTRY_ID(sink, nn_packet_read_u32(packet));
    s32 sequence = nn_packet_read_s32(packet);
    u64 at = nn_packet_read_u64(packet);
    u8 pause = nn_packet_read_u8(packet);

    struct camu_sink_entry *entry = get_entry_from_id(sink, id);
    if (!entry) goto out;
    nn_mutex_lock(&sink->lock);
    // As long as the list discards skips with a non-current sequence, this should hold true.
    al_assert(entry->sequence == sequence);
    log_trace("pause("ENTRY_FMT"), %s, audio_state: %hhu, video_state: %hhu.",
        ENTRY_ARG(entry), lia_pause_op_name(pause), AUDIO_STATE(entry), VIDEO_STATE(entry));
    if (sink->local) at = 0;
    switch (pause) {
    case LIANA_PAUSE_PAUSE: {
        entry->paused = true;
        camu_clock_pause(&entry->clock, at);
        log_info("Clock paused.");
        // Audio will be stopped in a BUFFER_PAUSED callback.
        // Video will be stopped in a CLOCK_PAUSED callback.
        break;
    }
    case LIANA_PAUSE_RESUME: {
        entry->paused = false;
        camu_clock_resume(&entry->clock, at);
        log_info("Clock resumed.");
        if (!VIDEO_ENDED_OR_EMPTY(entry) && !VIDEO_IS_STATIC(entry)) {
            queue_cmd(sink, CMD(START, .v.u = CAMU_SINK_VIDEO));
        }
        if (!AUDIO_ENDED_OR_EMPTY(entry)) {
            if (!sink->local) {
                camu_audio_buffer_resync(&entry->audio.buf);
            }
            queue_cmd(sink, CMD(START, .v.u = CAMU_SINK_AUDIO));
        }
        break;
    }
    }
    nn_mutex_unlock(&sink->lock);

out:
    nn_packet_stream_return_packet(conn->stream, packet);

    return false;
}

static bool seek_command_callback(void *userdata, struct nn_rpc_connection *conn,
    struct nn_packet *packet, struct nn_packet *rpacket)
{
    struct camu_sink *sink = (struct camu_sink *)userdata;
    (void)rpacket;

    u64 id = LOCAL_ENTRY_ID(sink, nn_packet_read_u32(packet));
    s32 sequence = nn_packet_read_s32(packet);
    u64 at = nn_packet_read_u64(packet);
    u64 pos = nn_packet_read_u64(packet);
    u32 reset_token = nn_packet_read_u32(packet);

    struct camu_sink_entry *entry = get_entry_from_id(sink, id);
    if (!entry) goto out;
    nn_mutex_lock(&sink->lock);
    entry->sequence = sequence;
    log_trace("seek("ENTRY_FMT", %.2f), reset_token: %u.", ENTRY_ARG(entry), pos / 1000000.0, reset_token);
    entry->reset_token = reset_token;
    nn_mutex_unlock(&sink->lock);
    // The rest of the seek is handled in CLIENT_REMOVE_BUFFERS/RESUME_AT/RECONNECTED.
    lia_client_seek(&entry->client, pos, at);

out:
    nn_packet_stream_return_packet(conn->stream, packet);

    return false;
}

static struct nn_rpc_command commands[] = {
    { .op = CAMU_SINK_SET, .callback = set_command_callback, .userdata = NULL },
    { .op = CAMU_SINK_PAUSE, .callback = pause_command_callback, .userdata = NULL },
    { .op = CAMU_SINK_SEEK, .callback = seek_command_callback, .userdata = NULL }
};

static void identify_callback(void *userdata, struct nn_rpc_connection *conn, struct nn_packet *packet)
{
    struct camu_sink *sink = (struct camu_sink *)userdata;
#ifndef CAMU_DIRECT_MODE
    if (sink->type == NNWT_SOCKET_UNIX) {
        log_info("Sink connected to %.*s.", al_str_x(&sink->addr));
    } else {
        log_info("Sink connected to %.*s:%hu.", al_str_x(&sink->addr), sink->port);
    }
#else
    (void)sink;
    log_info("Sink directly bridged to server.");
#endif
    nn_packet_stream_return_packet(conn->stream, packet);
}

static void identify_on_connection(struct camu_sink *sink)
{
    struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_IDENTIFY);
    nn_packet_write_u8(packet, CAMU_SINK);
    nn_packet_write_str(packet, &sink->name);
    nn_rpc_connection_command(sink->conn, packet, identify_callback, sink);
}

static void connection_callback(void *userdata, struct nn_rpc_connection *conn)
{
    struct camu_sink *sink = (struct camu_sink *)userdata;
    nn_timer_stop(&sink->reconnect_timer);
    if (sink->conn) al_assert(sink->conn == conn);
    sink->conn = conn;
    sink->connected = true;
    sink->connection_number = al_u16_inc_wrap(sink->connection_number);
    if (sink->connection_number == 0) sink->connection_number = 1;
    identify_on_connection(sink);
}

static void reconnect_timer_callback(void *userdata, struct nn_timer *timer)
{
    struct camu_sink *sink = (struct camu_sink *)userdata;
    (void)timer;
    if (sink->conn) {
        // If the client was connecting, reconnect() will force a disconnect before reconnecting.
        log_warn("Forcing reconnect due to timeout.");
    }
    sink->conn = nn_rpc_reconnect(&sink->client, &sink->addr, sink->port);
}

static void connection_closed_callback(void *userdata, struct nn_rpc_connection *conn)
{
    struct camu_sink *sink = (struct camu_sink *)userdata;
    bool reconnect = sink->conn != NULL;
    bool disconnected = sink->conn && sink->connection_number > 0;
    if (sink->conn) {
        al_assert(sink->conn == conn);
        sink->conn = NULL;
        sink->connected = false;
    } else {
        al_assert(!reconnect || !sink->connected);
    }
    if (reconnect) {
        if (disconnected) {
            log_warn("Connection to server closed, attempting reconnect...");
            nn_rpc_reconnect(&sink->client, &sink->addr, sink->port);
        } else {
            log_warn("Failed to connect to server, trying again...");
            nn_timer_again(&sink->reconnect_timer);
        }
    }
}

bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop,
    struct camu_mixer *mixer, struct camu_renderer *renderer)
{
    sink->loop = loop;
    nn_rpc_init(&sink->client, sink->loop, connection_callback, connection_closed_callback, sink);
    sink->conn = NULL;
    sink->connection_number = 0;
    nn_mutex_init(&sink->lock);
    nn_timer_init(&sink->reconnect_timer, sink->loop, reconnect_timer_callback, sink);
    // This is also effectively a timeout for attempted reconnects.
    nn_timer_set_repeat(&sink->reconnect_timer, NNWT_TS_FROM_USEC(1000000));
    nn_signal_init(&sink->queue_signal, sink->loop, queue_signal_callback, sink);
    nn_signal_start(&sink->queue_signal);
    camu_queue_init(sink->queue);
    sink->current = NULL;
    sink->target = NULL;
    sink->suspended = NULL;
    al_array_init(sink->previous);
    al_array_init(sink->entries);
    // Start high to exercise the wrapping path.
    sink->lru = SINK_LRU_MAX - al_random_int(0, ENTRY_MAX_AGE);
    mixer->callback = mixer_callback;
    mixer->userdata = sink;
    sink->audio.state = SINK_PAUSED;
    sink->audio.mixer = mixer;
    sink->video.state = SINK_PAUSED;
    sink->video.renderer = renderer;
    return true;
}

bool camu_sink_connect(struct camu_sink *sink, str *name, u8 type, str *addr, u16 port)
{
    al_str_clone(&sink->name, name);
    sink->type = type;
    al_str_clone(&sink->addr, addr);
    sink->port = port;
    sink->conn = nn_rpc_prepare_client(&sink->client);
    al_assert(sink->callback);
    for (u32 i = 0; i < ARRAY_SIZE(commands); i++) {
        commands[i].userdata = sink;
        nn_rpc_add_command(&sink->client, &commands[i]);
    }
#ifdef CAMU_DIRECT_MODE
    // Note that direct_connect() runs connection_callback() directly.
    nn_multiplex_direct_connect(sink->conn->stream, CAMU_MULTIPLEX_RPC);
#else
    nn_rpc_connect(&sink->client, CAMU_MULTIPLEX_RPC, sink->type, &sink->addr, sink->port);
#endif
    return true;
}

struct camu_sink_entry *camu_sink_get_current(struct camu_sink *sink)
{
    nn_mutex_lock(&sink->lock);
    return sink->current;
}

void camu_sink_return_current(struct camu_sink *sink)
{
    nn_mutex_unlock(&sink->lock);
}

void camu_sink_skip(struct camu_sink *sink, s32 n)
{
    nn_mutex_lock(&sink->lock);
    struct camu_sink_entry *current = get_entry_for_command(sink);
    nn_mutex_unlock(&sink->lock);
    queue_cmd(sink, CMD(SKIP, .v.i = n, .opaque = current));
}

void camu_sink_toggle_pause(struct camu_sink *sink)
{
    nn_mutex_lock(&sink->lock);
    struct camu_sink_entry *current = get_entry_for_command(sink);
    nn_mutex_unlock(&sink->lock);
    if (!current) return;
    bool armed_for_pause = false;
    camu_clock_external_pause(&current->clock);
    f64 pts = camu_clock_get_pts(&current->clock, 0.0, false, &armed_for_pause);
    queue_cmd(sink, CMD(TOGGLE_PAUSE, .v.f = pts, .opaque = current));
}

void camu_sink_seek(struct camu_sink *sink, void *value, u8 mode)
{
    nn_mutex_lock(&sink->lock);
    struct camu_sink_entry *current = get_entry_for_command(sink);
    u64 duration = 0;
    f64 pts = 0.0;
    if (current) {
        duration = current->client.duration;
        pts = camu_clock_get_last_pts(&current->clock);
    }
    nn_mutex_unlock(&sink->lock);
    if (!current || duration == 0) return;
    struct camu_sink_cmd cmd = {
        .op = SEEK,
        .opaque = current
    };
    switch (mode) {
    case CAMU_SEEK_POS: {
        u64 pos = *(u64 *)value;
        cmd.v.u = pos;
        break;
    }
    case CAMU_SEEK_RELATIVE: {
        f64 offset = *(f64 *)value;
        pts = MAX(pts + offset, 0.0);
        cmd.v.u = (u64)(pts * 1000000);
        break;
    }
    case CAMU_SEEK_PERCENT: {
        // There's probably a way to lose less precision here.
        f64 percent = *(f64 *)value;
        cmd.v.u = (u64)(duration * percent);
        break;
    }
    }
    queue_cmd(sink, cmd);
}

void camu_sink_reseek(struct camu_sink *sink)
{
    queue_cmd(sink, CMD(RESEEK));
}

void camu_sink_shuffle(struct camu_sink *sink)
{
    queue_cmd(sink, CMD(SHUFFLE));
}

void camu_sink_stop(struct camu_sink *sink)
{
    queue_cmds(sink, 3,
        CMD(STOP,          .v.u = CAMU_SINK_AUDIO),
        CMD(CLEAR_BUFFERS, .v.u = CAMU_SINK_AUDIO),
        CMD(CLOSE)
    );
}

void camu_sink_close(struct camu_sink *sink)
{
    nn_timer_stop(&sink->reconnect_timer);
    if (sink->conn) {
        struct nn_rpc_connection *conn = sink->conn;
        sink->conn = NULL; // Signal to connection_closed_callback() we're done.
        nn_rpc_conn_disconnect(conn);
    }
    struct camu_sink_entry *entry;
    al_array_foreach_rev(sink->entries, i, entry) {
        al_array_remove_at(sink->entries, i);
        maybe_disconnect_entry(entry);
    }
}

void camu_sink_free(struct camu_sink *sink)
{
    al_assert(!sink->entries.count);
    al_array_free(sink->entries);
    nn_rpc_free(&sink->client);
    camu_queue_free(sink->queue);
    nn_mutex_destroy(&sink->lock);
    al_str_free(&sink->name);
    al_str_free(&sink->addr);
}