diff --git a/tests/Makefile.am b/tests/Makefile.am index 08e006b3..c2e46e3b 100644 --- a/tests/Makefile.am +++ b/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) diff --git a/tools/chatgui-android/libutun_lite/instance_lite.c b/tools/chatgui-android/libutun_lite/instance_lite.c index 8a262ead..b12b0cc6 100644 --- a/tools/chatgui-android/libutun_lite/instance_lite.c +++ b/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 #include #include @@ -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); diff --git a/tools/chatgui-android/libutun_lite/stream_bridge.c b/tools/chatgui-android/libutun_lite/stream_bridge.c index 791c36a1..a999dc6c 100644 --- a/tools/chatgui-android/libutun_lite/stream_bridge.c +++ b/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);