Server-side fixes from DIRECT_MODE testing

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

1 files changed, 115 insertions, 48 deletions

diff --git a/src/liana/client.c b/src/liana/client.c
index 1e873c6..ca98194 100644
--- a/src/liana/client.c
+++ b/src/liana/client.c
@@ -1,4 +1,5 @@
 #define AL_LOG_SECTION "liana"
+//#define AL_LOG_ENABLE_TRACE
 #include <al/log.h>
 #include <nnwt/multiplex.h>
 
@@ -24,7 +25,14 @@ static void data_packet_callback(void *userdata, struct nn_packet_stream *stream
 {
     struct lia_client *client = (struct lia_client *)userdata;
     al_assert(client->vcr.data == stream);
-    lia_vcr_push_packet(&client->vcr, packet);
+    // The server might cut us off before sending any data, so don't attempt to
+    // recover unless we received at least one data packet.
+    if (!lia_vcr_push_packet(&client->vcr, packet)) {
+        nn_packet_stream_return_packet(stream, packet);
+    } else if (client->reconnect == RECONNECT_NONE) {
+        client->reconnect = RECONNECT_RECOVER;
+        lia_vcr_start(&client->vcr);
+    }
 }
 
 static s32 stream_compare(const void *a, const void *b)
@@ -45,7 +53,7 @@ static void collect_streams(struct lia_client *client, struct nn_packet *packet)
         struct camu_codec_stream stream = { 0 };
         str codec;
         nn_packet_read_str(packet, &codec);
-        stream.codec_info = camu_codec_info_by_name(&codec);
+        stream.codec_info = camu_codec_info_by_id(&codec);
         u8 mode = nn_packet_read_u8(packet);
         u8 type = nn_packet_read_u8(packet);
         u64 duration = nn_packet_read_u64(packet);
@@ -76,13 +84,14 @@ static void collect_streams(struct lia_client *client, struct nn_packet *packet)
             AVFormatContext *format_context = avformat_alloc_context();
             stream.av.stream = nn_packet_read_av_stream(format_context, av_codec, packet);
             switch (type) {
-            case CAMU_STREAM_ATTACHMENT:
+            case CAMU_STREAM_ATTACHMENT: {
                 // Assume all the data we need is in the AVStream object.
                 stream.type = CAMU_STREAM_ATTACHMENT;
                 client->callback(client->userdata, LIANA_CLIENT_CONFIGURE, &stream, NULL);
                 avformat_free_context(format_context);
                 continue;
             }
+            }
             stream.av.format_context = format_context;
             AVCodecParameters *codecpar = stream.av.stream->codecpar;
             switch (type) {
@@ -125,11 +134,11 @@ static void collect_streams(struct lia_client *client, struct nn_packet *packet)
 }
 
 static const char *stream_type_to_str[] = {
-    [CAMU_STREAM_AUDIO] = "audio",
-    [CAMU_STREAM_VIDEO] = "video",
-    [CAMU_STREAM_SUBTITLE] = "subtitle",
-    [CAMU_STREAM_ATTACHMENT] = "attachment",
-    [CAMU_STREAM_UNKNOWN] = "unknown"
+    [CAMU_STREAM_AUDIO] = "Audio",
+    [CAMU_STREAM_VIDEO] = "Video",
+    [CAMU_STREAM_SUBTITLE] = "Subtitles",
+    [CAMU_STREAM_ATTACHMENT] = "Attachment",
+    [CAMU_STREAM_UNKNOWN] = "Unknown"
 };
 
 static void parse_info_packet(struct lia_client *client, struct nn_packet *packet)
@@ -149,49 +158,85 @@ static void parse_info_packet(struct lia_client *client, struct nn_packet *packe
         struct camu_codec_stream *stream;
         al_array_foreach_ptr(client->streams, i, stream) {
             u8 type = stream->type;
+            s32 index = stream->index;
+            switch (type) {
+            case CAMU_STREAM_AUDIO:
+                if (index < prefs->index.audio_min) prefs->index.audio_min = index;
+                if (index > prefs->index.audio_max) prefs->index.audio_max = index;
+                break;
+            case CAMU_STREAM_SUBTITLE:
+                if (index < prefs->index.subtitles_min) prefs->index.subtitles_min = index;
+                if (index > prefs->index.subtitles_max) prefs->index.subtitles_max = index;
+                break;
+            }
             if ((selected & (1 << type)) || !(prefs->enabled & (1 << type))) {
                 continue;
             }
             const char *title = NULL;
 #ifdef CAMU_HAVE_FFMPEG
+            bool pref_is_index = false;
+            if (!accept_defaults) {
+                if (type == CAMU_STREAM_AUDIO && prefs->index.audio >= 0) {
+                    if (index != prefs->index.audio) {
+                        continue;
+                    }
+                    pref_is_index = true;
+                }
+                if (type == CAMU_STREAM_SUBTITLE && prefs->index.subtitles >= 0) {
+                    if (index != prefs->index.subtitles) {
+                        continue;
+                    }
+                    pref_is_index = true;
+                }
+            }
             if (stream->mode == CAMU_FFMPEG_COMPAT) {
                 AVDictionary *metadata = stream->av.stream->metadata;
                 const AVDictionaryEntry *title_entry = av_dict_get(metadata, "title", NULL, 0);
-                if (title_entry) {
-                    title = title_entry->value;
-                }
-                const AVDictionaryEntry *lang_entry = av_dict_get(metadata, "language", NULL, 0);
-                if (lang_entry && !accept_defaults) {
-                    u8 lang = CAMU_LANG_UNKNOWN;
-                    str s = al_str_cr(lang_entry->value);
-                    if      (al_str_eq(&s, &al_str_c("eng"))) lang = CAMU_LANG_ENGLISH;
-                    else if (al_str_eq(&s, &al_str_c("jpn"))) lang = CAMU_LANG_JAPANESE;
-                    if ((type == CAMU_STREAM_AUDIO    && lang != prefs->language.audio) ||
-                        (type == CAMU_STREAM_SUBTITLE && lang != prefs->language.subtitles)) {
-                        continue;
+                if (title_entry) title = title_entry->value;
+                if (!pref_is_index && !accept_defaults) {
+                    const AVDictionaryEntry *lang_entry = av_dict_get(metadata, "language", NULL, 0);
+                    if (lang_entry) {
+                        u8 lang = CAMU_LANG_UNKNOWN;
+                        str s = al_str_cr(lang_entry->value);
+                        if      (al_str_eq(&s, &al_str_c("eng"))) lang = CAMU_LANG_ENGLISH;
+                        else if (al_str_eq(&s, &al_str_c("jpn"))) lang = CAMU_LANG_JAPANESE;
+                        if ((type == CAMU_STREAM_AUDIO    && lang != prefs->language.audio) ||
+                            (type == CAMU_STREAM_SUBTITLE && lang != prefs->language.subtitles)) {
+                            continue;
+                        }
                     }
-                    break;
                 }
             }
 #endif
+            switch (type) {
+            case CAMU_STREAM_AUDIO:
+                prefs->index.audio = index;
+                break;
+            case CAMU_STREAM_VIDEO:
+                prefs->index.video = index;
+                break;
+            case CAMU_STREAM_SUBTITLE:
+                prefs->index.subtitles = index;
+                break;
+            }
             if (title) {
-                log_info("Selected %s stream (index: %u, title: %s).", stream_type_to_str[type], stream->index, title);
+                log_info("%s selected (index: %u, title: %s).", stream_type_to_str[type], index, title);
             } else {
-                log_info("Selected %s stream (index: %u).", stream_type_to_str[type], stream->index);
+                log_info("%s selected (index: %u).", stream_type_to_str[type], index);
             }
             selected |= (1 << type);
             struct lia_vcr_track *track = al_alloc_object(struct lia_vcr_track);
             track->stream = stream;
-            track->client = lia_handler_by_name(&handler)->create_client_handler();
-            track->client->callback = client->callback;
-            track->client->userdata = client->userdata;
-            if (!track->client->init(track->client, client->renderer, track->stream)) {
-                track->client->free(&track->client);
+            track->handler = lia_handler_by_name(&handler)->create_client_handler();
+            track->handler->callback = client->callback;
+            track->handler->userdata = client->userdata;
+            if (!track->handler->init(track->handler, client->renderer, track->stream)) {
+                track->handler->free(&track->handler);
                 al_free(track);
                 // This will NOT attempt to select another stream.
                 continue;
             }
-            client->mask |= 1 << stream->index;
+            client->mask |= 1 << index;
             client->callback(client->userdata, LIANA_CLIENT_CONFIGURE, stream, track);
             lia_vcr_add_track(&client->vcr, track);
         }
@@ -221,8 +266,12 @@ static void info_packet_callback(void *userdata, struct nn_packet_stream *stream
     struct nn_packet *rpacket = nn_packet_create();
     nn_packet_write_u64(rpacket, client->mask);
     nn_packet_stream_send_packet(stream, rpacket);
-    client->reconnect = RECONNECT_RECOVER;
-    lia_vcr_start(&client->vcr);
+}
+
+static void packet_dequeued_callback(void *userdata, struct nn_packet *packet)
+{
+    (void)userdata;
+    nn_packet_write_size(packet);
 }
 
 static void packet_sent_callback(void *userdata, struct nn_packet *packet)
@@ -234,6 +283,21 @@ static void packet_sent_callback(void *userdata, struct nn_packet *packet)
 static bool connection_callback(void *userdata, struct nn_packet_stream *stream)
 {
     struct lia_client *client = (struct lia_client *)userdata;
+
+    struct nn_socket *sock = &client->data.sock;
+    if (sock->type == NNWT_SOCKET_TCP) {
+#ifdef AL_LOG_ENABLE_TRACE
+        u32 rcvbuf = nn_socket_get_recv_buf(sock);
+        u32 sndbuf = nn_socket_get_send_buf(sock);
+#endif
+        nn_socket_set_recv_buf(sock, MB(2));
+        nn_socket_set_send_buf(sock, KB(8));
+#ifdef AL_LOG_ENABLE_TRACE
+        log_trace("rcvbuf: %u -> %u", rcvbuf, nn_socket_get_recv_buf(sock));
+        log_trace("sndbuf: %u -> %u", sndbuf, nn_socket_get_send_buf(sock));
+#endif
+    }
+
     if (client->reconnect == RECONNECT_SIGNAL_CLIENT) {
         client->reconnect = RECONNECT_NONE;
         // Even if client_seek() was called before the initial connection_callback(),
@@ -254,20 +318,19 @@ static bool connection_callback(void *userdata, struct nn_packet_stream *stream)
     } else {
         al_assert(client->connection_id == 0 && client->reconnect == RECONNECT_NONE);
     }
-    stream->packet_sent_callback = packet_sent_callback;
+
     struct nn_packet *packet = nn_packet_create();
     nn_packet_write_u32(packet, client->node_id);
     nn_packet_write_u32(packet, client->connection_id);
     nn_packet_write_u64(packet, client->mask);
     nn_packet_write_u64(packet, client->pos);
-    if (client->mask == 0) {
-        stream->packet_callback = info_packet_callback;
-    } else {
-        stream->packet_callback = data_packet_callback;
-        client->reconnect = RECONNECT_RECOVER;
-        lia_vcr_start(&client->vcr);
-    }
+
+    stream->packet_callback = (client->mask == 0) ? info_packet_callback : data_packet_callback;
+    stream->packet_dequeued_callback = packet_dequeued_callback;
+    stream->packet_sent_callback = packet_sent_callback;
+
     nn_packet_stream_send_packet(stream, packet);
+
     return true;
 }
 
@@ -290,7 +353,8 @@ static void connection_closed_callback(void *userdata, struct nn_packet_stream *
         lia_vcr_close_all(&client->vcr);
     }
 
-    // If reconnect = SIGNAL_CLIENT, we either never connected or recursed at the reconnect step.
+    // If reconnect = SIGNAL_CLIENT, we can guarantee that CLIENT_REMOVE_BUFFERS
+    // has been called after the most recent connection_callback().
     if (client->reconnect != RECONNECT_SIGNAL_CLIENT) {
         client->rec = (struct lia_reconnect_info){
             .reconnect = reconnect,
@@ -298,8 +362,8 @@ static void connection_closed_callback(void *userdata, struct nn_packet_stream *
             .mask = client->mask
         };
         al_array_init(client->rec.detached);
-        // CLIENT_REMOVE_BUFFERS should be allowed to run the event loop to wait and should
-        // attempt to maintain the same state if called consecutively.
+        // CLIENT_REMOVE_BUFFERS should be allowed to run the event loop to wait.
+        // It should also attempt to maintain the same state if called consecutively.
         client->callback(client->userdata, LIANA_CLIENT_REMOVE_BUFFERS, NULL, &client->rec);
         client->mask = client->rec.mask;
         struct camu_codec_stream *detached;
@@ -316,9 +380,11 @@ static void connection_closed_callback(void *userdata, struct nn_packet_stream *
     if (reconnect) {
         // If stream_reconnect() errors or is aborted, the client will be closed on recursion.
         client->reconnect = RECONNECT_SIGNAL_CLIENT;
-        // A client being seeked before an info packet is another reason mask may
-        // be unset here. In that case we don't want to forcefully close.
-        if (!client->rec.unconfigured && !client->mask) {
+        // It's possible the client was not configured yet. For example, if it was seeked
+        // before an info packet. In that case, we don't want to forcefully close it.
+        bool was_configured = !client->rec.unconfigured;
+        if (was_configured && client->mask == 0) {
+            // If mask was emptied by CLIENT_REMOVE_BUFFERS, close the client immediately.
             connection_closed_callback(userdata, stream);
         } else {
 #ifdef CAMU_DIRECT_MODE
@@ -337,12 +403,13 @@ void lia_client_connect(struct lia_client *client, struct nn_event_loop *loop,
 {
     client->loop = loop;
     client->node_id = node_id;
+    client->duration = LIANA_TIMESTAMP_INVALID;
+    lia_vcr_init(&client->vcr, client->loop, &client->data, node_id);
+    al_array_init(client->streams);
+    client->mask = 0;
     client->pos = pos;
     client->at = LIANA_TIMESTAMP_INVALID;
-    client->mask = 0;
-    al_array_init(client->streams);
     client->reconnect = RECONNECT_NONE;
-    lia_vcr_init(&client->vcr, client->loop, &client->data, node_id);
     al_str_clone(&client->addr, addr);
     client->port = port;
     client->connection_id = 0;