evgeny
ce1f23fe46
- stereo capture: Android AudioRecord stereo (CAMCORDER source for stereo, voice sources fall back to CAMCORDER) + desktop separate miniaudio capture/playback devices (no asymmetric duplex) - Opus stereo encode/decode, voice_jitter stereo (per-channel frame_size, slot-based SoundTouch) - radio_audio_start takes explicit capture_channels (stereo->mono fallback) - RADIO_MAX_OPUS 256->512 (128kbps stereo frames up to 320B) - radio_capture_stereo setting + desktop radio settings UI (opus mode/bitrate, capture stereo) - radio_send_to send stats logging (etcp_send ok/fail per second)
567 lines
25 KiB
C
567 lines
25 KiB
C
// 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 <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 "../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;
|
||
};
|
||
|
||
struct radio_instance {
|
||
radio_frame_fn frame_fn;
|
||
void* frame_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);
|
||
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_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) {
|
||
/* выключение прослушивания: снять всех активных говорящих — индикация должна погаснуть
|
||
* сразу, а не дожидаться 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);
|
||
}
|