Sink-side list sync resilience, vcr fix

Signed-off-by: Andrew Opalach <andrew@akon.city>

2 files changed, 58 insertions, 22 deletions

diff --git a/src/libsink/common.h b/src/libsink/common.h
index f8dd9fa..0b1a6c9 100644
--- a/src/libsink/common.h
+++ b/src/libsink/common.h
@@ -3,10 +3,11 @@
 #define CAMU_SINK_LOCAL 1
 
 enum {
-    CAMU_SINK_SET = 0,
+    CAMU_SINK_CLEAR = 0,
+    CAMU_SINK_SET,
     CAMU_SINK_BUFFER,
     CAMU_SINK_BUFFER_AND_QUEUE,
-    CAMU_SINK_CLEAR,
     CAMU_SINK_PAUSE,
-    CAMU_SINK_SEEK
+    CAMU_SINK_SEEK,
+    CAMU_SINK_DURATION
 };
diff --git a/src/libsink/sink.c b/src/libsink/sink.c
index 990cd69..b9554c4 100644
--- a/src/libsink/sink.c
+++ b/src/libsink/sink.c
@@ -43,10 +43,10 @@ enum {
 #define ENTRY_VIDEO_READY_OR_EMPTY(entry) true
 #else
 #define ENTRY_VIDEO_READY_OR_EMPTY(entry) \
-    (entry->video.state == BUFFER_INIT || entry->video.state == BUFFER_QUEUED || entry->video.state == BUFFER_ADDED)
+    ((entry)->video.state == BUFFER_INIT || (entry)->video.state == BUFFER_QUEUED || (entry)->video.state == BUFFER_ADDED)
 #endif
 #define ENTRY_AUDIO_READY_OR_EMPTY(entry) \
-    (entry->audio.state == BUFFER_INIT || entry->audio.state == BUFFER_QUEUED || entry->audio.state == BUFFER_ADDED)
+    ((entry)->audio.state == BUFFER_INIT || (entry)->audio.state == BUFFER_QUEUED || (entry)->audio.state == BUFFER_ADDED)
 
 #if defined CAMU_SCREEN_THREADED && defined CAMU_MIXER_THREADED
 #define BLOCKING_SLEEP(delay) aki_thread_sleep(delay)
@@ -57,7 +57,7 @@ enum {
 static bool entry_audio_buffer_held(struct camu_sink_entry *entry)
 {
 #ifdef CAMU_MIXER_THREADED
-    return al_atomic_load(u8)(&(entry)->audio.buf.ref, AL_ATOMIC_RELAXED) == 1;
+    return al_atomic_load(u8)(&entry->audio.buf.ref, AL_ATOMIC_RELAXED) == 1;
 #else
     (void)entry;
     return false;
@@ -68,7 +68,7 @@ static bool entry_audio_buffer_held(struct camu_sink_entry *entry)
 static bool entry_video_buffer_held(struct camu_sink_entry *entry)
 {
 #ifdef CAMU_SCREEN_THREADED
-    return al_atomic_load(u8)(&(entry)->video.buf.ref, AL_ATOMIC_RELAXED) == 1;
+    return al_atomic_load(u8)(&entry->video.buf.ref, AL_ATOMIC_RELAXED) == 1;
 #else
     (void)entry;
     return false;
@@ -90,6 +90,8 @@ static void remove_entry_audio_buffer(struct camu_sink *sink, struct camu_sink_e
     if (entry->audio.state == BUFFER_ADDED) {
         sink->callback(sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
         entry->audio.state = BUFFER_SET_OR_BUFFERED;
+    } else if (entry->audio.state == BUFFER_SET_OR_BUFFERED) {
+        entry->audio.state = BUFFER_CONFIGURED;
     }
 }
 
@@ -99,6 +101,8 @@ static void remove_entry_video_buffer(struct camu_sink *sink, struct camu_sink_e
     if (entry->video.state == BUFFER_ADDED) {
         sink->callback(sink->userdata, CAMU_SINK_REMOVE_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
         entry->video.state = BUFFER_SET_OR_BUFFERED;
+    } else if (entry->video.state == BUFFER_SET_OR_BUFFERED) {
+        entry->video.state = BUFFER_CONFIGURED;
     }
 }
 #endif
@@ -212,8 +216,12 @@ 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_SKIP);
-        //s32 sequence = sink->current ? sink->current->sequence : LIANA_SEQUENCE_ANY;
         s32 sequence = LIANA_SEQUENCE_ANY;
+        if (sink->target) {
+            sequence = sink->target->sequence;
+        } else if (sink->current) {
+            sequence = sink->current->sequence;
+        }
         aki_packet_write_s32(packet, sequence);
         aki_packet_write_s32(packet, cmd->value.i);
         aki_rpc_connection_command(sink->conn, packet, NULL, NULL);
@@ -305,17 +313,19 @@ static void maybe_remove_previous(struct camu_sink *sink)
 void add_audio_if_set_and_buffered(struct camu_sink_entry *entry)
 {
     u8 state = entry->audio.state;
+    if (state == BUFFER_ADDED || state == BUFFER_SET_OR_BUFFERED) {
+        camu_audio_buffer_unpause(&entry->audio.buf);
+        queue_cmd(entry->sink, (struct camu_sink_cmd){
+            .op = START,
+            .value.i = CAMU_SINK_AUDIO
+        });
+    }
     if (state == BUFFER_SET_OR_BUFFERED) {
         bool can_resume = ENTRY_VIDEO_READY_OR_EMPTY(entry);
-        camu_audio_buffer_unpause(&entry->audio.buf);
         entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_AUDIO, &entry->audio.buf);
         if (can_resume) {
             maybe_remove_previous(entry->sink);
         }
-        queue_cmd(entry->sink, (struct camu_sink_cmd){
-            .op = START,
-            .value.i = CAMU_SINK_AUDIO
-        });
         state = BUFFER_ADDED;
     } else if (state == BUFFER_CONFIGURED) {
         state = BUFFER_SET_OR_BUFFERED;
@@ -327,17 +337,20 @@ void add_audio_if_set_and_buffered(struct camu_sink_entry *entry)
 void add_video_if_set_and_buffered(struct camu_sink_entry *entry)
 {
     u8 state = entry->video.state;
+    if (state == BUFFER_ADDED || state == BUFFER_SET_OR_BUFFERED) {
+        bool single_frame = camu_video_buffer_is_single_frame(&entry->video.buf);
+        queue_cmd(entry->sink, (struct camu_sink_cmd){
+            .op = single_frame ? STOP : START,
+            .value.i = CAMU_SINK_VIDEO
+        });
+    }
     if (state == BUFFER_SET_OR_BUFFERED) {
         bool can_resume = ENTRY_AUDIO_READY_OR_EMPTY(entry);
         entry->sink->callback(entry->sink->userdata, CAMU_SINK_ADD_BUFFER, CAMU_SINK_VIDEO, &entry->video.buf);
         if (can_resume) {
             maybe_remove_previous(entry->sink);
         }
-        bool single_frame = camu_video_buffer_is_single_frame(&entry->video.buf);
-        queue_cmd(entry->sink, (struct camu_sink_cmd){
-            .op = single_frame ? STOP : START,
-            .value.i = CAMU_SINK_VIDEO
-        });
+
         state = BUFFER_ADDED;
     } else if (state == BUFFER_CONFIGURED) {
         state = BUFFER_SET_OR_BUFFERED;
@@ -453,6 +466,10 @@ static void evaluate_latency(struct camu_sink *sink, struct camu_sink_entry *ent
     if (entry->audio.state != BUFFER_INIT && entry->audio.state != BUFFER_QUEUED &&
         entry->video.state != BUFFER_INIT && entry->video.state != BUFFER_QUEUED) {
         camu_video_buffer_set_latency(&entry->video.buf, -camu_mixer_get_latency(sink->audio.mixer));
+    } else if (entry->audio.state != BUFFER_INIT && entry->audio.state != BUFFER_QUEUED) {
+#if !CAMU_SINK_LOCAL
+        camu_audio_buffer_set_latency(&entry->audio.buf, -camu_mixer_get_latency(sink->audio.mixer));
+#endif
     }
 #else
     (void)sink;
@@ -807,6 +824,10 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn
 #else
     switch (pause) {
     case LIANA_PAUSE_NONE:
+        if (sink->target) {
+            al_log_warn("sink", "Ignoring target on NONE.");
+            sink->target = NULL;
+        }
         if (sink->current) {
             al_array_push(sink->previous, sink->current);
         }
@@ -814,22 +835,35 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn
         sink->current = entry;
         break;
     case LIANA_PAUSE_RESUME:
-        camu_clock_resume(&entry->clock, at);
-        if (sink->current) {
+        if (sink->target) {
+            al_log_warn("sink", "Ignoring target on RESUME.");
+            sink->target = NULL;
+        }
+        if (entry == sink->current) {
+            camu_audio_buffer_unpause(&entry->audio.buf);
+        } else if (sink->current) {
             al_array_push(sink->previous, sink->current);
         }
+        camu_clock_resume(&entry->clock, at);
         set_or_queue_entry(entry);
         sink->current = entry;
         break;
     case LIANA_PAUSE_PAUSE:
-        if (sink->current) {
+        if (camu_clock_is_ended(&entry->clock)) {
+            // Server didn't know this entry was ended, but it is.
+            queue_cmd(sink, (struct camu_sink_cmd){
+                .op = END,
+                .value.i = entry->sequence
+            });
+            camu_clock_pause(&sink->current->clock, at);
+            sink->current = entry;
+        } else if (sink->current) {
             if (camu_clock_is_ended(&sink->current->clock)) {
                 // Server thought we weren't done, be we are.
                 al_array_push(sink->previous, sink->current);
                 set_or_queue_entry(entry);
                 sink->current = entry;
             } else {
-                al_printf("ay\n");
                 sink->target = entry;
                 camu_clock_pause(&sink->current->clock, at);
             }
@@ -843,6 +877,7 @@ static bool set_command_callback(void *userdata, struct aki_rpc_connection *conn
                 sink->target = NULL;
             } else {
                 sink->target = entry;
+                camu_audio_buffer_unpause(&entry->audio.buf);
             }
             camu_clock_pause(&prev_target->clock, at);
         } else if (sink->current) {