summaryrefslogtreecommitdiff
path: root/src/libsink
diff options
context:
space:
mode:
Diffstat (limited to 'src/libsink')
-rw-r--r--src/libsink/sink.c285
-rw-r--r--src/libsink/sink.h2
2 files changed, 201 insertions, 86 deletions
diff --git a/src/libsink/sink.c b/src/libsink/sink.c
index 1569286..f588cfa 100644
--- a/src/libsink/sink.c
+++ b/src/libsink/sink.c
@@ -38,51 +38,64 @@ enum {
// Command queue commands.
enum {
- START,
+ START = 0,
STOP,
+ ADD_BUFFER,
+ REMOVE_BUFFER,
+ CLEAR,
+ CLOSE,
TOGGLE_PAUSE,
SEEK,
SKIP,
SHUFFLE,
RESEEK,
UNSET,
- END,
- CLOSE
+ END
};
// Number of entries to keep buffered at one time.
#define ENTRY_MAX_AGE 7
-// If a buffer is still INIT or QUEUED after an entry is configured, it's "empty".
+// If a buffer is still INIT or QUEUED after the entry is configured, it's "empty".
#define BUFFER_EMPTY(buf) ((buf)->state == BUFFER_INIT || (buf)->state == BUFFER_QUEUED)
#define BUFFER_NOT_EMPTY(buf) (!BUFFER_EMPTY(buf))
-#ifdef CAMU_SINK_NO_VIDEO
-#define VIDEO_ADDED_OR_EMPTY(entry) true
-#else
-#define VIDEO_ADDED_OR_EMPTY(entry) ((entry)->video.state == BUFFER_ADDED || BUFFER_EMPTY(&(entry)->video))
+#define AUDIO_EMPTY(entry) BUFFER_EMPTY(&(entry)->audio)
+#ifndef CAMU_SINK_NO_VIDEO
+#define VIDEO_EMPTY(entry) BUFFER_EMPTY(&(entry)->video)
#endif
-#define AUDIO_ADDED_OR_EMPTY(entry) ((entry)->audio.state == BUFFER_ADDED || BUFFER_EMPTY(&(entry)->audio))
-#ifdef CAMU_SINK_NO_VIDEO
-#define VIDEO_ENDED_OR_EMPTY(entry) true
+#define AUDIO_NOT_EMPTY(entry) BUFFER_NOT_EMPTY(&(entry)->audio)
+#ifndef CAMU_SINK_NO_VIDEO
+#define VIDEO_NOT_EMPTY(entry) BUFFER_NOT_EMPTY(&(entry)->video)
+#endif
+
+#define AUDIO_ADDED_OR_EMPTY(entry) ((entry)->audio.state == BUFFER_ADDED || BUFFER_EMPTY(&(entry)->audio))
+#ifndef CAMU_SINK_NO_VIDEO
+#define VIDEO_ADDED_OR_EMPTY(entry) ((entry)->video.state == BUFFER_ADDED || BUFFER_EMPTY(&(entry)->video))
#else
-#define VIDEO_ENDED_OR_EMPTY(entry) ((entry)->video.state == BUFFER_ENDED || BUFFER_EMPTY(&(entry)->video))
+#define VIDEO_ADDED_OR_EMPTY(entry) true
#endif
+
#define AUDIO_ENDED_OR_EMPTY(entry) ((entry)->audio.state == BUFFER_ENDED || BUFFER_EMPTY(&(entry)->audio))
+#ifndef CAMU_SINK_NO_VIDEO
+#define VIDEO_ENDED_OR_EMPTY(entry) ((entry)->video.state == BUFFER_ENDED || BUFFER_EMPTY(&(entry)->video))
+#else
+#define VIDEO_ENDED_OR_EMPTY(entry) true
+#endif
-#ifdef CAMU_SINK_NO_VIDEO
-#define VIDEO_REMOVED_OR_EMPTY(entry) true
+#define AUDIO_NOT_ADDED(entry) ((entry)->audio.state != BUFFER_ADDED)
+#ifndef CAMU_SINK_NO_VIDEO
+#define VIDEO_NOT_ADDED(entry) ((entry)->video.state != BUFFER_ADDED)
#else
-#define VIDEO_REMOVED_OR_EMPTY(entry) ((entry)->video.state != BUFFER_ADDED)
+#define VIDEO_NOT_ADDED(entry) true
#endif
-#define AUDIO_REMOVED_OR_EMPTY(entry) ((entry)->audio.state != BUFFER_ADDED)
#ifndef CAMU_SINK_NO_VIDEO
-#define ENTRY_IS_SINGLE_FRAME(entry) camu_video_buffer_is_single_frame(&(entry)->video.buf)
+#define VIDEO_IS_SINGLE_FRAME(entry) camu_video_buffer_is_single_frame(&(entry)->video.buf)
#endif
-#if defined CAMU_SCREEN_THREADED && defined CAMU_MIXER_THREADED
+#if defined CAMU_SCREEN_THREADED && defined CAMU_MIXER_THREADED_START_STOP
#define BLOCKING_SLEEP(delay) nn_thread_sleep(delay)
#else
#define BLOCKING_SLEEP(delay) nn_event_loop_sleep(sink->loop, delay)
@@ -110,12 +123,48 @@ static inline bool entry_video_buffer_held(struct camu_sink_entry *entry)
}
#endif
+static void queue_cmd(struct camu_sink *sink, struct camu_sink_cmd cmd)
+{
+ camu_queue_push(sink->queue, cmd);
+ nn_signal_send(&sink->queue_signal);
+}
+
+static inline void add_entry_audio_buffer(struct camu_sink_entry *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, (struct camu_sink_cmd){
+ .op = ADD_BUFFER,
+ .value.i = CAMU_SINK_AUDIO,
+ .opaque = entry
+ });
+#endif
+}
+
+#ifndef CAMU_SINK_NO_VIDEO
+static inline void add_entry_video_buffer(struct camu_sink_entry *entry)
+{
+ struct camu_sink *sink = entry->sink;
+ sink->callback(sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
+}
+#endif
+
static void remove_entry_audio_buffer(struct camu_sink *sink, struct camu_sink_entry *entry)
{
al_assert(!entry->ended && entry->audio.state != BUFFER_ENDED);
al_assert(entry->audio.state != BUFFER_INIT);
if (entry->audio.state == BUFFER_ADDED) {
+#ifdef CAMU_MIXER_THREADED_START_STOP
sink->callback(sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
+#else
+ queue_cmd(sink, (struct camu_sink_cmd){
+ .op = REMOVE_BUFFER,
+ .value.i = CAMU_SINK_AUDIO,
+ .opaque = entry
+ });
+#endif
entry->audio.state = BUFFER_SET_OR_BUFFERED;
} else if (entry->audio.state == BUFFER_SET_OR_BUFFERED) {
entry->audio.state = BUFFER_CONFIGURED;
@@ -187,12 +236,12 @@ static void sink_local_pause(struct camu_sink *sink, struct camu_sink_entry *ent
{
if (camu_clock_is_paused(&entry->clock)) {
camu_clock_resume(&entry->clock, 0);
- if (!BUFFER_EMPTY(&entry->audio) && sink->audio.state == SINK_PAUSED) {
+ if (AUDIO_NOT_EMPTY(entry) && sink->audio.state == SINK_PAUSED) {
sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_AUDIO, NULL);
sink->audio.state = SINK_PLAYING;
}
#ifndef CAMU_SINK_NO_VIDEO
- if (!ENTRY_IS_SINGLE_FRAME(entry) && sink->video.state == SINK_PAUSED) {
+ if (!VIDEO_IS_SINGLE_FRAME(entry) && sink->video.state == SINK_PAUSED) {
sink->callback(sink->userdata, CAMU_SINK_START, CAMU_SINK_VIDEO, NULL);
sink->video.state = SINK_PLAYING;
}
@@ -200,7 +249,7 @@ static void sink_local_pause(struct camu_sink *sink, struct camu_sink_entry *ent
} else {
camu_clock_pause(&entry->clock, 0);
#ifndef CAMU_SINK_NO_VIDEO
- if (!ENTRY_IS_SINGLE_FRAME(entry) && sink->video.state == SINK_PLAYING) {
+ if (!VIDEO_IS_SINGLE_FRAME(entry) && sink->video.state == SINK_PLAYING) {
sink->callback(sink->userdata, CAMU_SINK_STOP, CAMU_SINK_VIDEO, NULL);
sink->video.state = SINK_PAUSED;
}
@@ -224,7 +273,6 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
{
switch (cmd->op) {
case START: {
- nn_mutex_lock(&sink->mutex);
switch (cmd->value.i) {
case CAMU_SINK_AUDIO:
if (sink->audio.state == SINK_PAUSED) {
@@ -241,11 +289,9 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
break;
#endif
}
- nn_mutex_unlock(&sink->mutex);
break;
}
case STOP: {
- nn_mutex_lock(&sink->mutex);
switch (cmd->value.i) {
case CAMU_SINK_AUDIO:
if (sink->audio.state == SINK_PLAYING) {
@@ -262,9 +308,54 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
break;
#endif
}
- nn_mutex_unlock(&sink->mutex);
break;
}
+ case ADD_BUFFER: {
+ struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
+ switch (cmd->value.i) {
+ case CAMU_SINK_AUDIO:
+ sink->callback(sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
+ break;
+#ifndef CAMU_SINK_NO_VIDEO
+ case CAMU_SINK_VIDEO:
+ sink->callback(sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
+ break;
+#endif
+ }
+ break;
+ }
+ case REMOVE_BUFFER: {
+ struct camu_sink_entry *entry = (struct camu_sink_entry *)cmd->opaque;
+ switch (cmd->value.i) {
+ case CAMU_SINK_AUDIO:
+ sink->callback(sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
+ break;
+#ifndef CAMU_SINK_NO_VIDEO
+ case CAMU_SINK_VIDEO:
+ sink->callback(sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
+ break;
+#endif
+ }
+ break;
+ }
+ case CLEAR: {
+ switch (cmd->value.i) {
+ case CAMU_SINK_AUDIO:
+ sink->callback(sink->userdata, CAMU_SINK_CLEAR, CAMU_SINK_AUDIO, NULL);
+ break;
+#ifndef CAMU_SINK_NO_VIDEO
+ case CAMU_SINK_VIDEO:
+ sink->callback(sink->userdata, CAMU_SINK_CLEAR, CAMU_SINK_VIDEO, NULL);
+ break;
+#endif
+ }
+ break;
+ }
+ case CLOSE: {
+ nn_signal_stop(&sink->queue_signal);
+ sink->callback(sink->userdata, CAMU_SINK_EXIT, 0, NULL);
+ return;
+ }
case SKIP: {
if (!sink->conn) return;
struct nn_packet *packet = nn_rpc_get_packet(&sink->client, CAMU_SERVER_LIST_ACTION);
@@ -338,11 +429,6 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
nn_rpc_connection_command(sink->conn, packet, NULL, NULL);
break;
}
- case CLOSE: {
- nn_signal_stop(&sink->queue_signal);
- sink->callback(sink->userdata, CAMU_SINK_EXIT, 0, NULL);
- return;
- }
}
}
@@ -358,17 +444,11 @@ 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);
- nn_signal_send(&sink->queue_signal);
-}
-
static void mixer_callback(void *userdata, u8 op)
{
struct camu_sink *sink = (struct camu_sink *)userdata;
if (op == CAMU_MIXER_EMPTY) {
- al_log_debug("sink", "Mixer empty.");
+ al_log_info("sink", "Mixer empty.");
queue_cmd(sink, (struct camu_sink_cmd){
.op = STOP,
.value.i = CAMU_SINK_AUDIO
@@ -510,31 +590,34 @@ void add_audio_if_set_and_buffered(struct camu_sink_entry *entry)
entry->audio.state = BUFFER_SET_OR_BUFFERED;
} else if (entry->audio.state == BUFFER_SET_OR_BUFFERED) {
entry->audio.state = BUFFER_ADDED;
-#ifndef CAMU_SINK_LOCAL
+ bool ignore_video = true;
+#ifndef CAMU_SINK_NO_VIDEO
+ ignore_video = VIDEO_EMPTY(entry) || VIDEO_IS_SINGLE_FRAME(entry);
+#endif
+ camu_audio_buffer_set_no_video(&entry->audio.buf, ignore_video);
+#ifdef CAMU_SINK_LOCAL
+ // If we're local we don't have to worry about syncing audio-only entries.
+ camu_audio_buffer_set_ignore_desync(&entry->audio.buf, ignore_video);
+#else
+ // Force resync.
camu_audio_buffer_unpause(&entry->audio.buf);
#endif
if (VIDEO_ADDED_OR_EMPTY(entry)) {
- entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
+ add_entry_audio_buffer(entry);
queue_cmd(entry->sink, (struct camu_sink_cmd){
.op = entry->audio.buffer_paused ? STOP : START,
.value.i = CAMU_SINK_AUDIO
});
#ifndef CAMU_SINK_NO_VIDEO
- bool stop_video = false;
- // Single frames are unconditionally added in add_video_if_set_and_buffered().
- if (!ENTRY_IS_SINGLE_FRAME(entry)) {
- if (entry->video.state == BUFFER_ADDED) {
- entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
- } else {
- // Entry has no video.
- stop_video = true;
- }
+ if (!ignore_video) {
+ // Single frames are unconditionally added in add_video_if_set_and_buffered().
+ add_entry_video_buffer(entry);
}
#endif
maybe_remove_previous(entry->sink);
#ifndef CAMU_SINK_NO_VIDEO
- // This must come after maybe_remove_previous().
- if (stop_video) {
+ if (VIDEO_EMPTY(entry)) {
+ // This must come after maybe_remove_previous().
queue_cmd(entry->sink, (struct camu_sink_cmd){
.op = STOP,
.value.i = CAMU_SINK_VIDEO
@@ -558,19 +641,19 @@ void add_video_if_set_and_buffered(struct camu_sink_entry *entry)
entry->video.state = BUFFER_SET_OR_BUFFERED;
} else if (entry->video.state == BUFFER_SET_OR_BUFFERED) {
entry->video.state = BUFFER_ADDED;
- bool single_frame = ENTRY_IS_SINGLE_FRAME(entry);
+ bool single_frame = VIDEO_IS_SINGLE_FRAME(entry);
if (AUDIO_ADDED_OR_EMPTY(entry)) {
if (entry->audio.state == BUFFER_ADDED) {
- entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
+ add_entry_audio_buffer(entry);
queue_cmd(entry->sink, (struct camu_sink_cmd){
.op = entry->audio.buffer_paused ? STOP : START,
.value.i = CAMU_SINK_AUDIO
});
}
- entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
+ add_entry_video_buffer(entry);
maybe_remove_previous(entry->sink);
} else if (single_frame) {
- entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
+ add_entry_video_buffer(entry);
}
// Stopping video here is needed if skipping from a video to an image.
queue_cmd(entry->sink, (struct camu_sink_cmd){
@@ -591,11 +674,11 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
maybe_add_to_previous(sink, current, target);
} else {
#ifndef CAMU_SINK_NO_VIDEO
- if (ENTRY_IS_SINGLE_FRAME(current)) {
+ if (VIDEO_IS_SINGLE_FRAME(current)) {
remove_entry_video_buffer(sink, current);
}
#endif
- al_assert(AUDIO_REMOVED_OR_EMPTY(current) && VIDEO_REMOVED_OR_EMPTY(current));
+ al_assert(AUDIO_NOT_ADDED(current) && VIDEO_NOT_ADDED(current));
}
if (sink->reconnecting) {
@@ -607,7 +690,7 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *target)
if (!target->ended) {
add_or_queue_entry(target);
#ifndef CAMU_SINK_NO_VIDEO
- } else if (ENTRY_IS_SINGLE_FRAME(target)) {
+ } else if (VIDEO_IS_SINGLE_FRAME(target)) {
add_video_if_set_and_buffered(target);
#endif
}
@@ -620,6 +703,7 @@ static void switch_to(struct camu_sink *sink, struct camu_sink_entry *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;
+ al_assert(target != current);
if (!current || current->ended) {
switch_to(sink, target);
} else {
@@ -693,11 +777,11 @@ static void audio_buffer_callback(void *userdata, u8 op)
// This assert likely doesn't matter due to the handling of the ENDED state.
al_assert(entry->audio.state == BUFFER_SET_OR_BUFFERED);
entry->audio.state = BUFFER_ENDED;
- bool run_queue = VIDEO_ENDED_OR_EMPTY(entry);
+ bool end_entry = VIDEO_ENDED_OR_EMPTY(entry);
#ifndef CAMU_SINK_NO_VIDEO
- run_queue = run_queue || ENTRY_IS_SINGLE_FRAME(entry);
+ end_entry = end_entry || VIDEO_IS_SINGLE_FRAME(entry);
#endif
- if (run_queue) {
+ if (end_entry) {
end_entry_and_advance_queue(sink, entry);
}
nn_mutex_unlock(&sink->mutex);
@@ -731,7 +815,10 @@ static void video_buffer_callback(void *userdata, u8 op)
bool swapped = false;
nn_mutex_lock(&sink->mutex);
al_log_info("sink", "Video EOF.");
- if (!ENTRY_IS_SINGLE_FRAME(entry)) {
+ if (AUDIO_NOT_EMPTY(entry)) {
+ camu_audio_buffer_set_no_video(&entry->audio.buf, true);
+ }
+ if (!VIDEO_IS_SINGLE_FRAME(entry)) {
if (entry->video.state == BUFFER_ADDED) {
remove_entry_video_buffer(sink, entry);
}
@@ -785,7 +872,7 @@ static void evaluate_latency(struct camu_sink *sink, struct camu_sink_entry *ent
{
#ifdef CAMU_SINK_LOCAL
#ifndef CAMU_SINK_NO_VIDEO
- if (BUFFER_NOT_EMPTY(&entry->audio) && BUFFER_NOT_EMPTY(&entry->video)) {
+ if (AUDIO_NOT_EMPTY(entry) && VIDEO_NOT_EMPTY(entry)) {
f64 audio = camu_mixer_get_latency(sink->audio.mixer);
s32 frames = audio / entry->video.buf.avg_frame_duration;
frames -= sink->video.renderer->get_latency(sink->video.renderer);
@@ -800,7 +887,7 @@ static void evaluate_latency(struct camu_sink *sink, struct camu_sink_entry *ent
// latency directly into the audio buffer.
f64 audio = camu_mixer_get_latency(sink->audio.mixer);
#ifndef CAMU_SINK_NO_VIDEO
- if (BUFFER_NOT_EMPTY(&entry->video)) {
+ if (VIDEO_NOT_EMPTY(entry)) {
s32 frames = audio / entry->video.buf.avg_frame_duration;
frames += sink->video.renderer->get_latency(sink->video.renderer);
camu_video_buffer_set_latency(&entry->video.buf, frames);
@@ -810,6 +897,19 @@ static void evaluate_latency(struct camu_sink *sink, struct camu_sink_entry *ent
#endif
}
+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 client_callback(void *userdata, u8 op, struct camu_codec_stream *stream, void *opaque)
{
struct camu_sink_entry *entry = (struct camu_sink_entry *)userdata;
@@ -833,7 +933,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
} else {
assert(false);
}
- if (VIDEO_ADDED_OR_EMPTY(entry)) evaluate_latency(sink, entry);
+ if (VIDEO_ADDED_OR_EMPTY(entry)) {
+ evaluate_latency(sink, entry);
+ }
nn_mutex_unlock(&sink->mutex);
break;
#ifndef CAMU_SINK_NO_VIDEO
@@ -852,7 +954,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
} else {
assert(false);
}
- if (AUDIO_ADDED_OR_EMPTY(entry)) evaluate_latency(sink, entry);
+ if (AUDIO_ADDED_OR_EMPTY(entry)) {
+ evaluate_latency(sink, entry);
+ }
nn_mutex_unlock(&sink->mutex);
break;
case CAMU_STREAM_SUBTITLE:
@@ -879,14 +983,14 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
struct camu_codec_frame *frame = (struct camu_codec_frame *)opaque;
switch (stream->type) {
case CAMU_STREAM_AUDIO:
- if (BUFFER_NOT_EMPTY(&entry->audio)) {
+ if (AUDIO_NOT_EMPTY(entry)) {
camu_audio_buffer_push(&entry->audio.buf, frame);
return;
}
break;
#ifndef CAMU_SINK_NO_VIDEO
case CAMU_STREAM_VIDEO:
- if (BUFFER_NOT_EMPTY(&entry->video)) {
+ if (VIDEO_NOT_EMPTY(entry)) {
camu_video_buffer_push(&entry->video.buf, frame);
return;
}
@@ -917,11 +1021,6 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
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.
- // - it was seeked.
- maybe_remove_from_previous(sink, entry);
-
if (reconnect) {
entry->ended = false;
if (entry == sink->current) {
@@ -933,6 +1032,11 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
}
}
+ // 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.
+ // - it was seeked.
+ maybe_remove_from_previous(sink, entry);
+
bool skip_audio = sink->audio.state == SINK_PAUSED;
if (entry->audio.state == BUFFER_ADDED) {
remove_entry_audio_buffer(sink, entry);
@@ -941,11 +1045,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
if (entry->audio.state > BUFFER_CONFIGURED) {
entry->audio.state = BUFFER_CONFIGURED;
}
+
#ifndef CAMU_SINK_NO_VIDEO
- bool ignore_video = BUFFER_EMPTY(&entry->video);
- if (!ignore_video) {
- ignore_video = reconnect && ENTRY_IS_SINGLE_FRAME(entry);
- }
+ bool ignore_video = VIDEO_EMPTY(entry) || (reconnect && VIDEO_IS_SINGLE_FRAME(entry));
if (!ignore_video) {
if (entry->video.state == BUFFER_ADDED) {
remove_entry_video_buffer(sink, entry);
@@ -956,6 +1058,9 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
}
#endif
+ // Resolve any queued REMOVE_BUFFER requests before waiting.
+ run_queue_by_opaque(sink, entry);
+
nn_mutex_unlock(&sink->mutex);
while ( // Block until buffers are removed.
@@ -967,11 +1072,11 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
) { BLOCKING_SLEEP(NNWT_TS_FROM_USEC(2000)); }
if (reconnect) {
- if (BUFFER_NOT_EMPTY(&entry->audio)) {
+ if (AUDIO_NOT_EMPTY(entry)) {
camu_audio_buffer_reset(&entry->audio.buf);
}
#ifndef CAMU_SINK_NO_VIDEO
- if (BUFFER_NOT_EMPTY(&entry->video) && !ignore_video) {
+ if (VIDEO_NOT_EMPTY(entry) && !ignore_video) {
camu_video_buffer_reset(&entry->video.buf);
}
#endif
@@ -998,14 +1103,16 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
case LIANA_CLIENT_RECONNECTED: {
nn_mutex_lock(&sink->mutex);
if (entry == sink->reconnecting) {
+ al_assert(entry == sink->current);
if (entry->audio.state > BUFFER_QUEUED) {
add_audio_if_set_and_buffered(entry);
}
#ifndef CAMU_SINK_NO_VIDEO
- if (entry->video.state > BUFFER_QUEUED && !ENTRY_IS_SINGLE_FRAME(entry)) {
+ if (entry->video.state > BUFFER_QUEUED && !VIDEO_IS_SINGLE_FRAME(entry)) {
add_video_if_set_and_buffered(entry);
}
#endif
+ sink->reconnecting = NULL;
}
nn_mutex_unlock(&sink->mutex);
break;
@@ -1013,7 +1120,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
case LIANA_CLIENT_EOF: {
switch (stream->type) {
case CAMU_STREAM_AUDIO: {
- if (BUFFER_NOT_EMPTY(&entry->audio)) {
+ if (AUDIO_NOT_EMPTY(entry)) {
camu_audio_buffer_flush(&entry->audio.buf);
}
break;
@@ -1021,7 +1128,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
#ifndef CAMU_SINK_NO_VIDEO
case CAMU_STREAM_VIDEO: {
// Single frames are immediately flushed inside the buffer.
- if (BUFFER_NOT_EMPTY(&entry->video) && !ENTRY_IS_SINGLE_FRAME(entry)) {
+ if (VIDEO_NOT_EMPTY(entry) && !VIDEO_IS_SINGLE_FRAME(entry)) {
camu_video_buffer_flush(&entry->video.buf);
}
break;
@@ -1053,16 +1160,20 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
sink->queued = NULL;
}
+ bool removed;
+ al_array_check_remove(sink->entries, entry, removed);
+
+ nn_mutex_unlock(&sink->mutex);
+
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;
- al_array_check_remove(sink->entries, entry, removed);
- nn_mutex_unlock(&sink->mutex);
al_free(entry);
+
al_log_info("sink", "Entry closed by %s.", removed ? "disconnect" : "cleanup");
+
break;
}
}
@@ -1440,7 +1551,7 @@ void camu_sink_toggle_pause(struct camu_sink *sink)
if (current) {
queue_cmd(sink, (struct camu_sink_cmd){
.op = TOGGLE_PAUSE,
- .value.f = camu_clock_get_pts(&current->clock, 0.0, true),
+ .value.f = camu_clock_get_pts(&current->clock, 0.0, false),
.opaque = current
});
}
@@ -1488,6 +1599,10 @@ void camu_sink_stop(struct camu_sink *sink)
.value.i = CAMU_SINK_AUDIO
});
queue_cmd(sink, (struct camu_sink_cmd){
+ .op = CLEAR,
+ .value.i = CAMU_SINK_AUDIO
+ });
+ queue_cmd(sink, (struct camu_sink_cmd){
.op = CLOSE
});
}
diff --git a/src/libsink/sink.h b/src/libsink/sink.h
index c08f763..10fc182 100644
--- a/src/libsink/sink.h
+++ b/src/libsink/sink.h
@@ -26,9 +26,9 @@ enum {
enum {
CAMU_SINK_ADD_BUFFER = 0,
CAMU_SINK_REMOVE_BUFFER,
- CAMU_SINK_SWAP_BUFFER,
CAMU_SINK_START,
CAMU_SINK_STOP,
+ CAMU_SINK_CLEAR,
CAMU_SINK_EXIT
};
generated by cgit v1.2.3-101-g0448 (git 2.54.0) at 2026-07-25 19:36:01 +0000