Extend server resource loading, work on client

Also, cleanup in preparation for changing code style.

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

4 files changed, 120 insertions, 57 deletions

diff --git a/src/liana/client.c b/src/liana/client.c
index a841f7a..ad365b3 100644
--- a/src/liana/client.c
+++ b/src/liana/client.c
@@ -62,7 +62,7 @@ static void parse_info_packet(struct lia_client *client, struct aki_packet *pack
                 client->mask |= 1 << index;
                 break;
             case CAMU_STREAM_ATTACHMENT:
-                // Assume we have all the data we need in the AVStream object.
+                // Assume all the data we need is in the AVStream object.
                 break;
             default:
                 continue;
diff --git a/src/liana/list.c b/src/liana/list.c
index 23eabc3..073cdf0 100644
--- a/src/liana/list.c
+++ b/src/liana/list.c
@@ -77,11 +77,26 @@ static bool assume_ended(struct lia_list_entry *entry, u64 at)
     return false;
 }
 
-static bool entry_load_and_get_duration(struct lia_list *list, struct lia_list_entry *entry)
+static bool entry_load_and_get_duration(struct lia_list *list, struct lia_list_entry *entry, s32 sequence, bool *error)
 {
-    bool loaded;
-    list->callback(list->userdata, LIANA_LOAD_ENTRY, entry, &loaded);
-    if (loaded) {
+    u8 status;
+    list->callback(list->userdata, LIANA_LOAD_ENTRY, entry, &status);
+    if (status == LIANA_ENTRY_ERRORED) {
+        if (sequence >= 0) {
+            al_array_remove_at(list->entries, (u32)sequence);
+            if (sequence == list->current && (u32)list->current == list->entries.size) {
+                // This maps to the behavior of only skipping ahead on errors.
+                list->current--;
+                list->idle = true;
+            }
+        }
+        *error = true;
+        u8 meta = LIANA_META_FAILED;
+        list->callback(list->userdata, LIANA_LIST_META, entry, &meta);
+        return false;
+    }
+    *error = false;
+    if (status == LIANA_ENTRY_LOADED) {
         list->callback(list->userdata, LIANA_GET_DURATION, entry, &entry->duration);
         return true;
     }
@@ -90,12 +105,13 @@ static bool entry_load_and_get_duration(struct lia_list *list, struct lia_list_e
 
 static bool handle_add_sink(struct lia_list *list, struct lia_list_sink *sink)
 {
-    if (list->current >= 0) {
-        struct lia_list_entry *current = al_array_at(list->entries, list->current);
-        if (!entry_load_and_get_duration(list, current)) {
+    struct lia_list_entry *current;
+    if (list->current >= 0 && !list->idle && !(current = al_array_at(list->entries, list->current))->ended) {
+        bool error;
+        if (!entry_load_and_get_duration(list, current, list->current, &error)) {
+            if (error) pump_queue(list);
             return false;
         }
-        sink->set = list->current;
         u64 now = aki_get_timestamp();
         u8 pause;
         u64 at = LIANA_TIMESTAMP_INVALID;
@@ -121,6 +137,7 @@ static bool handle_add_sink(struct lia_list *list, struct lia_list_sink *sink)
             .pause = pause,
             .ended = ended
         };
+        sink->set = list->current;
         sink->callback(sink->userdata, LIANA_SINK_SET, current, list->current, &time);
     } else {
         sink->set = -1;
@@ -145,8 +162,9 @@ static void handle_remove_sink(struct lia_list *list, void *userdata)
 static bool handle_add(struct lia_list *list, struct lia_list_entry *entry)
 {
     if (list->idle) {
-        if (!entry_load_and_get_duration(list, entry)) {
-            return false;
+        bool error;
+        if (!entry_load_and_get_duration(list, entry, -1, &error)) {
+            return error;
         }
         list->current++;
         list->idle = false;
@@ -163,7 +181,8 @@ static bool handle_add(struct lia_list *list, struct lia_list_entry *entry)
             sink->set = list->current;
             sink->callback(sink->userdata, LIANA_SINK_SET, entry, list->current, &time);
         }
-        al_log_info("list", "Now playing: %ls.", AL_WSTR_PRINTF(&entry->name));
+        u8 meta = LIANA_META_PLAYING;
+        list->callback(list->userdata, LIANA_LIST_META, entry, &meta);
     } else {
         /*
         if (list->queued == -1) {
@@ -185,7 +204,8 @@ static bool handle_add(struct lia_list *list, struct lia_list_entry *entry)
         */
             entry->start = LIANA_TIMESTAMP_INVALID;
         //}
-        // meta queued
+        u8 meta = LIANA_META_QUEUED;
+        list->callback(list->userdata, LIANA_LIST_META, entry, &meta);
     }
     al_array_push(list->entries, entry);
     return true;
@@ -202,7 +222,7 @@ static void adjust_current(struct lia_list *list, struct lia_list_entry *previou
             if (i == (u32)list->current) return;
             cmd->op = SKIPTO;
             cmd->sequence = i;
-            cmd->i = list->current;
+            cmd->value.i = list->current;
             list->current = i;
             break;
         }
@@ -241,6 +261,18 @@ static struct lia_list_entry *get_entry_from_sequence(struct lia_list *list, s32
     return al_array_at(list->entries, sequence);
 }
 
+static struct lia_list_entry *get_entry_from_id(struct lia_list *list, u32 id, s32 *sequence)
+{
+    struct lia_list_entry *entry;
+    al_array_foreach(list->entries, i, entry) {
+        if (entry->id == id) {
+            *sequence = (s32)i;
+            return entry;
+        }
+    }
+    return NULL;
+}
+
 // TODO:
 //if (current->start != LIANA_TIMESTAMP_INVALID && current->start > ts - LIANA_BASE_PING) {
 
@@ -260,7 +292,10 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index)
     al_assert(current && !current->held);
     if (!target) return true;
     al_assert(current != target);
-    if (!entry_load_and_get_duration(list, target)) {
+    bool error;
+    if (!entry_load_and_get_duration(list, target, index, &error)) {
+        // index might point to a different entry after an error.
+        if (error) pump_queue(list);
         return false;
     }
 
@@ -281,7 +316,7 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index)
         // An ended entry may never have been paused, but a non-ended entry that wasn't set
         // cannot be unpaused. Checking assume_ended(target) should be safe here as long
         // as it can't go from true to false (consideration for seek?).
-        al_assert(assume_ended(target, now) || target->paused_at != LIANA_TIMESTAMP_INVALID);
+        //al_assert(assume_ended(target, now) || target->paused_at != LIANA_TIMESTAMP_INVALID);
     }
 
     // These are not equivalent to current/target->ended.
@@ -342,7 +377,8 @@ static bool handle_skipto(struct lia_list *list, s32 sequence, s32 index)
         sink->callback(sink->userdata, LIANA_SINK_SET, target, index, &time);
     }
 
-    al_log_info("list", "Now playing: %ls.", AL_WSTR_PRINTF(&target->name));
+    u8 meta = LIANA_META_PLAYING;
+    list->callback(list->userdata, LIANA_LIST_META, target, &meta);
 
     return true;
 }
@@ -353,8 +389,8 @@ static bool handle_skip(struct lia_list *list, s32 sequence, s32 n)
     struct lia_list_cmd *cmd = list->cmd;
     cmd->op = SKIPTO;
     cmd->sequence = sequence;
-    cmd->i = sequence + n;
-    return handle_skipto(list, cmd->sequence, cmd->i);
+    cmd->value.i = sequence + n;
+    return handle_skipto(list, cmd->sequence, cmd->value.i);
 }
 
 static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts)
@@ -408,19 +444,19 @@ static void handle_toggle_pause(struct lia_list *list, s32 sequence, f64 pts)
     }
 }
 
