summaryrefslogtreecommitdiff
path: root/src/libsink
diff options
context:
space:
mode:
Diffstat (limited to 'src/libsink')
-rw-r--r--src/libsink/sink.c338
-rw-r--r--src/libsink/sink.h20
2 files changed, 176 insertions, 182 deletions
diff --git a/src/libsink/sink.c b/src/libsink/sink.c
index bd33f46..984a41c 100644
--- a/src/libsink/sink.c
+++ b/src/libsink/sink.c
@@ -50,7 +50,7 @@ enum {
};
// Number of entries to keep buffered at one time.
-#define ENTRY_MAX_AGE 6
+#define ENTRY_MAX_AGE 7
// If a buffer is still INIT or QUEUED after an entry is configured, it's "empty".
#define BUFFER_EMPTY(buf) ((buf)->state == BUFFER_INIT || (buf)->state == BUFFER_QUEUED)
@@ -82,9 +82,9 @@ enum {
#endif
#if defined CAMU_SCREEN_THREADED && defined CAMU_MIXER_THREADED
-#define BLOCKING_SLEEP(delay) aki_thread_sleep(delay)
+#define BLOCKING_SLEEP(delay) nn_thread_sleep(delay)
#else
-#define BLOCKING_SLEEP(delay) aki_event_loop_sleep(sink->loop, delay)
+#define BLOCKING_SLEEP(delay) nn_event_loop_sleep(sink->loop, delay)
#endif
static inline bool entry_audio_buffer_held(struct camu_sink_entry *entry)
@@ -223,7 +223,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
{
switch (cmd->op) {
case START: {
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
switch (cmd->value.i) {
case CAMU_SINK_AUDIO:
if (sink->audio.state == SINK_PAUSED) {
@@ -240,11 +240,11 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
break;
#endif
}
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
break;
}
case STOP: {
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
switch (cmd->value.i) {
case CAMU_SINK_AUDIO:
if (sink->audio.state == SINK_PLAYING) {
@@ -261,25 +261,25 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
break;
#endif
}
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
break;
}
case SKIP: {
if (!sink->conn) return;
- struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
- aki_packet_write_str(packet, &sink->default_list);
- aki_packet_write_u8(packet, CAMU_LIST_SKIP);
- aki_packet_write_s32(packet, get_sequence_for_command(sink));
- aki_packet_write_s32(packet, (s32)cmd->value.i);
- aki_rpc_connection_command(sink->conn, packet, NULL, NULL);
+ 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));
+ nn_packet_write_s32(packet, (s32)cmd->value.i);
+ nn_rpc_connection_command(sink->conn, packet, NULL, NULL);
break;
}
case SHUFFLE: {
if (!sink->conn) return;
- struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
- aki_packet_write_str(packet, &sink->default_list);
- aki_packet_write_u8(packet, CAMU_LIST_SHUFFLE);
- aki_rpc_connection_command(sink->conn, packet, NULL, NULL);
+ 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 TOGGLE_PAUSE: {
@@ -287,30 +287,30 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
sink_local_pause(sink, (struct camu_sink_entry *)cmd->opaque);
#else
if (!sink->conn) return;
- struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
- aki_packet_write_str(packet, &sink->default_list);
- aki_packet_write_u8(packet, CAMU_LIST_TOGGLE_PAUSE);
- aki_packet_write_s32(packet, get_sequence_for_command(sink));
- aki_packet_write_f64(packet, cmd->value.f);
- aki_rpc_connection_command(sink->conn, packet, NULL, NULL);
+ 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));
+ nn_packet_write_f64(packet, cmd->value.f);
+ nn_rpc_connection_command(sink->conn, packet, NULL, NULL);
#endif
break;
}
case SEEK: {
if (!sink->conn) return;
- struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
- aki_packet_write_str(packet, &sink->default_list);
- aki_packet_write_u8(packet, CAMU_LIST_SEEK);
+ 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);
struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
if (entry) {
- aki_packet_write_s32(packet, entry->sequence);
- aki_packet_write_u32(packet, entry->id);
+ nn_packet_write_s32(packet, entry->sequence);
+ nn_packet_write_u32(packet, entry->id);
} else {
- aki_packet_write_s32(packet, LIANA_SEQUENCE_ANY);
- aki_packet_write_u32(packet, 0);
+ nn_packet_write_s32(packet, LIANA_SEQUENCE_ANY);
+ nn_packet_write_u32(packet, 0);
}
- aki_packet_write_f64(packet, cmd->value.f);
- aki_rpc_connection_command(sink->conn, packet, NULL, NULL);
+ nn_packet_write_f64(packet, cmd->value.f);
+ nn_rpc_connection_command(sink->conn, packet, NULL, NULL);
break;
}
case RESEEK: {
@@ -320,23 +320,23 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
}
case UNSET: {
if (!sink->conn) return;
- struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
- aki_packet_write_str(packet, &sink->default_list);
- aki_packet_write_u8(packet, CAMU_LIST_UNSET);
- aki_rpc_connection_command(sink->conn, packet, NULL, NULL);
+ 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_UNSET);
+ nn_rpc_connection_command(sink->conn, packet, NULL, NULL);
break;
}
case END: {
if (!sink->conn) return;
- struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
- aki_packet_write_str(packet, &sink->default_list);
- aki_packet_write_u8(packet, CAMU_LIST_END);
- aki_packet_write_u32(packet, (u32)cmd->value.u);
- aki_rpc_connection_command(sink->conn, packet, NULL, NULL);
+ 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, (u32)cmd->value.u);
+ nn_rpc_connection_command(sink->conn, packet, NULL, NULL);
break;
}
case CLOSE: {
- aki_signal_stop(&sink->queue_signal);
+ nn_signal_stop(&sink->queue_signal);
sink->callback(sink->userdata, CAMU_SINK_EXIT, 0, NULL);
return;
}
@@ -358,7 +358,7 @@ static void queue_signal_callback(void *userdata)
static void queue_cmd(struct camu_sink *sink, struct camu_sink_cmd cmd)
{
camu_queue_push(sink->queue, cmd);
- aki_signal_send(&sink->queue_signal);
+ nn_signal_send(&sink->queue_signal);
}
static void mixer_callback(void *userdata, u8 op)
@@ -373,7 +373,7 @@ static void mixer_callback(void *userdata, u8 op)
}
}
-bool camu_sink_init(struct camu_sink *sink, struct aki_event_loop *loop,
+bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop,
struct camu_mixer *mixer
#ifndef CAMU_SINK_NO_VIDEO
, struct camu_renderer *renderer
@@ -381,9 +381,9 @@ bool camu_sink_init(struct camu_sink *sink, struct aki_event_loop *loop,
)
{
sink->loop = loop;
- aki_mutex_init(&sink->mutex);
- aki_signal_init(&sink->queue_signal, sink->loop, queue_signal_callback, sink);
- aki_signal_start(&sink->queue_signal);
+ nn_mutex_init(&sink->mutex);
+ nn_signal_init(&sink->queue_signal, sink->loop, queue_signal_callback, sink);
+ nn_signal_start(&sink->queue_signal);
camu_queue_init(sink->queue);
sink->queued = NULL;
sink->current = NULL;
@@ -415,13 +415,7 @@ static void maybe_remove_previous(struct camu_sink *sink)
// An obvious example of this is at the point an entry gets freed. See LIANA_CLIENT_REMOVE_BUFFERS.
static void maybe_remove_from_previous(struct camu_sink *sink, struct camu_sink_entry *entry)
{
- struct camu_sink_entry *rentry;
- // al_array_remove_all()
- al_array_foreach_rev(sink->previous, i, rentry) {
- if (rentry == entry) {
- al_array_remove_at(sink->previous, i);
- }
- }
+ al_array_remove_all(sink->previous, entry);
}
// Every call to maybe_add_to_previous() must map to a remove_entry_buffers().
@@ -430,10 +424,10 @@ static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry
{
al_assert(previous != target && !previous->ended);
- struct camu_sink_entry *rentry;
- al_array_foreach_rev(sink->previous, i, rentry) {
+ struct camu_sink_entry *entry;
+ al_array_foreach_rev(sink->previous, i, entry) {
// If the entry we are about to set is in previous, run the queue.
- if (rentry == target) {
+ if (entry == target) {
maybe_remove_previous(sink);
break;
}
@@ -441,7 +435,12 @@ static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry
// If neither the entries audio or video buffer is ADDED, we don't care about adding it
// to previous (waiting for the next added entry to remove it).
- if (!(previous->audio.state == BUFFER_ADDED || previous->video.state == BUFFER_ADDED)) {
+#ifndef CAMU_SINK_NO_VIDEO
+ if (!(previous->audio.state == BUFFER_ADDED || previous->video.state == BUFFER_ADDED))
+#else
+ if (!(previous->audio.state == BUFFER_ADDED))
+#endif
+ {
remove_entry_buffers(sink, previous);
return;
}
@@ -500,7 +499,7 @@ static void maybe_cleanup_old_entries(struct camu_sink *sink)
void add_audio_if_set_and_buffered(struct camu_sink_entry *entry)
{
al_assert(!entry->ended);
- al_assert(entry->audio.state != BUFFER_ADDED && entry->audio.state != BUFFER_QUEUED);
+ al_assert(entry->audio.state != BUFFER_ADDED);
if (entry->audio.state == BUFFER_ENDED) {
al_log_warn("sink", "Tried to add an ended audio buffer.");
return;
@@ -548,7 +547,7 @@ void add_audio_if_set_and_buffered(struct camu_sink_entry *entry)
#ifndef CAMU_SINK_NO_VIDEO
void add_video_if_set_and_buffered(struct camu_sink_entry *entry)
{
- al_assert(entry->video.state != BUFFER_ADDED && entry->video.state != BUFFER_QUEUED);
+ al_assert(entry->video.state != BUFFER_ADDED);
if (entry->video.state == BUFFER_ENDED) {
al_log_warn("sink", "Tried to add an ended video buffer.");
return;
@@ -585,6 +584,7 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
if (sink->current) {
struct camu_sink_entry *current = sink->current;
al_assert(current != target);
+
if (!current->ended) {
if (target->ended) {
remove_entry_buffers(sink, current);
@@ -599,8 +599,9 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
#endif
al_assert(AUDIO_REMOVED_OR_EMPTY(current) && VIDEO_REMOVED_OR_EMPTY(current));
}
+
if (sink->reconnecting) {
- al_assert(sink->reconnecting == sink->current);
+ al_assert(sink->reconnecting == current);
sink->reconnecting = NULL;
}
}
@@ -656,9 +657,9 @@ static void audio_buffer_callback(void *userdata, u8 op)
struct camu_sink *sink = entry->sink;
switch (op) {
case CAMU_BUFFER_BUFFERED:
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
add_audio_if_set_and_buffered(entry);
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
break;
case CAMU_BUFFER_CORK:
lia_vcr_cork(entry->audio.track);
@@ -667,7 +668,7 @@ static void audio_buffer_callback(void *userdata, u8 op)
lia_vcr_uncork(entry->audio.track);
break;
case CAMU_BUFFER_PAUSED:
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
al_log_info("sink", "Audio buffer paused.");
if (!entry->audio.ignore_paused) {
queue_cmd(sink, (struct camu_sink_cmd){
@@ -675,11 +676,11 @@ static void audio_buffer_callback(void *userdata, u8 op)
.value.i = CAMU_SINK_AUDIO
});
}
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
break;
case CAMU_BUFFER_EOF: {
lia_vcr_cork(entry->audio.track);
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
al_log_info("sink", "Audio EOF.");
if (entry->audio.state == BUFFER_ADDED) {
remove_entry_audio_buffer(sink, entry);
@@ -694,7 +695,7 @@ static void audio_buffer_callback(void *userdata, u8 op)
if (run_queue) {
end_entry_and_advance_queue(sink, entry);
}
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
break;
}
}
@@ -707,9 +708,9 @@ static void video_buffer_callback(void *userdata, u8 op)
struct camu_sink *sink = entry->sink;
switch (op) {
case CAMU_BUFFER_BUFFERED: {
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
add_video_if_set_and_buffered(entry);
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
break;
}
case CAMU_BUFFER_CORK:
@@ -721,7 +722,7 @@ static void video_buffer_callback(void *userdata, u8 op)
case CAMU_BUFFER_EOF: {
lia_vcr_cork(entry->video.track);
bool swapped = false;
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
al_log_info("sink", "Video EOF.");
if (!ENTRY_IS_SINGLE_FRAME(entry)) {
if (entry->video.state == BUFFER_ADDED) {
@@ -733,7 +734,7 @@ static void video_buffer_callback(void *userdata, u8 op)
swapped = end_entry_and_advance_queue(sink, entry);
}
}
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
if (!swapped) {
queue_cmd(sink, (struct camu_sink_cmd){
.op = STOP,
@@ -762,14 +763,12 @@ 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) {
- aki_mutex_lock(&sink->mutex);
- if (entry == sink->current) {
- if (sink->target) {
- switch_to(sink, sink->target);
- sink->target = NULL;
- }
+ nn_mutex_lock(&sink->mutex);
+ if (entry == sink->current && sink->target) {
+ switch_to(sink, sink->target);
+ sink->target = NULL;
}
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
}
}
@@ -817,7 +816,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
lia_client_disconnect(&entry->client);
return;
}
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
if (entry->audio.state == BUFFER_QUEUED) {
entry->audio.state = BUFFER_SET_OR_BUFFERED;
} else if (entry->audio.state == BUFFER_INIT) {
@@ -826,7 +825,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
assert(false);
}
if (VIDEO_ADDED_OR_EMPTY(entry)) evaluate_latency(sink, entry);
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
break;
#ifndef CAMU_SINK_NO_VIDEO
case CAMU_STREAM_VIDEO:
@@ -836,7 +835,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
lia_client_disconnect(&entry->client);
return;
}
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
if (entry->video.state == BUFFER_QUEUED) {
entry->video.state = BUFFER_SET_OR_BUFFERED;
} else if (entry->video.state == BUFFER_INIT) {
@@ -845,7 +844,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
assert(false);
}
if (AUDIO_ADDED_OR_EMPTY(entry)) evaluate_latency(sink, entry);
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
break;
case CAMU_STREAM_SUBTITLE:
#ifdef CAMU_HAVE_FFMPEG
@@ -907,7 +906,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
case LIANA_CLIENT_REMOVE_BUFFERS: {
bool reconnect = *(bool *)opaque;
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
// We need to call this if entry was added to previous then,
// - it's being cleaned up after (ENTRY_MAX_AGE - 1) entries were added but none buffered.
@@ -917,11 +916,10 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
if (reconnect) {
entry->ended = false;
if (entry == sink->current) {
+ sink->reconnecting = entry;
if (sink->target) {
switch_to(sink, sink->target);
sink->target = NULL;
- } else {
- sink->reconnecting = sink->current;
}
}
}
@@ -949,7 +947,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
}
#endif
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
while ( // Block until buffers are removed.
#ifndef CAMU_SINK_NO_VIDEO
@@ -957,7 +955,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
#else
(!skip_audio && entry_audio_buffer_held(entry))
#endif
- ) { BLOCKING_SLEEP(AKI_TS_FROM_USEC(2500)); }
+ ) { BLOCKING_SLEEP(NNWT_TS_FROM_USEC(2500)); }
if (reconnect) {
if (BUFFER_NOT_EMPTY(&entry->audio)) {
@@ -974,13 +972,13 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
}
case LIANA_CLIENT_RESUME_AT: {
struct lia_timing *time = (struct lia_timing *)opaque;
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
camu_clock_seek(&entry->clock, time->seek_pos / 1000000.0, time->at);
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
break;
}
case LIANA_CLIENT_RECONNECTED: {
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
if (entry == sink->reconnecting) {
if (entry->audio.state > BUFFER_QUEUED) {
add_audio_if_set_and_buffered(entry);
@@ -991,7 +989,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
}
#endif
}
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
break;
}
case LIANA_CLIENT_EOF: {
@@ -1015,11 +1013,16 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
break;
}
case LIANA_CLIENT_CLOSED: {
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
+
if (entry == sink->target) {
maybe_unset_current(sink);
sink->target = NULL;
} else if (entry == sink->current) {
+ if (sink->reconnecting) {
+ al_assert(sink->reconnecting == sink->current);
+ sink->reconnecting = NULL;
+ }
// current's buffers will already be removed.
sink->current = NULL;
if (sink->target) {
@@ -1031,26 +1034,17 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
}/* else if (entry == sink->queued) {
sink->queued = NULL;
}*/
+
lia_client_free(&entry->client);
camu_audio_buffer_free(&entry->audio.buf);
#ifndef CAMU_SINK_NO_VIDEO
camu_video_buffer_free(&entry->video.buf);
#endif
- bool removed = false;
- struct camu_sink_entry *rentry;
- al_array_foreach(sink->entries, i, rentry) {
- if (rentry == entry) {
- al_array_remove_at(sink->entries, i);
- removed = true;
- al_log_info("sink", "Entry closed by disconnect.");
- break;
- }
- }
+ bool removed;
+ al_array_check_remove(sink->entries, entry, removed);
+ nn_mutex_unlock(&sink->mutex);
al_free(entry);
- if (!removed) {
- al_log_info("sink", "Entry closed by cleanup.");
- }
- aki_mutex_unlock(&sink->mutex);
+ al_log_info("sink", "Entry closed by %s.", removed ? "disconnect" : "cleanup");
break;
}
}
@@ -1097,16 +1091,16 @@ static struct camu_sink_entry *get_entry_from_id(struct camu_sink *sink, u32 id)
return NULL;
}
-static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn,
- struct aki_packet *packet, struct aki_packet *rpacket)
+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)conn;
(void)rpacket;
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
- u8 op = aki_packet_read_u8(packet);
+ u8 op = nn_packet_read_u8(packet);
if (op == LIANA_SINK_UNSET) {
maybe_unset_current(sink);
goto out;
@@ -1114,18 +1108,18 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn
// Liana node info.
str addr;
- aki_packet_read_str(packet, &addr);
- u16 port = aki_packet_read_u16(packet);
- u32 node_id = aki_packet_read_u32(packet);
+ nn_packet_read_str(packet, &addr);
+ u16 port = nn_packet_read_u16(packet);
+ u32 node_id = nn_packet_read_u32(packet);
// List entry info.
- u32 id = aki_packet_read_u32(packet);
- s32 sequence = aki_packet_read_s32(packet);
- u64 at = aki_packet_read_u64(packet);
- u64 seek_pos = aki_packet_read_u64(packet);
- u8 pause = aki_packet_read_u8(packet);
- bool previous_ended = aki_packet_read_bool(packet);
- bool ended = aki_packet_read_bool(packet);
+ u32 id = nn_packet_read_u32(packet);
+ s32 sequence = nn_packet_read_s32(packet);
+ u64 at = nn_packet_read_u64(packet);
+ u64 seek_pos = nn_packet_read_u64(packet);
+ u8 pause = nn_packet_read_u8(packet);
+ bool previous_ended = nn_packet_read_bool(packet);
+ bool ended = nn_packet_read_bool(packet);
(void)previous_ended;
bool created = false;
@@ -1177,7 +1171,7 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn
if (sink->target) {
sink->target = NULL;
} else {
- al_assert(sink->current != entry);
+ al_assert(entry != sink->current);
}
if (entry != sink->current) {
switch_to(sink, entry);
@@ -1188,7 +1182,7 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn
if (sink->target) {
sink->target = NULL;
} else {
- al_assert(sink->current != entry);
+ al_assert(entry != sink->current);
}
if (entry != sink->current) {
switch_to(sink, entry);
@@ -1240,32 +1234,32 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn
#endif
out:
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
if (op != LIANA_SINK_BUFFER) {
maybe_cleanup_old_entries(sink);
}
- aki_packet_free(packet);
+ nn_packet_free(packet);
return false;
}
-static bool pause_command_callback(void *userdata, struct aki_rpc_connection *conn,
- struct aki_packet *packet, struct aki_packet *rpacket)
+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)conn;
(void)rpacket;
- u32 id = aki_packet_read_u32(packet);
- s32 sequence = aki_packet_read_s32(packet);
- u64 at = aki_packet_read_u64(packet);
- u8 pause = aki_packet_read_u8(packet);
+ u32 id = 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;
al_assert(entry->sequence == sequence);
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
#ifdef CAMU_SINK_LOCAL
(void)at;
(void)pause;
@@ -1287,25 +1281,25 @@ static bool pause_command_callback(void *userdata, struct aki_rpc_connection *co
break;
}
#endif
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
out:
- aki_packet_free(packet);
+ nn_packet_free(packet);
return false;
}
-static bool seek_command_callback(void *userdata, struct aki_rpc_connection *conn,
- struct aki_packet *packet, struct aki_packet *rpacket)
+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)conn;
(void)rpacket;
- u32 id = aki_packet_read_u32(packet);
- s32 sequence = aki_packet_read_s32(packet);
+ u32 id = nn_packet_read_u32(packet);
+ s32 sequence = nn_packet_read_s32(packet);
(void)sequence;
- u64 at = aki_packet_read_u64(packet);
- u64 pos = aki_packet_read_u64(packet);
+ u64 at = nn_packet_read_u64(packet);
+ u64 pos = nn_packet_read_u64(packet);
struct camu_sink_entry *entry = get_entry_from_id(sink, id);
if (!entry) goto out;
@@ -1317,81 +1311,81 @@ static bool seek_command_callback(void *userdata, struct aki_rpc_connection *con
lia_client_seek(&entry->client, pos, at);
out:
- aki_packet_free(packet);
+ nn_packet_free(packet);
return false;
}
-static struct aki_rpc_command commands[] = {
+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 idd_callback(void *userdata, struct aki_packet *packet)
+static void idd_callback(void *userdata, struct nn_packet *packet)
{
struct camu_sink *sink = (struct camu_sink *)userdata;
(void)sink;
- aki_packet_free(packet);
+ nn_packet_free(packet);
}
-static void connection_callback(void *userdata, struct aki_rpc_connection *conn)
+static void connection_callback(void *userdata, struct nn_rpc_connection *conn)
{
struct camu_sink *sink = (struct camu_sink *)userdata;
sink->conn = conn;
- aki_timer_stop(&sink->reconnect_timer);
- struct aki_packet *packet = aki_rpc_get_packet(&sink->client, CAMU_SERVER_IDENTIFY);
- aki_packet_write_u8(packet, CAMU_SINK);
- aki_packet_write_str(packet, &sink->name);
- aki_rpc_connection_command(sink->conn, packet, idd_callback, sink);
+ nn_timer_stop(&sink->reconnect_timer);
+ 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, idd_callback, sink);
}
-static void reconnect_timer_callback(void *userdata, struct aki_timer *timer)
+static void reconnect_timer_callback(void *userdata, struct nn_timer *timer)
{
struct camu_sink *sink = (struct camu_sink *)userdata;
(void)timer;
- aki_rpc_reconnect(&sink->client, &sink->addr, sink->port);
+ nn_rpc_reconnect(&sink->client, &sink->addr, sink->port);
}
-static void connection_closed_callback(void *userdata, struct aki_rpc_connection *conn)
+static void connection_closed_callback(void *userdata, struct nn_rpc_connection *conn)
{
struct camu_sink *sink = (struct camu_sink *)userdata;
if (sink->conn) {
al_assert(sink->conn == conn);
sink->conn = NULL;
}
- aki_timer_again(&sink->reconnect_timer);
+ nn_timer_again(&sink->reconnect_timer);
}
bool camu_sink_connect(struct camu_sink *sink, u8 type, str *addr, u16 port, str *name)
{
al_str_clone(&sink->name, name);
sink->type = type;
- aki_timer_init(&sink->reconnect_timer, sink->loop, reconnect_timer_callback, sink);
- aki_timer_set_repeat(&sink->reconnect_timer, AKI_TS_FROM_USEC(1000000));
- aki_rpc_init(&sink->client, sink->loop, connection_callback, connection_closed_callback, sink);
+ nn_timer_init(&sink->reconnect_timer, sink->loop, reconnect_timer_callback, sink);
+ nn_timer_set_repeat(&sink->reconnect_timer, NNWT_TS_FROM_USEC(1000000));
+ nn_rpc_init(&sink->client, sink->loop, connection_callback, connection_closed_callback, sink);
for (u32 i = 0; i < ARRAY_SIZE(commands); i++) {
commands[i].userdata = sink;
al_assert(sink->callback);
- aki_rpc_add_command(&sink->client, &commands[i]);
+ nn_rpc_add_command(&sink->client, &commands[i]);
}
- if (!aki_rpc_prepare_client(&sink->client, sink->type, CAMU_MULTIPLEX_RPC)) {
+ if (!nn_rpc_prepare_client(&sink->client, sink->type, CAMU_MULTIPLEX_RPC)) {
return false;
}
al_str_clone(&sink->addr, addr);
sink->port = port;
- aki_rpc_connect(&sink->client, &sink->addr, sink->port);
+ nn_rpc_connect(&sink->client, &sink->addr, sink->port);
return true;
}
struct camu_sink_entry *camu_sink_get_current(struct camu_sink *sink)
{
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
return sink->current;
}
void camu_sink_return_current(struct camu_sink *sink)
{
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
}
void camu_sink_skip(struct camu_sink *sink, s32 n)
@@ -1411,9 +1405,9 @@ void camu_sink_shuffle(struct camu_sink *sink)
void camu_sink_toggle_pause(struct camu_sink *sink)
{
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
struct camu_sink_entry *current = sink->current;
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
if (current) {
queue_cmd(sink, (struct camu_sink_cmd){
.op = TOGGLE_PAUSE,
@@ -1425,9 +1419,9 @@ void camu_sink_toggle_pause(struct camu_sink *sink)
void camu_sink_seek(struct camu_sink *sink, f64 precent)
{
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
struct camu_sink_entry *current = sink->current;
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
queue_cmd(sink, (struct camu_sink_cmd){
.op = SEEK,
.value.f = precent,
@@ -1437,9 +1431,9 @@ void camu_sink_seek(struct camu_sink *sink, f64 precent)
void camu_sink_reseek(struct camu_sink *sink)
{
- aki_mutex_lock(&sink->mutex);
+ nn_mutex_lock(&sink->mutex);
struct camu_sink_entry *current = sink->current;
- aki_mutex_unlock(&sink->mutex);
+ nn_mutex_unlock(&sink->mutex);
queue_cmd(sink, (struct camu_sink_cmd){
.op = RESEEK,
.opaque = current
@@ -1471,9 +1465,9 @@ void camu_sink_stop(struct camu_sink *sink)
void camu_sink_close(struct camu_sink *sink)
{
- aki_timer_stop(&sink->reconnect_timer);
- aki_timer_disable(&sink->reconnect_timer);
- if (sink->conn) aki_rpc_conn_disconnect(sink->conn);
+ nn_timer_stop(&sink->reconnect_timer);
+ nn_timer_disable(&sink->reconnect_timer);
+ if (sink->conn) nn_rpc_conn_disconnect(sink->conn);
struct camu_sink_entry *entry;
al_array_foreach_rev(sink->entries, i, entry) {
al_array_remove_at(sink->entries, i);
@@ -1488,9 +1482,9 @@ void camu_sink_free(struct camu_sink *sink)
al_free(entry);
}
al_array_free(sink->entries);
- aki_rpc_free(&sink->client);
+ nn_rpc_free(&sink->client);
camu_queue_free(sink->queue);
- aki_mutex_destroy(&sink->mutex);
+ nn_mutex_destroy(&sink->mutex);
al_str_free(&sink->addr);
al_str_free(&sink->name);
}
diff --git a/src/libsink/sink.h b/src/libsink/sink.h
index 48b2b87..4f719d2 100644
--- a/src/libsink/sink.h
+++ b/src/libsink/sink.h
@@ -2,9 +2,9 @@
#include <al/str.h>
#include <al/array.h>
-#include <aki/rpc2.h>
-#include <aki/signal.h>
-#include <aki/timer.h>
+#include <nnwt/rpc2.h>
+#include <nnwt/signal.h>
+#include <nnwt/timer.h>
#include "../util/queue.h"
@@ -69,16 +69,16 @@ struct camu_sink_cmd {
};
struct camu_sink {
- struct aki_event_loop *loop;
+ struct nn_event_loop *loop;
str name;
u8 type;
str addr;
u16 port;
- struct aki_rpc client;
- struct aki_rpc_connection *conn;
- struct aki_mutex mutex;
- struct aki_timer reconnect_timer;
- struct aki_signal queue_signal;
+ struct nn_rpc client;
+ struct nn_rpc_connection *conn;
+ struct nn_mutex mutex;
+ struct nn_timer reconnect_timer;
+ struct nn_signal queue_signal;
queue(struct camu_sink_cmd) queue;
str default_list;
struct camu_sink_entry *current;
@@ -102,7 +102,7 @@ struct camu_sink {
void *userdata;
};
-bool camu_sink_init(struct camu_sink *sink, struct aki_event_loop *loop,
+bool camu_sink_init(struct camu_sink *sink, struct nn_event_loop *loop,
struct camu_mixer *mixer
#ifndef CAMU_SINK_NO_VIDEO
, struct camu_renderer *renderer
generated by cgit v1.2.3-101-g0448 (git 2.54.0) at 2026-07-25 18:14:17 +0000