From db7264e271c7079b85d82208beba341490b53918 Mon Sep 17 00:00:00 2001 From: evgeny Date: Fri, 16 Oct 2026 01:47:44 +0200 Subject: [PATCH] Suppress source radio playback only while viewing audible streams --- src/radio/radio_audio.c | 2 +- src/radio/radio_headless.c | 2 +- src/stream/stream.c | 62 +++++++++++++++++++++++++++++++++----- 3 files changed, 57 insertions(+), 9 deletions(-) diff --git a/src/radio/radio_audio.c b/src/radio/radio_audio.c index 050a495d..f169848e 100644 --- a/src/radio/radio_audio.c +++ b/src/radio/radio_audio.c @@ -1173,7 +1173,7 @@ void radio_audio_on_frame(struct UTUN_INSTANCE* inst, uint64_t group_id, if (!fin && (!opus || len <= 0)) return; uint64_t now = get_time_tb(); - int muted = radio_member_muted(inst, group_id, src_node_id); + int muted = radio_member_silenced(inst, group_id, src_node_id); pthread_mutex_lock(&g_mtx); if (g_active && group_id == g_group_id) { if (len > 0) { g_last_rx_tb = now; g_last_rx_node = src_node_id; } /* busy-сигнал: чужой аудио-кадр */ diff --git a/src/radio/radio_headless.c b/src/radio/radio_headless.c index 7d6b5a4c..3153df0e 100644 --- a/src/radio/radio_headless.c +++ b/src/radio/radio_headless.c @@ -227,7 +227,7 @@ static void rh_frame_cb(struct UTUN_INSTANCE* inst, uint64_t group_id, /* активный канал: первый кадр фиксирует group_id (узел слушает один канал) */ if (rh->group_id == 0) rh->group_id = group_id; if (group_id != rh->group_id) return; - if (radio_member_muted(inst, group_id, src_node_id)) { + if (radio_member_silenced(inst, group_id, src_node_id)) { if (seq == 0) DEBUG_DEBUG(DEBUG_CATEGORY_RADIO, "%s: muted burst grp=%016llx src=%016llx stream=%u", RH_ID, (unsigned long long)group_id, (unsigned long long)src_node_id, stream_id); return; diff --git a/src/stream/stream.c b/src/stream/stream.c index 890b7edd..38ea3f2f 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -11,6 +11,7 @@ #include "../utun_instance.h" #include "../chat/chat_core.h" #include "../chat/chat_event.h" +#include "../radio/radio.h" #include "stream_member.h" #include "stream_source.h" #include "stream_priv.h" @@ -39,6 +40,7 @@ struct stream_session { unsigned owners, relay_owners, holds; int anchored, decoded, closing, invalidated, finished, end_delivered, reason; int restarting, restart_gap, notify_close; + int audio_enabled, radio_muted; }; struct STREAM_HANDLE { @@ -193,6 +195,25 @@ static struct stream_session *session_find(struct stream_service *service, uint6 return NULL; } +int stream_radio_muted(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id) { + struct stream_session *s = inst && inst->stream ? session_find(inst->stream, group_id, source_id) : NULL; + return s && s->radio_muted; +} + +/* Курсор означает просмотр. Raw readers, publisher и relay не отключают рацию. */ +static void session_radio_update(struct stream_session *s) { + int muted = 0; + if (!s->closing && !s->invalidated && !s->finished && !s->source && s->audio_enabled) + for (struct STREAM_HANDLE *h = s->service->handles; h; h = h->next) + if (h->session == s && h->media_cursor && !h->relay && !h->publisher) { muted = 1; break; } + if (muted == s->radio_muted) return; + s->radio_muted = muted; + DEBUG_INFO(DEBUG_CATEGORY_RADIO, "stream viewer mute group=%llx source=%llx epoch=%llx muted=%d audio=%d", + (unsigned long long)s->descriptor.group_id, (unsigned long long)s->descriptor.source_node_id, + (unsigned long long)s->descriptor.source_epoch, muted, s->audio_enabled); + radio_refresh_member_mute(s->service->inst, s->descriptor.group_id, s->descriptor.source_node_id); +} + /* Две опции заменяются одной подписанной записью; список отражает экземпляры, не отдельные handles. */ static int service_publish_options(struct stream_service *service, struct TOPO_GROUP *group) { if (!group) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream options group missing"); return STREAM_ERR_STATE; } @@ -212,7 +233,7 @@ static int service_publish_options(struct stream_service *service, struct TOPO_G state.subscriptions[at] = state.subscriptions[at - 1]; at--; } state.subscriptions[at] = entry; - if (s->source && !s->finished) { state.source = s->descriptor; state.published = 1; } + if (s->source && !s->finished) { state.source = s->descriptor; state.published = 1; state.audio_enabled = s->audio_enabled; } } char source[MS_USERINFO_MAX + 1], subscriptions[MS_USERINFO_MAX + 1]; int result = stream_member_format(&state, source, subscriptions); @@ -263,6 +284,7 @@ static void session_close(struct stream_session *s, int reason, int update_profi if (s->closing) return; s->notify_close = notify && s->owners > s->relay_owners; s->closing = 1; s->reason = reason; s->owners = 0; + session_radio_update(s); stream_optimizer_destroy(s->optimizer); s->optimizer = NULL; stream_network_cancel(s->service->network, &s->descriptor); for (struct STREAM_HANDLE *h = s->service->handles; h; h = h->next) @@ -501,6 +523,8 @@ static void member_changed(uint64_t node_id, const char *tags, const char *chann || state.source.block_size != s->descriptor.block_size)))) { s->invalidated = 1; s->reason = STREAM_CLOSE_SOURCE_CHANGED; } + if (!own && result == STREAM_OK) s->audio_enabled = state.published && state.audio_enabled; + session_radio_update(s); } if (!changed || service->reconcile) return; service->reconcile = uasync_call_soon(service->inst->ua, service, service_reconcile); @@ -594,7 +618,7 @@ void stream_set_event_callback(struct UTUN_INSTANCE *inst, stream_event_fn callb /* Подготовить все ресурсы до публикации member-опций; rollback не оставляет видимую подписку. */ static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descriptor *d, const uint8_t key[32], int publisher, int automatic, - int relay, + int relay, int audio_enabled, struct STREAM_HANDLE **out) { struct stream_service *service = inst->stream; struct stream_session *s = session_find(service, d->group_id, d->source_node_id); @@ -609,6 +633,7 @@ static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descript s = u_calloc(1, sizeof(*s)); if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream session allocation failed"); u_free(h); return STREAM_ERR_MEMORY; } s->service = service; s->descriptor = *d; memcpy(s->public_key, key, 32); + s->audio_enabled = audio_enabled; s->output = u_malloc((uint32_t)stream_input_size(d)); s->metrics = u_calloc(d->n, sizeof(*s->metrics)); int result = s->output && s->metrics ? STREAM_OK : STREAM_ERR_MEMORY; @@ -688,14 +713,14 @@ int stream_publish_open(struct UTUN_INSTANCE *inst, uint64_t group_id, const str struct stream_session *previous = session_find(inst->stream, group_id, inst->node_id); if (previous && previous->source && previous->end_valid) session_close(previous, STREAM_CLOSE_SOURCE_CHANGED, 0, 1); - return session_open(inst, &d, inst->my_ed25519_pubkey, 1, 0, 0, out); + return session_open(inst, &d, inst->my_ed25519_pubkey, 1, 0, 0, 0, out); } static int subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, int automatic, struct STREAM_HANDLE **out) { int result = stream_open_check(inst, group_id, out); if (result != STREAM_OK) return result; struct stream_session *s = session_find(inst->stream, group_id, source_id); - if (s) return session_open(inst, &s->descriptor, s->public_key, 0, automatic, 0, out); + if (s) return session_open(inst, &s->descriptor, s->public_key, 0, automatic, 0, s->audio_enabled, out); struct stream_member_state state; uint8_t key[32]; result = stream_member_load(inst, group_id, source_id, &state, key); if (result != STREAM_OK) return result; @@ -703,7 +728,7 @@ static int subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_ DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "source not yet published source=%llx", (unsigned long long)source_id); return STREAM_WAIT; } - return session_open(inst, &state.source, key, 0, automatic, 0, out); + return session_open(inst, &state.source, key, 0, automatic, 0, state.audio_enabled, out); } /* Запрос старой эпохи не может незаметно подписать суперузел на новый запуск. */ @@ -712,7 +737,7 @@ int stream_relay_open(struct UTUN_INSTANCE *inst, const struct stream_descriptor if (result != STREAM_OK) return result; struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id); if (s && s->descriptor.source_epoch == key->source_epoch) - return session_open(inst, &s->descriptor, s->public_key, 0, 1, 1, out); + return session_open(inst, &s->descriptor, s->public_key, 0, 1, 1, s->audio_enabled, out); struct stream_member_state state; uint8_t public_key[32]; result = stream_member_load(inst, key->group_id, key->source_node_id, &state, public_key); if (result != STREAM_OK) return result; @@ -721,7 +746,7 @@ int stream_relay_open(struct UTUN_INSTANCE *inst, const struct stream_descriptor (unsigned long long)key->source_node_id, (unsigned long long)key->source_epoch, state.published, (unsigned long long)state.source.source_epoch); return STREAM_ERR_STATE; } - return session_open(inst, &state.source, public_key, 0, 1, 1, out); + return session_open(inst, &state.source, public_key, 0, 1, 1, state.audio_enabled, out); } int stream_subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, struct STREAM_HANDLE **out) { @@ -758,6 +783,7 @@ void stream_close(struct STREAM_HANDLE *h) { } struct stream_session *s = h->session; if (s) { + session_radio_update(s); int was_relay = s->owners == s->relay_owners; if ((h->publisher && !s->end_valid) || s->owners == 1) session_close(s, STREAM_CLOSE_STOPPED, 1, !h->relay); else { @@ -804,6 +830,7 @@ int stream_media_enable(struct STREAM_HANDLE *h, unsigned tracks) { if (tracks && !h->media_cursor) { int result = stream_media_cursor_create(&h->media_cursor, s->media, tracks); if (result != STREAM_OK) return result; + session_radio_update(s); } session_media_schedule(s); return STREAM_OK; @@ -898,6 +925,27 @@ int stream_publish_block(struct STREAM_HANDLE *h, const uint8_t *data, size_t le return result; } +int stream_publish_set_audio_enabled(struct STREAM_HANDLE *h, int enabled) { + if (!stream_handle_active(h) || !h->publisher || h->session->finished || (enabled != 0 && enabled != 1)) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream audio change invalid publisher/state enabled=%d", enabled); return STREAM_ERR_STATE; + } + struct stream_session *s = h->session; + if (s->audio_enabled == enabled) return STREAM_OK; + s->holds++; + int previous = s->audio_enabled; + s->audio_enabled = enabled; + int result = service_publish_options(s->service, stream_group(s->service->inst, s->descriptor.group_id)); + if (result != STREAM_OK || s->closing) { + s->audio_enabled = previous; + if (result == STREAM_OK) result = STREAM_ERR_STATE; + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream audio state publication failed result=%d", result); + } else DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream audio group=%llx epoch=%llx enabled=%d", + (unsigned long long)s->descriptor.group_id, (unsigned long long)s->descriptor.source_epoch, enabled); + s->holds--; + if (s->closing && !s->holds && !s->delivery) session_free(s); + return result; +} + int stream_publish_finish(struct STREAM_HANDLE *h) { if (!stream_handle_active(h) || !h->publisher) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream finish not a publisher"); return STREAM_ERR_STATE; } struct stream_session *s = h->session;