1052 lines
50 KiB
C
1052 lines
50 KiB
C
// call.c — P2P аудио-звонок: сигналинг + медиа (per-instance)
|
||
//
|
||
// Полный цикл: INVITE → RINGING → ACCEPT → conn_mgr (прямой линк) → ACTIVE →
|
||
// медиа (CALL_MEDIA) → HANGUP/ENDED. Звонок из CHAT-группы или DM — модуль
|
||
// оперирует (group_id, peer_node_id), маршрутизация через etcp_router.
|
||
//
|
||
// Сигналинг идёт с ROUTE_CRYPTO_SIGN|ROUTE_CRYPTO_ENCRYPT, медиа — без подписи.
|
||
// Реестр сессий: ll_queue с hash-индексом по call_id (8 байт).
|
||
//
|
||
// Thread-модель: всё в uasync-потоке (single-threaded). Освобождение сессии
|
||
// отложено через uasync_call_soon, чтобы не было UAF при завершении из
|
||
// коллбэков conn_mgr/таймеров.
|
||
|
||
#include "call.h"
|
||
#include "call_proto.h"
|
||
#include "../dm/dm_core.h"
|
||
#include "../utun_instance.h"
|
||
#include "../chat/chat_event.h"
|
||
#include "../routing_layer/etcp_router.h"
|
||
#include "../routing_layer/route_crypto.h"
|
||
#include "../routing_layer/conn_mgr.h"
|
||
#include "../routing_layer/topo_group.h"
|
||
#include "../routing_layer/topo_node.h"
|
||
#include "../transport_layer/etcp_api.h"
|
||
#include "../transport_layer/etcp.h"
|
||
#include "../../lib/mem.h"
|
||
#include "../../lib/ll_queue.h"
|
||
#include "../../lib/debug_config.h"
|
||
#include "../../lib/u_async.h"
|
||
#include "../../lib/platform_compat.h"
|
||
|
||
#include <string.h>
|
||
#include <stdlib.h>
|
||
#include <stdio.h>
|
||
|
||
#define CALL_ID "call"
|
||
|
||
enum call_state {
|
||
CALL_ST_IDLE = 0,
|
||
CALL_ST_OUTGOING, /* caller: INVITE отправлен */
|
||
CALL_ST_RINGING, /* caller: получил RINGING; callee: отправил RINGING */
|
||
CALL_ST_ACTIVE, /* вызов принят; доставка через роутер независимо от готовности CM */
|
||
CALL_ST_ENDED,
|
||
};
|
||
|
||
enum call_role { CALL_ROLE_CALLER = 0, CALL_ROLE_CALLEE };
|
||
|
||
struct call_session {
|
||
struct ll_entry ll; /* dummy-заголовок; структура хранится в entry->data */
|
||
uint64_t call_id; /* hash-индекс (8 байт) */
|
||
uint64_t group_id;
|
||
uint64_t peer_node_id;
|
||
uint8_t role; /* CALL_ROLE_* */
|
||
uint8_t state; /* CALL_ST_* */
|
||
uint8_t reason; /* CALL_REASON_* (при ENDED) */
|
||
struct call_ctx* ctx; /* обратная ссылка (для коллбэков conn_mgr/таймеров) */
|
||
struct CONN_MGR_HANDLE* cm_handle;
|
||
void* ring_timer; /* caller: 45с ожидания ответа; callee: 2с watchdog кодограмм */
|
||
void* invite_timer; /* caller: кодограмма INVITE каждые 300мс (OUTGOING+RINGING) */
|
||
uint8_t no_route_strikes; /* caller: подряд ретрансмитов без маршрута (fail-fast) */
|
||
void* traffic_timer; /* ACTIVE: 20с без медиа (запускается по первому медиа) */
|
||
void* teardown_timer; /* отложенный разрыв conn_mgr (500мс после завершения) */
|
||
void* stats_timer; /* Раз в 1с — путь подключения; в ACTIVE также CALL_STATS. */
|
||
uint8_t connection_reported, connection_type;
|
||
uint64_t connection_next_hop;
|
||
uint64_t last_media_tb; /* время последнего CALL_MEDIA (0.1ms) */
|
||
uint16_t tx_media_seq; /* seq следующего исходящего CALL_MEDIA */
|
||
struct ll_queue* tx_q; /* FIFO закодированных CALL_MEDIA (backpressure) */
|
||
struct queue_waiter_handle send_waiter; /* waiter на send_q роутера */
|
||
uint32_t c_tx_frames; /* отправлено медиа-кадров */
|
||
uint32_t c_rx_frames; /* принято медиа-кадров */
|
||
uint32_t c_tx_dropped; /* дропнуто при переполнении tx_q */
|
||
uint32_t c_rx_dropped; /* дропнуто при переполнении jitter-буфера */
|
||
};
|
||
|
||
struct call_ctx {
|
||
struct UTUN_INSTANCE* inst;
|
||
int initialized;
|
||
struct ll_queue* sessions; /* struct call_session, hash по call_id */
|
||
uint32_t c_calls_started; /* суммарно начато звонков */
|
||
uint32_t c_calls_ended; /* суммарно завершено звонков */
|
||
call_media_recv_fn media_recv_fn; /* приём Opus-кадров пира */
|
||
void* media_recv_arg;
|
||
const struct call_audio_ops* audio_ops; /* аудио-движок (GUI-сборки), NULL в демоне */
|
||
};
|
||
|
||
/* ───────────────────────────────────────────────────────────────────────── */
|
||
|
||
static struct call_ctx* call_of(struct UTUN_INSTANCE* inst) {
|
||
return inst ? inst->call : NULL;
|
||
}
|
||
|
||
static const char* call_state_name(uint8_t st) {
|
||
switch (st) {
|
||
case CALL_ST_IDLE: return "IDLE";
|
||
case CALL_ST_OUTGOING: return "OUTGOING";
|
||
case CALL_ST_RINGING: return "RINGING";
|
||
case CALL_ST_ACTIVE: return "ACTIVE";
|
||
case CALL_ST_ENDED: return "ENDED";
|
||
default: return "?";
|
||
}
|
||
}
|
||
|
||
static const char* call_subcmd_name(uint8_t sc) {
|
||
switch (sc) {
|
||
case CALL_SUBCMD_INVITE: return "INVITE";
|
||
case CALL_SUBCMD_RINGING: return "RINGING";
|
||
case CALL_SUBCMD_ACCEPT: return "ACCEPT";
|
||
case CALL_SUBCMD_DECLINE: return "DECLINE";
|
||
case CALL_SUBCMD_BUSY: return "BUSY";
|
||
case CALL_SUBCMD_CANCEL: return "CANCEL";
|
||
case CALL_SUBCMD_HANGUP: return "HANGUP";
|
||
case CALL_SUBCMD_MEDIA: return "MEDIA";
|
||
default: return "?";
|
||
}
|
||
}
|
||
|
||
/* ─────────────────── отправка ─────────────────── */
|
||
|
||
static void call_stats_timer_cb(void* arg);
|
||
|
||
/* data[0] = subcmd. etcp_route_send забирает владение entry. */
|
||
static int call_send(struct call_ctx* ctx, uint64_t group_id, uint64_t dst,
|
||
const uint8_t* data, size_t len, int mode) {
|
||
struct ll_entry* e = queue_entry_new(0);
|
||
if (!e) return -1;
|
||
e->dgram = u_malloc(1 + len);
|
||
if (!e->dgram) { queue_entry_free(e); return -1; }
|
||
e->dgram[0] = ETCP_RT_ID_CALL;
|
||
if (len) memcpy(e->dgram + 1, data, len);
|
||
e->len = (uint16_t)(1 + len);
|
||
int rc = etcp_route_send(ctx->inst, group_id, dst, e, 1, mode);
|
||
if (rc != 0) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: send FAIL rc=%d subcmd=%s(0x%02x) dst=0x%016llx group=0x%016llx",
|
||
CALL_ID, rc, call_subcmd_name(len ? data[0] : 0), len ? data[0] : 0,
|
||
(unsigned long long)dst, (unsigned long long)group_id);
|
||
}
|
||
return rc;
|
||
}
|
||
|
||
/* Собрать и отправить INVITE для сессии вызывающего. */
|
||
static int call_send_invite(struct call_ctx* ctx, struct call_session* s) {
|
||
uint8_t buf[sizeof(struct call_pkt_invite)];
|
||
struct call_pkt_invite* inv = (struct call_pkt_invite*)buf;
|
||
inv->subcmd = CALL_SUBCMD_INVITE;
|
||
inv->call_id = s->call_id;
|
||
inv->group_id = s->group_id;
|
||
inv->num_tracks = 1;
|
||
inv->tracks[0].type = CALL_TRACK_AUDIO;
|
||
inv->tracks[0].codec = CALL_CODEC_OPUS;
|
||
inv->tracks[0].sample_rate = CALL_OPUS_SAMPLE_RATE;
|
||
inv->tracks[0].channels = 1;
|
||
inv->tracks[0].bitrate = 32000;
|
||
return call_send(ctx, s->group_id, s->peer_node_id, buf, sizeof(buf), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
|
||
}
|
||
|
||
/* ─────────────────── реестр сессий ─────────────────── */
|
||
|
||
static struct call_session* call_session_find(struct call_ctx* ctx, uint64_t call_id) {
|
||
if (!ctx || !ctx->sessions) return NULL;
|
||
struct ll_entry* e = queue_find_data_by_index(ctx->sessions, &call_id);
|
||
return e ? (struct call_session*)e->data : NULL;
|
||
}
|
||
|
||
static struct call_session* call_session_new(struct call_ctx* ctx, uint64_t call_id,
|
||
uint64_t group_id, uint64_t peer, uint8_t role) {
|
||
if (!ctx || !ctx->sessions) return NULL;
|
||
struct ll_entry* qe = queue_entry_new(sizeof(struct call_session));
|
||
if (!qe) return NULL;
|
||
struct call_session* s = (struct call_session*)qe->data;
|
||
memset(s, 0, sizeof(*s));
|
||
s->call_id = call_id;
|
||
s->group_id = group_id;
|
||
s->peer_node_id = peer;
|
||
s->role = role;
|
||
s->state = CALL_ST_IDLE;
|
||
s->ctx = ctx;
|
||
s->tx_q = queue_new(ctx->inst->ua, 0, 0, 0, "call_txq");
|
||
if (!s->tx_q) { queue_entry_free(qe); return NULL; }
|
||
// При провале queue_data_put_with_index сам освобождает qe (вместе с s), поэтому
|
||
// сохраняем tx_q заранее: queue_entry_free(qe) повторно дал бы double-free.
|
||
struct ll_queue* tx_q = s->tx_q;
|
||
if (queue_data_put_with_index(ctx->sessions, qe) != 0) {
|
||
queue_free(tx_q);
|
||
return NULL;
|
||
}
|
||
return s;
|
||
}
|
||
|
||
static void call_session_free_cb(void* arg) {
|
||
struct ll_entry* e = (struct ll_entry*)arg;
|
||
queue_entry_free(e);
|
||
}
|
||
|
||
/* Освободить медиа-объекты сессии: отменить waiter на send_q роутера и
|
||
* дренировать tx_q (без отправки). Безопасно вызывать из uasync-потока. */
|
||
static void call_session_cleanup(struct call_ctx* ctx, struct call_session* s) {
|
||
etcp_router_cancel_send_ready(ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL, &s->send_waiter);
|
||
if (s->tx_q) {
|
||
struct ll_entry* e;
|
||
while ((e = queue_data_get(s->tx_q)) != NULL) {
|
||
queue_dgram_free(e);
|
||
queue_entry_free(e);
|
||
}
|
||
queue_free(s->tx_q);
|
||
s->tx_q = NULL;
|
||
}
|
||
}
|
||
|
||
/* Убрать сессию из реестра и освободить отложенно (не из коллбэков conn_mgr/таймеров). */
|
||
static void call_session_remove_and_free(struct call_ctx* ctx, struct call_session* s) {
|
||
call_session_cleanup(ctx, s);
|
||
struct ll_entry* e = queue_find_data_by_index(ctx->sessions, &s->call_id);
|
||
if (e) {
|
||
queue_remove_data(ctx->sessions, e);
|
||
uasync_call_soon(ctx->inst->ua, e, call_session_free_cb);
|
||
}
|
||
}
|
||
|
||
/* ─────────────────── события в GUI ─────────────────── */
|
||
|
||
static void call_event_id(struct call_ctx* ctx, int type, uint64_t call_id, uint8_t reason) {
|
||
uint8_t buf[16];
|
||
size_t off = 0;
|
||
memcpy(buf + off, &call_id, 8); off += 8;
|
||
buf[off++] = reason;
|
||
chat_event_post(ctx->inst, type, buf, (int)off);
|
||
}
|
||
|
||
static void call_event_incoming(struct call_ctx* ctx, uint64_t group_id, uint64_t caller, uint64_t call_id) {
|
||
char ch[32];
|
||
snprintf(ch, sizeof(ch), "%llu", (unsigned long long)group_id);
|
||
size_t ch_len = strlen(ch);
|
||
uint8_t buf[1 + 32 + 8 + 8];
|
||
size_t off = 0;
|
||
buf[off++] = (uint8_t)ch_len;
|
||
memcpy(buf + off, ch, ch_len); off += ch_len;
|
||
memcpy(buf + off, &caller, 8); off += 8;
|
||
memcpy(buf + off, &call_id, 8); off += 8;
|
||
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) {
|
||
off = snprintf(text, sizeof(text), "BGP: нет маршрута");
|
||
} else if (hc == 1) {
|
||
off = snprintf(text, sizeof(text), "BGP: прямой");
|
||
} else {
|
||
off = snprintf(text, sizeof(text), "BGP: через ");
|
||
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);
|
||
}
|
||
|
||
/* Тот же выбор next hop, что для CALL_MEDIA, включая BGP fallback до готовности CM. */
|
||
static void call_post_connection(struct call_session* s) {
|
||
struct ETCP_ROUTER_PATH path = {0};
|
||
if (s->state != CALL_ST_ENDED)
|
||
path = etcp_router_get_path(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL);
|
||
if (s->connection_reported && s->connection_type == path.type && s->connection_next_hop == path.next_hop_node_id) return;
|
||
s->connection_reported = 1; s->connection_type = path.type; s->connection_next_hop = path.next_hop_node_id;
|
||
uint8_t buf[17];
|
||
memcpy(buf, &s->call_id, 8); buf[8] = path.type; memcpy(buf + 9, &path.next_hop_node_id, 8);
|
||
static const char* names[] = { "NONE", "DIRECT", "REVERSE", "INDIRECT" };
|
||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: connection id=%016llx peer=%016llx type=%s via=%016llx",
|
||
CALL_ID, (unsigned long long)s->call_id, (unsigned long long)s->peer_node_id, names[path.type],
|
||
(unsigned long long)path.next_hop_node_id);
|
||
chat_event_post(s->ctx->inst, CHAT_EVT_CALL_CONNECTION, buf, sizeof(buf));
|
||
if (s->state != CALL_ST_ENDED) call_post_bgp_path(s->ctx, s);
|
||
}
|
||
|
||
static void call_watch_connection(struct call_session* s) {
|
||
call_post_connection(s);
|
||
if (s->state == CALL_ST_ENDED || s->stats_timer) return;
|
||
s->stats_timer = uasync_set_timeout(s->ctx->inst->ua, 10000, s, call_stats_timer_cb, "call_stats");
|
||
if (!s->stats_timer) DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: connection timer failed id=%016llx",
|
||
CALL_ID, (unsigned long long)s->call_id);
|
||
}
|
||
|
||
/* ─────────────────── завершение ─────────────────── */
|
||
|
||
/* Отложенный разрыв: единственная точка, где закрывается conn_mgr (после того,
|
||
* как финальный сигналинг — HANGUP/DECLINE — гарантированно долетел). */
|
||
static void call_teardown_timer_cb(void* arg) {
|
||
struct call_session* s = (struct call_session*)arg;
|
||
if (!s || !s->ctx) return;
|
||
struct call_ctx* ctx = s->ctx;
|
||
s->teardown_timer = NULL;
|
||
if (s->cm_handle) { conn_mgr_close(s->cm_handle); s->cm_handle = NULL; }
|
||
call_session_remove_and_free(ctx, s);
|
||
}
|
||
|
||
static void call_end(struct call_ctx* ctx, struct call_session* s, uint8_t reason) {
|
||
if (!ctx || !s || s->state == CALL_ST_ENDED) return;
|
||
|
||
s->state = CALL_ST_ENDED;
|
||
s->reason = reason;
|
||
ctx->c_calls_ended++;
|
||
|
||
if (s->ring_timer) { uasync_cancel_timeout(ctx->inst->ua, s->ring_timer); s->ring_timer = NULL; }
|
||
if (s->invite_timer) { uasync_cancel_timeout(ctx->inst->ua, s->invite_timer); s->invite_timer = NULL; }
|
||
if (s->traffic_timer) { uasync_cancel_timeout(ctx->inst->ua, s->traffic_timer); s->traffic_timer = NULL; }
|
||
if (s->stats_timer) { uasync_cancel_timeout(ctx->inst->ua, s->stats_timer); s->stats_timer = NULL; }
|
||
|
||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: call ENDED id=%016llx reason=%d role=%s rx=%u tx=%u tx_dropped=%u",
|
||
CALL_ID, (unsigned long long)s->call_id, reason,
|
||
s->role == CALL_ROLE_CALLER ? "caller" : "callee",
|
||
s->c_rx_frames, s->c_tx_frames, s->c_tx_dropped);
|
||
|
||
call_post_connection(s);
|
||
call_event_id(ctx, CHAT_EVT_CALL_ENDED, s->call_id, reason);
|
||
|
||
/* Закрытие запроса CM откладываем на 500мс для отправки финального сигналинга. */
|
||
if (s->cm_handle) {
|
||
s->teardown_timer = uasync_set_timeout(ctx->inst->ua, CALL_TEARDOWN_DELAY_TB, s, call_teardown_timer_cb, "call_teardown");
|
||
} else {
|
||
call_session_remove_and_free(ctx, s);
|
||
}
|
||
}
|
||
|
||
/* ─────────────────── conn_mgr ─────────────────── */
|
||
|
||
static void call_traffic_timeout_cb(void* arg);
|
||
|
||
/* CM готовит оптимальный путь; звонок всегда отправляет через роутер с автоматическим fallback. */
|
||
static void call_conn_cb(struct CONN_MGR_HANDLE* h, uint64_t node_id, uint64_t group_id,
|
||
enum conn_mgr_event event, void* arg) {
|
||
struct call_session* s = arg;
|
||
if (!s || s->state == CALL_ST_ENDED) return;
|
||
if (event == CONN_EVENT_UP) {
|
||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: CM path ready id=%016llx peer=%016llx", CALL_ID,
|
||
(unsigned long long)s->call_id, (unsigned long long)node_id);
|
||
} else {
|
||
DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: CM event=%d id=%016llx peer=%016llx; delivery continues via router",
|
||
CALL_ID, event, (unsigned long long)s->call_id, (unsigned long long)node_id);
|
||
}
|
||
call_watch_connection(s);
|
||
(void)h; (void)group_id;
|
||
}
|
||
|
||
static void call_open_cm(struct call_session* s) {
|
||
if (s->cm_handle) return;
|
||
if (conn_mgr_open(s->ctx->inst, s->group_id, s->peer_node_id, call_conn_cb, s, &s->cm_handle) != 0)
|
||
DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: CM request failed id=%016llx; using existing routing", CALL_ID,
|
||
(unsigned long long)s->call_id);
|
||
}
|
||
|
||
static void call_activate(struct call_session* s) {
|
||
if (s->state == CALL_ST_ACTIVE || s->state == CALL_ST_ENDED) return;
|
||
struct call_ctx* ctx = s->ctx;
|
||
s->state = CALL_ST_ACTIVE;
|
||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: accepted id=%016llx -> ACTIVE (router delivery)", CALL_ID,
|
||
(unsigned long long)s->call_id);
|
||
call_event_id(ctx, CHAT_EVT_CALL_ACCEPTED, s->call_id, 0);
|
||
call_watch_connection(s);
|
||
s->traffic_timer = uasync_set_timeout(ctx->inst->ua, CALL_NO_TRAFFIC_TB, s, call_traffic_timeout_cb, "call_traffic");
|
||
}
|
||
|
||
/* ─────────────────── таймеры ─────────────────── */
|
||
|
||
static void call_ring_timeout_cb(void* arg) {
|
||
struct call_session* s = (struct call_session*)arg;
|
||
if (!s || !s->ctx) return;
|
||
struct call_ctx* ctx = s->ctx;
|
||
s->ring_timer = NULL;
|
||
if (s->state == CALL_ST_ENDED) return;
|
||
|
||
DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: ring timeout id=%016llx -> CANCEL",
|
||
CALL_ID, (unsigned long long)s->call_id);
|
||
uint8_t pkt[sizeof(struct call_pkt_cancel)];
|
||
struct call_pkt_cancel* c = (struct call_pkt_cancel*)pkt;
|
||
c->subcmd = CALL_SUBCMD_CANCEL;
|
||
c->call_id = s->call_id;
|
||
call_send(ctx, s->group_id, s->peer_node_id, pkt, sizeof(pkt), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
|
||
call_end(ctx, s, CALL_REASON_RING_TIMEOUT);
|
||
}
|
||
|
||
/* Периодическая кодограмма INVITE (caller) каждые 300мс на всём протяжении вызова
|
||
* (OUTGOING + RINGING) — «сердцебиение»: покрывает потерю INVITE/RINGING, держит
|
||
* маршрут и кормит watchdog вызываемого (2с без кодограммы → вызов сорван).
|
||
* При стойком отсутствии маршрута — fail-fast CONNECT_FAIL. */
|
||
static void call_invite_retry_cb(void* arg) {
|
||
struct call_session* s = (struct call_session*)arg;
|
||
if (!s || !s->ctx) return;
|
||
struct call_ctx* ctx = s->ctx;
|
||
s->invite_timer = NULL;
|
||
if (s->state == CALL_ST_ENDED) return;
|
||
if (s->state != CALL_ST_OUTGOING && s->state != CALL_ST_RINGING) return; /* ACTIVE — звонок идёт */
|
||
|
||
if (etcp_router_has_route(ctx->inst, s->group_id, s->peer_node_id)) {
|
||
s->no_route_strikes = 0;
|
||
DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "%s: INVITE codegram id=%016llx (route ok)", CALL_ID, (unsigned long long)s->call_id);
|
||
call_send_invite(ctx, s);
|
||
} else if (++s->no_route_strikes >= CALL_NO_ROUTE_MAX_STRIKES) {
|
||
DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: no route after %d strikes id=%016llx -> CONNECT_FAIL",
|
||
CALL_ID, s->no_route_strikes, (unsigned long long)s->call_id);
|
||
uint8_t buf[9];
|
||
memcpy(buf, &s->call_id, 8);
|
||
buf[8] = 2; /* err: peer unreachable (no route) */
|
||
chat_event_post(ctx->inst, CHAT_EVT_CALL_ERROR, buf, sizeof(buf));
|
||
call_end(ctx, s, CALL_REASON_CONNECT_FAIL);
|
||
return;
|
||
} else {
|
||
DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "%s: INVITE codegram id=%016llx no route (strike %d/%d)",
|
||
CALL_ID, (unsigned long long)s->call_id, s->no_route_strikes, CALL_NO_ROUTE_MAX_STRIKES);
|
||
}
|
||
|
||
s->invite_timer = uasync_set_timeout(ctx->inst->ua, CALL_INVITE_RETRY_TB, s, call_invite_retry_cb, "call_invite_retry");
|
||
}
|
||
|
||
/* Вызываемый: 2с без кодограммы INVITE → вызывающий пропал (отмена потерялась) — закрываем. */
|
||
static void call_ring_dead_cb(void* arg) {
|
||
struct call_session* s = (struct call_session*)arg;
|
||
if (!s || !s->ctx) return;
|
||
struct call_ctx* ctx = s->ctx;
|
||
s->ring_timer = NULL;
|
||
if (s->state != CALL_ST_RINGING) return;
|
||
|
||
DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: ring dead (2с без кодограммы) id=%016llx -> end",
|
||
CALL_ID, (unsigned long long)s->call_id);
|
||
call_end(ctx, s, CALL_REASON_CANCEL);
|
||
}
|
||
|
||
static void call_traffic_timeout_cb(void* arg) {
|
||
struct call_session* s = (struct call_session*)arg;
|
||
if (!s || !s->ctx) return;
|
||
struct call_ctx* ctx = s->ctx;
|
||
s->traffic_timer = NULL;
|
||
if (s->state == CALL_ST_ENDED) return;
|
||
|
||
DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: no traffic 20s id=%016llx -> HANGUP",
|
||
CALL_ID, (unsigned long long)s->call_id);
|
||
uint8_t pkt[sizeof(struct call_pkt_hangup)];
|
||
struct call_pkt_hangup* hu = (struct call_pkt_hangup*)pkt;
|
||
hu->subcmd = CALL_SUBCMD_HANGUP;
|
||
hu->call_id = s->call_id;
|
||
hu->reason = CALL_REASON_NO_TRAFFIC;
|
||
call_send(ctx, s->group_id, s->peer_node_id, pkt, sizeof(pkt), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
|
||
call_end(ctx, s, CALL_REASON_NO_TRAFFIC);
|
||
}
|
||
|
||
/* Периодическое событие CALL_STATS (раз в 1с): rtt_ms, buffer_ms, tempo_x100. */
|
||
static void call_post_stats(struct call_session* s) {
|
||
struct call_ctx* ctx = s->ctx;
|
||
|
||
int buffer_ms = 0, tempo_x100 = 100;
|
||
struct call_jitter_range range = {0, 0, 60};
|
||
uint32_t dropped = 0, underruns = 0;
|
||
if (ctx->audio_ops && ctx->audio_ops->get_stats)
|
||
ctx->audio_ops->get_stats(s->call_id, &buffer_ms, &tempo_x100, &dropped, &underruns, &range);
|
||
|
||
uint16_t rtt_tb = 0;
|
||
struct ETCP_ROUTER_CONN* rconn =
|
||
etcp_router_conn_get(ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL);
|
||
if (rconn && rconn->recv_conn) rtt_tb = rconn->recv_conn->rtt_last;
|
||
uint16_t rtt = rtt_tb / 10; /* ETCP хранит RTT в 0.1мс; CALL_STATS передаёт миллисекунды. */
|
||
DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "%s: stats id=%016llx recv_peer=%016llx rtt_tb=%u rtt_ms=%u buffer=%dms tempo=%d dropped=%u underruns=%u",
|
||
CALL_ID, (unsigned long long)s->call_id,
|
||
(unsigned long long)(rconn && rconn->recv_conn ? rconn->recv_conn->peer_node_id : 0),
|
||
rtt_tb, rtt, buffer_ms, tempo_x100, dropped, underruns);
|
||
|
||
uint8_t buf[24];
|
||
size_t off = 0;
|
||
memcpy(buf + off, &s->call_id, 8); off += 8;
|
||
buf[off++] = (uint8_t)(rtt >> 8); buf[off++] = (uint8_t)(rtt & 0xFF);
|
||
uint16_t bms = (uint16_t)(buffer_ms < 0 ? 0 : (buffer_ms > 0xFFFF ? 0xFFFF : buffer_ms));
|
||
buf[off++] = (uint8_t)(bms >> 8); buf[off++] = (uint8_t)(bms & 0xFF);
|
||
uint16_t t = (uint16_t)(tempo_x100 < 0 ? 0 : (tempo_x100 > 0xFFFF ? 0xFFFF : tempo_x100));
|
||
buf[off++] = (uint8_t)(t >> 8); buf[off++] = (uint8_t)(t & 0xFF);
|
||
uint16_t dr = (uint16_t)(dropped > 0xFFFF ? 0xFFFF : dropped);
|
||
buf[off++] = (uint8_t)(dr >> 8); buf[off++] = (uint8_t)(dr & 0xFF);
|
||
uint16_t un = (uint16_t)(underruns > 0xFFFF ? 0xFFFF : underruns);
|
||
buf[off++] = (uint8_t)(un >> 8); buf[off++] = (uint8_t)(un & 0xFF);
|
||
int values[] = {range.min_ms, range.max_ms, range.reserve_ms};
|
||
for (size_t i = 0; i < 3; i++) {
|
||
uint16_t value = (uint16_t)values[i];
|
||
buf[off++] = (uint8_t)(value >> 8); buf[off++] = (uint8_t)value;
|
||
}
|
||
chat_event_post(ctx->inst, CHAT_EVT_CALL_STATS, buf, (int)off);
|
||
}
|
||
|
||
static void call_stats_timer_cb(void* arg) {
|
||
struct call_session* s = arg;
|
||
if (!s || !s->ctx) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: invalid stats timer session", CALL_ID); return; }
|
||
s->stats_timer = NULL;
|
||
if (s->state == CALL_ST_ENDED) return;
|
||
if (s->state == CALL_ST_ACTIVE) call_post_stats(s);
|
||
call_watch_connection(s);
|
||
}
|
||
|
||
/* ─────────────────── обработчики приёма ─────────────────── */
|
||
|
||
static void call_handle_invite(struct call_ctx* ctx, uint64_t from, uint64_t group,
|
||
const uint8_t* b, size_t n) {
|
||
if (n < sizeof(struct call_pkt_invite) - sizeof(struct call_pkt_track)) {
|
||
DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: INVITE too short n=%zu", CALL_ID, n);
|
||
return;
|
||
}
|
||
uint64_t call_id, inv_group;
|
||
memcpy(&call_id, b + 1, 8);
|
||
memcpy(&inv_group, b + 9, 8);
|
||
|
||
/* кодограмма-дубликат (тот же call_id): пока звоним — пере-взводим watchdog и
|
||
* идемпотентно отвечаем RINGING; в разговоре (ACTIVE) — игнорируем. */
|
||
struct call_session* s = call_session_find(ctx, call_id);
|
||
if (s) {
|
||
if (s->role == CALL_ROLE_CALLEE && s->state == CALL_ST_RINGING) {
|
||
if (s->ring_timer) { uasync_cancel_timeout(ctx->inst->ua, s->ring_timer); s->ring_timer = NULL; }
|
||
s->ring_timer = uasync_set_timeout(ctx->inst->ua, CALL_RING_DEAD_TB, s, call_ring_dead_cb, "call_ring_dead");
|
||
DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "%s: INVITE codegram id=%016llx -> re-RINGING + re-arm dead", CALL_ID, (unsigned long long)call_id);
|
||
uint8_t pkt[sizeof(struct call_pkt_ringing)];
|
||
struct call_pkt_ringing* r = (struct call_pkt_ringing*)pkt;
|
||
r->subcmd = CALL_SUBCMD_RINGING;
|
||
r->call_id = call_id;
|
||
call_send(ctx, s->group_id, from, pkt, sizeof(pkt), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
|
||
} else {
|
||
DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "%s: INVITE codegram id=%016llx ignored (state=%s)",
|
||
CALL_ID, (unsigned long long)call_id, call_state_name(s->state));
|
||
}
|
||
return;
|
||
}
|
||
|
||
/* занятость: есть другая активная сессия */
|
||
struct ll_entry* e = ctx->sessions ? ctx->sessions->head : NULL;
|
||
while (e) {
|
||
struct call_session* other = (struct call_session*)e->data;
|
||
if (other->state != CALL_ST_ENDED) {
|
||
DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: busy — active call id=%016llx, reply BUSY to id=%016llx",
|
||
CALL_ID, (unsigned long long)other->call_id, (unsigned long long)call_id);
|
||
uint8_t pkt[sizeof(struct call_pkt_busy)];
|
||
struct call_pkt_busy* bz = (struct call_pkt_busy*)pkt;
|
||
bz->subcmd = CALL_SUBCMD_BUSY;
|
||
bz->call_id = call_id;
|
||
call_send(ctx, group, from, pkt, sizeof(pkt), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
|
||
return;
|
||
}
|
||
e = e->next;
|
||
}
|
||
|
||
s = call_session_new(ctx, call_id, group, from, CALL_ROLE_CALLEE);
|
||
if (!s) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: session alloc failed id=%016llx", CALL_ID, (unsigned long long)call_id);
|
||
return;
|
||
}
|
||
s->state = CALL_ST_RINGING;
|
||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: INVITE id=%016llx from=0x%016llx group=0x%016llx (declared group=0x%016llx)",
|
||
CALL_ID, (unsigned long long)call_id, (unsigned long long)from,
|
||
(unsigned long long)group, (unsigned long long)inv_group);
|
||
|
||
call_event_incoming(ctx, group, from, call_id);
|
||
call_watch_connection(s);
|
||
|
||
/* watchdog: вызывающий шлёт кодограммы INVITE каждые 300мс; 2с тишины → вызов сорван */
|
||
s->ring_timer = uasync_set_timeout(ctx->inst->ua, CALL_RING_DEAD_TB, s, call_ring_dead_cb, "call_ring_dead");
|
||
|
||
uint8_t pkt[sizeof(struct call_pkt_ringing)];
|
||
struct call_pkt_ringing* r = (struct call_pkt_ringing*)pkt;
|
||
r->subcmd = CALL_SUBCMD_RINGING;
|
||
r->call_id = call_id;
|
||
call_send(ctx, group, from, pkt, sizeof(pkt), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
|
||
}
|
||
|
||
static void call_handle_ringing(struct call_ctx* ctx, uint64_t from, const uint8_t* b, size_t n) {
|
||
if (n < 9) return;
|
||
uint64_t call_id; memcpy(&call_id, b + 1, 8);
|
||
struct call_session* s = call_session_find(ctx, call_id);
|
||
if (!s || s->role != CALL_ROLE_CALLER || s->state == CALL_ST_ENDED) return;
|
||
if (s->state == CALL_ST_OUTGOING) {
|
||
s->state = CALL_ST_RINGING;
|
||
/* invite_timer не трогаем — кодограмма INVITE продолжается и в RINGING */
|
||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: RINGING id=%016llx", CALL_ID, (unsigned long long)call_id);
|
||
call_event_id(ctx, CHAT_EVT_CALL_RINGING, call_id, 0);
|
||
}
|
||
(void)from;
|
||
}
|
||
|
||
static void call_handle_accept(struct call_ctx* ctx, uint64_t from, const uint8_t* b, size_t n) {
|
||
if (n < 9) return;
|
||
uint64_t call_id; memcpy(&call_id, b + 1, 8);
|
||
struct call_session* s = call_session_find(ctx, call_id);
|
||
if (!s || s->role != CALL_ROLE_CALLER || s->state == CALL_ST_ENDED || s->state == CALL_ST_ACTIVE) return;
|
||
|
||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: ACCEPT id=%016llx", CALL_ID, (unsigned long long)call_id);
|
||
if (s->ring_timer) { uasync_cancel_timeout(ctx->inst->ua, s->ring_timer); s->ring_timer = NULL; }
|
||
if (s->invite_timer) { uasync_cancel_timeout(ctx->inst->ua, s->invite_timer); s->invite_timer = NULL; }
|
||
call_open_cm(s);
|
||
call_activate(s);
|
||
(void)from;
|
||
}
|
||
|
||
static void call_handle_decline(struct call_ctx* ctx, uint64_t from, const uint8_t* b, size_t n) {
|
||
if (n < 10) return;
|
||
uint64_t call_id; memcpy(&call_id, b + 1, 8);
|
||
uint8_t reason = b[9];
|
||
struct call_session* s = call_session_find(ctx, call_id);
|
||
if (!s || s->state == CALL_ST_ENDED) return;
|
||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: DECLINE id=%016llx reason=%d", CALL_ID, (unsigned long long)call_id, reason);
|
||
call_event_id(ctx, CHAT_EVT_CALL_DECLINED, call_id, reason);
|
||
call_end(ctx, s, CALL_REASON_DECLINE);
|
||
(void)from;
|
||
}
|
||
|
||
static void call_handle_busy(struct call_ctx* ctx, uint64_t from, const uint8_t* b, size_t n) {
|
||
if (n < 9) return;
|
||
uint64_t call_id; memcpy(&call_id, b + 1, 8);
|
||
struct call_session* s = call_session_find(ctx, call_id);
|
||
if (!s || s->state == CALL_ST_ENDED) return;
|
||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: BUSY id=%016llx", CALL_ID, (unsigned long long)call_id);
|
||
call_end(ctx, s, CALL_REASON_BUSY);
|
||
(void)from;
|
||
}
|
||
|
||
static void call_handle_cancel(struct call_ctx* ctx, uint64_t from, const uint8_t* b, size_t n) {
|
||
if (n < 9) return;
|
||
uint64_t call_id; memcpy(&call_id, b + 1, 8);
|
||
struct call_session* s = call_session_find(ctx, call_id);
|
||
if (!s || s->state == CALL_ST_ENDED) return;
|
||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: CANCEL id=%016llx", CALL_ID, (unsigned long long)call_id);
|
||
call_end(ctx, s, CALL_REASON_CANCEL);
|
||
(void)from;
|
||
}
|
||
|
||
static void call_handle_hangup(struct call_ctx* ctx, uint64_t from, const uint8_t* b, size_t n) {
|
||
if (n < 10) return;
|
||
uint64_t call_id; memcpy(&call_id, b + 1, 8);
|
||
uint8_t reason = b[9];
|
||
struct call_session* s = call_session_find(ctx, call_id);
|
||
if (!s || s->state == CALL_ST_ENDED) return;
|
||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: HANGUP id=%016llx reason=%d", CALL_ID, (unsigned long long)call_id, reason);
|
||
call_end(ctx, s, CALL_REASON_REMOTE_HANGUP);
|
||
(void)from;
|
||
}
|
||
|
||
static void call_handle_media(struct call_ctx* ctx, uint64_t from, const uint8_t* b, size_t n) {
|
||
if (n <= CALL_MEDIA_HDR_SIZE) { DEBUG_WARN(DEBUG_CATEGORY_CALL, "call media truncated bytes=%zu", n); return; }
|
||
uint64_t call_id; memcpy(&call_id, b + 1, 8);
|
||
struct call_session* s = call_session_find(ctx, call_id);
|
||
if (!s || s->state != CALL_ST_ACTIVE) {
|
||
/* медиа вне ACTIVE — игнорируем (гонка с завершением) */
|
||
return;
|
||
}
|
||
struct call_pkt_media packet;
|
||
memcpy(&packet, b, sizeof(packet));
|
||
if (from != s->peer_node_id || packet.track_id != CALL_TRACK_AUDIO || packet.codec_delay_samples > CALL_OPUS_SAMPLE_RATE
|
||
|| (packet.capture_pts_us != INT64_MIN && (packet.capture_pts_us < 0 || packet.capture_pts_us > INT64_MAX / 2
|
||
|| !packet.capture_epoch))) {
|
||
DEBUG_WARN(DEBUG_CATEGORY_CALL, "call media invalid peer=%016llx expected=%016llx track=%u pts=%lld epoch=%llu delay=%u",
|
||
(unsigned long long)from, (unsigned long long)s->peer_node_id, packet.track_id,
|
||
(long long)packet.capture_pts_us, (unsigned long long)packet.capture_epoch, packet.codec_delay_samples);
|
||
return;
|
||
}
|
||
struct call_capture_time timing = {packet.capture_pts_us, packet.capture_epoch, packet.codec_delay_samples};
|
||
s->c_rx_frames++;
|
||
s->last_media_tb = get_time_tb();
|
||
|
||
uint16_t seq = 0;
|
||
uint32_t ts_ms = 0;
|
||
memcpy(&seq, b + 10, 2);
|
||
memcpy(&ts_ms, b + 12, 4);
|
||
|
||
/* аудио-движок (декод + джиттер-буфер) — если зарегистрирован */
|
||
if (ctx->audio_ops && ctx->audio_ops->on_media)
|
||
ctx->audio_ops->on_media(ctx->inst, call_id, seq, ts_ms,
|
||
b + CALL_MEDIA_HDR_SIZE, (int)(n - CALL_MEDIA_HDR_SIZE), &timing);
|
||
|
||
/* релей Opus-кадра потребителю (headless-аудио-сокет) */
|
||
if (ctx->media_recv_fn)
|
||
ctx->media_recv_fn(ctx->inst, call_id, b + CALL_MEDIA_HDR_SIZE,
|
||
(int)(n - CALL_MEDIA_HDR_SIZE), ctx->media_recv_arg);
|
||
|
||
/* watchdog: пере-взводим таймер no-traffic на каждый принятый кадр */
|
||
if (s->traffic_timer) { uasync_cancel_timeout(ctx->inst->ua, s->traffic_timer); s->traffic_timer = NULL; }
|
||
s->traffic_timer = uasync_set_timeout(ctx->inst->ua, CALL_NO_TRAFFIC_TB, s, call_traffic_timeout_cb, "call_traffic");
|
||
(void)from;
|
||
}
|
||
|
||
/* ─────────────────── приём (etcp_router) ─────────────────── */
|
||
|
||
static void call_etcp_recv(struct ETCP_CONN* conn, struct ll_entry* entry) {
|
||
if (!entry || !conn || !entry->dgram || entry->len < ROUTER_SVC_PAYLOAD_OFF + 1) {
|
||
if (entry) { queue_dgram_free(entry); queue_entry_free(entry); }
|
||
return;
|
||
}
|
||
struct call_ctx* ctx = call_of(conn->instance);
|
||
if (!ctx || !ctx->initialized) { queue_dgram_free(entry); queue_entry_free(entry); return; }
|
||
|
||
const uint8_t* d = entry->dgram;
|
||
uint64_t from; memcpy(&from, d + ROUTER_SVC_SRC_OFF, 8);
|
||
uint64_t group; memcpy(&group, d + ROUTER_SVC_GROUP_OFF, 8);
|
||
uint8_t subcmd = d[ROUTER_SVC_PAYLOAD_OFF];
|
||
const uint8_t* body = d + ROUTER_SVC_PAYLOAD_OFF;
|
||
size_t blen = entry->len - ROUTER_SVC_PAYLOAD_OFF;
|
||
|
||
switch (subcmd) {
|
||
case CALL_SUBCMD_INVITE: call_handle_invite(ctx, from, group, body, blen); break;
|
||
case CALL_SUBCMD_RINGING: call_handle_ringing(ctx, from, body, blen); break;
|
||
case CALL_SUBCMD_ACCEPT: call_handle_accept(ctx, from, body, blen); break;
|
||
case CALL_SUBCMD_DECLINE: call_handle_decline(ctx, from, body, blen); break;
|
||
case CALL_SUBCMD_BUSY: call_handle_busy(ctx, from, body, blen); break;
|
||
case CALL_SUBCMD_CANCEL: call_handle_cancel(ctx, from, body, blen); break;
|
||
case CALL_SUBCMD_HANGUP: call_handle_hangup(ctx, from, body, blen); break;
|
||
case CALL_SUBCMD_MEDIA: call_handle_media(ctx, from, body, blen); break;
|
||
default:
|
||
DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: unknown subcmd=0x%02x from=0x%016llx len=%zu",
|
||
CALL_ID, subcmd, (unsigned long long)from, blen);
|
||
break;
|
||
}
|
||
queue_dgram_free(entry);
|
||
queue_entry_free(entry);
|
||
}
|
||
|
||
/* ─────────────────── публичный API ─────────────────── */
|
||
|
||
int call_init(struct UTUN_INSTANCE* inst) {
|
||
if (!inst) return -1;
|
||
if (inst->call && inst->call->initialized) return 0;
|
||
|
||
struct call_ctx* ctx = u_calloc(1, sizeof(*ctx));
|
||
if (!ctx) return -1;
|
||
ctx->inst = inst;
|
||
inst->call = ctx;
|
||
|
||
ctx->sessions = queue_new(inst->ua, 256, offsetof(struct call_session, call_id), 8, "call_sessions");
|
||
if (!ctx->sessions) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: queue_new(sessions) failed", CALL_ID);
|
||
u_free(ctx); inst->call = NULL;
|
||
return -1;
|
||
}
|
||
if (etcp_router_bind(inst, ETCP_RT_ID_CALL, call_etcp_recv) != 0) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: etcp_router_bind(0x%02x) failed", CALL_ID, ETCP_RT_ID_CALL);
|
||
queue_free(ctx->sessions);
|
||
u_free(ctx); inst->call = NULL;
|
||
return -1;
|
||
}
|
||
|
||
ctx->initialized = 1;
|
||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: initialized (svc=0x%02x)", CALL_ID, ETCP_RT_ID_CALL);
|
||
return 0;
|
||
}
|
||
|
||
void call_destroy(struct UTUN_INSTANCE* inst) {
|
||
struct call_ctx* ctx = call_of(inst);
|
||
if (!ctx || !ctx->initialized) return;
|
||
|
||
/* завершить все активные сессии (без отправки — мы выключаемся) */
|
||
while (ctx->sessions && ctx->sessions->head) {
|
||
struct ll_entry* e = ctx->sessions->head;
|
||
struct call_session* s = (struct call_session*)e->data;
|
||
s->state = CALL_ST_ENDED;
|
||
if (s->ring_timer) { uasync_cancel_timeout(inst->ua, s->ring_timer); s->ring_timer = NULL; }
|
||
if (s->invite_timer) { uasync_cancel_timeout(inst->ua, s->invite_timer); s->invite_timer = NULL; }
|
||
if (s->traffic_timer) { uasync_cancel_timeout(inst->ua, s->traffic_timer); s->traffic_timer = NULL; }
|
||
if (s->stats_timer) { uasync_cancel_timeout(inst->ua, s->stats_timer); s->stats_timer = NULL; }
|
||
if (s->teardown_timer) { uasync_cancel_timeout(inst->ua, s->teardown_timer); s->teardown_timer = NULL; }
|
||
if (s->cm_handle) { conn_mgr_close(s->cm_handle); s->cm_handle = NULL; }
|
||
call_session_cleanup(ctx, s);
|
||
queue_remove_data(ctx->sessions, e);
|
||
queue_entry_free(e);
|
||
}
|
||
queue_free(ctx->sessions);
|
||
etcp_router_unbind(inst, ETCP_RT_ID_CALL);
|
||
inst->call = NULL;
|
||
uint32_t started = ctx->c_calls_started, ended = ctx->c_calls_ended;
|
||
u_free(ctx);
|
||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: destroyed (started=%u ended=%u)", CALL_ID, started, ended);
|
||
}
|
||
|
||
int call_is_online(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t peer_node_id) {
|
||
if (!inst || !ch_id || !ch_id[0] || !inst->topo_groups) return 0;
|
||
if (peer_node_id == inst->node_id) return 1;
|
||
uint64_t group_id = strtoull(ch_id, NULL, 10);
|
||
struct TOPO_GROUP* g = topo_groups_find(inst->topo_groups, group_id);
|
||
if (!g) return 0;
|
||
return topo_node_find_by_id(g, peer_node_id) != NULL;
|
||
}
|
||
|
||
void call_set_media_recv_cb(struct UTUN_INSTANCE* inst, call_media_recv_fn fn, void* arg) {
|
||
struct call_ctx* ctx = call_of(inst);
|
||
if (!ctx) return;
|
||
ctx->media_recv_fn = fn;
|
||
ctx->media_recv_arg = arg;
|
||
}
|
||
|
||
void call_set_audio_ops(struct UTUN_INSTANCE* inst, const struct call_audio_ops* ops) {
|
||
struct call_ctx* ctx = call_of(inst);
|
||
if (!ctx) return;
|
||
ctx->audio_ops = ops;
|
||
}
|
||
|
||
/* ─────────────────── TX медиа: tx_q + backpressure ─────────────────── */
|
||
|
||
static void call_tx_flush(struct call_session* s);
|
||
|
||
static void call_tx_drain_cb(struct ll_queue* q, void* arg) {
|
||
(void)q;
|
||
struct call_session* s = (struct call_session*)arg;
|
||
if (!s || !s->ctx) return;
|
||
if (s->state == CALL_ST_ENDED) return;
|
||
call_tx_flush(s);
|
||
}
|
||
|
||
/* Дренируем tx_q, пока роутер готов; иначе — одноразовый waiter (re-arm в drain_cb). */
|
||
static void call_tx_flush(struct call_session* s) {
|
||
struct call_ctx* ctx = s->ctx;
|
||
if (s->state == CALL_ST_ENDED) return;
|
||
/* гарантируем router conn — иначе has_room/on_send_ready не видят его */
|
||
etcp_router_conn_get(ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL);
|
||
while (s->tx_q && s->tx_q->head) {
|
||
if (!etcp_router_send_q_has_room(ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL)) {
|
||
etcp_router_on_send_ready(ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL,
|
||
&s->send_waiter, call_tx_drain_cb, s);
|
||
return;
|
||
}
|
||
struct ll_entry* e = queue_data_get(s->tx_q);
|
||
if (etcp_route_send(ctx->inst, s->group_id, s->peer_node_id, e, 1, 0) == 0)
|
||
s->c_tx_frames++;
|
||
else DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: media enqueue failed id=%016llx", CALL_ID, (unsigned long long)s->call_id);
|
||
}
|
||
}
|
||
|
||
int call_send_media(struct UTUN_INSTANCE* inst, uint64_t call_id, const uint8_t* opus, int len,
|
||
const struct call_capture_time* timing) {
|
||
struct call_ctx* ctx = call_of(inst);
|
||
if (!ctx || !ctx->initialized || !opus || len <= 0) return -1;
|
||
struct call_session* s = call_session_find(ctx, call_id);
|
||
if (!s || s->state != CALL_ST_ACTIVE) return -1;
|
||
if (CALL_MEDIA_HDR_SIZE + (size_t)len > 1024) return -1;
|
||
if (timing && (timing->delay_samples > CALL_OPUS_SAMPLE_RATE || (timing->pts_us != INT64_MIN
|
||
&& (timing->pts_us < 0 || timing->pts_us > INT64_MAX / 2 || !timing->epoch)))) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "call media invalid capture metadata pts=%lld epoch=%llu delay=%u",
|
||
(long long)timing->pts_us, (unsigned long long)timing->epoch, timing->delay_samples);
|
||
return -1;
|
||
}
|
||
|
||
uint8_t buf[1024];
|
||
struct call_pkt_media* m = (struct call_pkt_media*)buf;
|
||
m->subcmd = CALL_SUBCMD_MEDIA;
|
||
m->call_id = call_id;
|
||
m->track_id = CALL_TRACK_AUDIO;
|
||
m->seq = s->tx_media_seq++;
|
||
m->ts_ms = (uint32_t)(get_time_tb() / 10);
|
||
m->capture_pts_us = timing ? timing->pts_us : INT64_MIN;
|
||
m->capture_epoch = timing ? timing->epoch : 0;
|
||
m->codec_delay_samples = timing ? timing->delay_samples : 0;
|
||
memcpy(buf + CALL_MEDIA_HDR_SIZE, opus, (size_t)len);
|
||
size_t mlen = CALL_MEDIA_HDR_SIZE + (size_t)len;
|
||
|
||
/* entry: [svc_id][media-пакет] — кладём в tx_q (переполнение → дроп старейшего) */
|
||
struct ll_entry* e = queue_entry_new(0);
|
||
if (!e) return -1;
|
||
e->dgram = u_malloc(1 + mlen);
|
||
if (!e->dgram) { queue_entry_free(e); return -1; }
|
||
e->dgram[0] = ETCP_RT_ID_CALL;
|
||
memcpy(e->dgram + 1, buf, mlen);
|
||
e->len = (uint16_t)(1 + mlen);
|
||
|
||
if (queue_entry_count(s->tx_q) >= CALL_MAX_TX_FRAMES) {
|
||
struct ll_entry* old = queue_data_get(s->tx_q);
|
||
if (old) {
|
||
queue_dgram_free(old); queue_entry_free(old); s->c_tx_dropped++;
|
||
if (s->c_tx_dropped == 1 || s->c_tx_dropped % 50 == 0)
|
||
DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: TX overflow id=%016llx queued=%zu dropped=%u", CALL_ID,
|
||
(unsigned long long)call_id, (size_t)queue_entry_count(s->tx_q), s->c_tx_dropped);
|
||
}
|
||
}
|
||
queue_data_put(s->tx_q, e);
|
||
|
||
call_tx_flush(s);
|
||
return 0;
|
||
}
|
||
|
||
void call_start(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t peer_node_id, uint64_t out_call_id) {
|
||
struct call_ctx* ctx = call_of(inst);
|
||
if (!ctx || !ctx->initialized || !ch_id || !ch_id[0]) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: call_start bad args (ctx=%p ch_id=%s)", CALL_ID, (void*)ctx, ch_id ? ch_id : "(null)");
|
||
return;
|
||
}
|
||
if (peer_node_id == inst->node_id) {
|
||
DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: call_start self-call rejected", CALL_ID);
|
||
return;
|
||
}
|
||
|
||
uint64_t group_id = 0;
|
||
char route_ch[64];
|
||
if (!strncmp(ch_id, "dm:", 3)) {
|
||
uint64_t peer = 0;
|
||
int available = dm_call_route(inst, ch_id + 3, &peer, &group_id);
|
||
if (!available || peer != peer_node_id) {
|
||
DEBUG_WARN(DEBUG_CATEGORY_CALL, "PM call rejected conv=%s requested_peer=%llu actual_peer=%llu available=%d",
|
||
ch_id + 3, (unsigned long long)peer_node_id, (unsigned long long)peer, available);
|
||
uint8_t evt[9];
|
||
memcpy(evt, &out_call_id, 8); evt[8] = 1;
|
||
chat_event_post(inst, CHAT_EVT_CALL_ERROR, evt, sizeof(evt));
|
||
return;
|
||
}
|
||
snprintf(route_ch, sizeof(route_ch), "%llu", (unsigned long long)group_id);
|
||
ch_id = route_ch;
|
||
} else group_id = strtoull(ch_id, NULL, 10);
|
||
if (!call_is_online(inst, ch_id, peer_node_id)) {
|
||
DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: peer 0x%016llx offline in group 0x%016llx — call not started",
|
||
CALL_ID, (unsigned long long)peer_node_id, (unsigned long long)group_id);
|
||
uint8_t buf[9];
|
||
memcpy(buf, &out_call_id, 8);
|
||
buf[8] = 1; /* err: peer offline */
|
||
chat_event_post(inst, CHAT_EVT_CALL_ERROR, buf, sizeof(buf));
|
||
return;
|
||
}
|
||
|
||
struct call_session* s = call_session_new(ctx, out_call_id, group_id, peer_node_id, CALL_ROLE_CALLER);
|
||
if (!s) {
|
||
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: session alloc failed id=%016llx", CALL_ID, (unsigned long long)out_call_id);
|
||
return;
|
||
}
|
||
s->state = CALL_ST_OUTGOING;
|
||
ctx->c_calls_started++;
|
||
|
||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: call_start id=%016llx peer=0x%016llx group=0x%016llx",
|
||
CALL_ID, (unsigned long long)out_call_id, (unsigned long long)peer_node_id, (unsigned long long)group_id);
|
||
|
||
call_open_cm(s);
|
||
if (call_send_invite(ctx, s) != 0) {
|
||
call_end(ctx, s, CALL_REASON_CONNECT_FAIL);
|
||
return;
|
||
}
|
||
call_watch_connection(s);
|
||
s->ring_timer = uasync_set_timeout(inst->ua, CALL_RING_TIMEOUT_TB, s, call_ring_timeout_cb, "call_ring");
|
||
s->invite_timer = uasync_set_timeout(inst->ua, CALL_INVITE_RETRY_TB, s, call_invite_retry_cb, "call_invite_retry");
|
||
}
|
||
|
||
void call_accept(struct UTUN_INSTANCE* inst, uint64_t call_id) {
|
||
struct call_ctx* ctx = call_of(inst);
|
||
if (!ctx || !ctx->initialized) return;
|
||
struct call_session* s = call_session_find(ctx, call_id);
|
||
if (!s || s->role != CALL_ROLE_CALLEE || s->state == CALL_ST_ENDED || s->state == CALL_ST_ACTIVE) return;
|
||
|
||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: call_accept id=%016llx -> ACCEPT + conn_mgr_open",
|
||
CALL_ID, (unsigned long long)call_id);
|
||
if (s->ring_timer) { uasync_cancel_timeout(ctx->inst->ua, s->ring_timer); s->ring_timer = NULL; } /* watchdog кодограмм */
|
||
|
||
uint8_t pkt[sizeof(struct call_pkt_accept)];
|
||
struct call_pkt_accept* a = (struct call_pkt_accept*)pkt;
|
||
a->subcmd = CALL_SUBCMD_ACCEPT;
|
||
a->call_id = call_id;
|
||
call_send(ctx, s->group_id, s->peer_node_id, pkt, sizeof(pkt), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
|
||
|
||
call_open_cm(s);
|
||
call_activate(s);
|
||
}
|
||
|
||
void call_decline(struct UTUN_INSTANCE* inst, uint64_t call_id) {
|
||
struct call_ctx* ctx = call_of(inst);
|
||
if (!ctx || !ctx->initialized) return;
|
||
struct call_session* s = call_session_find(ctx, call_id);
|
||
if (!s || s->role != CALL_ROLE_CALLEE || s->state == CALL_ST_ENDED) return;
|
||
|
||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: call_decline id=%016llx", CALL_ID, (unsigned long long)call_id);
|
||
uint8_t pkt[sizeof(struct call_pkt_decline)];
|
||
struct call_pkt_decline* d = (struct call_pkt_decline*)pkt;
|
||
d->subcmd = CALL_SUBCMD_DECLINE;
|
||
d->call_id = call_id;
|
||
d->reason = CALL_REASON_DECLINE;
|
||
call_send(ctx, s->group_id, s->peer_node_id, pkt, sizeof(pkt), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
|
||
|
||
call_event_id(ctx, CHAT_EVT_CALL_DECLINED, call_id, CALL_REASON_DECLINE);
|
||
call_end(ctx, s, CALL_REASON_DECLINE);
|
||
}
|
||
|
||
void call_hangup(struct UTUN_INSTANCE* inst, uint64_t call_id) {
|
||
struct call_ctx* ctx = call_of(inst);
|
||
if (!ctx || !ctx->initialized) return;
|
||
struct call_session* s = call_session_find(ctx, call_id);
|
||
if (!s || s->state == CALL_ST_ENDED) return;
|
||
|
||
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: call_hangup id=%016llx", CALL_ID, (unsigned long long)call_id);
|
||
uint8_t pkt[sizeof(struct call_pkt_hangup)];
|
||
struct call_pkt_hangup* hu = (struct call_pkt_hangup*)pkt;
|
||
hu->subcmd = CALL_SUBCMD_HANGUP;
|
||
hu->call_id = call_id;
|
||
hu->reason = CALL_REASON_LOCAL_HANGUP;
|
||
call_send(ctx, s->group_id, s->peer_node_id, pkt, sizeof(pkt), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT);
|
||
|
||
call_end(ctx, s, CALL_REASON_LOCAL_HANGUP);
|
||
}
|
||
|
||
/* ─────────────────── trampoline-функции (GUI → uasync) ─────────────────── */
|
||
|
||
void call_start_trampoline(void* arg) {
|
||
struct call_start_arg* a = (struct call_start_arg*)arg;
|
||
call_start(a->inst, a->ch_id, a->peer_node_id, a->call_id);
|
||
u_free(a);
|
||
}
|
||
|
||
void call_accept_trampoline(void* arg) {
|
||
struct call_id_arg* a = (struct call_id_arg*)arg;
|
||
call_accept(a->inst, a->call_id);
|
||
u_free(a);
|
||
}
|
||
|
||
void call_decline_trampoline(void* arg) {
|
||
struct call_id_arg* a = (struct call_id_arg*)arg;
|
||
call_decline(a->inst, a->call_id);
|
||
u_free(a);
|
||
}
|
||
|
||
void call_hangup_trampoline(void* arg) {
|
||
struct call_id_arg* a = (struct call_id_arg*)arg;
|
||
call_hangup(a->inst, a->call_id);
|
||
u_free(a);
|
||
}
|
||
|
||
void call_send_media_trampoline(void* arg) {
|
||
struct call_media_arg* a = (struct call_media_arg*)arg;
|
||
call_send_media(a->inst, a->call_id, a->opus, a->len, &a->timing);
|
||
u_free(a);
|
||
}
|