Extend server resource loading, work on client

Also, cleanup in preparation for changing code style.

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

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;