1 files changed, 30 insertions, 17 deletions

diff --git a/src/libsink/sink.c b/src/libsink/sink.c
index f792001..1569286 100644
--- a/src/libsink/sink.c
+++ b/src/libsink/sink.c
@@ -1,4 +1,5 @@
 #include <al/log.h>
+#include <nnwt/multiplex.h>
 
 #include "../server/common.h"
 
@@ -683,7 +684,7 @@ static void audio_buffer_callback(void *userdata, u8 op)
         nn_mutex_unlock(&sink->mutex);
         break;
     case CAMU_BUFFER_EOF: {
-        lia_vcr_cork(entry->audio.track);
+        lia_vcr_set_buffered(entry->audio.track);
         nn_mutex_lock(&sink->mutex);
         al_log_info("sink", "Audio EOF.");
         if (entry->audio.state == BUFFER_ADDED) {
@@ -726,7 +727,7 @@ static void video_buffer_callback(void *userdata, u8 op)
         lia_vcr_uncork(entry->video.track);
         break;
     case CAMU_BUFFER_EOF: {
-        lia_vcr_cork(entry->video.track);
+        lia_vcr_set_buffered(entry->video.track);
         bool swapped = false;
         nn_mutex_lock(&sink->mutex);
         al_log_info("sink", "Video EOF.");
@@ -963,7 +964,7 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
 #else
             (!skip_audio && entry_audio_buffer_held(entry))
 #endif
-        ) { BLOCKING_SLEEP(NNWT_TS_FROM_USEC(2500)); }
+        ) { BLOCKING_SLEEP(NNWT_TS_FROM_USEC(2000)); }
 
         if (reconnect) {
             if (BUFFER_NOT_EMPTY(&entry->audio)) {
@@ -981,7 +982,16 @@ static void client_callback(void *userdata, u8 op, struct camu_codec_stream *str
     case LIANA_CLIENT_RESUME_AT: {
         struct lia_timing *time = (struct lia_timing *)opaque;
         nn_mutex_lock(&sink->mutex);
+#if defined LIANA_LIST_SCUFFED_LOOP && !defined CAMU_SINK_NO_VIDEO
+        struct camu_video_buffer *buf = &entry->video.buf;
+        if (time->seek_pos == 0Lu && buf->last_pts >= 0.0) {
+            camu_clock_offset(&entry->clock, buf->last_pts);
+        } else {
+            camu_clock_seek(&entry->clock, time->seek_pos / 1000000.0, time->at);
+        }
+#else
         camu_clock_seek(&entry->clock, time->seek_pos / 1000000.0, time->at);
+#endif
         nn_mutex_unlock(&sink->mutex);
         break;
     }
@@ -1250,7 +1260,8 @@ out:
         maybe_cleanup_old_entries(sink);
     }
 
-    nn_packet_free(packet);
+    nn_packet_stream_return_packet(conn->stream, packet);
+
     return false;
 }
 
@@ -1295,7 +1306,8 @@ static bool pause_command_callback(void *userdata, struct nn_rpc_connection *con
     nn_mutex_unlock(&sink->mutex);
 
 out:
-    nn_packet_free(packet);
+    nn_packet_stream_return_packet(conn->stream, packet);
+
     return false;
 }
 
@@ -1314,10 +1326,7 @@ static bool seek_command_callback(void *userdata, struct nn_rpc_connection *conn
     u32 reset_id = nn_packet_read_u32(packet);
 
     struct camu_sink_entry *entry = get_entry_from_id(sink, id);
-    if (!entry) {
-        goto out;
-    }
-
+    if (!entry) goto out;
     entry->reset_id = reset_id;
 
 #ifdef CAMU_SINK_LOCAL
@@ -1327,7 +1336,8 @@ static bool seek_command_callback(void *userdata, struct nn_rpc_connection *conn
     lia_client_seek(&entry->client, pos, at);
 
 out:
-    nn_packet_free(packet);
+    nn_packet_stream_return_packet(conn->stream, packet);
+
     return false;
 }
 
@@ -1337,11 +1347,11 @@ static struct nn_rpc_command commands[] = {
     { .op = CAMU_SINK_SEEK, .callback = seek_command_callback, .userdata = NULL }
 };
 
-static void idd_callback(void *userdata, struct nn_packet *packet)
+static void idd_callback(void *userdata, struct nn_rpc_connection *conn, struct nn_packet *packet)
 {
     struct camu_sink *sink = (struct camu_sink *)userdata;
     (void)sink;
-    nn_packet_free(packet);
+    nn_packet_stream_return_packet(conn->stream, packet);
 }
 
 static void connection_callback(void *userdata, struct nn_rpc_connection *conn)
@@ -1384,12 +1394,15 @@ bool camu_sink_connect(struct camu_sink *sink, u8 type, str *addr, u16 port, str
         al_assert(sink->callback);
         nn_rpc_add_command(&sink->client, &commands[i]);
     }
-    if (!nn_rpc_prepare_client(&sink->client, sink->type, CAMU_MULTIPLEX_RPC)) {
-        return false;
-    }
+    nn_rpc_prepare_client(&sink->client);
     al_str_clone(&sink->addr, addr);
     sink->port = port;
-    nn_rpc_connect(&sink->client, &sink->addr, sink->port);
+#ifdef CAMU_DIRECT_MODE
+    struct nn_rpc_connection *conn = sink->client.conn;
+    nn_multiplex_direct_connect(conn->stream, CAMU_MULTIPLEX_RPC);
+#else
+    nn_rpc_connect(&sink->client, CAMU_MULTIPLEX_RPC, sink->type, &sink->addr, sink->port);
+#endif
     return true;
 }
 
@@ -1427,7 +1440,7 @@ void camu_sink_toggle_pause(struct camu_sink *sink)
     if (current) {
         queue_cmd(sink, (struct camu_sink_cmd){
             .op = TOGGLE_PAUSE,
-            .value.f = camu_clock_get_pts(&current->clock, 0.0),
+            .value.f = camu_clock_get_pts(&current->clock, 0.0, true),
             .opaque = current
         });
     }