radio: PTT-радио в каналах (src/radio + chatgui + chatgui-android)
Новый модуль src/radio (radio/radio_audio/radio_headless) — push-to-talk радио поверх topo-групп, доставка аудио в канал, headless-режим и тесты. Интеграция в chatgui (ptt_hotkey_manager, radio_audio_engine, радионастройки, вызовы) и chatgui-android (RadioAudioEngine, RadioOverlayService, UI).
This commit is contained in:
evgeny committed
1 parent
398078a28f
commit
b96ed8bd00
69 files changed
+5372
-51
No files matched your search
@@ -97,6 +97,8 @@ utun_CORE_SOURCES = \
|
||||
dm/dm_mailbox.c \
|
||||
call/call.c \
|
||||
call/call_headless.c \
|
||||
radio/radio.c \
|
||||
radio/radio_headless.c \
|
||||
broadcast.c
|
||||
|
||||
# libutun: all core sources except main()
|
||||
@@ -193,6 +195,8 @@ libutun_a_SOURCES = \
|
||||
dm/dm_mailbox.c \
|
||||
call/call.c \
|
||||
call/call_headless.c \
|
||||
radio/radio.c \
|
||||
radio/radio_headless.c \
|
||||
broadcast.c
|
||||
libutun_a_CFLAGS = $(utun_CFLAGS)
|
||||
|
||||
|
||||
@@ -219,6 +219,52 @@ static void call_event_incoming(struct call_ctx* ctx, uint64_t group_id, uint64_
|
||||
chat_event_post(ctx->inst, CHAT_EVT_CALL_INCOMING, buf, (int)off);
|
||||
}
|
||||
|
||||
/* ─────────────────── BGP-маршрут до пира ─────────────────── */
|
||||
|
||||
/* Формирует строку BGP-пути до peer_node_id и постит CHAT_EVT_CALL_PATH ([call_id:8][text:var]).
|
||||
* hop_list = [peer, m1, ..., mN] (mN — наш прямой сосед), поэтому идём от нас к пиру (в обратном порядке).
|
||||
* Формат hop'а: "[%04X] имя" (короткий адрес + имя, как в chat_status). */
|
||||
static void call_post_bgp_path(struct call_ctx* ctx, struct call_session* s) {
|
||||
if (!ctx || !s || !ctx->inst || !ctx->inst->topo_groups) return;
|
||||
struct TOPO_GROUP* g = topo_groups_find(ctx->inst->topo_groups, s->group_id);
|
||||
if (!g) return;
|
||||
struct TOPO_GROUP_NODE* nq = topo_node_find_by_id(g, s->peer_node_id);
|
||||
if (!nq) return;
|
||||
|
||||
uint8_t hc = 0;
|
||||
uint64_t* hops = topo_node_best_hop_list(nq, &hc, NULL);
|
||||
uint16_t cr = topo_get_chain_rtt(nq);
|
||||
|
||||
char text[512];
|
||||
int off = 0;
|
||||
if (hc <= 1) {
|
||||
off = snprintf(text, sizeof(text), "маршрут: прямой");
|
||||
} else {
|
||||
off = snprintf(text, sizeof(text), "маршрут: через ");
|
||||
for (int i = hc - 1; i >= 1 && off < (int)sizeof(text) - 48; i--) {
|
||||
char hop[48];
|
||||
struct TOPO_NODE* ni = topo_node_registry_find(ctx->inst->topo_groups, hops[i]);
|
||||
if (ni && ni->node_name && ni->node_name[0])
|
||||
snprintf(hop, sizeof(hop), "[%04X] %s", (unsigned)(hops[i] & 0xFFFF), ni->node_name);
|
||||
else
|
||||
snprintf(hop, sizeof(hop), "[%04X]", (unsigned)(hops[i] & 0xFFFF));
|
||||
if (i != hc - 1) off += snprintf(text + off, sizeof(text) - (size_t)off, " → ");
|
||||
off += snprintf(text + off, sizeof(text) - (size_t)off, "%s", hop);
|
||||
}
|
||||
}
|
||||
if (cr != 0xFFFF)
|
||||
off += snprintf(text + off, sizeof(text) - (size_t)off, " (rtt %ums)", cr / 10);
|
||||
|
||||
uint8_t buf[8 + 512];
|
||||
size_t boff = 0;
|
||||
memcpy(buf + boff, &s->call_id, 8); boff += 8;
|
||||
memcpy(buf + boff, text, (size_t)off); boff += (size_t)off;
|
||||
chat_event_post(ctx->inst, CHAT_EVT_CALL_PATH, buf, (int)boff);
|
||||
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: path id=%016llx hc=%u rtt=%u: %s",
|
||||
CALL_ID, (unsigned long long)s->call_id, hc, cr, text);
|
||||
}
|
||||
|
||||
/* ─────────────────── завершение ─────────────────── */
|
||||
|
||||
/* Отложенный разрыв: единственная точка, где закрывается conn_mgr (после того,
|
||||
@@ -279,6 +325,7 @@ static void call_conn_cb(struct CONN_MGR_HANDLE* h, uint64_t node_id, uint64_t g
|
||||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: conn UP id=%016llx peer=0x%016llx -> ACTIVE",
|
||||
CALL_ID, (unsigned long long)s->call_id, (unsigned long long)node_id);
|
||||
call_event_id(ctx, CHAT_EVT_CALL_ACCEPTED, s->call_id, 0);
|
||||
call_post_bgp_path(ctx, s);
|
||||
if (!s->stats_timer)
|
||||
s->stats_timer = uasync_set_timeout(ctx->inst->ua, 10000, s, call_stats_timer_cb, "call_stats");
|
||||
} else if (event == CONN_EVENT_DOWN) {
|
||||
@@ -447,6 +494,7 @@ static void call_handle_invite(struct call_ctx* ctx, uint64_t from, uint64_t gro
|
||||
(unsigned long long)group, (unsigned long long)inv_group);
|
||||
|
||||
call_event_incoming(ctx, group, from, call_id);
|
||||
call_post_bgp_path(ctx, s);
|
||||
|
||||
uint8_t pkt[sizeof(struct call_pkt_ringing)];
|
||||
struct call_pkt_ringing* r = (struct call_pkt_ringing*)pkt;
|
||||
@@ -789,6 +837,7 @@ void call_start(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t peer_nod
|
||||
call_end(ctx, s, CALL_REASON_CONNECT_FAIL);
|
||||
return;
|
||||
}
|
||||
call_post_bgp_path(ctx, s);
|
||||
s->ring_timer = uasync_set_timeout(inst->ua, CALL_RING_TIMEOUT_TB, s, call_ring_timeout_cb, "call_ring");
|
||||
}
|
||||
|
||||
|
||||
@@ -20,6 +20,7 @@ function(utun_setup_voice LIB_DIR CALL_DIR)
|
||||
${CALL_DIR}/call_audio.c
|
||||
${CALL_DIR}/call_tones.c
|
||||
${CALL_DIR}/voice_jitter.cpp
|
||||
${_src_dir}/radio/radio_audio.c
|
||||
${_st_dir}/source/SoundTouch/SoundTouch.cpp
|
||||
${_st_dir}/source/SoundTouch/TDStretch.cpp
|
||||
${_st_dir}/source/SoundTouch/RateTransposer.cpp
|
||||
|
||||
@@ -39,8 +39,15 @@ void chat_event_post(struct UTUN_INSTANCE* inst, int type, const uint8_t* data,
|
||||
[22]="MEMBER_DETAIL", [23]="MEMBER_LIST",
|
||||
[24]="MEMBER_UPDATED", [25]="MEMBER_REMOVED",
|
||||
[26]="BGP_INFO", [27]="CONNECTING_NODES",
|
||||
[28]="ADMIN_KEY_RECEIVED", [29]="INVITE_LINK_READY",
|
||||
[30]="INVITE_CANDIDATES", [31]="CHANNEL_DELETED",
|
||||
[32]="DM_MSG_RECEIVED", [33]="DM_CONV_UPDATED",
|
||||
[34]="CALL_INCOMING", [35]="CALL_RINGING", [36]="CALL_ACCEPTED",
|
||||
[37]="CALL_ENDED", [38]="CALL_DECLINED", [39]="CALL_STATS",
|
||||
[40]="CALL_ERROR", [41]="DM_MESSAGES", [42]="CALL_PATH",
|
||||
[43]="RADIO_TALK",
|
||||
};
|
||||
const char* n = (type >= 1 && type <= 27) ? names[type] : "?";
|
||||
const char* n = (type >= 1 && type <= 43) ? names[type] : "?";
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_CHAT, "chat_event: %s(%d) data=%d bytes", n, type, len);
|
||||
if (data && len > 0) log_dump(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_CHAT, " ", data, len);
|
||||
}
|
||||
@@ -62,6 +62,8 @@ extern "C" {
|
||||
#define CHAT_EVT_CALL_STATS 39 /* [call_id:8][rtt_ms:2][buffer_ms:2][tempo_x100:2] */
|
||||
#define CHAT_EVT_CALL_ERROR 40 /* [call_id:8][err:1][text:var] */
|
||||
#define CHAT_EVT_DM_MESSAGES 41 /* [conv_id_len:1][conv_id:var][count:2][msg:var]* — расшифрованные сообщения беседы */
|
||||
#define CHAT_EVT_CALL_PATH 42 /* [call_id:8][text:var] — BGP-маршрут до пира звонка */
|
||||
#define CHAT_EVT_RADIO_TALK 43 /* [group_id:8][src_node_id:8][stream_id:2][on:1] — кто-то начал/закончил говорить */
|
||||
|
||||
typedef void (*chat_event_handler_fn)(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int len);
|
||||
|
||||
|
||||
@@ -21,6 +21,7 @@
|
||||
#include "../dm/dm_crypto.h"
|
||||
#include "../call/call.h"
|
||||
#include "../call/call_proto.h"
|
||||
#include "../radio/radio.h"
|
||||
#include "../transport_layer/secure_channel.h"
|
||||
#include "../transport_layer/etcp.h"
|
||||
#include "../transport_layer/etcp_connections.h"
|
||||
@@ -728,12 +729,43 @@ static void hc_handle_call_hangup(struct headless_client* cli, int id, const cha
|
||||
send_response(cli, id, "{\"hungup\":true}", NULL);
|
||||
}
|
||||
|
||||
/* ── Рация (PTT walkie-talkie) ── */
|
||||
|
||||
static void hc_handle_radio_on(struct headless_client* cli, int id, const char* json) {
|
||||
char ch[64];
|
||||
if (json_get_str(json, "ch", ch, sizeof(ch)) < 0) { send_response(cli, id, NULL, "missing 'ch' param"); return; }
|
||||
uint64_t group_id = strtoull(ch, NULL, 10);
|
||||
if (!group_id || !cli->hc->inst) { send_response(cli, id, NULL, "invalid ch"); return; }
|
||||
radio_set_active(cli->hc->inst, group_id, 1);
|
||||
char resp[128]; snprintf(resp, sizeof(resp), "{\"radio\":true,\"ch\":\"%llu\"}", (unsigned long long)group_id);
|
||||
send_response(cli, id, resp, NULL);
|
||||
}
|
||||
|
||||
static void hc_handle_radio_off(struct headless_client* cli, int id, const char* json) {
|
||||
char ch[64];
|
||||
if (json_get_str(json, "ch", ch, sizeof(ch)) < 0) { send_response(cli, id, NULL, "missing 'ch' param"); return; }
|
||||
uint64_t group_id = strtoull(ch, NULL, 10);
|
||||
if (!group_id || !cli->hc->inst) { send_response(cli, id, NULL, "invalid ch"); return; }
|
||||
radio_set_active(cli->hc->inst, group_id, 0);
|
||||
char resp[128]; snprintf(resp, sizeof(resp), "{\"radio\":false,\"ch\":\"%llu\"}", (unsigned long long)group_id);
|
||||
send_response(cli, id, resp, NULL);
|
||||
}
|
||||
|
||||
static void hc_handle_radio_subscribers(struct headless_client* cli, int id, const char* json) {
|
||||
char ch[64];
|
||||
if (json_get_str(json, "ch", ch, sizeof(ch)) < 0) { send_response(cli, id, NULL, "missing 'ch' param"); return; }
|
||||
uint64_t group_id = strtoull(ch, NULL, 10);
|
||||
int n = radio_subscribers(cli->hc->inst, group_id);
|
||||
char resp[128]; snprintf(resp, sizeof(resp), "{\"subscribers\":%d,\"ch\":\"%llu\"}", n, (unsigned long long)group_id);
|
||||
send_response(cli, id, resp, NULL);
|
||||
}
|
||||
|
||||
/* ── Command dispatch ── */
|
||||
|
||||
typedef void (*hc_cmd_fn)(struct headless_client* cli, int id, const char* json);
|
||||
|
||||
static const char* cmd_names[] = { "ping", "status", "channels", "members", "messages", "send", "attach", "invite", "invite_nodes", "connect", "create_channel", "invite_to", "subscribe", "dm_start", "dm_list", "dm_messages", "dm_send", "call_start", "call_accept", "call_decline", "call_hangup", "quit", NULL };
|
||||
static hc_cmd_fn cmd_handlers[] = { hc_handle_ping, hc_handle_status, hc_handle_channels, hc_handle_members, hc_handle_messages, hc_handle_send, hc_handle_attach, hc_handle_invite, hc_handle_invite_nodes, hc_handle_connect, hc_handle_create_channel, hc_handle_invite_to, hc_handle_subscribe, hc_handle_dm_start, hc_handle_dm_list, hc_handle_dm_messages, hc_handle_dm_send, hc_handle_call_start, hc_handle_call_accept, hc_handle_call_decline, hc_handle_call_hangup, hc_handle_quit };
|
||||
static const char* cmd_names[] = { "ping", "status", "channels", "members", "messages", "send", "attach", "invite", "invite_nodes", "connect", "create_channel", "invite_to", "subscribe", "dm_start", "dm_list", "dm_messages", "dm_send", "call_start", "call_accept", "call_decline", "call_hangup", "radio_on", "radio_off", "radio_subscribers", "quit", NULL };
|
||||
static hc_cmd_fn cmd_handlers[] = { hc_handle_ping, hc_handle_status, hc_handle_channels, hc_handle_members, hc_handle_messages, hc_handle_send, hc_handle_attach, hc_handle_invite, hc_handle_invite_nodes, hc_handle_connect, hc_handle_create_channel, hc_handle_invite_to, hc_handle_subscribe, hc_handle_dm_start, hc_handle_dm_list, hc_handle_dm_messages, hc_handle_dm_send, hc_handle_call_start, hc_handle_call_accept, hc_handle_call_decline, hc_handle_call_hangup, hc_handle_radio_on, hc_handle_radio_off, hc_handle_radio_subscribers, hc_handle_quit };
|
||||
|
||||
static void hc_handle_command(struct headless_client* cli, const char* json) {
|
||||
if (!cli || !json) return;
|
||||
|
||||
@@ -838,6 +838,11 @@ static int parse_chatserver(const char *key, const char *value, struct global_co
|
||||
global->call_audio_bind[sizeof(global->call_audio_bind) - 1] = '\0';
|
||||
return 0;
|
||||
}
|
||||
if (strcmp(key, "radio_audio_bind") == 0) {
|
||||
strncpy(global->radio_audio_bind, value, sizeof(global->radio_audio_bind) - 1);
|
||||
global->radio_audio_bind[sizeof(global->radio_audio_bind) - 1] = '\0';
|
||||
return 0;
|
||||
}
|
||||
return -1;
|
||||
}
|
||||
|
||||
|
||||
@@ -223,6 +223,7 @@ struct global_config {
|
||||
uint64_t chatserver_storage_unit_size; // bytes, лимит одного файла (0 = отключено)
|
||||
char headless_control_bind[64]; // ip:port для headless chat CLI (пусто = отключено)
|
||||
char call_audio_bind[64]; // ip:port для headless аудио-сокета звонков (пусто = отключено)
|
||||
char radio_audio_bind[64]; // ip:port для headless аудио-сокета рации (пусто = отключено)
|
||||
struct chat_setting_state chat_settings; // chat-настройки из [chatserver] (per-instance)
|
||||
|
||||
// Reality-камуфляж ([reality] section)
|
||||
|
||||
@@ -0,0 +1,432 @@
|
||||
// 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 /* максимум одновременно говорящих источников */
|
||||
|
||||
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; /* дропнуто некорректных пакетов */
|
||||
struct radio_talker { uint64_t src; uint16_t stream; } talkers[RADIO_MAX_TALKERS]; /* активные говорящие */
|
||||
};
|
||||
|
||||
struct radio_instance {
|
||||
radio_frame_fn frame_fn;
|
||||
void* frame_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;
|
||||
|
||||
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;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_RADIO, "%s: destroy grp=%016llx (tx=%u rx=%u fwd=%u loop=%u nosub=%u bad=%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);
|
||||
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;
|
||||
}
|
||||
|
||||
/* ── отправка ── */
|
||||
|
||||
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) return;
|
||||
uint8_t* pkt = u_malloc(1 + payload_len);
|
||||
if (!pkt) { queue_entry_free(ee); 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); }
|
||||
}
|
||||
|
||||
/* Есть ли за 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) {
|
||||
for (int i = 0; i < RADIO_MAX_TALKERS; i++) {
|
||||
if (ctx->talkers[i].src == src) {
|
||||
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;
|
||||
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;
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/* ── приём ── */
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
/* доставка локально, если мы подписаны */
|
||||
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 */
|
||||
uint64_t new_hops[RADIO_MAX_HOPS];
|
||||
memcpy(new_hops, hops, (size_t)hop_count * 8);
|
||||
new_hops[hop_count] = 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;
|
||||
}
|
||||
|
||||
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; }
|
||||
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;
|
||||
}
|
||||
|
||||
/* ── передача (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) radio_talk_begin(inst, group_id);
|
||||
|
||||
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);
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
// radio.h — рация (PTT walkie-talkie): мультикаст голоса по CHAT-группе.
|
||||
//
|
||||
// Флудит Opus-кадры всем подписчикам канала (узлы с TOPO_FLAG_RADIO в NODEINFO).
|
||||
// Передача полудуплексная: радио включается на прослушивание, PTT — на передачу.
|
||||
// Все функции — в uasync-потоке (кроме radio_set_frame_cb, который ставит
|
||||
// обработчик приёма: его вызывает radio_audio в момент init).
|
||||
#ifndef RADIO_H
|
||||
#define RADIO_H
|
||||
|
||||
#include <stdint.h>
|
||||
|
||||
struct TOPO_GROUP;
|
||||
struct UTUN_INSTANCE;
|
||||
|
||||
#ifdef __cplusplus
|
||||
extern "C" {
|
||||
#endif
|
||||
|
||||
/* ── жизненный цикл ── */
|
||||
|
||||
int radio_init(struct TOPO_GROUP* group); /* из topo_group_create */
|
||||
void radio_destroy(struct TOPO_GROUP* group); /* из topo_group_destroy */
|
||||
|
||||
int radio_init_instance(struct UTUN_INSTANCE* inst); /* etcp_bind(ETCP_ID_RADIO) */
|
||||
void radio_destroy_instance(struct UTUN_INSTANCE* inst); /* etcp_unbind */
|
||||
|
||||
/* ── приём Opus-кадра пира (для аудио-движка radio_audio) ── */
|
||||
|
||||
typedef void (*radio_frame_fn)(struct UTUN_INSTANCE* inst, uint64_t group_id,
|
||||
uint64_t src_node_id, uint16_t stream_id, uint16_t seq,
|
||||
uint8_t fin, const uint8_t* opus, int len, void* arg);
|
||||
void radio_set_frame_cb(struct UTUN_INSTANCE* inst, radio_frame_fn fn, void* arg);
|
||||
|
||||
/* ── передача (PTT), uasync-поток ── */
|
||||
|
||||
int radio_talk_begin(struct UTUN_INSTANCE* inst, uint64_t group_id); /* новая передача, seq=0 */
|
||||
int radio_talk_send(struct UTUN_INSTANCE* inst, uint64_t group_id, const uint8_t* opus, int len);
|
||||
int radio_talk_end(struct UTUN_INSTANCE* inst, uint64_t group_id); /* FIN-кадр */
|
||||
|
||||
/* Включить/выключить прослушивание рации канала (флаг + BGP-рассылка). */
|
||||
void radio_set_active(struct UTUN_INSTANCE* inst, uint64_t group_id, int on);
|
||||
int radio_is_active(struct UTUN_INSTANCE* inst, uint64_t group_id);
|
||||
|
||||
/* Число подписчиков рации канала (без себя) — для индикатора в GUI. */
|
||||
int radio_subscribers(struct UTUN_INSTANCE* inst, uint64_t group_id);
|
||||
|
||||
/* ── trampoline-структуры (GUI/аудио-поток → uasync) ── */
|
||||
|
||||
struct radio_talk_arg {
|
||||
struct UTUN_INSTANCE* inst;
|
||||
uint64_t group_id;
|
||||
};
|
||||
|
||||
struct radio_media_arg {
|
||||
struct UTUN_INSTANCE* inst;
|
||||
uint64_t group_id;
|
||||
uint16_t len;
|
||||
uint8_t opus[256];
|
||||
};
|
||||
|
||||
struct radio_set_active_arg {
|
||||
struct UTUN_INSTANCE* inst;
|
||||
uint64_t group_id;
|
||||
int on;
|
||||
};
|
||||
|
||||
void radio_talk_begin_trampoline(void* arg);
|
||||
void radio_talk_end_trampoline(void* arg);
|
||||
void radio_talk_send_trampoline(void* arg);
|
||||
void radio_set_active_trampoline(void* arg);
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
#endif
|
||||
#endif /* RADIO_H */
|
||||
@@ -0,0 +1,299 @@
|
||||
// radio_audio.c — сервисный аудио-движок рации (Opus TX + per-source jitter + микшер).
|
||||
//
|
||||
// Таблица говорящих: per-source инстанс (Opus-декодер + voice_jitter с time-stretch),
|
||||
// ключ = src_node_id. Один пир — один поток: новый stream_id сбрасывает инстанс
|
||||
// (не создаём дубль). Инстанс удаляется по FIN (после дренажа буфера) или по таймауту
|
||||
// (нет кадров 2с — потерян FIN / пир отвалился).
|
||||
//
|
||||
// RX: radio.c → radio_set_frame_cb → radio_audio_on_frame (uasync) → vj_push (кодированный кадр);
|
||||
// аудио-поток radio_audio_pull_pcm → vj_pull (декод + SoundTouch stretch) → микшер.
|
||||
// TX: аудио-поток radio_audio_feed_pcm → encode → uasync_post(radio_talk_send_trampoline).
|
||||
// Микшер избыточных источников — насыщающее суммирование (soft-clip). Полудуплекс.
|
||||
// Синглтон на один активный group_id (GUI слушает один канал), защищён мьютексом.
|
||||
|
||||
#include "radio_audio.h"
|
||||
#include "radio.h"
|
||||
#include "../call/voice_jitter.h"
|
||||
#include "../utun_instance.h"
|
||||
#include "../../lib/opus_codec.h"
|
||||
#include "../../lib/debug_config.h"
|
||||
#include "../../lib/mem.h"
|
||||
#include "../../lib/u_async.h"
|
||||
#include "../../lib/platform_compat.h"
|
||||
|
||||
#include <pthread.h>
|
||||
#include <string.h>
|
||||
|
||||
#define RADIO_AUDIO_ID "radio_audio"
|
||||
#define RADIO_AUDIO_BITRATE 32000
|
||||
#define RADIO_AUDIO_MAX_SOURCES 16
|
||||
#define RADIO_AUDIO_MAX_FRAMES 50 /* ~1000мс @20мс */
|
||||
#define RADIO_AUDIO_MIX_SAMPLES 960 /* максимум сэмплов за pull */
|
||||
#define RADIO_AUDIO_SOURCE_TIMEOUT_TB 20000 /* 2с без кадров → источник завершён */
|
||||
|
||||
struct radio_source {
|
||||
uint64_t src_node_id;
|
||||
uint16_t stream_id;
|
||||
uint8_t used;
|
||||
uint8_t ending; /* получен FIN — дренируем и удаляем */
|
||||
uint64_t last_frame_tb;
|
||||
opus_codec_decoder_t* dec; /* decode-аргумент для vj (stateless per call) */
|
||||
struct vj* vj; /* jitter-буфер с time-stretch */
|
||||
};
|
||||
|
||||
static pthread_mutex_t g_mtx = PTHREAD_MUTEX_INITIALIZER;
|
||||
static struct UTUN_INSTANCE* g_inst = NULL;
|
||||
static int g_active = 0;
|
||||
static uint64_t g_group_id = 0;
|
||||
static opus_codec_encoder_t* g_encoder = NULL;
|
||||
static struct radio_source g_sources[RADIO_AUDIO_MAX_SOURCES];
|
||||
static int32_t g_mix[RADIO_AUDIO_MIX_SAMPLES];
|
||||
|
||||
/* vj decode-коллбэк: дёргается только из аудио-потока (vj_pull). */
|
||||
static int radio_audio_decode_cb(void* arg, const uint8_t* enc, int len, int16_t* pcm, int max_samples) {
|
||||
opus_codec_decoder_t* dec = (opus_codec_decoder_t*)arg;
|
||||
return opus_codec_decode(dec, enc, len, pcm, max_samples);
|
||||
}
|
||||
|
||||
/* ── таблица говорящих ── */
|
||||
|
||||
static struct radio_source* radio_src_find(uint64_t src) {
|
||||
for (int i = 0; i < RADIO_AUDIO_MAX_SOURCES; i++)
|
||||
if (g_sources[i].used && g_sources[i].src_node_id == src) return &g_sources[i];
|
||||
return NULL;
|
||||
}
|
||||
|
||||
static void radio_src_free(struct radio_source* s) {
|
||||
if (s->vj) { vj_destroy(s->vj); s->vj = NULL; }
|
||||
if (s->dec) { opus_codec_decoder_destroy(s->dec); s->dec = NULL; }
|
||||
s->used = 0; s->ending = 0; s->last_frame_tb = 0;
|
||||
}
|
||||
|
||||
/* Получить/создать инстанс источника. Один пир — один инстанс: если stream_id
|
||||
* сменился (новая передача PTT того же пира) — сбрасываем jitter, не дублируем. */
|
||||
static struct radio_source* radio_src_acquire(uint64_t src, uint16_t stream) {
|
||||
struct radio_source* s = radio_src_find(src);
|
||||
if (s) {
|
||||
if (s->stream_id != stream) {
|
||||
s->stream_id = stream;
|
||||
s->ending = 0;
|
||||
vj_destroy(s->vj);
|
||||
s->vj = vj_create(RADIO_AUDIO_SAMPLE_RATE, 1, RADIO_AUDIO_FRAME_SAMPLES, RADIO_AUDIO_FRAME_MS,
|
||||
RADIO_AUDIO_MAX_FRAMES, radio_audio_decode_cb, s->dec);
|
||||
if (!s->vj) { radio_src_free(s); return NULL; }
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_RADIO, "%s: source reset src=%016llx stream=%u", RADIO_AUDIO_ID, (unsigned long long)src, stream);
|
||||
}
|
||||
return s;
|
||||
}
|
||||
for (int i = 0; i < RADIO_AUDIO_MAX_SOURCES; i++) {
|
||||
if (g_sources[i].used) continue;
|
||||
s = &g_sources[i];
|
||||
s->dec = opus_codec_decoder_create(RADIO_AUDIO_SAMPLE_RATE, 1);
|
||||
if (!s->dec) { DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: decoder create failed", RADIO_AUDIO_ID); return NULL; }
|
||||
s->vj = vj_create(RADIO_AUDIO_SAMPLE_RATE, 1, RADIO_AUDIO_FRAME_SAMPLES, RADIO_AUDIO_FRAME_MS,
|
||||
RADIO_AUDIO_MAX_FRAMES, radio_audio_decode_cb, s->dec);
|
||||
if (!s->vj) { opus_codec_decoder_destroy(s->dec); s->dec = NULL; return NULL; }
|
||||
s->src_node_id = src;
|
||||
s->stream_id = stream;
|
||||
s->used = 1; s->ending = 0; s->last_frame_tb = 0;
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_RADIO, "%s: new source src=%016llx stream=%u", RADIO_AUDIO_ID, (unsigned long long)src, stream);
|
||||
return s;
|
||||
}
|
||||
DEBUG_WARN(DEBUG_CATEGORY_RADIO, "%s: too many sources (>=%d), drop src=%016llx",
|
||||
RADIO_AUDIO_ID, RADIO_AUDIO_MAX_SOURCES, (unsigned long long)src);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
/* ── lifecycle ── */
|
||||
|
||||
int radio_audio_init(struct UTUN_INSTANCE* inst) {
|
||||
if (!inst) return -1;
|
||||
if (g_inst == inst) return 0;
|
||||
g_inst = inst;
|
||||
radio_set_frame_cb(inst, radio_audio_on_frame, NULL);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_RADIO, "%s: initialized", RADIO_AUDIO_ID);
|
||||
return 0;
|
||||
}
|
||||
|
||||
void radio_audio_destroy(struct UTUN_INSTANCE* inst) {
|
||||
if (!inst || g_inst != inst) return;
|
||||
radio_set_frame_cb(inst, NULL, NULL);
|
||||
g_inst = NULL;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_RADIO, "%s: destroyed", RADIO_AUDIO_ID);
|
||||
}
|
||||
|
||||
int radio_audio_start(uint64_t group_id) {
|
||||
if (group_id == 0) { DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: start bad group_id=0", RADIO_AUDIO_ID); return -1; }
|
||||
|
||||
pthread_mutex_lock(&g_mtx);
|
||||
if (g_active && g_group_id == group_id) { pthread_mutex_unlock(&g_mtx); return 0; }
|
||||
if (g_active) { pthread_mutex_unlock(&g_mtx); radio_audio_stop(); pthread_mutex_lock(&g_mtx); }
|
||||
|
||||
opus_codec_encoder_t* enc = opus_codec_encoder_create(RADIO_AUDIO_SAMPLE_RATE, 1);
|
||||
if (!enc) { pthread_mutex_unlock(&g_mtx); DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: encoder create failed", RADIO_AUDIO_ID); return -1; }
|
||||
opus_codec_encoder_bitrate_set(enc, RADIO_AUDIO_BITRATE);
|
||||
|
||||
g_encoder = enc;
|
||||
g_group_id = group_id;
|
||||
g_active = 1;
|
||||
memset(g_sources, 0, sizeof(g_sources));
|
||||
pthread_mutex_unlock(&g_mtx);
|
||||
|
||||
DEBUG_INFO(DEBUG_CATEGORY_RADIO, "%s: started grp=%016llx", RADIO_AUDIO_ID, (unsigned long long)group_id);
|
||||
return 0;
|
||||
}
|
||||
|
||||
void radio_audio_stop(void) {
|
||||
opus_codec_encoder_t* enc = NULL;
|
||||
int was_active = 0;
|
||||
|
||||
pthread_mutex_lock(&g_mtx);
|
||||
if (g_active) {
|
||||
was_active = 1;
|
||||
g_active = 0;
|
||||
g_group_id = 0;
|
||||
enc = g_encoder; g_encoder = NULL;
|
||||
for (int i = 0; i < RADIO_AUDIO_MAX_SOURCES; i++)
|
||||
if (g_sources[i].used) radio_src_free(&g_sources[i]);
|
||||
}
|
||||
pthread_mutex_unlock(&g_mtx);
|
||||
|
||||
if (was_active) {
|
||||
if (enc) opus_codec_encoder_destroy(enc);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_RADIO, "%s: stopped", RADIO_AUDIO_ID);
|
||||
}
|
||||
}
|
||||
|
||||
int radio_audio_active(void) {
|
||||
int a = 0;
|
||||
pthread_mutex_lock(&g_mtx);
|
||||
a = g_active;
|
||||
pthread_mutex_unlock(&g_mtx);
|
||||
return a;
|
||||
}
|
||||
|
||||
/* ── PTT (GUI → uasync) ── */
|
||||
|
||||
static void radio_audio_post_talk(struct UASYNC* ua, uint64_t group_id, void (*fn)(void*)) {
|
||||
struct radio_talk_arg* a = (struct radio_talk_arg*)u_malloc(sizeof(*a));
|
||||
if (!a) return;
|
||||
a->inst = g_inst;
|
||||
a->group_id = group_id;
|
||||
uasync_post(ua, fn, a);
|
||||
}
|
||||
|
||||
void radio_audio_talk_begin(uint64_t group_id) {
|
||||
if (!g_inst || !g_inst->ua) return;
|
||||
radio_audio_post_talk(g_inst->ua, group_id, radio_talk_begin_trampoline);
|
||||
}
|
||||
|
||||
void radio_audio_talk_end(uint64_t group_id) {
|
||||
if (!g_inst || !g_inst->ua) return;
|
||||
radio_audio_post_talk(g_inst->ua, group_id, radio_talk_end_trampoline);
|
||||
}
|
||||
|
||||
/* ── TX (аудио-поток) ── */
|
||||
|
||||
int radio_audio_feed_pcm(uint64_t group_id, const int16_t* pcm, int count) {
|
||||
if (!pcm || count != RADIO_AUDIO_FRAME_SAMPLES) return -1;
|
||||
|
||||
uint8_t opus[256];
|
||||
struct UASYNC* ua = NULL;
|
||||
struct UTUN_INSTANCE* inst = NULL;
|
||||
int len = -1;
|
||||
|
||||
pthread_mutex_lock(&g_mtx);
|
||||
if (!g_active || group_id != g_group_id || !g_encoder) {
|
||||
pthread_mutex_unlock(&g_mtx);
|
||||
return -1;
|
||||
}
|
||||
len = opus_codec_encode(g_encoder, pcm, count, opus, (int)sizeof(opus));
|
||||
ua = g_inst ? g_inst->ua : NULL;
|
||||
inst = g_inst;
|
||||
pthread_mutex_unlock(&g_mtx);
|
||||
|
||||
if (len <= 0) return -1;
|
||||
if (!ua || !inst) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: feed_pcm: instance not set", RADIO_AUDIO_ID);
|
||||
return -1;
|
||||
}
|
||||
|
||||
struct radio_media_arg* a = (struct radio_media_arg*)u_malloc(sizeof(*a));
|
||||
if (!a) return -1;
|
||||
a->inst = inst;
|
||||
a->group_id = group_id;
|
||||
a->len = (uint16_t)len;
|
||||
memcpy(a->opus, opus, (size_t)len);
|
||||
uasync_post(ua, radio_talk_send_trampoline, a);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* ── RX (uasync-поток): кодированный кадр → vj_push ── */
|
||||
|
||||
void radio_audio_on_frame(struct UTUN_INSTANCE* inst, uint64_t group_id,
|
||||
uint64_t src_node_id, uint16_t stream_id, uint16_t seq,
|
||||
uint8_t fin, const uint8_t* opus, int len, void* arg) {
|
||||
(void)inst; (void)seq; (void)arg;
|
||||
if (!opus || len <= 0) return;
|
||||
|
||||
pthread_mutex_lock(&g_mtx);
|
||||
if (g_active && group_id == g_group_id) {
|
||||
struct radio_source* s = radio_src_acquire(src_node_id, stream_id);
|
||||
if (s) {
|
||||
vj_push(s->vj, opus, len);
|
||||
s->last_frame_tb = get_time_tb();
|
||||
if (fin) s->ending = 1;
|
||||
}
|
||||
}
|
||||
pthread_mutex_unlock(&g_mtx);
|
||||
}
|
||||
|
||||
/* ── RX микшер (аудио-поток) ── */
|
||||
|
||||
int radio_audio_pull_pcm(uint64_t group_id, int16_t* out, int max_samples) {
|
||||
if (!out || max_samples <= 0) return 0;
|
||||
if (max_samples > RADIO_AUDIO_MIX_SAMPLES) max_samples = RADIO_AUDIO_MIX_SAMPLES;
|
||||
|
||||
int active = 0;
|
||||
pthread_mutex_lock(&g_mtx);
|
||||
if (g_active && group_id == g_group_id) active = 1;
|
||||
pthread_mutex_unlock(&g_mtx);
|
||||
if (!active) { memset(out, 0, (size_t)max_samples * sizeof(int16_t)); return max_samples; }
|
||||
|
||||
memset(g_mix, 0, (size_t)max_samples * sizeof(int32_t));
|
||||
|
||||
uint64_t now = get_time_tb();
|
||||
|
||||
pthread_mutex_lock(&g_mtx);
|
||||
for (int i = 0; i < RADIO_AUDIO_MAX_SOURCES; i++) {
|
||||
struct radio_source* s = &g_sources[i];
|
||||
if (!s->used) continue;
|
||||
/* таймаут источника (нет кадров 2с — потерян FIN/обрыв передающего).
|
||||
* только для НЕ-ending: ending-источник дренируется до пустого буфера. */
|
||||
if (!s->ending && s->last_frame_tb && (now - s->last_frame_tb > RADIO_AUDIO_SOURCE_TIMEOUT_TB)) {
|
||||
DEBUG_WARN(DEBUG_CATEGORY_RADIO, "%s: source timeout src=%016llx", RADIO_AUDIO_ID, (unsigned long long)s->src_node_id);
|
||||
radio_src_free(s);
|
||||
continue;
|
||||
}
|
||||
int16_t tmp[RADIO_AUDIO_MIX_SAMPLES];
|
||||
int n = vj_pull(s->vj, tmp, max_samples);
|
||||
for (int k = 0; k < n; k++) g_mix[k] += tmp[k];
|
||||
if (s->ending && n == 0) {
|
||||
uint32_t dropped = 0, underruns = 0;
|
||||
vj_get_stats(s->vj, NULL, NULL, &dropped, &underruns);
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_RADIO, "%s: source ended src=%016llx dropped=%u underruns=%u",
|
||||
RADIO_AUDIO_ID, (unsigned long long)s->src_node_id, dropped, underruns);
|
||||
radio_src_free(s);
|
||||
}
|
||||
}
|
||||
pthread_mutex_unlock(&g_mtx);
|
||||
|
||||
/* насыщающее суммирование (soft-clip) в int16 */
|
||||
for (int k = 0; k < max_samples; k++) {
|
||||
int32_t v = g_mix[k];
|
||||
if (v > 32767) v = 32767;
|
||||
else if (v < -32768) v = -32768;
|
||||
out[k] = (int16_t)v;
|
||||
}
|
||||
return max_samples;
|
||||
}
|
||||
@@ -0,0 +1,63 @@
|
||||
// radio_audio.h — сервисный аудио-движок рации (единый для desktop/Android/headless).
|
||||
//
|
||||
// Владеет Opus encoder (TX), per-source декодерами и микшером (RX). Приём:
|
||||
// radio.c → radio_set_frame_cb → radio_audio_on_frame (uasync) декодирует Opus-кадр
|
||||
// и кладёт PCM в ring-буфер источника; аудио-поток radio_audio_pull_pcm микширует
|
||||
// PCM всех источников. Передача: аудио-поток radio_audio_feed_pcm кодирует PCM и
|
||||
// постит radio_talk_send в uasync. Микшер избыточных источников — насыщающее
|
||||
// суммирование (soft-clip). Полудуплекс: радио включено — слушаем, PTT — говорим.
|
||||
//
|
||||
// Потоки (как call_audio):
|
||||
// - radio_audio_start / talk_begin / talk_end / stop — GUI-поток;
|
||||
// - radio_audio_feed_pcm / pull_pcm — аудио-поток;
|
||||
// - radio_audio_on_frame — uasync-поток (из radio.c).
|
||||
// Общее состояние — под мьютексом. Синглтон на один активный group_id.
|
||||
|
||||
#ifndef RADIO_AUDIO_H
|
||||
#define RADIO_AUDIO_H
|
||||
|
||||
#include <stdint.h>
|
||||
|
||||
struct UTUN_INSTANCE;
|
||||
|
||||
#ifdef __cplusplus
|
||||
extern "C" {
|
||||
#endif
|
||||
|
||||
#define RADIO_AUDIO_SAMPLE_RATE 48000
|
||||
#define RADIO_AUDIO_FRAME_MS 20
|
||||
#define RADIO_AUDIO_FRAME_SAMPLES (RADIO_AUDIO_SAMPLE_RATE * RADIO_AUDIO_FRAME_MS / 1000) /* 960 */
|
||||
|
||||
/* Зарегистрировать аудио-движок в radio.c (идемпотентно). */
|
||||
int radio_audio_init(struct UTUN_INSTANCE* inst);
|
||||
void radio_audio_destroy(struct UTUN_INSTANCE* inst);
|
||||
|
||||
/* GUI-поток: начать прослушивание канала (encoder + реестр источников). */
|
||||
int radio_audio_start(uint64_t group_id);
|
||||
|
||||
/* GUI-поток: прекратить прослушивание (освободить всё). */
|
||||
void radio_audio_stop(void);
|
||||
|
||||
/* GUI-поток: PTT зажата/отпущена — постит radio_talk_begin/end в uasync. */
|
||||
void radio_audio_talk_begin(uint64_t group_id);
|
||||
void radio_audio_talk_end(uint64_t group_id);
|
||||
|
||||
/* аудио-поток: закодировать PCM-кадр (ровно RADIO_AUDIO_FRAME_SAMPLES) и отправить. */
|
||||
int radio_audio_feed_pcm(uint64_t group_id, const int16_t* pcm, int count);
|
||||
|
||||
/* аудио-поток: выдать до max_samples микшированного PCM (источники микшируются,
|
||||
* источники без данных дают тишину). Возвращает число сэмплов (обычно max). */
|
||||
int radio_audio_pull_pcm(uint64_t group_id, int16_t* out, int max_samples);
|
||||
|
||||
/* 1 = аудио-движок активен (радио канала включено). */
|
||||
int radio_audio_active(void);
|
||||
|
||||
/* uasync-поток: принятый Opus-кадр источника (декодируем + кладём в ring). */
|
||||
void radio_audio_on_frame(struct UTUN_INSTANCE* inst, uint64_t group_id,
|
||||
uint64_t src_node_id, uint16_t stream_id, uint16_t seq,
|
||||
uint8_t fin, const uint8_t* opus, int len, void* arg);
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
#endif
|
||||
#endif /* RADIO_AUDIO_H */
|
||||
@@ -0,0 +1,491 @@
|
||||
// radio_headless.c — headless-аудио-сокет рации (PCM, без тактирования).
|
||||
//
|
||||
// TCP-сокет с бинарным фреймингом [type:1][group_id:8][len:2][data:len].
|
||||
// Клиент привязывается к каналу (HELLO) и обменивается готовым PCM (int16 mono 48k):
|
||||
// клиент → ядро: FRAME (960 сэмплов) — кодируется в Opus и флудится подписчикам;
|
||||
// клиент → ядро: PTT_START / PTT_STOP — radio_talk_begin / radio_talk_end;
|
||||
// ядро → клиент: микшированный PCM всех говорящих (FRAME), таймер 20мс.
|
||||
// Таблица говорящих: per-source (src_node_id) декодер + PCM-ring; один пир — один
|
||||
// поток; инстанс удаляется по FIN (дренаж) или по таймауту 2с. Без time-stretch —
|
||||
// тактирование на стороне клиента (как в call_headless).
|
||||
//
|
||||
// Управление подпиской (вкл/выкл рации) — на control-сокете (chat_headless_control):
|
||||
// radio_on / radio_off.
|
||||
#include "radio_headless.h"
|
||||
#include "radio.h"
|
||||
#include "../utun_instance.h"
|
||||
#include "../../lib/opus_codec.h"
|
||||
#include "../../lib/u_async.h"
|
||||
#include "../../lib/mem.h"
|
||||
#include "../../lib/socket_compat.h"
|
||||
#include "../../lib/debug_config.h"
|
||||
#include "../../lib/platform_compat.h"
|
||||
|
||||
#include <string.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <errno.h>
|
||||
|
||||
#define RH_ID "radio_headless"
|
||||
|
||||
#define RH_RECV_BUF_SIZE 65536
|
||||
#define RH_SEND_BUF_SIZE 65536
|
||||
#define RH_MAX_CLIENTS 16
|
||||
#define RH_MAX_FRAME 2048
|
||||
|
||||
#define RH_SAMPLE_RATE 48000
|
||||
#define RH_FRAME_SAMPLES 960 /* 20мс @48k */
|
||||
#define RH_PCM_FRAME_BYTES (RH_FRAME_SAMPLES * (int)sizeof(int16_t)) /* 1920 */
|
||||
#define RH_BITRATE 32000
|
||||
|
||||
#define RH_MAX_SOURCES 16
|
||||
#define RH_RING_FRAMES 25 /* ~500мс @20мс */
|
||||
#define RH_RING_SAMPLES (RH_RING_FRAMES * RH_FRAME_SAMPLES)
|
||||
#define RH_MIX_TIMER_TB 200 /* 20мс — период микширования */
|
||||
#define RH_SOURCE_TIMEOUT_TB 20000 /* 2с без кадров → источник завершён */
|
||||
|
||||
struct rh_source {
|
||||
uint64_t src_node_id;
|
||||
uint16_t stream_id;
|
||||
uint8_t used;
|
||||
uint8_t ending;
|
||||
uint64_t last_frame_tb;
|
||||
opus_codec_decoder_t* dec;
|
||||
int16_t* ring; /* u_malloc(RH_RING_SAMPLES) */
|
||||
int rd, wr, count;
|
||||
};
|
||||
|
||||
struct radio_headless_client {
|
||||
socket_t fd;
|
||||
void* socket_id;
|
||||
uint64_t group_id; /* 0 = не привязан */
|
||||
uint8_t recv_buf[RH_RECV_BUF_SIZE];
|
||||
int recv_len;
|
||||
uint8_t send_buf[RH_SEND_BUF_SIZE];
|
||||
int send_len;
|
||||
int send_offset;
|
||||
int closing;
|
||||
struct radio_headless* rh;
|
||||
struct radio_headless_client* next;
|
||||
};
|
||||
|
||||
struct radio_headless {
|
||||
struct UASYNC* ua;
|
||||
struct UTUN_INSTANCE* inst;
|
||||
struct radio_headless_client* clients;
|
||||
int client_count;
|
||||
socket_t listen_fd;
|
||||
void* listen_sock_id;
|
||||
int running;
|
||||
uint64_t group_id; /* активный канал (0 = не задан) */
|
||||
uint64_t enc_group_id;
|
||||
opus_codec_encoder_t* enc; /* TX-кодер активного канала */
|
||||
struct rh_source sources[RH_MAX_SOURCES];
|
||||
void* mix_timer;
|
||||
int32_t mix[RH_FRAME_SAMPLES];
|
||||
/* диагностика */
|
||||
uint32_t c_enc_frames, c_dec_frames, c_enc_dropped, c_dec_dropped;
|
||||
};
|
||||
|
||||
static struct radio_headless* rh_of(struct UTUN_INSTANCE* inst) {
|
||||
return inst ? (struct radio_headless*)inst->radio_headless : NULL;
|
||||
}
|
||||
|
||||
static void rh_close_client(struct radio_headless_client* cli);
|
||||
|
||||
/* ── отправка фрейма клиенту (буферизация как в call_headless) ── */
|
||||
|
||||
static void rh_send_raw(struct radio_headless_client* cli, const uint8_t* data, size_t len) {
|
||||
struct radio_headless* rh = cli->rh;
|
||||
if (!cli || cli->closing || cli->fd == SOCKET_INVALID || !data || len == 0) return;
|
||||
if (cli->send_len > 0) {
|
||||
size_t space = sizeof(cli->send_buf) - (size_t)cli->send_len;
|
||||
if (len > space) return;
|
||||
memcpy(cli->send_buf + cli->send_len, data, len);
|
||||
cli->send_len += (int)len;
|
||||
return;
|
||||
}
|
||||
ssize_t r = send(cli->fd, data, (int)len, 0);
|
||||
if (r < 0 && (socket_get_error() == ERR_AGAIN || socket_get_error() == ERR_WOULDBLOCK)) {
|
||||
if (len <= sizeof(cli->send_buf)) { memcpy(cli->send_buf, data, len); cli->send_len = (int)len; cli->send_offset = 0; }
|
||||
if (rh && rh->ua) uasync_set_socket_write(rh->ua, cli->socket_id, 1);
|
||||
return;
|
||||
}
|
||||
if (r >= 0 && (size_t)r < len) {
|
||||
size_t remain = len - (size_t)r;
|
||||
if (remain <= sizeof(cli->send_buf)) { memcpy(cli->send_buf, data + r, remain); cli->send_len = (int)remain; cli->send_offset = 0; }
|
||||
if (rh && rh->ua) uasync_set_socket_write(rh->ua, cli->socket_id, 1);
|
||||
}
|
||||
if (r < 0) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: send failed err=%d, closing client", RH_ID, socket_get_error());
|
||||
rh_close_client(cli);
|
||||
}
|
||||
}
|
||||
|
||||
static void rh_write_frame(struct radio_headless_client* cli, uint8_t type, uint64_t group_id,
|
||||
const uint8_t* data, uint16_t len) {
|
||||
uint8_t hdr[RADIO_AUDIO_HDR_SIZE];
|
||||
hdr[0] = type;
|
||||
memcpy(hdr + 1, &group_id, 8);
|
||||
hdr[9] = (uint8_t)(len >> 8);
|
||||
hdr[10] = (uint8_t)(len & 0xFF);
|
||||
rh_send_raw(cli, hdr, sizeof(hdr));
|
||||
if (len) rh_send_raw(cli, data, len);
|
||||
}
|
||||
|
||||
/* ── Opus-кодер активного канала ── */
|
||||
|
||||
static opus_codec_encoder_t* rh_encoder(struct radio_headless* rh, uint64_t group_id) {
|
||||
if (rh->enc && rh->enc_group_id == group_id) return rh->enc;
|
||||
if (rh->enc) { opus_codec_encoder_destroy(rh->enc); rh->enc = NULL; }
|
||||
rh->enc = opus_codec_encoder_create(RH_SAMPLE_RATE, 1);
|
||||
if (rh->enc) {
|
||||
opus_codec_encoder_bitrate_set(rh->enc, RH_BITRATE);
|
||||
rh->enc_group_id = group_id;
|
||||
} else {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: encoder create failed", RH_ID);
|
||||
}
|
||||
return rh->enc;
|
||||
}
|
||||
|
||||
/* ── таблица говорящих ── */
|
||||
|
||||
static struct rh_source* rh_src_find(struct radio_headless* rh, uint64_t src) {
|
||||
for (int i = 0; i < RH_MAX_SOURCES; i++)
|
||||
if (rh->sources[i].used && rh->sources[i].src_node_id == src) return &rh->sources[i];
|
||||
return NULL;
|
||||
}
|
||||
|
||||
static void rh_src_free(struct rh_source* s) {
|
||||
if (s->dec) { opus_codec_decoder_destroy(s->dec); s->dec = NULL; }
|
||||
if (s->ring) { u_free(s->ring); s->ring = NULL; }
|
||||
s->used = 0; s->ending = 0; s->rd = s->wr = s->count = 0;
|
||||
}
|
||||
|
||||
static struct rh_source* rh_src_acquire(struct radio_headless* rh, uint64_t src, uint16_t stream) {
|
||||
struct rh_source* s = rh_src_find(rh, src);
|
||||
if (s) {
|
||||
if (s->stream_id != stream) {
|
||||
s->stream_id = stream;
|
||||
s->ending = 0; s->rd = s->wr = s->count = 0;
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_RADIO, "%s: source reset src=%016llx stream=%u", RH_ID, (unsigned long long)src, stream);
|
||||
}
|
||||
return s;
|
||||
}
|
||||
for (int i = 0; i < RH_MAX_SOURCES; i++) {
|
||||
if (rh->sources[i].used) continue;
|
||||
s = &rh->sources[i];
|
||||
s->dec = opus_codec_decoder_create(RH_SAMPLE_RATE, 1);
|
||||
if (!s->dec) { DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: decoder create failed", RH_ID); return NULL; }
|
||||
s->ring = u_malloc(RH_RING_SAMPLES * sizeof(int16_t));
|
||||
if (!s->ring) { opus_codec_decoder_destroy(s->dec); s->dec = NULL; return NULL; }
|
||||
s->src_node_id = src;
|
||||
s->stream_id = stream;
|
||||
s->used = 1; s->ending = 0; s->rd = s->wr = s->count = 0;
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_RADIO, "%s: new source src=%016llx stream=%u", RH_ID, (unsigned long long)src, stream);
|
||||
return s;
|
||||
}
|
||||
DEBUG_WARN(DEBUG_CATEGORY_RADIO, "%s: too many sources (>=%d), drop src=%016llx", RH_ID, RH_MAX_SOURCES, (unsigned long long)src);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
static void rh_src_push(struct rh_source* s, const int16_t* pcm, int n) {
|
||||
int cap = RH_RING_SAMPLES;
|
||||
if (n > cap) { pcm += n - cap; n = cap; }
|
||||
if (s->count + n > cap) {
|
||||
int overflow = s->count + n - cap;
|
||||
s->rd = (s->rd + overflow) % cap;
|
||||
s->count -= overflow;
|
||||
}
|
||||
for (int i = 0; i < n; i++) {
|
||||
s->ring[s->wr] = pcm[i];
|
||||
s->wr = (s->wr + 1) % cap;
|
||||
}
|
||||
s->count += n;
|
||||
}
|
||||
|
||||
/* ── приём Opus-кадра пира (frame cb) → декод → ring источника ── */
|
||||
|
||||
static void rh_frame_cb(struct UTUN_INSTANCE* inst, uint64_t group_id,
|
||||
uint64_t src_node_id, uint16_t stream_id, uint16_t seq,
|
||||
uint8_t fin, const uint8_t* opus, int len, void* arg) {
|
||||
(void)seq;
|
||||
struct radio_headless* rh = (struct radio_headless*)arg;
|
||||
if (!rh || !rh->running || !opus || len <= 0) return;
|
||||
|
||||
/* активный канал: первый кадр фиксирует group_id (узел слушает один канал) */
|
||||
if (rh->group_id == 0) rh->group_id = group_id;
|
||||
if (group_id != rh->group_id) return;
|
||||
|
||||
struct rh_source* s = rh_src_acquire(rh, src_node_id, stream_id);
|
||||
if (!s) return;
|
||||
int16_t pcm[RH_FRAME_SAMPLES];
|
||||
int n = opus_codec_decode(s->dec, opus, len, pcm, RH_FRAME_SAMPLES);
|
||||
if (n <= 0) { rh->c_dec_dropped++; return; }
|
||||
rh_src_push(s, pcm, n);
|
||||
rh->c_dec_frames++;
|
||||
s->last_frame_tb = get_time_tb();
|
||||
if (fin) s->ending = 1;
|
||||
(void)inst;
|
||||
}
|
||||
|
||||
/* ── периодический микшер: sum источников → PCM клиентам ── */
|
||||
|
||||
static void rh_mix_tick(void* arg) {
|
||||
struct radio_headless* rh = (struct radio_headless*)arg;
|
||||
rh->mix_timer = NULL;
|
||||
if (!rh || !rh->running) return;
|
||||
|
||||
if (rh->group_id != 0) {
|
||||
memset(rh->mix, 0, sizeof(rh->mix));
|
||||
uint64_t now = get_time_tb();
|
||||
int any = 0;
|
||||
|
||||
for (int i = 0; i < RH_MAX_SOURCES; i++) {
|
||||
struct rh_source* s = &rh->sources[i];
|
||||
if (!s->used) continue;
|
||||
if (!s->ending && s->last_frame_tb && (now - s->last_frame_tb > RH_SOURCE_TIMEOUT_TB)) {
|
||||
DEBUG_WARN(DEBUG_CATEGORY_RADIO, "%s: source timeout src=%016llx", RH_ID, (unsigned long long)s->src_node_id);
|
||||
rh_src_free(s);
|
||||
continue;
|
||||
}
|
||||
int take = s->count < RH_FRAME_SAMPLES ? s->count : RH_FRAME_SAMPLES;
|
||||
for (int k = 0; k < take; k++) {
|
||||
rh->mix[k] += s->ring[s->rd];
|
||||
s->rd = (s->rd + 1) % RH_RING_SAMPLES;
|
||||
}
|
||||
s->count -= take;
|
||||
if (take > 0) any = 1;
|
||||
if (s->ending && s->count == 0) {
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_RADIO, "%s: source ended src=%016llx", RH_ID, (unsigned long long)s->src_node_id);
|
||||
rh_src_free(s);
|
||||
}
|
||||
}
|
||||
|
||||
/* шлём PCM клиентам только когда есть что микшировать (тишина — неявно) */
|
||||
if (any) {
|
||||
int16_t pcm[RH_FRAME_SAMPLES];
|
||||
for (int k = 0; k < RH_FRAME_SAMPLES; k++) {
|
||||
int32_t v = rh->mix[k];
|
||||
if (v > 32767) v = 32767;
|
||||
else if (v < -32768) v = -32768;
|
||||
pcm[k] = (int16_t)v;
|
||||
}
|
||||
for (struct radio_headless_client* c = rh->clients; c; c = c->next) {
|
||||
if (c->group_id == rh->group_id && !c->closing)
|
||||
rh_write_frame(c, RADIO_AUDIO_FRAME, rh->group_id, (const uint8_t*)pcm, RH_PCM_FRAME_BYTES);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
rh->mix_timer = uasync_set_timeout(rh->ua, RH_MIX_TIMER_TB, rh, rh_mix_tick, "rh_mix");
|
||||
}
|
||||
|
||||
/* ── разбор входящих фреймов ── */
|
||||
|
||||
static void rh_parse_frames(struct radio_headless_client* cli) {
|
||||
struct radio_headless* rh = cli->rh;
|
||||
uint8_t* p = cli->recv_buf;
|
||||
int avail = cli->recv_len;
|
||||
int consumed = 0;
|
||||
while (avail - consumed >= RADIO_AUDIO_HDR_SIZE) {
|
||||
uint8_t type = p[consumed];
|
||||
uint64_t group_id; memcpy(&group_id, p + consumed + 1, 8);
|
||||
uint16_t flen = (uint16_t)((p[consumed + 9] << 8) | p[consumed + 10]);
|
||||
if (flen > RH_MAX_FRAME) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: frame too big %u — closing", RH_ID, flen);
|
||||
rh_close_client(cli);
|
||||
return;
|
||||
}
|
||||
if (avail - consumed < (int)(RADIO_AUDIO_HDR_SIZE + flen)) break;
|
||||
const uint8_t* data = p + consumed + RADIO_AUDIO_HDR_SIZE;
|
||||
|
||||
switch (type) {
|
||||
case RADIO_AUDIO_HELLO:
|
||||
cli->group_id = group_id;
|
||||
if (rh->group_id == 0) rh->group_id = group_id;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_RADIO, "%s: client bound to group=0x%016llx", RH_ID, (unsigned long long)group_id);
|
||||
break;
|
||||
case RADIO_AUDIO_PTT_START:
|
||||
radio_talk_begin(rh->inst, group_id);
|
||||
break;
|
||||
case RADIO_AUDIO_PTT_STOP:
|
||||
radio_talk_end(rh->inst, group_id);
|
||||
break;
|
||||
case RADIO_AUDIO_FRAME:
|
||||
if (flen != RH_PCM_FRAME_BYTES) {
|
||||
rh->c_enc_dropped++;
|
||||
DEBUG_WARN(DEBUG_CATEGORY_RADIO, "%s: bad PCM frame size=%u (expect %d) — dropped",
|
||||
RH_ID, flen, RH_PCM_FRAME_BYTES);
|
||||
} else {
|
||||
opus_codec_encoder_t* enc = rh_encoder(rh, group_id);
|
||||
uint8_t opus[256];
|
||||
int olen = enc ? opus_codec_encode(enc, (const int16_t*)data, RH_FRAME_SAMPLES, opus, (int)sizeof(opus)) : -1;
|
||||
if (olen <= 0) {
|
||||
rh->c_enc_dropped++;
|
||||
DEBUG_WARN(DEBUG_CATEGORY_RADIO, "%s: encode failed (olen=%d) group=0x%016llx",
|
||||
RH_ID, olen, (unsigned long long)group_id);
|
||||
} else {
|
||||
rh->c_enc_frames++;
|
||||
if (radio_talk_send(rh->inst, group_id, opus, olen) != 0)
|
||||
DEBUG_WARN(DEBUG_CATEGORY_RADIO, "%s: talk_send failed group=0x%016llx len=%d",
|
||||
RH_ID, (unsigned long long)group_id, olen);
|
||||
}
|
||||
}
|
||||
break;
|
||||
default:
|
||||
DEBUG_WARN(DEBUG_CATEGORY_RADIO, "%s: unknown frame type=0x%02x — closing", RH_ID, type);
|
||||
rh_close_client(cli);
|
||||
return;
|
||||
}
|
||||
consumed += RADIO_AUDIO_HDR_SIZE + flen;
|
||||
}
|
||||
if (consumed > 0) {
|
||||
memmove(cli->recv_buf, cli->recv_buf + consumed, (size_t)(cli->recv_len - consumed));
|
||||
cli->recv_len -= consumed;
|
||||
}
|
||||
}
|
||||
|
||||
/* ── socket callbacks ── */
|
||||
|
||||
static void rh_write_callback(socket_t fd, void* arg) {
|
||||
struct radio_headless_client* cli = (struct radio_headless_client*)arg;
|
||||
struct radio_headless* rh = cli ? cli->rh : NULL;
|
||||
if (!cli || cli->closing || fd != cli->fd) return;
|
||||
if (cli->send_len <= 0) { if (rh && rh->ua) uasync_set_socket_write(rh->ua, cli->socket_id, 0); return; }
|
||||
ssize_t r = send(cli->fd, cli->send_buf + cli->send_offset, cli->send_len - cli->send_offset, 0);
|
||||
if (r < 0 && (socket_get_error() == ERR_AGAIN || socket_get_error() == ERR_WOULDBLOCK)) return;
|
||||
if (r <= 0) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: send failed (r=%zd err=%d) — closing client", RH_ID, r, socket_get_error());
|
||||
rh_close_client(cli);
|
||||
return;
|
||||
}
|
||||
cli->send_offset += (int)r;
|
||||
if (cli->send_offset >= cli->send_len) { cli->send_len = 0; cli->send_offset = 0; if (rh && rh->ua) uasync_set_socket_write(rh->ua, cli->socket_id, 0); }
|
||||
}
|
||||
|
||||
static void rh_close_client(struct radio_headless_client* cli) {
|
||||
struct radio_headless* rh = cli ? cli->rh : NULL;
|
||||
if (!cli || cli->closing) return;
|
||||
cli->closing = 1;
|
||||
if (rh && rh->ua && cli->socket_id) uasync_remove_socket_t(rh->ua, cli->fd);
|
||||
if (cli->fd != SOCKET_INVALID) { socket_close_wrapper(cli->fd); cli->fd = SOCKET_INVALID; }
|
||||
DEBUG_INFO(DEBUG_CATEGORY_RADIO, "%s: client closed group=0x%016llx", RH_ID, (unsigned long long)cli->group_id);
|
||||
}
|
||||
|
||||
static void rh_read_callback(socket_t fd, void* arg) {
|
||||
struct radio_headless_client* cli = (struct radio_headless_client*)arg;
|
||||
if (!cli || cli->closing) return;
|
||||
uint8_t tmp[4096];
|
||||
ssize_t r = recv(fd, tmp, (int)sizeof(tmp), 0);
|
||||
if (r <= 0) { rh_close_client(cli); return; }
|
||||
if (cli->recv_len + (int)r > RH_RECV_BUF_SIZE) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: recv buffer overflow — closing", RH_ID);
|
||||
rh_close_client(cli);
|
||||
return;
|
||||
}
|
||||
memcpy(cli->recv_buf + cli->recv_len, tmp, (size_t)r);
|
||||
cli->recv_len += (int)r;
|
||||
rh_parse_frames(cli);
|
||||
}
|
||||
|
||||
static void rh_accept_callback(socket_t fd, void* arg) {
|
||||
struct radio_headless* rh = (struct radio_headless*)arg;
|
||||
struct sockaddr_storage addr; socklen_t alen = sizeof(addr);
|
||||
socket_t cfd = accept(fd, (struct sockaddr*)&addr, &alen);
|
||||
if (cfd == SOCKET_INVALID) {
|
||||
int e = socket_get_error();
|
||||
if (e != ERR_AGAIN && e != ERR_WOULDBLOCK) DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: accept failed: %s", RH_ID, socket_strerror(e));
|
||||
return;
|
||||
}
|
||||
socket_set_nonblocking(cfd);
|
||||
if (rh->client_count >= RH_MAX_CLIENTS) { socket_close_wrapper(cfd); DEBUG_WARN(DEBUG_CATEGORY_RADIO, "%s: max clients reached", RH_ID); return; }
|
||||
|
||||
struct radio_headless_client* cli = u_calloc(1, sizeof(*cli));
|
||||
if (!cli) { socket_close_wrapper(cfd); return; }
|
||||
cli->fd = cfd;
|
||||
cli->rh = rh;
|
||||
cli->socket_id = uasync_add_socket_t(rh->ua, cfd, rh_read_callback, rh_write_callback, NULL, cli);
|
||||
if (!cli->socket_id) { u_free(cli); socket_close_wrapper(cfd); DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: failed to register client fd", RH_ID); return; }
|
||||
|
||||
cli->next = rh->clients; rh->clients = cli; rh->client_count++;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_RADIO, "%s: client connected fd=%d total=%d", RH_ID, (int)cfd, rh->client_count);
|
||||
}
|
||||
|
||||
/* ── периодическая очистка закрытых клиентов ── */
|
||||
|
||||
static void rh_cleanup(void* arg) {
|
||||
struct radio_headless* rh = (struct radio_headless*)arg;
|
||||
struct radio_headless_client** prev = &rh->clients;
|
||||
while (*prev) {
|
||||
if ((*prev)->closing) {
|
||||
struct radio_headless_client* dead = *prev; *prev = dead->next;
|
||||
if (dead->fd != SOCKET_INVALID) socket_close_wrapper(dead->fd);
|
||||
u_free(dead); rh->client_count--;
|
||||
} else prev = &(*prev)->next;
|
||||
}
|
||||
if (rh->ua) uasync_set_timeout(rh->ua, 5000, rh, rh_cleanup, "rh_cleanup");
|
||||
}
|
||||
|
||||
/* ── init / destroy ── */
|
||||
|
||||
int radio_headless_init(struct UASYNC* ua, struct UTUN_INSTANCE* inst,
|
||||
const char* bind_ip, int port) {
|
||||
if (!ua || !inst || !bind_ip || port <= 0 || port > 65535) return -1;
|
||||
struct radio_headless* rh = rh_of(inst);
|
||||
if (rh && rh->running) return 0;
|
||||
|
||||
if (!rh) {
|
||||
rh = u_calloc(1, sizeof(*rh));
|
||||
if (!rh) return -1;
|
||||
inst->radio_headless = rh;
|
||||
}
|
||||
rh->ua = ua; rh->inst = inst;
|
||||
|
||||
rh->listen_fd = socket(AF_INET, SOCK_STREAM, 0);
|
||||
if (rh->listen_fd == SOCKET_INVALID) { DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: socket() failed: %s", RH_ID, socket_strerror(socket_get_error())); return -1; }
|
||||
socket_set_reuseaddr(rh->listen_fd, 1);
|
||||
socket_set_nonblocking(rh->listen_fd);
|
||||
|
||||
struct sockaddr_in sin; memset(&sin, 0, sizeof(sin));
|
||||
sin.sin_family = AF_INET; sin.sin_port = htons((uint16_t)port);
|
||||
if (inet_pton(AF_INET, bind_ip, &sin.sin_addr) != 1) { socket_close_wrapper(rh->listen_fd); rh->listen_fd = SOCKET_INVALID; return -1; }
|
||||
if (bind(rh->listen_fd, (struct sockaddr*)&sin, sizeof(sin)) < 0) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_RADIO, "%s: bind(%s:%d) failed: %s", RH_ID, bind_ip, port, socket_strerror(socket_get_error()));
|
||||
socket_close_wrapper(rh->listen_fd); rh->listen_fd = SOCKET_INVALID; return -1;
|
||||
}
|
||||
if (listen(rh->listen_fd, 5) < 0) { socket_close_wrapper(rh->listen_fd); rh->listen_fd = SOCKET_INVALID; return -1; }
|
||||
|
||||
rh->listen_sock_id = uasync_add_socket_t(ua, rh->listen_fd, rh_accept_callback, NULL, NULL, rh);
|
||||
if (!rh->listen_sock_id) { socket_close_wrapper(rh->listen_fd); rh->listen_fd = SOCKET_INVALID; return -1; }
|
||||
|
||||
radio_set_frame_cb(inst, rh_frame_cb, rh);
|
||||
uasync_set_timeout(ua, 5000, rh, rh_cleanup, "rh_cleanup");
|
||||
rh->mix_timer = uasync_set_timeout(ua, RH_MIX_TIMER_TB, rh, rh_mix_tick, "rh_mix");
|
||||
|
||||
rh->running = 1;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_RADIO, "%s: audio socket listening on %s:%d", RH_ID, bind_ip, port);
|
||||
return 0;
|
||||
}
|
||||
|
||||
void radio_headless_destroy(struct UTUN_INSTANCE* inst) {
|
||||
struct radio_headless* rh = rh_of(inst);
|
||||
if (!rh || !rh->running) return;
|
||||
rh->running = 0;
|
||||
radio_set_frame_cb(inst, NULL, NULL);
|
||||
if (rh->mix_timer && rh->ua) { uasync_cancel_timeout(rh->ua, rh->mix_timer); rh->mix_timer = NULL; }
|
||||
if (rh->listen_sock_id && rh->ua) uasync_remove_socket_t(rh->ua, rh->listen_fd);
|
||||
if (rh->listen_fd != SOCKET_INVALID) { socket_close_wrapper(rh->listen_fd); rh->listen_fd = SOCKET_INVALID; }
|
||||
if (rh->enc) { opus_codec_encoder_destroy(rh->enc); rh->enc = NULL; }
|
||||
for (int i = 0; i < RH_MAX_SOURCES; i++) if (rh->sources[i].used) rh_src_free(&rh->sources[i]);
|
||||
struct radio_headless_client* c = rh->clients;
|
||||
while (c) {
|
||||
struct radio_headless_client* n = c->next;
|
||||
if (c->fd != SOCKET_INVALID) socket_close_wrapper(c->fd);
|
||||
u_free(c);
|
||||
c = n;
|
||||
}
|
||||
rh->clients = NULL; rh->client_count = 0;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_RADIO, "%s: audio socket destroyed (enc=%u dec=%u enc_drop=%u dec_drop=%u)",
|
||||
RH_ID, rh->c_enc_frames, rh->c_dec_frames, rh->c_enc_dropped, rh->c_dec_dropped);
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
// radio_headless.h — headless-аудио-сокет рации (PCM, без тактирования).
|
||||
//
|
||||
// Отдельный TCP-сокет, через который headless-клиент (CLI/скрипт) обменивается
|
||||
// готовым PCM с ядром рации. Ядро кодирует PCM→Opus (TX), принимает Opus от пиров,
|
||||
// микширует их в общий PCM (RX, простой микшер без time-stretch — тактирование на
|
||||
// стороне клиента, как в call_headless). Управление рацией/PTT — через фреймы
|
||||
// PTT_START/PTT_STOP и команды radio_on/radio_off на control-сокете.
|
||||
//
|
||||
// Бинарный фрейм (обе стороны):
|
||||
// [type:1][group_id:8][len:2][data:len]
|
||||
// type = 0x01 HELLO (клиент → ядро: привязаться к group_id, data пусто)
|
||||
// type = 0x02 PTT_START (клиент → ядро: radio_talk_begin, data пусто)
|
||||
// type = 0x03 PTT_STOP (клиент → ядро: radio_talk_end, data пусто)
|
||||
// type = 0x04 FRAME (PCM int16 mono 48k, 960 сэмплов = 1920 байт)
|
||||
// клиент → ядро = PCM захвата (кодируется и флудится подписчикам);
|
||||
// ядро → клиент = микшированный PCM всех говорящих
|
||||
#ifndef RADIO_HEADLESS_H
|
||||
#define RADIO_HEADLESS_H
|
||||
|
||||
#include <stdint.h>
|
||||
|
||||
struct UASYNC;
|
||||
struct UTUN_INSTANCE;
|
||||
|
||||
#define RADIO_AUDIO_HELLO 0x01
|
||||
#define RADIO_AUDIO_PTT_START 0x02
|
||||
#define RADIO_AUDIO_PTT_STOP 0x03
|
||||
#define RADIO_AUDIO_FRAME 0x04
|
||||
#define RADIO_AUDIO_HDR_SIZE 11 /* type(1) + group_id(8) + len(2) */
|
||||
#define RADIO_AUDIO_PCM_FRAME_BYTES (960 * (int)sizeof(int16_t)) /* 1920 — PCM int16 mono 48k 20мс */
|
||||
|
||||
#ifdef __cplusplus
|
||||
extern "C" {
|
||||
#endif
|
||||
|
||||
int radio_headless_init(struct UASYNC* ua, struct UTUN_INSTANCE* inst,
|
||||
const char* bind_ip, int port);
|
||||
void radio_headless_destroy(struct UTUN_INSTANCE* inst);
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
#endif
|
||||
#endif /* RADIO_HEADLESS_H */
|
||||
@@ -0,0 +1,39 @@
|
||||
// radio_proto.h — wire-формат и константы рации (PTT walkie-talkie мультикаст).
|
||||
//
|
||||
// Транспорт: raw ETCP (etcp_bind/etcp_send), первый байт кодограммы = ETCP_ID_RADIO.
|
||||
// Пакет флудится по senders_list CHAT-группы с hop-list против циклов и
|
||||
// RPF-обрезанием ветвей без подписчиков (узел с radio==1 в NODEINFO).
|
||||
//
|
||||
// Формат payload (после dgram[0] = ETCP_ID_RADIO):
|
||||
// [group_id:8][src_node_id:8][stream_id:2][seq:2][flags:1][hop_count:1]
|
||||
// [hop_list:8*hop_count][opus...]
|
||||
// flags: bit0 = FIN (последний кадр передачи)
|
||||
// stream_id — случайный идентификатор передачи (talk-burst), seq нумерует кадры от 0.
|
||||
#ifndef RADIO_PROTO_H
|
||||
#define RADIO_PROTO_H
|
||||
|
||||
#include <stdint.h>
|
||||
|
||||
#ifdef __cplusplus
|
||||
extern "C" {
|
||||
#endif
|
||||
|
||||
#define RADIO_FLAG_FIN 0x01
|
||||
|
||||
#define RADIO_MAX_HOPS 16
|
||||
#define RADIO_MAX_OPUS 256
|
||||
|
||||
#define RADIO_HDR_SIZE 22 /* group(8)+src(8)+stream(2)+seq(2)+flags(1)+hop_count(1), без hop_list */
|
||||
|
||||
#define RADIO_OFF_GROUP 0
|
||||
#define RADIO_OFF_SRC 8
|
||||
#define RADIO_OFF_STREAM 16
|
||||
#define RADIO_OFF_SEQ 18
|
||||
#define RADIO_OFF_FLAGS 20
|
||||
#define RADIO_OFF_HOPS 21
|
||||
#define RADIO_OFF_HOPLIST 22
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
#endif
|
||||
#endif /* RADIO_PROTO_H */
|
||||
@@ -27,6 +27,7 @@
|
||||
#include "topo_group_connect.h"
|
||||
#include "topo_recovery.h"
|
||||
#include "broadcast.h"
|
||||
#include "../radio/radio.h"
|
||||
#include "../chat/chat_setting.h"
|
||||
|
||||
#define TOPO_GROUP_UTUN 0x8000000000000000ULL
|
||||
@@ -355,6 +356,7 @@ static struct TOPO_GROUP* topo_group_create(struct UTUN_INSTANCE* instance, uint
|
||||
group->conn_mgr = conn_mgr_init(group);
|
||||
|
||||
broadcast_init(group);
|
||||
radio_init(group);
|
||||
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "TOPO group created: group_id=%016llx", (unsigned long long)group_id);
|
||||
return group;
|
||||
@@ -374,6 +376,9 @@ static void topo_group_destroy(struct TOPO_GROUP* group) {
|
||||
broadcast_destroy(group);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 2b broadcast_destroy done");
|
||||
|
||||
radio_destroy(group);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[GRP_DESTROY] 2c radio_destroy done");
|
||||
|
||||
struct ll_entry* e;
|
||||
while ((e = queue_data_get(group->senders_list)) != NULL) {
|
||||
struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e;
|
||||
@@ -628,6 +633,48 @@ void topo_group_set_nat_check_local(struct TOPO_GROUP* group, int allow) {
|
||||
nat_detection_set_allow_local(group->instance->nat_det, allow);
|
||||
}
|
||||
|
||||
/* ── Рация (PTT walkie-talkie, per-CHAT-группа) ── */
|
||||
|
||||
void topo_group_set_radio(struct TOPO_GROUP* group, int on) {
|
||||
if (!group || !group->instance) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "set_radio: invalid args"); return; }
|
||||
if (group->group_type != TOPO_GROUP_TYPE_CHAT) {
|
||||
DEBUG_WARN(DEBUG_CATEGORY_BGP, "set_radio: only CHAT groups support radio grp=%016llx type=%u",
|
||||
(unsigned long long)group->group_id, group->group_type);
|
||||
return;
|
||||
}
|
||||
uint8_t v = on ? 1 : 0;
|
||||
if (group->radio_active == v) {
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "set_radio: unchanged (%u) grp=%016llx ch=%s", v,
|
||||
(unsigned long long)group->group_id, group->channel_id);
|
||||
return;
|
||||
}
|
||||
group->radio_active = v;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "set_radio: %s grp=%016llx ch=%s",
|
||||
v ? "ON" : "OFF", (unsigned long long)group->group_id, group->channel_id);
|
||||
topo_group_update_my_nodeinfo(group->instance, group);
|
||||
if (group->senders_list) {
|
||||
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 && group->local_node)
|
||||
topo_group_send_nodeinfo(group, group->local_node, item->conn, 0);
|
||||
se = se->next;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
int topo_group_radio_subscribers(struct TOPO_GROUP* group) {
|
||||
if (!group || !group->nodes) return 0;
|
||||
int n = 0;
|
||||
struct ll_entry* e = group->nodes->head;
|
||||
while (e) {
|
||||
struct TOPO_GROUP_NODE* nq = (struct TOPO_GROUP_NODE*)e;
|
||||
if (nq->radio && nq->node_id != group->instance->node_id) n++;
|
||||
e = e->next;
|
||||
}
|
||||
return n;
|
||||
}
|
||||
|
||||
/* Устанавливает колбэк сохранения узла в БД (используется chatgui/member_sync). */
|
||||
void topo_groups_set_node_updated_cb(struct TOPO_GROUPS* groups, topo_node_updated_fn fn) {
|
||||
if (groups) groups->node_updated_cb = fn;
|
||||
@@ -981,6 +1028,7 @@ int topo_group_process_nodeinfo(struct TOPO_GROUP* group, struct ETCP_CONN* from
|
||||
(unsigned long long)node_id, (unsigned long long)group->group_id, group->group_type, group->channel_id);
|
||||
|
||||
nodeinfo1->paths = paths; nodeinfo1->last_ver = new_ver;
|
||||
nodeinfo1->radio = (ni->flags & TOPO_FLAG_RADIO) ? 1 : 0;
|
||||
|
||||
uint64_t hop_list[MAX_HOPS];
|
||||
if (new_hop_count > 0) memcpy(hop_list, new_hop_list, new_hop_count * 8);
|
||||
|
||||
@@ -51,6 +51,7 @@ struct CONN_MGR;
|
||||
struct TOPO_RECOVERY_CTX;
|
||||
struct TOPO_GROUP_CONNECT;
|
||||
struct broadcast_ctx;
|
||||
struct radio_ctx;
|
||||
|
||||
/** Callback when a node's info (pubkeys, addresses) is persisted in the DB */
|
||||
typedef void (*topo_node_updated_fn)(struct UTUN_INSTANCE* inst, uint64_t node_id,
|
||||
@@ -191,6 +192,8 @@ struct TOPO_GROUP {
|
||||
struct TOPO_RECOVERY_CTX* recovery_list; // список активных процедур восстановления
|
||||
struct TOPO_GROUP_CONNECT* connect; // авто-подключение к узлам группы (CHAT)
|
||||
struct broadcast_ctx* broadcast; // broadcast protocol per-group context
|
||||
struct radio_ctx* radio; // radio (PTT walkie-talkie) per-group context
|
||||
uint8_t radio_active; // 1 = я слушаю рацию этого канала (TOPO_FLAG_RADIO в NODEINFO)
|
||||
struct topo_node_cbk_entry* node_cbks; // цепочка подписчиков на события узлов
|
||||
};
|
||||
|
||||
@@ -346,7 +349,16 @@ void topo_groups_set_node_updated_cb(struct TOPO_GROUPS* groups, topo_node_updat
|
||||
/** Разрешить/запретить NAT check для локальных подсетей (тестовый хелпер, делегат в nat_detection) */
|
||||
void topo_group_set_nat_check_local(struct TOPO_GROUP* group, int allow);
|
||||
|
||||
/* ── BGP node event callbacks ── */
|
||||
/* ── Рация (PTT walkie-talkie, per-CHAT-группа) ── */
|
||||
|
||||
/* Включить/выключить прослушивание рации канала: выставляет local_node->radio и
|
||||
* рассылает NODEINFO (TOPO_FLAG_RADIO) через BGP. Работает только для CHAT-групп. */
|
||||
void topo_group_set_radio(struct TOPO_GROUP* group, int on);
|
||||
|
||||
/* Число узлов группы, слушающих рацию (radio==1), без учёта себя. */
|
||||
int topo_group_radio_subscribers(struct TOPO_GROUP* group);
|
||||
|
||||
/* BGP node event callbacks ── */
|
||||
void topo_group_add_node_cbk(struct TOPO_GROUP* group, topo_node_event_fn fn, void* arg);
|
||||
void topo_group_remove_node_cbk(struct TOPO_GROUP* group, topo_node_event_fn fn, void* arg);
|
||||
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
/*
|
||||
* topo_group_connect.c — авто-подключение к узлам группы (бесконечный цикл до успеха)
|
||||
* topo_group_connect.c — авто-подключение к узлам группы (бесконечный цикл до цели)
|
||||
*
|
||||
* Phase 1: одновременный запуск node_conn_direct_open для пиров с connected=1.
|
||||
* Таймаут = TGC_DIRECT_TIMEOUT_MS. Если хоть один подключился → done.
|
||||
* Таймаут = TGC_DIRECT_TIMEOUT_MS. Если цель достигнута → done.
|
||||
*
|
||||
* Phase 2: последовательный перебор (supernode, затем public/EIM адреса).
|
||||
* После каждого TIMEOUT — пауза перед следующей попыткой.
|
||||
@@ -12,9 +12,15 @@
|
||||
*
|
||||
* После исчерпания Phase 3 → пауза → cycle_restart → Phase 1 (бесконечно).
|
||||
*
|
||||
* Цель (tgc_goal_reached): подключены хотя бы к одному суперузлу (node_type==4)
|
||||
* ЛИБО к 3+ не-мобильным клиентам (device_type != MOBILE, не суперузлам).
|
||||
* Пока цель не достигнута, успешное подключение НЕ останавливает цикл —
|
||||
* продолжается перебор следующих кандидатов.
|
||||
*
|
||||
* Пауза: активный режим — 1с; фоновый (Android standby) — standby_wait
|
||||
* (burst → сразу, sleep → до следующего burst).
|
||||
* При DOWN (active_conn_count → 0) — немедленный cycle_restart.
|
||||
* При DOWN, если цель перестала выполняться — немедленный cycle_restart
|
||||
* (уже установленные соединения не трогаются).
|
||||
* При повторном topo_group_connect_init — destroy + fresh init.
|
||||
*
|
||||
* Счётчик active_conn_count — число уникальных peer-узлов с реальными соединениями.
|
||||
@@ -33,6 +39,7 @@
|
||||
#include "../lib/u_async.h"
|
||||
#include "../lib/ll_queue.h"
|
||||
#include "etcp.h"
|
||||
#include "etcp_connections.h"
|
||||
#include "../chat/chat_event.h"
|
||||
#ifdef UTUN_HAVE_STANDBY
|
||||
#include "standby.h"
|
||||
@@ -48,6 +55,8 @@
|
||||
#define TGC_PHASE_PAUSE 4
|
||||
#define TGC_MAX_HANDLES 128
|
||||
#define TGC_PAUSE_ACTIVE_TB 10000 /* 1s пауза в активном режиме */
|
||||
#define TGC_GOAL_SUPERNODES 1 /* цель: подключены хотя бы к одному суперузлу */
|
||||
#define TGC_GOAL_NONMOBILE 3 /* цель: подключены к 3+ не-мобильным клиентам (не суперузлам) */
|
||||
|
||||
struct TOPO_GROUP_CONNECT {
|
||||
struct TOPO_GROUP* group;
|
||||
@@ -72,6 +81,7 @@ static void tgc_phase3_try_next(struct TOPO_GROUP_CONNECT* gc);
|
||||
static void tgc_phase1_timeout(void* arg);
|
||||
static void tgc_cycle_restart(struct TOPO_GROUP_CONNECT* gc);
|
||||
static void tgc_start_pause(struct TOPO_GROUP_CONNECT* gc, void (*cb)(void*), const char* label);
|
||||
static int tgc_goal_reached(struct TOPO_GROUP_CONNECT* gc);
|
||||
|
||||
static void tgc_callback(struct NODE_CONN_DIRECT* h, enum ncd_event event, void* arg);
|
||||
|
||||
@@ -97,6 +107,58 @@ static void tgc_notify_connecting(struct TOPO_GROUP_CONNECT* gc, const uint64_t*
|
||||
TGC_ID, gc->group->channel_id, count);
|
||||
}
|
||||
|
||||
/* ═══════════════════════════════════════════════════════════════════════
|
||||
* Условие достижения цели авто-подключения
|
||||
* ══════════════════════════════════════════════════════════════════════ */
|
||||
|
||||
/* Тип устройства пира из handshake (peer_device_type на первом UP-линке). */
|
||||
static uint8_t tgc_peer_device_type(struct ETCP_CONN* conn) {
|
||||
if (!conn || !conn->links) return CLIENT_TYPE_SERVER;
|
||||
return conn->links->peer_device_type;
|
||||
}
|
||||
|
||||
/**
|
||||
* Достигнута ли цель подключения к группе:
|
||||
* - подключены хотя бы к одному суперузлу (peers node_type==4), ЛИБО
|
||||
* - подключены к 3+ не-мобильным клиентам (device_type != MOBILE и не суперузел).
|
||||
*
|
||||
* Сканирует senders_list (активные BGP-пиры) с дедупом по node_id.
|
||||
* Состояние не хранится — пересчитывается на лету в редких точках принятия
|
||||
* решения (таймаут фазы, каждый успех, каждый down).
|
||||
*/
|
||||
static int tgc_goal_reached(struct TOPO_GROUP_CONNECT* gc) {
|
||||
if (!gc || !gc->group || !gc->group->senders_list) return 0;
|
||||
struct TOPO_GROUP* group = gc->group;
|
||||
sqlite3* db = group->instance->topo_sqlite_db;
|
||||
|
||||
uint64_t seen[TGC_MAX_HANDLES]; int seen_count = 0;
|
||||
int supernode_count = 0, nonmobile_count = 0;
|
||||
|
||||
struct ll_entry* e = group->senders_list->head;
|
||||
while (e) {
|
||||
struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)e->data;
|
||||
struct ETCP_CONN* conn = item ? item->conn : NULL;
|
||||
if (conn) {
|
||||
int dup = 0;
|
||||
for (int i = 0; i < seen_count; i++) if (seen[i] == conn->peer_node_id) { dup = 1; break; }
|
||||
if (!dup && seen_count < TGC_MAX_HANDLES) {
|
||||
seen[seen_count++] = conn->peer_node_id;
|
||||
int is_super = db && topo_node_sqlite_get_node_type(db, group->channel_id, conn->peer_node_id) == 4;
|
||||
int is_mobile = tgc_peer_device_type(conn) == CLIENT_TYPE_MOBILE;
|
||||
if (is_super) supernode_count++;
|
||||
else if (!is_mobile) nonmobile_count++;
|
||||
}
|
||||
}
|
||||
e = e->next;
|
||||
}
|
||||
|
||||
int reached = (supernode_count >= TGC_GOAL_SUPERNODES) || (nonmobile_count >= TGC_GOAL_NONMOBILE);
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "%s: goal ch=%s super=%d/%d nonmobile=%d/%d → %s",
|
||||
TGC_ID, group->channel_id, supernode_count, TGC_GOAL_SUPERNODES,
|
||||
nonmobile_count, TGC_GOAL_NONMOBILE, reached ? "reached" : "not reached");
|
||||
return reached;
|
||||
}
|
||||
|
||||
/* ═══════════════════════════════════════════════════════════════════════
|
||||
* Жизненный цикл
|
||||
* ══════════════════════════════════════════════════════════════════════ */
|
||||
@@ -259,8 +321,8 @@ void topo_group_connect_on_down(struct TOPO_GROUP* group, struct ETCP_CONN* conn
|
||||
if (db) topo_node_sqlite_set_connected(db, group->channel_id, peer, 0);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: DOWN peer=0x%016llx ch=%s active=%d",
|
||||
TGC_ID, (unsigned long long)peer, group->channel_id, gc->active_conn_count);
|
||||
if (gc->active_conn_count == 0) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: DOWN no connections — cycle restart ch=%s", TGC_ID, group->channel_id);
|
||||
if (!tgc_goal_reached(gc)) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: DOWN goal not reached — cycle restart ch=%s", TGC_ID, group->channel_id);
|
||||
tgc_cycle_restart(gc);
|
||||
}
|
||||
}
|
||||
@@ -272,8 +334,8 @@ void topo_group_connect_on_down(struct TOPO_GROUP* group, struct ETCP_CONN* conn
|
||||
/**
|
||||
* Единый коллбэк на результат каждой попытки подключения (успех/таймаут/обрыв).
|
||||
* В Phase 1 — просто считает результаты (итог подводит tgc_phase1_timeout).
|
||||
* В Phase 2/3 — успех завершает цикл; провал закрывает handle (чтобы не оставить
|
||||
* залипшее состояние) и ставит паузу перед следующей попыткой.
|
||||
* В Phase 2/3 — успех проверяет цель: достигнута → done, иначе следующий кандидат;
|
||||
* провал закрывает handle (чтобы не оставить залипшее состояние) и ставит паузу.
|
||||
*/
|
||||
/* Передача владения: после topo_group_new_conn (который взял свой NCD-handle)
|
||||
* закрываем connect-фазу handle, чтобы владельцем conn остался один BGP-хэндл
|
||||
@@ -312,9 +374,14 @@ static void tgc_callback(struct NODE_CONN_DIRECT* h, enum ncd_event event, void*
|
||||
break;
|
||||
case TGC_PHASE_TWO:
|
||||
if (ok) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase2 connected to 0x%016llx — done", TGC_ID, (unsigned long long)node_id);
|
||||
gc->phase = TGC_PHASE_DONE;
|
||||
tgc_notify_connecting(gc, NULL, 0);
|
||||
if (tgc_goal_reached(gc)) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase2 connected to 0x%016llx — goal reached, done", TGC_ID, (unsigned long long)node_id);
|
||||
gc->phase = TGC_PHASE_DONE;
|
||||
tgc_notify_connecting(gc, NULL, 0);
|
||||
} else {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase2 connected to 0x%016llx — goal not reached, try next", TGC_ID, (unsigned long long)node_id);
|
||||
tgc_phase2_try_next(gc);
|
||||
}
|
||||
} else {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase2 0x%016llx FAIL — pause then try next", TGC_ID, (unsigned long long)node_id);
|
||||
if (gc->cur_handle) { node_conn_direct_close(gc->cur_handle); gc->cur_handle = NULL; }
|
||||
@@ -323,9 +390,14 @@ static void tgc_callback(struct NODE_CONN_DIRECT* h, enum ncd_event event, void*
|
||||
break;
|
||||
case TGC_PHASE_THREE:
|
||||
if (ok) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase3 connected to 0x%016llx — done", TGC_ID, (unsigned long long)node_id);
|
||||
gc->phase = TGC_PHASE_DONE;
|
||||
tgc_notify_connecting(gc, NULL, 0);
|
||||
if (tgc_goal_reached(gc)) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase3 connected to 0x%016llx — goal reached, done", TGC_ID, (unsigned long long)node_id);
|
||||
gc->phase = TGC_PHASE_DONE;
|
||||
tgc_notify_connecting(gc, NULL, 0);
|
||||
} else {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase3 connected to 0x%016llx — goal not reached, try next", TGC_ID, (unsigned long long)node_id);
|
||||
tgc_phase3_try_next(gc);
|
||||
}
|
||||
} else {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase3 0x%016llx FAIL — pause then try next", TGC_ID, (unsigned long long)node_id);
|
||||
if (gc->cur_handle) { node_conn_direct_close(gc->cur_handle); gc->cur_handle = NULL; }
|
||||
@@ -368,8 +440,8 @@ static void tgc_start_pause(struct TOPO_GROUP_CONNECT* gc, void (*cb)(void*), co
|
||||
|
||||
/**
|
||||
* Таймаут параллельных попыток Phase 1 (2 секунды). Закрывает неудачные
|
||||
* handles и сбрасывает connected в БД. Если хоть один узел подключился —
|
||||
* цикл завершён; иначе переходим к Phase 2.
|
||||
* handles и сбрасывает connected в БД. Если цель достигнута — цикл завершён;
|
||||
* иначе переходим к Phase 2 (дополнительные попытки).
|
||||
*/
|
||||
static void tgc_phase1_timeout(void* arg) {
|
||||
struct TOPO_GROUP_CONNECT* gc = (struct TOPO_GROUP_CONNECT*)arg;
|
||||
@@ -400,12 +472,13 @@ static void tgc_phase1_timeout(void* arg) {
|
||||
}
|
||||
gc->handle_count = remaining;
|
||||
|
||||
if (gc->connected_count > 0) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase1 done — %d connected, skipping Phase 2", TGC_ID, gc->connected_count);
|
||||
if (tgc_goal_reached(gc)) {
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase1 done — goal reached (%d connected), skipping Phase 2", TGC_ID, gc->connected_count);
|
||||
gc->phase = TGC_PHASE_DONE;
|
||||
tgc_notify_connecting(gc, NULL, 0);
|
||||
return;
|
||||
}
|
||||
DEBUG_INFO(DEBUG_CATEGORY_BGP, "%s: Phase1 done — goal not reached (%d connected) → Phase 2", TGC_ID, gc->connected_count);
|
||||
gc->phase = TGC_PHASE_TWO;
|
||||
tgc_phase2_try_next(gc);
|
||||
}
|
||||
|
||||
@@ -1,19 +1,23 @@
|
||||
/**
|
||||
* @file topo_group_connect.h
|
||||
* @brief Авто-подключение к узлам группы — бесконечный цикл до успеха.
|
||||
* @brief Авто-подключение к узлам группы — бесконечный цикл до цели.
|
||||
*
|
||||
* При старте CHAT-группы (topo_group_connect_init):
|
||||
* Phase 1 — параллельный запуск ко всем пирам с connected=1 (таймаут 2с).
|
||||
* Если хоть один подключился → done.
|
||||
* Если цель достигнута → done.
|
||||
* Phase 2 — последовательный перебор (supernode, затем public/EIM).
|
||||
* После каждого TIMEOUT — пауза перед следующей попыткой.
|
||||
* Phase 3 — последовательный перебор локальных/strict NAT.
|
||||
* После каждого TIMEOUT — пауза перед следующей попыткой.
|
||||
* После исчерпания Phase 3 → пауза → cycle_restart → Phase 1 (бесконечно).
|
||||
*
|
||||
* Цель (tgc_goal_reached): подключены хотя бы к одному суперузлу (node_type==4)
|
||||
* ЛИБО к 3+ не-мобильным клиентам (device_type != MOBILE, не суперузлам).
|
||||
* Пока цель не достигнута, успех НЕ останавливает цикл.
|
||||
*
|
||||
* Пауза: активный режим — 1с; фоновый (Android standby) — standby_wait
|
||||
* (burst → сразу, sleep → до следующего burst).
|
||||
* При DOWN (active_conn_count → 0) — немедленный cycle_restart из любой фазы.
|
||||
* При DOWN, если цель перестала выполняться — немедленный cycle_restart.
|
||||
* При повторном вызове topo_group_connect_init — destroy + fresh init.
|
||||
*
|
||||
* active_conn_count — количество уникальных peer-узлов с живыми соединениями.
|
||||
|
||||
@@ -376,6 +376,7 @@ int topo_node_serialize(struct TOPO_NODE* ni, struct TOPO_GROUP_NODE* nq,
|
||||
msg.client_type = ni->client_type;
|
||||
msg.client_activity = ni->client_activity;
|
||||
msg.cumulative_rtt = cumulative_rtt;
|
||||
if (nq->radio) msg.flags |= TOPO_FLAG_RADIO;
|
||||
|
||||
size_t total = TOPOMSG_NODE_HDR_SIZE + (size_t)topo_node_dyn_size(&msg);
|
||||
if (total > out_max) return -1;
|
||||
@@ -853,6 +854,7 @@ int topo_group_update_my_nodeinfo(struct UTUN_INSTANCE* instance, struct TOPO_GR
|
||||
|| (vc6 != (group->local_node->subnets ? topo_list_count((struct _topo_head*)group->local_node->subnets->v6_subnets) : 0))
|
||||
|| (oni->client_type != instance->client_type)
|
||||
|| (oni->client_activity != instance->client_activity)
|
||||
|| (group->local_node->radio != group->radio_active)
|
||||
|| (memcmp(oni->ed25519_public_key, group->ed25519_public_key, SC_PUBKEY_SIZE) != 0);
|
||||
}
|
||||
}
|
||||
@@ -879,6 +881,7 @@ int topo_group_update_my_nodeinfo(struct UTUN_INSTANCE* instance, struct TOPO_GR
|
||||
ni->ver = (old_ver % 255) + 1;
|
||||
ni->client_type = instance->client_type;
|
||||
ni->client_activity = instance->client_activity;
|
||||
lq->radio = group->radio_active;
|
||||
memcpy(ni->public_key, instance->my_keys.public_key, SC_PUBKEY_SIZE);
|
||||
memcpy(ni->ed25519_public_key, group->ed25519_public_key, SC_PUBKEY_SIZE);
|
||||
if (name_len) { ni->node_name = u_malloc(name_len + 1); if (ni->node_name) { memcpy(ni->node_name, instance->name, name_len); ni->node_name[name_len] = 0; } }
|
||||
|
||||
@@ -82,6 +82,7 @@ struct UTUN_INSTANCE;
|
||||
#define NCONN_BGP (1 << 2) // узел виден в топологии (BGP)
|
||||
|
||||
#define TOPO_FLAG_SEND_SUBNETS 0x01
|
||||
#define TOPO_FLAG_RADIO 0x02
|
||||
#define TOPO_GROUP_UTUN 0x8000000000000000ULL
|
||||
#define TOPO_GROUP_TYPE_UTUN 1
|
||||
#define TOPO_GROUP_TYPE_CHAT 2
|
||||
@@ -211,6 +212,7 @@ struct TOPO_GROUP_NODE {
|
||||
struct TOPO_CONNECTIVITY connectivity; // результаты ping-проб (interface/nat/real)
|
||||
uint8_t conn_presence; // NODE_CONN_* — какие подключения есть в принципе
|
||||
uint8_t conn_up; // NODE_CONN_* — какие из них подняты
|
||||
uint8_t radio; // 1 = узел слушает рацию канала (TOPO_FLAG_RADIO)
|
||||
struct NODE_CONN_DIRECT* handle; // активный ncd-handle прямого соединения к узлу
|
||||
};
|
||||
|
||||
|
||||
@@ -1237,3 +1237,17 @@ int topo_node_sqlite_get_local_peers(sqlite3* db, const char* channel_id,
|
||||
*out_ids = ids; *out_count = cnt;
|
||||
return 0;
|
||||
}
|
||||
|
||||
int topo_node_sqlite_get_node_type(sqlite3* db, const char* channel_id, uint64_t node_id) {
|
||||
if (!db || !channel_id || !node_id) return -1;
|
||||
char peers_tbl[80]; peers_table_name(channel_id, peers_tbl, sizeof(peers_tbl));
|
||||
char sql[256];
|
||||
snprintf(sql, sizeof(sql), "SELECT node_type FROM \"%s\" WHERE node_id=?", peers_tbl);
|
||||
sqlite3_stmt* st = NULL;
|
||||
if (sqlite3_prepare_v2(db, sql, -1, &st, NULL) != SQLITE_OK) return -1;
|
||||
sqlite3_bind_int64(st, 1, (sqlite3_int64)node_id);
|
||||
int node_type = -1;
|
||||
if (sqlite3_step(st) == SQLITE_ROW) node_type = sqlite3_column_int(st, 0);
|
||||
sqlite3_finalize(st);
|
||||
return node_type;
|
||||
}
|
||||
@@ -112,4 +112,7 @@ int topo_node_sqlite_get_public_peers(sqlite3* db, const char* channel_id, uint6
|
||||
int topo_node_sqlite_get_local_peers(sqlite3* db, const char* channel_id, uint64_t** out_ids, int* out_count, uint64_t self_node_id);
|
||||
int topo_node_sqlite_get_supernode_peers(sqlite3* db, const char* channel_id, uint64_t** out_ids, int* out_count);
|
||||
|
||||
/* node_type узла в канале (peers_<ch>.node_type); -1 если нет записи/ошибка. */
|
||||
int topo_node_sqlite_get_node_type(sqlite3* db, const char* channel_id, uint64_t node_id);
|
||||
|
||||
#endif
|
||||
@@ -44,8 +44,8 @@ extern "C" {
|
||||
// (raw-пространство) — помечены «(raw)»; их значения занимают raw-реестр.
|
||||
//
|
||||
// raw: 0x00 DATA, 0x01 TOPO_ENTRY, 0x02 NAT_DETECTION, 0x03 SVC_ROUTE,
|
||||
// 0x04 NTP_TIME, 0x05 BROADCAST, 0x12 NCD_CONTROL, 0x13 GROUP_INVITE,
|
||||
// 0x20 DB_SYNC, 0x30 CHAT_SYNC, 0x31 MEMBER_SYNC
|
||||
// 0x04 NTP_TIME, 0x05 BROADCAST, 0x06 RADIO, 0x12 NCD_CONTROL,
|
||||
// 0x13 GROUP_INVITE, 0x20 DB_SYNC, 0x30 CHAT_SYNC, 0x31 MEMBER_SYNC
|
||||
// router: 0x00 DATA, 0x02 NAT, 0x04 TCP_PROXY_SERVER, 0x05 UDP_PROXY,
|
||||
// 0x06 ICMP_PROXY, 0x07 MEDIA_DELIVERY, 0x08 TCP_PROXY_CLIENT,
|
||||
// 0x11 CONN_MGR, 0x32 ADMIN_KEY, 0x33 JOIN, 0x34 DM,
|
||||
@@ -62,6 +62,7 @@ extern "C" {
|
||||
#define ETCP_ID_SVC_ROUTE 0x03 // транспорт роутера (etcp_router)
|
||||
#define ETCP_ID_NTP_TIME 0x04 // NTP time sync P2P
|
||||
#define ETCP_ID_BROADCAST 0x05 // broadcast-сообщения между пирами
|
||||
#define ETCP_ID_RADIO 0x06 // рация (PTT мультикаст голоса) между пирами
|
||||
|
||||
// === Router-сервисы (etcp_router_bind/etcp_route_send) ===
|
||||
#define ETCP_RT_ID_DATA 0x00 // routing.c — маршрутизация данных
|
||||
|
||||
+20
-1
@@ -23,6 +23,8 @@
|
||||
#include "call/call.h"
|
||||
#include "call/call_headless.h"
|
||||
#include "broadcast.h"
|
||||
#include "radio/radio.h"
|
||||
#include "radio/radio_headless.h"
|
||||
#include "stcp_server.h"
|
||||
#include "control_server.h"
|
||||
#include "transport_layer/node_conn_direct.h"
|
||||
@@ -465,7 +467,8 @@ void utun_instance_destroy(struct UTUN_INSTANCE *instance) {
|
||||
topo_groups_destroy(instance);
|
||||
}
|
||||
broadcast_destroy_instance(instance);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[DESTROY] F done — BGP + broadcast");
|
||||
radio_destroy_instance(instance);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[DESTROY] F done — BGP + broadcast + radio");
|
||||
|
||||
/* Phase F2: NAT transport — отвязываем и закрываем nat_tun ДО разборки
|
||||
сокетов/соединений, чтобы nat_transport_etcp_recv_cb не был вызван с
|
||||
@@ -497,6 +500,7 @@ void utun_instance_destroy(struct UTUN_INSTANCE *instance) {
|
||||
dm_core_destroy(instance);
|
||||
call_headless_destroy(instance);
|
||||
call_destroy(instance);
|
||||
radio_headless_destroy(instance);
|
||||
chat_sync_destroy(instance);
|
||||
chat_core_destroy(instance);
|
||||
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[DESTROY] H done — chat");
|
||||
@@ -780,6 +784,20 @@ int utun_instance_init(struct UTUN_INSTANCE *instance) {
|
||||
if (call_headless_init(instance->ua, instance, ip_str, port) != 0)
|
||||
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "Failed to init headless call audio socket (non-fatal)");
|
||||
}
|
||||
|
||||
// Initialize headless radio audio socket if configured in [chatserver]
|
||||
if (instance->config->global.radio_audio_bind[0] != '\0') {
|
||||
char ip_str[64] = "127.0.0.1"; int port = 10002;
|
||||
const char* colon = strrchr(instance->config->global.radio_audio_bind, ':');
|
||||
if (colon) {
|
||||
size_t ip_len = (size_t)(colon - instance->config->global.radio_audio_bind);
|
||||
if (ip_len < sizeof(ip_str)) { memcpy(ip_str, instance->config->global.radio_audio_bind, ip_len); ip_str[ip_len] = '\0'; }
|
||||
port = atoi(colon + 1);
|
||||
if (port <= 0 || port > 65535) port = 10002;
|
||||
}
|
||||
if (radio_headless_init(instance->ua, instance, ip_str, port) != 0)
|
||||
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "Failed to init headless radio audio socket (non-fatal)");
|
||||
}
|
||||
|
||||
// Start the main loop
|
||||
instance->running = 1;
|
||||
@@ -795,6 +813,7 @@ int utun_instance_init(struct UTUN_INSTANCE *instance) {
|
||||
}
|
||||
|
||||
broadcast_init_instance(instance);
|
||||
radio_init_instance(instance);
|
||||
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "Connections initialized successfully");
|
||||
return 0;
|
||||
|
||||
@@ -40,6 +40,8 @@ struct join_key_entry; /* chat_join.c */
|
||||
struct dm_state;
|
||||
struct dm_mb_state;
|
||||
struct call_ctx; /* call/call.c */
|
||||
struct radio_instance; /* radio/radio.c */
|
||||
struct radio_headless; /* radio/radio_headless.c */
|
||||
struct uasync_s;
|
||||
struct ROUTE_TABLE;
|
||||
struct ETCP_CONN;
|
||||
@@ -202,6 +204,8 @@ struct UTUN_INSTANCE {
|
||||
struct dm_mb_state* dm_mailbox; // dm/dm_mailbox.c (was g_mb)
|
||||
struct call_ctx* call; // call/call.c — P2P audio call
|
||||
void* call_audio; // call/call_headless.c — audio stream socket
|
||||
struct radio_instance* radio; // radio/radio.c — PTT walkie-talkie (frame cb + counters)
|
||||
struct radio_headless* radio_headless; // radio/radio_headless.c — audio stream socket
|
||||
struct chat_setting_state chat_settings; // chat_setting.c (was globals)
|
||||
chat_event_handler_fn chat_event_handler; // chat_event.c (was g_handler)
|
||||
void* headless; // chat_headless_control.c (was g_hc)
|
||||
|
||||
Reference in new issue
Block a user