summaryrefslogtreecommitdiff
path: root/src/libsink
diff options
context:
space:
mode:
Diffstat (limited to 'src/libsink')
-rw-r--r--src/libsink/sink.c105
-rw-r--r--src/libsink/sink.h3
2 files changed, 53 insertions, 55 deletions
diff --git a/src/libsink/sink.c b/src/libsink/sink.c
index 26c144b..bd33f46 100644
--- a/src/libsink/sink.c
+++ b/src/libsink/sink.c
@@ -270,7 +270,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
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, cmd->value.i);
+ aki_packet_write_s32(packet, (s32)cmd->value.i);
aki_rpc_connection_command(sink->conn, packet, NULL, NULL);
break;
}
@@ -301,7 +301,14 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
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);
- aki_packet_write_s32(packet, get_sequence_for_command(sink));
+ 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);
+ } else {
+ aki_packet_write_s32(packet, LIANA_SEQUENCE_ANY);
+ aki_packet_write_u32(packet, 0);
+ }
aki_packet_write_f64(packet, cmd->value.f);
aki_rpc_connection_command(sink->conn, packet, NULL, NULL);
break;
@@ -324,7 +331,7 @@ static void handle_sink_cmd(struct camu_sink *sink, struct camu_sink_cmd *cmd)
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_s32(packet, cmd->value.i);
+ aki_packet_write_u32(packet, (u32)cmd->value.u);
aki_rpc_connection_command(sink->conn, packet, NULL, NULL);
break;
}
@@ -404,11 +411,12 @@ static void maybe_remove_previous(struct camu_sink *sink)
}
// Due to the looseness of the previous queue, we may have to explicitly remove an entry
-// at a point if it becomes incorrect to attempt removing it's buffers.
+// if it becomes incorrect to attempt removing it's buffers.
// 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);
@@ -416,7 +424,9 @@ static void maybe_remove_from_previous(struct camu_sink *sink, struct camu_sink_
}
}
-static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry *previous, struct camu_sink_entry *target)
+// 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)
{
al_assert(previous != target && !previous->ended);
@@ -429,10 +439,9 @@ static void maybe_add_to_previous(struct camu_sink *sink, struct camu_sink_entry
}
}
- // Every call to maybe_add_to_previous() should map to a remove_entry_buffers().
- // We can take a shortcut here because pushing an entry to previous is
- // pointless if it's not currently added.
- if (!AUDIO_ADDED_OR_EMPTY(previous) && !VIDEO_ADDED_OR_EMPTY(previous)) {
+ // 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)) {
remove_entry_buffers(sink, previous);
return;
}
@@ -490,7 +499,8 @@ 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 && entry->audio.state != BUFFER_ADDED);
+ al_assert(!entry->ended);
+ al_assert(entry->audio.state != BUFFER_ADDED && entry->audio.state != BUFFER_QUEUED);
if (entry->audio.state == BUFFER_ENDED) {
al_log_warn("sink", "Tried to add an ended audio buffer.");
return;
@@ -538,7 +548,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);
+ al_assert(entry->video.state != BUFFER_ADDED && entry->video.state != BUFFER_QUEUED);
if (entry->video.state == BUFFER_ENDED) {
al_log_warn("sink", "Tried to add an ended video buffer.");
return;
@@ -589,6 +599,10 @@ 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);
+ sink->reconnecting = NULL;
+ }
}
if (!target->ended) {
@@ -621,6 +635,10 @@ static bool end_entry_and_advance_queue(struct camu_sink *sink, struct camu_sink
{
al_log_info("sink", "Entry ended.");
entry->ended = true;
+ queue_cmd(sink, (struct camu_sink_cmd){
+ .op = END,
+ .value.u = entry->id
+ });
maybe_remove_from_previous(sink, entry);
if (sink->target) {
switch_to(sink, sink->target);
@@ -666,8 +684,7 @@ static void audio_buffer_callback(void *userdata, u8 op)
if (entry->audio.state == BUFFER_ADDED) {
remove_entry_audio_buffer(sink, entry);
}
- // This assert likely doesn't matter, but should be kept
- // if it doesn't unnecessarialy trip.
+ // 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);
@@ -678,12 +695,6 @@ static void audio_buffer_callback(void *userdata, u8 op)
end_entry_and_advance_queue(sink, entry);
}
aki_mutex_unlock(&sink->mutex);
- if (run_queue) {
- queue_cmd(sink, (struct camu_sink_cmd){
- .op = END,
- .value.i = entry->sequence
- });
- }
break;
}
}
@@ -710,7 +721,6 @@ static void video_buffer_callback(void *userdata, u8 op)
case CAMU_BUFFER_EOF: {
lia_vcr_cork(entry->video.track);
bool swapped = false;
- bool run_queue = false;
aki_mutex_lock(&sink->mutex);
al_log_info("sink", "Video EOF.");
if (!ENTRY_IS_SINGLE_FRAME(entry)) {
@@ -721,7 +731,6 @@ static void video_buffer_callback(void *userdata, u8 op)
entry->video.state = BUFFER_ENDED;
if (AUDIO_ENDED_OR_EMPTY(entry)) {
swapped = end_entry_and_advance_queue(sink, entry);
- run_queue = true;
}
}
aki_mutex_unlock(&sink->mutex);
@@ -731,12 +740,6 @@ static void video_buffer_callback(void *userdata, u8 op)
.value.i = CAMU_SINK_VIDEO
});
}
- if (run_queue) {
- queue_cmd(sink, (struct camu_sink_cmd){
- .op = END,
- .value.i = entry->sequence
- });
- }
break;
}
}
@@ -905,15 +908,22 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
bool reconnect = *(bool *)opaque;
aki_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) {
- // This may not be correct if entry->ended can ever be set anywhere
- // besides end_entry_and_advance_queue().
entry->ended = false;
+ if (entry == sink->current) {
+ if (sink->target) {
+ switch_to(sink, sink->target);
+ sink->target = NULL;
+ } else {
+ sink->reconnecting = sink->current;
+ }
+ }
}
bool skip_audio = sink->audio.state == SINK_PAUSED;
@@ -938,6 +948,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
}
}
#endif
+
aki_mutex_unlock(&sink->mutex);
while ( // Block until buffers are removed.
@@ -970,7 +981,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
}
case LIANA_CLIENT_RECONNECTED: {
aki_mutex_lock(&sink->mutex);
- if (entry == sink->current) {
+ if (entry == sink->reconnecting) {
if (entry->audio.state > BUFFER_QUEUED) {
add_audio_if_set_and_buffered(entry);
}
@@ -1250,14 +1261,11 @@ static bool pause_command_callback(void *userdata, struct aki_rpc_connection *co
u64 at = aki_packet_read_u64(packet);
u8 pause = aki_packet_read_u8(packet);
- aki_mutex_lock(&sink->mutex);
struct camu_sink_entry *entry = get_entry_from_id(sink, id);
- if (!entry) {
- aki_mutex_unlock(&sink->mutex);
- goto out;
- }
+ if (!entry) goto out;
al_assert(entry->sequence == sequence);
+ aki_mutex_lock(&sink->mutex);
#ifdef CAMU_SINK_LOCAL
(void)at;
(void)pause;
@@ -1279,9 +1287,9 @@ static bool pause_command_callback(void *userdata, struct aki_rpc_connection *co
break;
}
#endif
-out:
aki_mutex_unlock(&sink->mutex);
+out:
aki_packet_free(packet);
return false;
}
@@ -1295,21 +1303,12 @@ static bool seek_command_callback(void *userdata, struct aki_rpc_connection *con
u32 id = aki_packet_read_u32(packet);
s32 sequence = aki_packet_read_s32(packet);
+ (void)sequence;
u64 at = aki_packet_read_u64(packet);
u64 pos = aki_packet_read_u64(packet);
- aki_mutex_lock(&sink->mutex);
struct camu_sink_entry *entry = get_entry_from_id(sink, id);
- if (!entry) {
- aki_mutex_unlock(&sink->mutex);
- goto out;
- }
- if (entry == sink->current && sink->target) {
- switch_to(sink, sink->target);
- sink->target = NULL;
- }
- al_assert(entry->sequence == sequence);
- aki_mutex_unlock(&sink->mutex);
+ if (!entry) goto out;
#ifdef CAMU_SINK_LOCAL
at = 0;
@@ -1429,13 +1428,11 @@ void camu_sink_seek(struct camu_sink *sink, f64 precent)
aki_mutex_lock(&sink->mutex);
struct camu_sink_entry *current = sink->current;
aki_mutex_unlock(&sink->mutex);
- if (current) {
- queue_cmd(sink, (struct camu_sink_cmd){
- .op = SEEK,
- .value.f = precent,
- .opaque = current
- });
- }
+ queue_cmd(sink, (struct camu_sink_cmd){
+ .op = SEEK,
+ .value.f = precent,
+ .opaque = current
+ });
}
void camu_sink_reseek(struct camu_sink *sink)
diff --git a/src/libsink/sink.h b/src/libsink/sink.h
index 674b684..48b2b87 100644
--- a/src/libsink/sink.h
+++ b/src/libsink/sink.h
@@ -64,7 +64,7 @@ struct camu_sink_entry {
struct camu_sink_cmd {
u8 op;
- union { s64 i; f64 f; } value;
+ union { s64 i; u64 u; f64 f; } value;
void *opaque;
};
@@ -84,6 +84,7 @@ struct camu_sink {
struct camu_sink_entry *current;
struct camu_sink_entry *queued;
struct camu_sink_entry *target;
+ struct camu_sink_entry *reconnecting;
array(struct camu_sink_entry *) previous;
array(struct camu_sink_entry *) entries;
u16 lru;
generated by cgit v1.2.3-101-g0448 (git 2.54.0) at 2026-07-25 15:59:54 +0000