jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
2286 Commits
15 Branches
2 Tags
184 MiB
 
 
 
 
 
 
Tree: 7e19cbaf3a
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from '7e19cbaf3a'
${ noResults }
utun2/src/radio/radio.c

675 lines
31 KiB
Raw Blame History

// radio.c — рация (PTT walkie-talkie): мультикаст голоса по CHAT-группе.
//
// Флудит Opus-кадры (ETCP_ID_RADIO) всем подписчикам канала — узлам с
// TOPO_FLAG_RADIO в NODEINFO. Пакет несёт hop_list (узлы, через которые он уже
// прошёл) для защиты от циклов и форвардится только в те ветви, за которыми есть
// хотя бы один подписчик (RPF-обрезание по BGP-путям). Кадры нумеруются от 0,
// последний кадр передачи помечается FIN.
//
// Приём: если мы подписаны (group->radio_active) — доставляем кадр в frame_fn
// (аудио-движок radio_audio). Независимо от подписки форвардим дальше.
// Все функции — в uasync-потоке.
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include "../lib/platform_compat.h"
#include "../lib/debug_config.h"
#include "../lib/mem.h"
#include "../lib/u_async.h"
#include "../lib/ll_queue.h"
#include "../transport_layer/etcp_api.h"
#include "../transport_layer/etcp.h"
#include "../routing_layer/topo_node.h"
#include "../routing_layer/topo_group.h"
#include "../routing_layer/topo_node_sqlite.h"
#include "../chat/chat_event.h"
#include "radio.h"
#include "radio_proto.h"
#define RADIO_ID "radio"
#define RADIO_MAX_TALKERS 16 /* максимум одновременно говорящих источников */
#define RADIO_DEDUP_MAX 128 /* ёмкость кэша дедупликации (≈2.5с @ 50 кадров/с) */
#define RADIO_DEDUP_TTL_TB 20000 /* 2с — окно дедупликации (совпадает с таймаутом источника) */
#define RADIO_TALKER_TIMEOUT_TB 20000 /* 2с без кадров — считаем передачу завершённой (FIN потерян/пир отвалился) */
/* Запись кэша дедупликации: ключ — (src, stream, seq), tb — момент прихода. */
struct radio_seen {
uint64_t src;
uint16_t stream;
uint16_t seq;
uint64_t tb;
};
struct radio_ctx {
struct TOPO_GROUP* group;
uint16_t stream_id; /* id текущей передачи (talk-burst) */
uint16_t seq; /* seq следующего исходящего кадра */
int talking; /* 1 = идёт передача (PTT зажата) */
uint32_t c_tx_frames; /* отправлено кадров */
uint32_t c_rx_frames; /* доставлено локально (мы подписаны) */
uint32_t c_fwd_frames; /* форварднуто кадров */
uint32_t c_loop_dropped; /* дропнуто из-за цикла (self в hop_list) */
uint32_t c_nosub_dropped; /* не отправлено: за соседом нет подписчиков */
uint32_t c_bad_dropped; /* дропнуто некорректных пакетов */
uint32_t c_dup_dropped; /* дропнуто дублей (src,stream,seq) пришедших по другому пути */
struct radio_talker { uint64_t src; uint16_t stream; uint64_t last_tb; } talkers[RADIO_MAX_TALKERS]; /* активные говорящие */
void* talker_timer; /* периодический sweep зависших говорящих (self-rearming) */
struct radio_seen seen[RADIO_DEDUP_MAX]; /* кэш дедупликации */
int seen_count;
uint64_t* muted_members; /* локальный кэш, принадлежит uasync-потоку */
int muted_count;
int muted_loaded;
};
struct radio_instance {
radio_frame_fn frame_fn;
void* frame_arg;
radio_mute_fn mute_fn;
void* mute_arg;
};
static void radio_talker_timer_cb(void* arg);
static struct radio_ctx* radio_of(struct TOPO_GROUP* group) {
return group ? group->radio : NULL;
}
static struct radio_instance* radio_inst(struct UTUN_INSTANCE* inst) {
return inst ? inst->radio : NULL;
}
int radio_init(struct TOPO_GROUP* group) {
if (!group) { DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: init null group", RADIO_ID); return -1; }
if (group->radio) { DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: already initialized grp=%016llx", RADIO_ID, (unsigned long long)group->group_id); return -1; }
struct radio_ctx* ctx = u_calloc(1, sizeof(struct radio_ctx));
if (!ctx) { DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: alloc failed", RADIO_ID); return -1; }
ctx->group = group;
ctx->stream_id = (uint16_t)(get_time_tb() & 0xFFFF);
group->radio = ctx;
ctx->talker_timer = uasync_set_timeout(group->instance->ua, RADIO_TALKER_TIMEOUT_TB, ctx, radio_talker_timer_cb, "radio_talker");
if (!ctx->talker_timer) DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: init talker_timer failed grp=%016llx", RADIO_ID, (unsigned long long)group->group_id);
DEBUG_INFO(DEBUG_CATEGORY_RADIO, "%s: init ok grp=%016llx", RADIO_ID, (unsigned long long)group->group_id);
return 0;
}
void radio_destroy(struct TOPO_GROUP* group) {
if (!group || !group->radio) return;
struct radio_ctx* ctx = group->radio;
if (ctx->talker_timer) { uasync_cancel_timeout(group->instance->ua, ctx->talker_timer); ctx->talker_timer = NULL; }
DEBUG_INFO(DEBUG_CATEGORY_RADIO, "%s: destroy grp=%016llx (tx=%u rx=%u fwd=%u loop=%u nosub=%u bad=%u dup=%u)",
RADIO_ID, (unsigned long long)group->group_id,
ctx->c_tx_frames, ctx->c_rx_frames, ctx->c_fwd_frames,
ctx->c_loop_dropped, ctx->c_nosub_dropped, ctx->c_bad_dropped, ctx->c_dup_dropped);
u_free(ctx->muted_members);
u_free(ctx);
group->radio = NULL;
}
/* ── сборка payload ── */
/* Пишет payload рации (начиная с group_id) в buf, возвращает длину. */
static size_t radio_build(uint8_t* buf, uint64_t group_id, uint64_t src, uint16_t stream,
uint16_t seq, uint8_t flags, const uint64_t* hops, uint8_t hop_count,
const uint8_t* opus, int opus_len) {
size_t off = 0;
memcpy(buf + off, &group_id, 8); off += 8;
memcpy(buf + off, &src, 8); off += 8;
memcpy(buf + off, &stream, 2); off += 2;
memcpy(buf + off, &seq, 2); off += 2;
buf[off++] = flags;
buf[off++] = hop_count;
if (hop_count) { memcpy(buf + off, hops, (size_t)hop_count * 8); off += (size_t)hop_count * 8; }
if (opus_len > 0) { memcpy(buf + off, opus, (size_t)opus_len); off += (size_t)opus_len; }
return off;
}
/* ── отправка ── */
/* статистика отправки: раз в секунду — сколько кадров реально ушло в etcp_send и сколько упало. */
static uint64_t s_send_stats_tb = 0;
static uint32_t s_send_ok = 0;
static uint32_t s_send_fail = 0;
static void radio_send_to(struct TOPO_GROUP* group, struct ETCP_CONN* conn,
const uint8_t* payload, size_t payload_len) {
struct ll_entry* ee = queue_entry_new(0);
if (!ee) {
DEBUG_WARN(DEBUG_CATEGORY_RADIO, "%s: send: queue_entry_new OOM", RADIO_ID);
return;
}
uint8_t* pkt = u_malloc(1 + payload_len);
if (!pkt) {
queue_entry_free(ee);
DEBUG_WARN(DEBUG_CATEGORY_RADIO, "%s: send: u_malloc OOM len=%zu", RADIO_ID, payload_len);
return;
}
pkt[0] = ETCP_ID_RADIO;
memcpy(pkt + 1, payload, payload_len);
ee->dgram = pkt;
ee->len = (size_t)(1 + payload_len);
if (etcp_send(conn, ee) != 0) {
u_free(pkt); queue_entry_free(ee);
s_send_fail++;
DEBUG_WARN(DEBUG_CATEGORY_RADIO, "%s: send: etcp_send FAILED conn=%s len=%zu",
RADIO_ID, conn ? conn->log_name : "?", (size_t)(1 + payload_len));
} else {
s_send_ok++;
}
uint64_t now = get_time_tb();
if (s_send_stats_tb == 0 || now - s_send_stats_tb >= 10000) { /* 1с (tb 0.1мс) */
DEBUG_INFO(DEBUG_CATEGORY_RADIO, "%s: send stats ok=%u/s fail=%u", RADIO_ID, s_send_ok, s_send_fail);
s_send_stats_tb = now;
s_send_ok = 0;
s_send_fail = 0;
}
}
/* Есть ли за conn (не считая уже пройденных hop_list) хотя бы один подписчик рации. */
static int radio_neighbor_has_subscriber(struct TOPO_GROUP* group, struct ETCP_CONN* conn,
const uint64_t* hop_list, uint8_t hop_count) {
if (!group || !conn || !group->nodes) return 0;
uint64_t self = group->instance ? group->instance->node_id : 0;
struct ll_entry* ne = group->nodes->head;
while (ne) {
struct TOPO_GROUP_NODE* nq = (struct TOPO_GROUP_NODE*)ne;
if (nq->radio && nq->node_id != self) {
int visited = 0;
for (int i = 0; i < hop_count; i++) if (hop_list[i] == nq->node_id) { visited = 1; break; }
if (!visited) {
struct ll_entry* pe = nq->paths ? nq->paths->head : NULL;
while (pe) {
struct TOPO_NODEPATH* p = (struct TOPO_NODEPATH*)pe;
if (p->conn == conn) return 1;
pe = pe->next;
}
}
}
ne = ne->next;
}
return 0;
}
/* Флудит payload (уже содержит group_id..hop_list..opus) всем соседям, кроме from_conn. */
static void radio_flood(struct TOPO_GROUP* group, struct ETCP_CONN* from_conn,
const uint8_t* payload, size_t payload_len,
const uint64_t* hop_list, uint8_t hop_count) {
struct radio_ctx* ctx = radio_of(group);
if (!ctx || !group->senders_list) return;
struct ll_entry* se = group->senders_list->head;
while (se) {
struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)se->data;
if (item && item->conn && item->conn != from_conn) {
if (radio_neighbor_has_subscriber(group, item->conn, hop_list, hop_count)) {
radio_send_to(group, item->conn, payload, payload_len);
ctx->c_fwd_frames++;
} else {
ctx->c_nosub_dropped++;
}
}
se = se->next;
}
}
/* ── индикация говорящих ── */
static void radio_talk_event(struct TOPO_GROUP* group, uint64_t src, uint16_t stream, uint8_t on) {
uint8_t buf[19];
size_t off = 0;
memcpy(buf + off, &group->group_id, 8); off += 8;
memcpy(buf + off, &src, 8); off += 8;
memcpy(buf + off, &stream, 2); off += 2;
buf[off++] = on;
chat_event_post(group->instance, CHAT_EVT_RADIO_TALK, buf, (int)off);
DEBUG_DEBUG(DEBUG_CATEGORY_RADIO, "%s: talk %s src=%016llx stream=%u",
RADIO_ID, on ? "START" : "STOP", (unsigned long long)src, stream);
}
/* зафиксировать начало передачи: пост on=1, если это новый поток источника */
static void radio_talker_on(struct radio_ctx* ctx, uint64_t src, uint16_t stream) {
uint64_t now = get_time_tb();
for (int i = 0; i < RADIO_MAX_TALKERS; i++) {
if (ctx->talkers[i].src == src) {
ctx->talkers[i].last_tb = now;
if (ctx->talkers[i].stream != stream) {
ctx->talkers[i].stream = stream;
radio_talk_event(ctx->group, src, stream, 1);
}
return;
}
}
for (int i = 0; i < RADIO_MAX_TALKERS; i++) {
if (ctx->talkers[i].src == 0) {
ctx->talkers[i].src = src;
ctx->talkers[i].stream = stream;
ctx->talkers[i].last_tb = now;
radio_talk_event(ctx->group, src, stream, 1);
return;
}
}
}
/* зафиксировать конец передачи: пост on=0, если источник был в таблице */
static void radio_talker_off(struct radio_ctx* ctx, uint64_t src) {
for (int i = 0; i < RADIO_MAX_TALKERS; i++) {
if (ctx->talkers[i].src == src) {
radio_talk_event(ctx->group, src, ctx->talkers[i].stream, 0);
ctx->talkers[i].src = 0;
ctx->talkers[i].stream = 0;
ctx->talkers[i].last_tb = 0;
return;
}
}
}
/* Снять говорящих, от которых дольше RADIO_TALKER_TIMEOUT_TB нет кадров:
* FIN потерян/не был отправлен (пир отвалился, выключил рацию во время разговора).
* Аналог таймаута источника в radio_audio.c — гарантирует, что индикация «говорит» гаснет. */
static void radio_talkers_sweep(struct radio_ctx* ctx) {
uint64_t now = get_time_tb();
for (int i = 0; i < RADIO_MAX_TALKERS; i++) {
if (!ctx->talkers[i].src) continue;
if (now - ctx->talkers[i].last_tb > RADIO_TALKER_TIMEOUT_TB) {
DEBUG_DEBUG(DEBUG_CATEGORY_RADIO, "%s: talker timeout src=%016llx (last frame %.0fms ago)",
RADIO_ID, (unsigned long long)ctx->talkers[i].src,
(double)(now - ctx->talkers[i].last_tb) / 10.0);
radio_talker_off(ctx, ctx->talkers[i].src);
}
}
}
static void radio_talker_timer_cb(void* arg) {
struct radio_ctx* ctx = (struct radio_ctx*)arg;
if (!ctx || !ctx->group) return;
radio_talkers_sweep(ctx);
ctx->talker_timer = uasync_set_timeout(ctx->group->instance->ua, RADIO_TALKER_TIMEOUT_TB, ctx, radio_talker_timer_cb, "radio_talker");
}
/* Дедупликация: вернуть 1, если (src,stream,seq) уже видели (дубль по другому пути),
* иначе зафиксировать в кэше и вернуть 0. Лениво чистит просроченные записи. */
static int radio_seen_dup(struct radio_ctx* ctx, uint64_t src, uint16_t stream, uint16_t seq, uint64_t now) {
int free_slot = -1, oldest = -1;
uint64_t oldest_tb = 0;
for (int i = 0; i < ctx->seen_count; i++) {
struct radio_seen* s = &ctx->seen[i];
if (s->tb && now - s->tb > RADIO_DEDUP_TTL_TB) { s->tb = 0; s->src = 0; }
if (!s->tb) { if (free_slot < 0) free_slot = i; continue; }
if (s->src == src && s->stream == stream && s->seq == seq) return 1;
if (oldest < 0 || s->tb < oldest_tb) { oldest = i; oldest_tb = s->tb; }
}
if (free_slot < 0) {
if (ctx->seen_count < RADIO_DEDUP_MAX) free_slot = ctx->seen_count++;
else free_slot = oldest;
}
ctx->seen[free_slot].src = src;
ctx->seen[free_slot].stream = stream;
ctx->seen[free_slot].seq = seq;
ctx->seen[free_slot].tb = now;
return 0;
}
/* ── приём ── */
static void radio_recv(struct TOPO_GROUP* group, struct ETCP_CONN* from_conn,
const uint8_t* payload, size_t payload_len) {
struct radio_ctx* ctx = radio_of(group);
if (!ctx || !from_conn || payload_len < RADIO_HDR_SIZE) return;
const uint8_t* p = payload;
uint64_t group_id, src;
uint16_t stream, seq;
uint8_t flags, hop_count;
memcpy(&group_id, p + RADIO_OFF_GROUP, 8);
memcpy(&src, p + RADIO_OFF_SRC, 8);
memcpy(&stream, p + RADIO_OFF_STREAM, 2);
memcpy(&seq, p + RADIO_OFF_SEQ, 2);
flags = p[RADIO_OFF_FLAGS];
hop_count = p[RADIO_OFF_HOPS];
if (group_id != group->group_id || hop_count > RADIO_MAX_HOPS
|| payload_len < RADIO_HDR_SIZE + (size_t)hop_count * 8) {
ctx->c_bad_dropped++;
DEBUG_WARN(DEBUG_CATEGORY_RADIO, "%s: bad pkt grp=%016llx hops=%u len=%zu from=%s",
RADIO_ID, (unsigned long long)group_id, hop_count, payload_len, from_conn->log_name);
return;
}
const uint64_t* hops = (const uint64_t*)(p + RADIO_OFF_HOPLIST);
const uint8_t* opus = p + RADIO_OFF_HOPLIST + (size_t)hop_count * 8;
int opus_len = (int)(payload_len - RADIO_OFF_HOPLIST - (size_t)hop_count * 8);
if (opus_len > RADIO_MAX_OPUS) { ctx->c_bad_dropped++; return; }
uint64_t self = group->instance->node_id;
/* цикл: мы уже проходили этот пакет */
for (int i = 0; i < hop_count; i++) {
if (hops[i] == self) {
ctx->c_loop_dropped++;
DEBUG_DEBUG(DEBUG_CATEGORY_RADIO, "%s: loop drop src=%016llx stream=%u seq=%u", RADIO_ID, (unsigned long long)src, stream, seq);
return;
}
}
/* дедупликация: (src,stream,seq) уже приходил по другому пути (mesh) */
if (radio_seen_dup(ctx, src, stream, seq, get_time_tb())) {
ctx->c_dup_dropped++;
DEBUG_DEBUG(DEBUG_CATEGORY_RADIO, "%s: dup drop src=%016llx stream=%u seq=%u", RADIO_ID, (unsigned long long)src, stream, seq);
return;
}
/* начало передачи (первый кадр burst): видно и на транзитных узлах (коло), не только у подписчиков */
if (seq == 0) {
DEBUG_DEBUG(DEBUG_CATEGORY_RADIO, "%s: recv burst src=%016llx stream=%u fin=%u len=%d sub=%d",
RADIO_ID, (unsigned long long)src, stream, (flags & RADIO_FLAG_FIN) ? 1 : 0,
opus_len, group->radio_active ? 1 : 0);
}
/* доставка локально, если мы подписаны */
if (group->radio_active && ctx->group) {
struct radio_instance* ri = radio_inst(group->instance);
if (ri && ri->frame_fn) {
ri->frame_fn(group->instance, group_id, src, stream, seq,
(flags & RADIO_FLAG_FIN) ? 1 : 0, opus, opus_len, ri->frame_arg);
ctx->c_rx_frames++;
}
/* индикация говорящих: старт/стоп передачи */
if (flags & RADIO_FLAG_FIN) radio_talker_off(ctx, src);
else radio_talker_on(ctx, src, stream);
}
/* форвардинг: новый hop_list = старый + self */
uint8_t new_count = hop_count + 1;
if (new_count >= RADIO_MAX_HOPS) {
DEBUG_DEBUG(DEBUG_CATEGORY_RADIO, "%s: max hops reached, no fwd src=%016llx", RADIO_ID, (unsigned long long)src);
return;
}
uint64_t new_hops[RADIO_MAX_HOPS];
memcpy(new_hops, hops, (size_t)hop_count * 8);
new_hops[hop_count] = self;
uint8_t out[RADIO_HDR_SIZE + RADIO_MAX_HOPS * 8 + RADIO_MAX_OPUS];
size_t out_len = radio_build(out, group_id, src, stream, seq, flags, new_hops, new_count, opus, opus_len);
radio_flood(group, from_conn, out, out_len, new_hops, new_count);
}
static void radio_recv_dispatcher(struct ETCP_CONN* from_conn, struct ll_entry* entry) {
if (!from_conn || !entry || entry->len < 1 + 8 + RADIO_HDR_SIZE) {
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
return;
}
struct UTUN_INSTANCE* instance = from_conn->instance;
if (!instance) { queue_dgram_free(entry); queue_entry_free(entry); return; }
uint8_t* dgram = entry->dgram;
if (dgram[0] != ETCP_ID_RADIO) { queue_dgram_free(entry); queue_entry_free(entry); return; }
uint64_t group_id;
memcpy(&group_id, dgram + 1, 8);
if (instance->topo_groups) {
struct TOPO_GROUP* group = topo_groups_find(instance->topo_groups, group_id);
if (group) radio_recv(group, from_conn, dgram + 1, entry->len - 1);
}
queue_dgram_free(entry); queue_entry_free(entry);
}
int radio_init_instance(struct UTUN_INSTANCE* inst) {
if (!inst) return -1;
if (inst->radio) return 0;
struct radio_instance* ri = u_calloc(1, sizeof(*ri));
if (!ri) { DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: alloc instance failed", RADIO_ID); return -1; }
inst->radio = ri;
if (etcp_bind(inst, ETCP_ID_RADIO, radio_recv_dispatcher) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: etcp_bind(ETCP_ID_RADIO) failed", RADIO_ID);
u_free(ri); inst->radio = NULL;
return -1;
}
DEBUG_INFO(DEBUG_CATEGORY_RADIO, "%s: instance initialized (bound ETCP_ID_RADIO)", RADIO_ID);
return 0;
}
void radio_destroy_instance(struct UTUN_INSTANCE* inst) {
if (!inst || !inst->radio) return;
etcp_unbind(inst, ETCP_ID_RADIO);
u_free(inst->radio);
inst->radio = NULL;
DEBUG_INFO(DEBUG_CATEGORY_RADIO, "%s: instance destroyed", RADIO_ID);
}
void radio_set_frame_cb(struct UTUN_INSTANCE* inst, radio_frame_fn fn, void* arg) {
struct radio_instance* ri = radio_inst(inst);
if (!ri) return;
ri->frame_fn = fn;
ri->frame_arg = arg;
}
void radio_set_mute_cb(struct UTUN_INSTANCE* inst, radio_mute_fn fn, void* arg) {
struct radio_instance* ri = radio_inst(inst);
if (!ri) { DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "radio: mute callback without instance"); return; }
ri->mute_fn = fn; ri->mute_arg = arg;
}
static int radio_load_muted_members(struct radio_ctx* ctx) {
struct UTUN_INSTANCE* inst = ctx->group->instance;
sqlite3* db = inst->topo_sqlite_db;
char sql[160];
snprintf(sql, sizeof(sql), "SELECT node_id FROM \"peers_%llu\" WHERE radio_muted<>0",
(unsigned long long)ctx->group->group_id);
sqlite3_stmt* st = NULL;
if (!db || sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "radio: load muted members grp=%016llx: %s",
(unsigned long long)ctx->group->group_id, db ? sqlite3_errmsg(db) : "no database"); return -1;
}
uint64_t* members = NULL;
int count = 0, rc;
while ((rc = sqlite3_step(st)) == SQLITE_ROW) {
uint64_t* next = u_realloc(members, (size_t)(count + 1) * sizeof(*members));
if (!next) {
DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "radio: muted member cache allocation failed");
u_free(members); sqlite3_finalize(st); return -1;
}
members = next; members[count++] = (uint64_t)sqlite3_column_int64(st, 0);
}
sqlite3_finalize(st);
if (rc != SQLITE_DONE) {
DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "radio: read muted members: %s", sqlite3_errmsg(db)); u_free(members); return -1;
}
u_free(ctx->muted_members); ctx->muted_members = members; ctx->muted_count = count; ctx->muted_loaded = 1;
DEBUG_DEBUG(DEBUG_CATEGORY_RADIO, "radio: loaded local mutes grp=%016llx count=%d",
(unsigned long long)ctx->group->group_id, count);
return 0;
}
int radio_member_muted(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t node_id) {
if (!inst || !inst->topo_groups) return 0;
struct radio_ctx* ctx = radio_of(topo_groups_find(inst->topo_groups, group_id));
if (!ctx) return 0;
for (int i = 0; i < ctx->muted_count; i++) if (ctx->muted_members[i] == node_id) return 1;
return 0;
}
void radio_forget_member_mute(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t node_id) {
if (!inst || !inst->topo_groups) return;
struct radio_ctx* ctx = radio_of(topo_groups_find(inst->topo_groups, group_id));
if (!ctx) return;
for (int i = 0; i < ctx->muted_count; i++) if (ctx->muted_members[i] == node_id) {
ctx->muted_members[i] = ctx->muted_members[--ctx->muted_count];
DEBUG_DEBUG(DEBUG_CATEGORY_RADIO, "radio: forget removed member mute grp=%016llx node=%016llx",
(unsigned long long)group_id, (unsigned long long)node_id);
return;
}
}
int radio_set_member_muted(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t node_id, int muted) {
struct TOPO_GROUP* group = inst && inst->topo_groups ? topo_groups_find(inst->topo_groups, group_id) : NULL;
struct radio_ctx* ctx = radio_of(group);
if (!ctx || group->group_type != TOPO_GROUP_TYPE_CHAT || !node_id || node_id == inst->node_id || (muted != 0 && muted != 1)) {
DEBUG_WARN(DEBUG_CATEGORY_RADIO, "radio: invalid local mute grp=%016llx node=%016llx muted=%d",
(unsigned long long)group_id, (unsigned long long)node_id, muted); return -1;
}
if (!ctx->muted_loaded && radio_load_muted_members(ctx) < 0) return -1;
int slot = -1;
for (int i = 0; i < ctx->muted_count; i++) if (ctx->muted_members[i] == node_id) { slot = i; break; }
/* Выделить память до записи БД: после commit применение не может отказать. */
if (muted && slot < 0) {
uint64_t* next = u_realloc(ctx->muted_members, (size_t)(ctx->muted_count + 1) * sizeof(*next));
if (!next) { DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "radio: grow muted member cache failed"); return -1; }
ctx->muted_members = next;
}
sqlite3* db = inst->topo_sqlite_db;
char sql[160];
snprintf(sql, sizeof(sql), "UPDATE \"peers_%llu\" SET radio_muted=? WHERE node_id=?", (unsigned long long)group_id);
sqlite3_stmt* st = NULL;
if (!db || sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "radio: prepare local mute: %s", db ? sqlite3_errmsg(db) : "no database"); return -1;
}
sqlite3_bind_int(st, 1, muted); sqlite3_bind_int64(st, 2, (sqlite3_int64)node_id);
int rc = sqlite3_step(st); sqlite3_finalize(st);
if (rc != SQLITE_DONE || sqlite3_changes(db) != 1) {
DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "radio: save local mute grp=%016llx node=%016llx rc=%d changed=%d: %s",
(unsigned long long)group_id, (unsigned long long)node_id, rc, sqlite3_changes(db), sqlite3_errmsg(db)); return -1;
}
if (muted && slot < 0) ctx->muted_members[ctx->muted_count++] = node_id;
else if (!muted && slot >= 0) ctx->muted_members[slot] = ctx->muted_members[--ctx->muted_count];
struct radio_instance* ri = radio_inst(inst);
if (ri && ri->mute_fn) ri->mute_fn(inst, group_id, node_id, muted, ri->mute_arg);
DEBUG_INFO(DEBUG_CATEGORY_RADIO, "radio: local member mute grp=%016llx node=%016llx muted=%d",
(unsigned long long)group_id, (unsigned long long)node_id, muted);
return 0;
}
/* ── подписка (вкл/выкл прослушивание) ── */
void radio_set_active(struct UTUN_INSTANCE* inst, uint64_t group_id, int on) {
if (!inst || !inst->topo_groups) return;
struct TOPO_GROUP* group = topo_groups_find(inst->topo_groups, group_id);
if (!group) { DEBUG_WARN(DEBUG_CATEGORY_RADIO, "%s: set_active group 0x%016llx not found", RADIO_ID, (unsigned long long)group_id); return; }
if (on && group->group_type == TOPO_GROUP_TYPE_CHAT) {
struct radio_ctx* ctx = radio_of(group);
if (!ctx || radio_load_muted_members(ctx) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "radio: subscription rejected because local mutes could not be loaded"); return;
}
}
if (!on) {
/* выключение прослушивания: снять всех активных говорящих — индикация должна погаснуть
* сразу, а не дожидаться FIN/таймаута */
struct radio_ctx* ctx = radio_of(group);
if (ctx) {
for (int i = 0; i < RADIO_MAX_TALKERS; i++)
if (ctx->talkers[i].src) radio_talker_off(ctx, ctx->talkers[i].src);
}
}
topo_group_set_radio(group, on);
}
int radio_is_active(struct UTUN_INSTANCE* inst, uint64_t group_id) {
if (!inst || !inst->topo_groups) return 0;
struct TOPO_GROUP* group = topo_groups_find(inst->topo_groups, group_id);
return group ? group->radio_active : 0;
}
int radio_subscribers(struct UTUN_INSTANCE* inst, uint64_t group_id) {
if (!inst || !inst->topo_groups) return 0;
struct TOPO_GROUP* group = topo_groups_find(inst->topo_groups, group_id);
return group ? topo_group_radio_subscribers(group) : 0;
}
/* Число дропнутых дублей (src,stream,seq) — для диагностики/тестов. */
uint32_t radio_dup_dropped(struct UTUN_INSTANCE* inst, uint64_t group_id) {
if (!inst || !inst->topo_groups) return 0;
struct TOPO_GROUP* group = topo_groups_find(inst->topo_groups, group_id);
struct radio_ctx* ctx = radio_of(group);
return ctx ? ctx->c_dup_dropped : 0;
}
/* ── передача (PTT) ── */
int radio_talk_begin(struct UTUN_INSTANCE* inst, uint64_t group_id) {
if (!inst || !inst->topo_groups) return -1;
struct TOPO_GROUP* group = topo_groups_find(inst->topo_groups, group_id);
if (!group) { DEBUG_WARN(DEBUG_CATEGORY_RADIO, "%s: talk_begin group 0x%016llx not found", RADIO_ID, (unsigned long long)group_id); return -1; }
struct radio_ctx* ctx = radio_of(group);
if (!ctx) return -1;
ctx->stream_id = (uint16_t)(ctx->stream_id + 1);
ctx->seq = 0;
ctx->talking = 1;
DEBUG_DEBUG(DEBUG_CATEGORY_RADIO, "%s: talk_begin grp=%016llx stream=%u", RADIO_ID, (unsigned long long)group_id, ctx->stream_id);
return 0;
}
int radio_talk_send(struct UTUN_INSTANCE* inst, uint64_t group_id, const uint8_t* opus, int len) {
if (!inst || !inst->topo_groups || !opus || len < 0 || len > RADIO_MAX_OPUS) return -1;
struct TOPO_GROUP* group = topo_groups_find(inst->topo_groups, group_id);
if (!group) return -1;
struct radio_ctx* ctx = radio_of(group);
if (!ctx) return -1;
if (!ctx->talking) {
/* burst не открыт: хвостовой кадр, пришедший после radio_talk_end (гонка релиза PTT) —
* не переоткрываем передачу, иначе получится burst без FIN и «говорит» зависнет у пиров. */
DEBUG_DEBUG(DEBUG_CATEGORY_RADIO, "%s: send dropped: no open burst grp=%016llx", RADIO_ID, (unsigned long long)group_id);
return 0;
}
uint64_t self = inst->node_id;
uint64_t hops[1] = { self };
uint8_t out[RADIO_HDR_SIZE + 8 + RADIO_MAX_OPUS];
size_t out_len = radio_build(out, group_id, self, ctx->stream_id, ctx->seq, 0, hops, 1, opus, len);
ctx->seq++;
ctx->c_tx_frames++;
radio_flood(group, NULL, out, out_len, hops, 1);
return 0;
}
int radio_talk_end(struct UTUN_INSTANCE* inst, uint64_t group_id) {
if (!inst || !inst->topo_groups) return -1;
struct TOPO_GROUP* group = topo_groups_find(inst->topo_groups, group_id);
if (!group) return -1;
struct radio_ctx* ctx = radio_of(group);
if (!ctx) return -1;
if (!ctx->talking) return 0;
uint64_t self = inst->node_id;
uint64_t hops[1] = { self };
uint8_t out[RADIO_HDR_SIZE + 8];
size_t out_len = radio_build(out, group_id, self, ctx->stream_id, ctx->seq,
RADIO_FLAG_FIN, hops, 1, NULL, 0);
ctx->seq++;
ctx->c_tx_frames++;
ctx->talking = 0;
DEBUG_DEBUG(DEBUG_CATEGORY_RADIO, "%s: talk_end grp=%016llx stream=%u frames=%u",
RADIO_ID, (unsigned long long)group_id, ctx->stream_id, ctx->c_tx_frames);
radio_flood(group, NULL, out, out_len, hops, 1);
return 0;
}
/* ── trampoline-функции (GUI/аудио-поток → uasync) ── */
void radio_talk_begin_trampoline(void* arg) {
struct radio_talk_arg* a = (struct radio_talk_arg*)arg;
radio_talk_begin(a->inst, a->group_id);
u_free(a);
}
void radio_talk_end_trampoline(void* arg) {
struct radio_talk_arg* a = (struct radio_talk_arg*)arg;
radio_talk_end(a->inst, a->group_id);
u_free(a);
}
void radio_talk_send_trampoline(void* arg) {
struct radio_media_arg* a = (struct radio_media_arg*)arg;
radio_talk_send(a->inst, a->group_id, a->opus, a->len);
u_free(a);
}
void radio_set_active_trampoline(void* arg) {
struct radio_set_active_arg* a = (struct radio_set_active_arg*)arg;
radio_set_active(a->inst, a->group_id, a->on);
u_free(a);
}