Bind Android media queues to core lifetime and source epochs

tests/Makefile.am

@@ -257,7 +257,7 @@ test_android_replies_SOURCES = test_android_replies.c \
     ../tools/chatgui-android/libutun_lite/instance_lite.c ../tools/chatgui-android/libutun_lite/utun_config_api.c \
     ../tools/chatgui-android/libutun_lite/attachment_sender.c ../tools/chatgui-android/libutun_lite/photo_sender.c \
     ../tools/chatgui-android/libutun_lite/video_sender.c ../tools/chatgui-android/libutun_lite/voice_recorder.c \
-    ../tools/chatgui-android/libutun_lite/standby.c
+    ../tools/chatgui-android/libutun_lite/standby.c ../tools/chatgui-android/libutun_lite/stream_bridge.c
 test_android_replies_CPPFLAGS = $(AM_CPPFLAGS) -I$(top_srcdir)/src/chat -I$(top_srcdir)/src/media_async
 test_android_replies_CFLAGS = $(AM_CFLAGS) -ffunction-sections -fdata-sections
 test_android_replies_LDADD = $(LIBUTUN) $(COMMON_LIBS)

tools/chatgui-android/libutun_lite/instance_lite.c

@@ -29,6 +29,7 @@
 #include "../jni_bridge/android_udp_log.h"
 #include "../jni_bridge/android_jni_bridge.h"
 #include "standby.h"
+#include "stream_bridge.h"
 #include <pthread.h>
 #include <stdio.h>
 #include <string.h>
@@ -311,6 +312,7 @@ static void* instance_thread(void* arg) {
         return NULL;
     }
 
+    android_stream_attach(g_inst);
     /* sockets are now created — sync my addresses to DB */
     chat_core_sync_my_addresses(g_inst);
     fire_local_sockets_event();
@@ -355,6 +357,7 @@ static void* instance_thread(void* arg) {
         __atomic_store_n(&g_running, 0, __ATOMIC_RELEASE);
         struct UTUN_INSTANCE* old_inst = g_inst;
         g_inst = NULL;
+        android_stream_detach(old_inst);
         utun_instance_destroy(old_inst);
         my_inst = NULL;
         uasync_print_resources(g_ua, "AFTER_DESTROY");
@@ -387,6 +390,7 @@ static void* instance_thread(void* arg) {
 
         /* Recreate core and chat independently of UTUN. */
         if (utun_core_start(g_inst) != 0 || chat_service_start(g_inst) != 0) { IL_LOGE("poll exit: core/chat start failed"); break; }
+        android_stream_attach(g_inst);
         chat_core_sync_my_addresses(g_inst);
         fire_local_sockets_event();
         if (g_inst->config->global.name[0]) chat_core_update_my_name(g_inst, g_inst->config->global.name);
@@ -410,6 +414,7 @@ static void* instance_thread(void* arg) {
         if (my_inst) {
             chat_event_post(my_inst, CHAT_EVT_SERVICE_STOPPED, NULL, 0);
             g_inst = NULL;
+            android_stream_detach(my_inst);
             utun_instance_destroy(my_inst);
             my_inst = NULL;
         }
@@ -429,7 +434,7 @@ static void* instance_thread(void* arg) {
            но собственные объекты этого потока надо закрыть, иначе течёт fd. */
         IL_LOGI("cleanup (stale thread gen=%d vs %d): destroying own inst=%p ua=%p to avoid fd leak",
                 my_gen, g_generation, (void*)my_inst, (void*)my_ua);
-        if (my_inst) { utun_instance_destroy(my_inst); my_inst = NULL; }
+        if (my_inst) { android_stream_detach(my_inst); utun_instance_destroy(my_inst); my_inst = NULL; }
         if (my_ua) { udp_log_detach(my_ua); uasync_destroy(my_ua, 0); my_ua = NULL; }
     }
     __atomic_store_n(&g_thread_running, 0, __ATOMIC_RELEASE);

tools/chatgui-android/libutun_lite/stream_bridge.c

@@ -24,7 +24,7 @@ struct media_record {
 };
 struct media_queue { struct media_record *head, *tail; size_t bytes; unsigned count; };
 struct android_media_session {
-    uint64_t id, group, source, next_frame, last_pts[2], drain_until;
+    uint64_t id, group, source, epoch, next_frame, last_pts[2], drain_until;
     struct STREAM_HANDLE *handle;
     struct stream_media_receiver *receiver;
     struct stream_media_format format;
@@ -110,7 +110,8 @@ static void media_event(struct UTUN_INSTANCE *inst, const struct stream_event *e
     (void)arg;
     pthread_mutex_lock(&media_lock);
     if (inst == media_inst && viewer.handle && event->descriptor.group_id == viewer.group
-        && event->descriptor.source_node_id == viewer.source) {
+        && event->descriptor.source_node_id == viewer.source && event->descriptor.source_epoch == viewer.epoch
+        && !viewer.stopping && (event->type != STREAM_EVENT_CLOSED || !stream_handle_active(viewer.handle))) {
         if (event->type == STREAM_EVENT_DATA) {
             int result = stream_media_receiver_feed(viewer.receiver, event->data, event->length);
             if (result != STREAM_OK || viewer.rx_overflow) {
@@ -182,10 +183,16 @@ static void media_process(void *arg) {
     for (unsigned i = 0; i < 2; i++) {
         struct android_media_session *s = sessions[i];
         if (s->opening) {
+            if (s->stopping) { media_close(s, 5); continue; }
             s->opening = 0;
             int result = s->source ? stream_subscribe_open(media_inst, s->group, s->source, &s->handle)
                                    : stream_publish_open(media_inst, s->group, NULL, &s->handle);
             if (result == STREAM_OK && s->source) result = stream_media_receiver_create(&s->receiver, media_frame, s);
+            struct stream_info info;
+            if (result == STREAM_OK) {
+                result = stream_get_info(s->handle, &info);
+                if (result == STREAM_OK) s->epoch = info.descriptor.source_epoch;
+            }
             if (result != STREAM_OK) {
                 DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open failed id=%llu group=%llx source=%llx result=%d",
                             (unsigned long long)s->id, (unsigned long long)s->group, (unsigned long long)s->source, result);