diff --git a/src/stream/stream.c b/src/stream/stream.c index 38731c5b..bc923919 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -54,6 +54,15 @@ struct STREAM_HANDLE { struct stream_media_cursor *media_cursor; }; +struct source_watch { + struct source_watch *next; + struct stream_service *service; + uint64_t owner, group; + void *pending; + char *snapshot; + int valid; +}; + struct stream_service { struct UTUN_INSTANCE *inst; struct stream_session *sessions; @@ -62,10 +71,7 @@ struct stream_service { stream_event_fn callback; void *arg; void *reconcile; - void *sources_pending; - uint64_t sources_group; - char *sources_snapshot; - int sources_valid; + struct source_watch *watches; int stopping, dispatching, registered, initializing; }; @@ -125,14 +131,15 @@ int stream_sources_json(struct UTUN_INSTANCE *inst, uint64_t group_id, char **ou /* Один отложенный снимок объединяет Merkle/BGP-события; неизменный список повторно не отправляется. */ static void sources_deliver(void *arg) { - struct stream_service *service = arg; - service->sources_pending = NULL; - if (service->stopping || !service->sources_group) return; + struct source_watch *watch = arg; + struct stream_service *service = watch->service; + watch->pending = NULL; + if (service->stopping) return; char *json = NULL; - int result = stream_sources_json(service->inst, service->sources_group, &json); - int valid = result == STREAM_OK || result == STREAM_WAIT; + int result = stream_sources_json(service->inst, watch->group, &json); + int valid = result == STREAM_OK; const char *list = json ? json : "[]"; - if (service->sources_snapshot && valid == service->sources_valid && !strcmp(service->sources_snapshot, list)) { + if (watch->snapshot && valid == watch->valid && !strcmp(watch->snapshot, list)) { u_free(json); return; } size_t length = strlen(list); @@ -140,30 +147,46 @@ static void sources_deliver(void *arg) { if (!event) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source event allocation failed bytes=%zu", length); u_free(json); return; } - stream_write_be(event, service->sources_group, 8); event[8] = (uint8_t)valid; + stream_write_be(event, watch->group, 8); event[8] = (uint8_t)valid; memcpy(event + 9, list, length); char *snapshot = json ? json : u_strdup(list); if (!snapshot) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source snapshot allocation failed"); - u_free(service->sources_snapshot); service->sources_snapshot = snapshot; service->sources_valid = valid; + u_free(watch->snapshot); watch->snapshot = snapshot; watch->valid = valid; DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream sources changed group=%llx valid=%d bytes=%zu", - (unsigned long long)service->sources_group, valid, length); + (unsigned long long)watch->group, valid, length); chat_event_post(service->inst, CHAT_EVT_STREAM_SOURCES, event, (int)(length + 9)); u_free(event); } /* Наблюдение принадлежит сервису и снимается до освобождения ядра. */ -int stream_watch_sources(struct UTUN_INSTANCE *inst, uint64_t group_id) { +int stream_watch_sources(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t owner) { struct stream_service *service = inst ? inst->stream : NULL; - if (!service || service->stopping) { + if (!service || (service->stopping && group_id)) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream source watch without service group=%llx", (unsigned long long)group_id); return STREAM_ERR_STATE; } - service->sources_group = group_id; - u_free(service->sources_snapshot); service->sources_snapshot = NULL; - if (service->sources_pending) uasync_call_soon_cancel(inst->ua, service->sources_pending); - service->sources_pending = group_id ? uasync_call_soon(inst->ua, service, sources_deliver) : NULL; - DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source watch group=%llx", (unsigned long long)group_id); - if (group_id && !service->sources_pending) { + struct source_watch **link = &service->watches; + while (*link && (*link)->owner != owner) link = &(*link)->next; + struct source_watch *watch = *link; + if (!group_id) { + if (watch) { + if (watch->pending) uasync_call_soon_cancel(inst->ua, watch->pending); + *link = watch->next; u_free(watch->snapshot); u_free(watch); + } + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source watch removed owner=%llu", (unsigned long long)owner); + return STREAM_OK; + } + if (!watch) { + watch = u_calloc(1, sizeof(*watch)); + if (!watch) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source watch allocation failed"); return STREAM_ERR_MEMORY; } + watch->service = service; watch->owner = owner; *link = watch; + } + watch->group = group_id; + u_free(watch->snapshot); watch->snapshot = NULL; + if (!watch->pending) watch->pending = uasync_call_soon(inst->ua, watch, sources_deliver); + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream source watch group=%llx owner=%llu", + (unsigned long long)group_id, (unsigned long long)owner); + if (!watch->pending) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source watch scheduling failed"); return STREAM_ERR_MEMORY; } return STREAM_OK; @@ -492,10 +515,12 @@ static void member_changed(uint64_t node_id, const char *tags, const char *chann (void)tags; struct stream_service *service = arg; if (service->stopping) return; - struct TOPO_GROUP *watched = service->sources_group ? stream_group(service->inst, service->sources_group) : NULL; - if (watched && channel && !strcmp(watched->channel_id, channel) && !service->sources_pending) { - service->sources_pending = uasync_call_soon(service->inst->ua, service, sources_deliver); - if (!service->sources_pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source refresh scheduling failed"); + for (struct source_watch *watch = service->watches; watch; watch = watch->next) { + struct TOPO_GROUP *watched = stream_group(service->inst, watch->group); + if (watched && channel && !strcmp(watched->channel_id, channel) && !watch->pending) { + watch->pending = uasync_call_soon(service->inst->ua, watch, sources_deliver); + if (!watch->pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream source refresh scheduling failed"); + } } if (!service->sessions) return; stream_network_membership_changed(service->network, node_id, channel); @@ -579,8 +604,7 @@ void stream_service_destroy(struct UTUN_INSTANCE *inst) { stream_network_destroy(service->network); service->network = NULL; if (service->registered) member_sync_remove_props_cbk(inst, member_changed, service); if (service->reconcile) uasync_call_soon_cancel(inst->ua, service->reconcile); - if (service->sources_pending) uasync_call_soon_cancel(inst->ua, service->sources_pending); - u_free(service->sources_snapshot); + while (service->watches) stream_watch_sources(inst, 0, service->watches->owner); while (service->sessions) { struct stream_session *s = service->sessions; if (!s->closing) session_close(s, STREAM_CLOSE_SERVICE, 1, 0); @@ -597,9 +621,11 @@ void stream_service_destroy(struct UTUN_INSTANCE *inst) { void stream_cancel_group(struct UTUN_INSTANCE *inst, uint64_t group_id) { struct stream_service *service = inst ? inst->stream : NULL; if (!service) return; - if (service->sources_group == group_id && !service->sources_pending) { - service->sources_pending = uasync_call_soon(inst->ua, service, sources_deliver); - if (!service->sources_pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream cancelled group source refresh failed"); + for (struct source_watch *watch = service->watches; watch; watch = watch->next) { + if (watch->group == group_id && !watch->pending) { + watch->pending = uasync_call_soon(inst->ua, watch, sources_deliver); + if (!watch->pending) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream cancelled group source refresh failed"); + } } struct stream_session *s = service->sessions; while (s) { diff --git a/src/stream/stream.h b/src/stream/stream.h index 6f8b9ea7..51227c9e 100644 --- a/src/stream/stream.h +++ b/src/stream/stream.h @@ -64,9 +64,11 @@ void stream_set_event_callback(struct UTUN_INSTANCE *inst, stream_event_fn callb /* Проверенные online-публикации: JSON с id(decimal string), epoch(hex string), name, self. * Только uasync; *out заранее NULL, результат освобождается u_free(). WAIT — группа недоступна. */ int stream_sources_json(struct UTUN_INSTANCE *inst, uint64_t group_id, char **out); -/* Один выбранный канал frontend: начальный снимок и изменения через CHAT_EVT_STREAM_SOURCES. - * group_id=0 снимает наблюдение. Не создаёт подписку/буфер и не удерживает ACTIVE. Только uasync. */ -int stream_watch_sources(struct UTUN_INSTANCE *inst, uint64_t group_id); +/* Независимое наблюдение owner: начальный снимок и изменения через CHAT_EVT_STREAM_SOURCES. + * group_id=0 снимает только этого owner; owner=0 разрешён для выбранного канала frontend. + * Недоступный каталог имеет valid=0 и не означает завершение источников. + * Не создаёт подписку/буфер и не удерживает ACTIVE. Только uasync. */ +int stream_watch_sources(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t owner); /* Локальный owner удерживает эффективный ACTIVE; желаемый frontend-режим ядро восстанавливает после последнего close. */ int stream_service_needs_activity(const struct UTUN_INSTANCE *inst); /* Только локальный viewer этой группы/эпохи подавляет рацию источника со включённым звуком. */ diff --git a/tools/chatgui-android/libutun_lite/stream_bridge.c b/tools/chatgui-android/libutun_lite/stream_bridge.c index b9d0440d..204f94f2 100644 --- a/tools/chatgui-android/libutun_lite/stream_bridge.c +++ b/tools/chatgui-android/libutun_lite/stream_bridge.c @@ -168,7 +168,7 @@ static void media_process(void *arg) { if (!media_inst) { pthread_mutex_unlock(&media_lock); return; } if (watch_pending) { watch_pending = 0; - if (stream_watch_sources(media_inst, watch_group) != STREAM_OK) + if (stream_watch_sources(media_inst, watch_group, 0) != STREAM_OK) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream source watch failed group=%llx", (unsigned long long)watch_group); } if (viewer.opening) { @@ -247,7 +247,7 @@ void android_stream_detach(struct UTUN_INSTANCE *inst) { if (media_post) uasync_cancel_post(inst->ua, media_post); if (media_timer) uasync_cancel_timeout(inst->ua, media_timer); media_post = NULL; media_timer = NULL; - stream_watch_sources(inst, 0); + stream_watch_sources(inst, 0, 0); stream_set_event_callback(inst, NULL, NULL); queue_clear(&output_queue); stream_publisher_destroy(transmitter); transmitter = NULL; memset(&publisher, 0, sizeof(publisher));