-// TODO: Take entry id instead of sequence for seek() and end()?
-
-static void handle_seek(struct lia_list *list, s32 sequence, f64 percent)
+static void handle_seek(struct lia_list *list, s32 sequence, u32 id, f64 percent)
 {
-    if (sequence == LIANA_SEQUENCE_ANY) sequence = list->current;
-    if (sequence < 0) return;
-    if (sequence != list->current) {
-        return;
+    struct lia_list_entry *entry;
+    if (sequence == LIANA_SEQUENCE_ANY) {
+        sequence = list->current;
+        if (sequence < 0) return;
+        entry = get_entry_from_sequence(list, sequence);
+        al_assert(entry);
+    } else {
+        entry = get_entry_from_id(list, id, &sequence);
+        if (!entry) return;
     }
-    struct lia_list_entry *entry = get_entry_from_sequence(list, sequence);
-    al_assert(entry);
     if (entry->duration == LIANA_TIMESTAMP_INVALID) {
-        // Until any kind of live resource buffering.
         return;
     }
     entry->ended = false;
@@ -428,8 +464,9 @@ static void handle_seek(struct lia_list *list, s32 sequence, f64 percent)
     u64 now = aki_get_timestamp();
     u64 pos = (u64)(entry->duration * percent);
     u64 at = now + LIANA_BASE_DELAY;
+    u8 pause = entry->paused_at == LIANA_TIMESTAMP_INVALID ? LIANA_PAUSE_RESUME : LIANA_PAUSE_NONE;
     entry->offset = pos;
-    if (entry->paused_at == LIANA_TIMESTAMP_INVALID) {
+    if (pause == LIANA_PAUSE_RESUME) {
         entry->start = at;
     }
 
@@ -440,31 +477,34 @@ static void handle_seek(struct lia_list *list, s32 sequence, f64 percent)
     struct lia_timing time = {
         .at = at,
         .seek_pos = pos,
-        .pause = LIANA_PAUSE_NONE,
+        .pause = pause,
         .ended = false
     };
 
     struct lia_list_sink *sink;
     al_array_foreach(list->sinks, i, sink) {
+        if (sequence == list->current && sink->set != sequence) {
+            sink->set = sequence;
+            sink->callback(sink->userdata, LIANA_SINK_SET, entry, sequence, &time);
+        }
         sink->callback(sink->userdata, LIANA_SINK_SEEK, entry, sequence, &time);
     }
 }
 
-static bool handle_end(struct lia_list *list, s32 sequence)
+static bool handle_end(struct lia_list *list, s32 id)
 {
-    al_assert(sequence != LIANA_SEQUENCE_ANY);
+    s32 sequence;
+    struct lia_list_entry *entry = get_entry_from_id(list, id, &sequence);
+    if (!entry) return true;
 
-    s32 size = (s32)list->entries.size;
-    struct lia_list_entry *entry = al_array_at(list->entries, sequence);
-    al_assert(entry);
     if (entry->ended) {
         al_log_warn("list", "Got end() from an already ended resource, ignoring.");
         return true;
     }
     entry->ended = true;
-    // TODO: Calculate duration for live resources?
     entry->offset = entry->duration;
 
+    s32 size = (s32)list->entries.size;
     if (sequence == list->current) {
         s32 next = sequence + 1;
         if (list->queued >= 0) {
@@ -476,12 +516,13 @@ static bool handle_end(struct lia_list *list, s32 sequence)
                 sink->queued = -1;
             }
             struct lia_list_entry *current = al_array_at(list->entries, list->current);
-            al_log_info("list", "Now playing: %ls.", AL_WSTR_PRINTF(&current->name));
+            u8 meta = LIANA_META_PLAYING;
+            list->callback(list->userdata, LIANA_LIST_META, current, &meta);
         } else if (next < size) {
             struct lia_list_cmd *cmd = list->cmd;
             cmd->op = SKIPTO;
             cmd->sequence = sequence;
-            cmd->i = next;
+            cmd->value.i = next;
             pump_queue(list);
             return false;
         } else {
@@ -567,13 +608,13 @@ void pump_queue(struct lia_list *list)
         handle_unset(list);
         break;
     case SKIPTO:
-        if (!handle_skipto(list, cmd->sequence, cmd->i)) {
+        if (!handle_skipto(list, cmd->sequence, cmd->value.i)) {
             // Target entry not loaded.
             return;
         }
         break;
     case SKIP:
-        if (!handle_skip(list, cmd->sequence, cmd->i)) {
+        if (!handle_skip(list, cmd->sequence, cmd->value.i)) {
             // Converted to skipto and entry not loaded.
             return;
         }
@@ -582,10 +623,10 @@ void pump_queue(struct lia_list *list)
         handle_toggle_pause(list, cmd->sequence, cmd->f);
         break;
     case SEEK:
-        handle_seek(list, cmd->sequence, cmd->f);
+        handle_seek(list, cmd->sequence, cmd->value.u, cmd->f);
         break;
     case END:
-        if (!handle_end(list, cmd->sequence)) {
+        if (!handle_end(list, cmd->value.u)) {
             // End was converted to a skip.
             return;
         }
@@ -665,7 +706,7 @@ void lia_list_skipto(struct lia_list *list, s32 sequence, s32 index)
     struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd);
     cmd->op = SKIPTO;
     cmd->sequence = sequence;
-    cmd->i = index;
+    cmd->value.i = index;
     al_array_push(list->queue, cmd);
     pump_queue(list);
 }
@@ -675,7 +716,7 @@ void lia_list_skip(struct lia_list *list, s32 sequence, s32 n)
     struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd);
     cmd->op = SKIP;
     cmd->sequence = sequence;
-    cmd->i = n;
+    cmd->value.i = n;
     al_array_push(list->queue, cmd);
     pump_queue(list);
 }
@@ -690,21 +731,22 @@ void lia_list_toggle_pause(struct lia_list *list, s32 sequence, f64 pts)
     pump_queue(list);
 }
 
-void lia_list_seek(struct lia_list *list, s32 sequence, f64 percent)
+void lia_list_seek(struct lia_list *list, s32 sequence, u32 id, f64 percent)
 {
     struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd);
     cmd->op = SEEK;
     cmd->sequence = sequence;
+    cmd->value.u = id;
     cmd->f = percent;
     al_array_push(list->queue, cmd);
     pump_queue(list);
 }
 
-void lia_list_end(struct lia_list *list, s32 sequence)
+void lia_list_end(struct lia_list *list, u32 id)
 {
     struct lia_list_cmd *cmd = al_alloc_object(struct lia_list_cmd);
     cmd->op = END;
-    cmd->sequence = sequence;
+    cmd->value.u = id;
     al_array_push(list->queue, cmd);
     pump_queue(list);
 }
diff --git a/src/liana/list.h b/src/liana/list.h
index 90f8c02..8d47b72 100644
--- a/src/liana/list.h
+++ b/src/liana/list.h
@@ -26,7 +26,22 @@ enum {
 enum {
     LIANA_LOAD_ENTRY = 0,
     LIANA_GET_DURATION,
-    LIANA_UNLOAD_ENTRY
+    LIANA_UNLOAD_ENTRY,
+    LIANA_LIST_META
+};
+
+enum {
+    LIANA_ENTRY_PREPARING = 0,
+    LIANA_ENTRY_PREPARED,
+    LIANA_ENTRY_LOADING,
+    LIANA_ENTRY_LOADED,
+    LIANA_ENTRY_ERRORED
+};
+
+enum {
+    LIANA_META_PLAYING = 0,
+    LIANA_META_QUEUED,
+    LIANA_META_FAILED
 };
 
 // NOTE: To handle an entry being queued right before a skip, keep a global
@@ -75,7 +90,7 @@ struct lia_list_cmd {
     void *userdata;
     struct lia_list_entry *entry;
     s32 sequence;
-    s32 i;
+    union { s32 i; u32 u; } value;
     f64 f;
 };
 
@@ -106,8 +121,8 @@ void lia_list_unset(struct lia_list *list);
 void lia_list_skipto(struct lia_list *list, s32 sequence, s32 i);
 void lia_list_skip(struct lia_list *list, s32 sequence, s32 n);
 void lia_list_toggle_pause(struct lia_list *list, s32 sequence, f64 pts);
-void lia_list_seek(struct lia_list *list, s32 sequence, f64 percent);
-void lia_list_end(struct lia_list *list, s32 sequence);
+void lia_list_seek(struct lia_list *list, s32 sequence, u32 id, f64 percent);
+void lia_list_end(struct lia_list *list, u32 id);
 
 void lia_list_reverse(struct lia_list *list);
 void lia_list_sort(struct lia_list *list);
diff --git a/src/liana/vcr.c b/src/liana/vcr.c
index 435fcba..c4bf583 100644
--- a/src/liana/vcr.c
+++ b/src/liana/vcr.c
@@ -24,8 +24,8 @@ static void signal_callback(void *userdata)
 
 static void reset_metrics(struct lia_vcr *vcr)
 {
-    vcr->metric.current_frame = 0;
-    vcr->metric.last_report_ts = 0;
+    vcr->metric.current_frame = 0Lu;
+    vcr->metric.last_report_ts = 0Lu;
 }
 
 void lia_vcr_init(struct lia_vcr *vcr, struct aki_event_loop *loop, struct aki_packet_stream *data)
@@ -162,10 +162,15 @@ static void update_metrics(struct lia_vcr *vcr, u32 size)
     }
     u64 diff;
     if ((diff = now - vcr->metric.last_report_ts) > 1000000Lu) {
-        f32 kbps = (vcr->metric.current_frame / 125.f) / (diff / 1000000.f);
-        al_log_info("vcr", "Receiving packets at %.2fkbps.", kbps);
-        vcr->metric.current_frame = 0;
         vcr->metric.last_report_ts = now;
+        u64 frame = vcr->metric.current_frame;
+        vcr->metric.current_frame = 0Lu;
+        if (diff > 2500000Lu) {
+            al_log_debug("vcr", "Ignoring %lu bytes in metrics.", frame);
+            return;
+        }
+        f32 kbps = (frame / 125.f) / (diff / 1000000.f);
+        al_log_info("vcr", "Receiving packets at %.2fkbps.", kbps);
     }
 }
 
@@ -263,6 +268,7 @@ void lia_vcr_flush(struct lia_vcr *vcr)
     }
     aki_signal_stop(&vcr->signal);
     al_atomic_store(u64)(&vcr->count, 0, AL_ATOMIC_RELAXED);
+    reset_metrics(vcr);
     if (vcr->expand == VCR_EXPAND_COMPLETE) {
         vcr->mark.low = 0;
         vcr->expand = VCR_EXPAND_GROWN;