call: P2P аудиозвонки — сигналинг, релей медиа, аудио-контур, headless/desktop/android

- src/call/: сигналинг (call.c, call_proto.h, ETCP_RT_ID_CALL 0x36) + релей медиа
  + headless аудио-сокет (call_headless.c, call_audio_bind) + Opus/jitter/тоны
- chat: события CHAT_EVT_CALL_* (34–40), CHAT_EVT_DM_MESSAGES (41), DM-трамплины
- chatgui: call window, miniaudio I/O, audio_frame_ring, test_audio_jitter
- chatgui-android: CallScreen/IncomingCallDialog, аудио-роутинг, JNI, рингтон
- tests: test_call, test_call_headless
This commit is contained in:
evgeny committed 2026-09-16 10:17:57 +03:00
1 parent 0157f7f96b
commit b11e77e4c0
78 files changed
+7053 -70

No files matched your search

+5
View File
@@ -95,6 +95,8 @@ utun_CORE_SOURCES = \
dm/dm_core.c \
dm/dm_crypto.c \
dm/dm_mailbox.c \
call/call.c \
call/call_headless.c \
broadcast.c
# libutun: all core sources except main()
@@ -189,6 +191,8 @@ libutun_a_SOURCES = \
dm/dm_core.c \
dm/dm_crypto.c \
dm/dm_mailbox.c \
call/call.c \
call/call_headless.c \
broadcast.c
libutun_a_CFLAGS = $(utun_CFLAGS)
@@ -206,6 +210,7 @@ utun_CFLAGS = \
-I$(top_srcdir)/src/media_delivery \
-I$(top_srcdir)/src/media_async \
-I$(top_srcdir)/src/video \
-I$(top_srcdir)/src/call \
-I$(top_srcdir)/src/uip \
-g \
-DUSE_SQLITE \
+180
View File
@@ -0,0 +1,180 @@
// audio_jitter.cpp — адаптивный джиттер-буфер с time-stretch (SoundTouch).
//
// PCM-кадры (int16, 960 сэмплов @20мс) кладутся из uasync-потока в кольцо и
// вытаскиваются аудио-потоком через SoundTouch с плавно меняющимся темпом:
// глубина < target → 1.0x, глубина растёт → ускорение до 1.6x (догон live).
// Кольцо защищено мьютексом (SPSC); SoundTouch трогает только аудио-поток.
#include "audio_jitter.h"
#include "soundtouch/SoundTouch.h"
#include <atomic>
#include <cmath>
#include <cstring>
#include <mutex>
#include <vector>
static const double kTargetMs = 60.0; // целевая глубина буфера
static const double kMaxTempo = 1.6;
static const double kMinTempo = 1.0;
static const double kMaxDepthMs = 1000.0; // при этой глубине tempo == kMaxTempo
static const double kEmaAlpha = 0.2; // сглаживание глубины
static const double kTempoStep = 0.01; // макс. изменение темпа за один pull (без кликов)
struct ajb {
int sr = 0;
int ch = 0;
int frame_samples = 0;
int frame_ms = 0;
int cap = 0;
std::mutex mtx;
std::vector<int16_t> ring; // cap * frame_samples
int head = 0;
int count = 0;
uint32_t dropped = 0;
soundtouch::SoundTouch st; // только аудио-поток
std::vector<int16_t> frame_buf; // scratch: вытащенный кадр
std::vector<float> fin; // scratch: кадр в float
std::vector<float> fout; // scratch: выход float
double ema_depth_ms = 0.0;
double tempo = 1.0;
std::atomic<int> tempo_x100{100};
};
static double ajb_compute_tempo(double depth_ms) {
if (depth_ms <= kTargetMs) return kMinTempo;
double x = (depth_ms - kTargetMs) / (kMaxDepthMs - kTargetMs);
if (x > 1.0) x = 1.0;
return kMinTempo + (kMaxTempo - kMinTempo) * x;
}
/* Вытащить один кадр из кольца (под мьютексом). Возвращает 0 если пусто. */
static int ajb_pop_frame(struct ajb* j, int16_t* out) {
std::lock_guard<std::mutex> lk(j->mtx);
if (j->count == 0) return 0;
std::memcpy(out, &j->ring[(size_t)j->head * j->frame_samples],
(size_t)j->frame_samples * sizeof(int16_t));
j->head = (j->head + 1) % j->cap;
j->count--;
return 1;
}
extern "C" {
struct ajb* ajb_create(int sample_rate, int channels, int frame_samples, int frame_ms, int max_frames) {
if (sample_rate <= 0 || channels <= 0 || frame_samples <= 0 || frame_ms <= 0 || max_frames <= 0)
return nullptr;
ajb* j = new ajb();
j->sr = sample_rate;
j->ch = channels;
j->frame_samples = frame_samples;
j->frame_ms = frame_ms;
j->cap = max_frames;
j->ring.assign((size_t)max_frames * frame_samples, 0);
j->frame_buf.assign(frame_samples, 0);
j->fin.assign(frame_samples, 0.0f);
j->st.setSampleRate((uint)sample_rate);
j->st.setChannels((uint)channels);
j->st.setTempo(kMinTempo);
j->st.setSetting(SETTING_USE_QUICKSEEK, 1);
j->st.setSetting(SETTING_SEQUENCE_MS, 20);
j->st.setSetting(SETTING_SEEKWINDOW_MS, 10);
j->st.setSetting(SETTING_OVERLAP_MS, 4);
j->st.setSetting(SETTING_USE_AA_FILTER, 0);
return j;
}
void ajb_destroy(struct ajb* j) {
if (!j) return;
delete j;
}
void ajb_reset(struct ajb* j) {
if (!j) return;
{
std::lock_guard<std::mutex> lk(j->mtx);
j->head = 0;
j->count = 0;
}
j->st.clear();
j->st.setTempo(kMinTempo);
j->tempo = kMinTempo;
j->ema_depth_ms = 0.0;
j->tempo_x100.store(100);
}
void ajb_push(struct ajb* j, const int16_t* pcm, int samples) {
if (!j || !pcm || samples != j->frame_samples) return;
std::lock_guard<std::mutex> lk(j->mtx);
if (j->count == j->cap) {
j->head = (j->head + 1) % j->cap; // дроп старейшего
j->count--;
j->dropped++;
}
int idx = (j->head + j->count) % j->cap;
std::memcpy(&j->ring[(size_t)idx * j->frame_samples], pcm, (size_t)samples * sizeof(int16_t));
j->count++;
}
int ajb_pull(struct ajb* j, int16_t* out, int max_samples) {
if (!j || !out || max_samples <= 0) return 0;
/* 1) глубина (сглаженная) → целевой темп (плавно) */
int cnt;
{
std::lock_guard<std::mutex> lk(j->mtx);
cnt = j->count;
}
double depth_ms = (double)cnt * j->frame_ms;
j->ema_depth_ms = kEmaAlpha * depth_ms + (1.0 - kEmaAlpha) * j->ema_depth_ms;
double target = ajb_compute_tempo(j->ema_depth_ms);
double d = target - j->tempo;
if (d > kTempoStep) d = kTempoStep;
else if (d < -kTempoStep) d = -kTempoStep;
j->tempo += d;
j->st.setTempo(j->tempo);
j->tempo_x100.store((int)std::lround(j->tempo * 100.0));
/* 2) кормим SoundTouch, пока он не сможет выдать max_samples (или кольцо пусто) */
int guard = 0;
while (j->st.numSamples() < (uint)max_samples && guard++ < j->cap + 4) {
if (!ajb_pop_frame(j, j->frame_buf.data())) break;
for (int i = 0; i < j->frame_samples; i++)
j->fin[(size_t)i] = (float)j->frame_buf[(size_t)i] / 32768.0f;
j->st.putSamples(j->fin.data(), (uint)j->frame_samples);
}
/* 3) забираем готовое */
j->fout.resize((size_t)max_samples);
uint n = j->st.receiveSamples(j->fout.data(), (uint)max_samples);
for (uint i = 0; i < n; i++) {
float v = j->fout[i] * 32767.0f;
if (v > 32767.0f) v = 32767.0f;
else if (v < -32768.0f) v = -32768.0f;
out[i] = (int16_t)std::lrintf(v);
}
return (int)n;
}
void ajb_get_stats(struct ajb* j, int* buffer_ms, int* tempo_x100) {
if (!j) return;
if (buffer_ms) {
std::lock_guard<std::mutex> lk(j->mtx);
*buffer_ms = j->count * j->frame_ms;
}
if (tempo_x100) *tempo_x100 = j->tempo_x100.load();
}
uint32_t ajb_dropped(struct ajb* j) {
if (!j) return 0;
std::lock_guard<std::mutex> lk(j->mtx);
return j->dropped;
}
} // extern "C"
+51
View File
@@ -0,0 +1,51 @@
#ifndef AUDIO_JITTER_H
#define AUDIO_JITTER_H
// audio_jitter.h — адаптивный джиттер-буфер с time-stretch (SoundTouch), C API.
//
// Хранит декодированные PCM-кадры (по frame_samples сэмплов). На pull применяет
// tempo: при росте глубины буфера (сетевая пауза → burst) воспроизведение плавно
// ускоряется (до 1.6x) и догоняет live, не теряя данные (до max_frames кадров).
//
// Потоки:
// - ajb_push — uasync-поток (producer);
// - ajb_pull — аудио-поток (consumer, единственный владелец SoundTouch);
// - ajb_get_stats — любой поток.
// Кольцо кадров защищено мьютексом (SPSC, дёшево: ~50 кадров/с).
#include <stdint.h>
#ifdef __cplusplus
extern "C" {
#endif
struct ajb;
/* Создать буфер. frame_samples — сэмплов в кадре (960 @20мс/48k), frame_ms —
* длительность кадра в мс, max_frames — ёмкость (50 = 1000мс). */
struct ajb* ajb_create(int sample_rate, int channels, int frame_samples, int frame_ms, int max_frames);
void ajb_destroy(struct ajb* j);
/* Сброс: очистить кольцо и внутреннее состояние SoundTouch (темп → 1.0). */
void ajb_reset(struct ajb* j);
/* producer: положить декодированный PCM-кадр (ровно frame_samples сэмплов).
* При переполнении дропает старейший кадр (счётчик внутри). */
void ajb_push(struct ajb* j, const int16_t* pcm, int samples);
/* consumer: выдать до max_samples PCM с учётом текущего темпа. Возвращает число
* выданных сэмплов; может быть < max_samples при нехватке данных (андерфлоу). */
int ajb_pull(struct ajb* j, int16_t* out, int max_samples);
/* Диагностика: текущая глубина буфера (мс) и темп (×100, 100 = 1.0x). */
void ajb_get_stats(struct ajb* j, int* buffer_ms, int* tempo_x100);
/* Число дропнутых кадров при переполнении (для диагностики/тестов). */
uint32_t ajb_dropped(struct ajb* j);
#ifdef __cplusplus
}
#endif
#endif /* AUDIO_JITTER_H */
+877
View File
@@ -0,0 +1,877 @@
// 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 "../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_CONNECTING, /* обе стороны: conn_mgr_open в процессе */
CALL_ST_ACTIVE, /* обе стороны: соединение поднято */
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с ожидания ответа */
void* traffic_timer; /* ACTIVE: 20с без медиа (запускается по первому медиа) */
void* teardown_timer; /* отложенный разрыв conn_mgr (500мс после завершения) */
void* stats_timer; /* ACTIVE: раз в 1с — CALL_STATS (rtt/buffer/tempo) */
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_CONNECTING: return "CONNECTING";
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;
}
/* ─────────────────── реестр сессий ─────────────────── */
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; }
if (queue_data_put_with_index(ctx->sessions, qe) != 0) {
queue_free(s->tx_q);
queue_entry_free(qe);
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);
}
/* ─────────────────── завершение ─────────────────── */
/* Отложенный разрыв: единственная точка, где закрывается 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->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_event_id(ctx, CHAT_EVT_CALL_ENDED, s->call_id, reason);
/* Разрыв conn_mgr — отложенно (500мс), чтобы HANGUP/DECLINE успел долететь.
* Без cm_handle рвать нечего — освобождаем сразу (отложенно). */
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_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 = (struct call_session*)arg;
if (!s || !s->ctx) return;
struct call_ctx* ctx = s->ctx;
if (s->state == CALL_ST_ENDED) {
/* сессия завершена, ждёт teardown: conn_mgr сообщает об обрыве — handle больше
* не трогаем (entry чистится самим conn_mgr / удалённой стороной) */
if (event == CONN_EVENT_DOWN || event == CONN_EVENT_TIMEOUT) s->cm_handle = NULL;
return;
}
if (event == CONN_EVENT_UP) {
s->cm_handle = h;
s->state = CALL_ST_ACTIVE;
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);
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) {
DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: conn DOWN id=%016llx peer=0x%016llx (temporary, keep alive)",
CALL_ID, (unsigned long long)s->call_id, (unsigned long long)node_id);
} else { /* CONN_EVENT_TIMEOUT */
DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: conn TIMEOUT id=%016llx peer=0x%016llx -> fail",
CALL_ID, (unsigned long long)s->call_id, (unsigned long long)node_id);
/* entry уже чистится вызывающим (monitor) — handle не закрываем */
s->cm_handle = NULL;
call_end(ctx, s, CALL_REASON_CONNECT_FAIL);
}
}
/* ─────────────────── таймеры ─────────────────── */
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);
}
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_stats_timer_cb(void* arg) {
struct call_session* s = (struct call_session*)arg;
if (!s || !s->ctx) return;
struct call_ctx* ctx = s->ctx;
s->stats_timer = NULL;
if (s->state != CALL_ST_ACTIVE) return;
int buffer_ms = 0, tempo_x100 = 100;
if (ctx->audio_ops && ctx->audio_ops->get_stats)
ctx->audio_ops->get_stats(s->call_id, &buffer_ms, &tempo_x100);
uint16_t rtt = 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 = rconn->recv_conn->rtt_last;
uint8_t buf[14];
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);
chat_event_post(ctx->inst, CHAT_EVT_CALL_STATS, buf, (int)off);
s->stats_timer = uasync_set_timeout(ctx->inst->ua, 10000, s, call_stats_timer_cb, "call_stats");
}
/* ─────────────────── потеря соединения ─────────────────── */
/* На обрыв ETCP-соединения с пиром завершаем ACTIVE-звонки (robust fallback:
* HANGUP может не успеть долететь, если закрывающая сторона рвёт conn_mgr). */
static void call_on_conn_status(struct ETCP_CONN* conn, int status, void* arg) {
struct UTUN_INSTANCE* inst = (struct UTUN_INSTANCE*)arg;
struct call_ctx* ctx = call_of(inst);
if (!ctx || !ctx->initialized || !conn) return;
if (status != ETCP_CONN_STATUS_DOWN && status != ETCP_CONN_STATUS_DELETE) return;
uint64_t peer = conn->peer_node_id;
if (!peer) return;
struct ll_entry* e = ctx->sessions ? ctx->sessions->head : NULL;
while (e) {
struct ll_entry* next = e->next;
struct call_session* s = (struct call_session*)e->data;
if (s->peer_node_id == peer) {
if (s->state == CALL_ST_ACTIVE) {
DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: peer conn lost id=%016llx peer=0x%016llx -> end",
CALL_ID, (unsigned long long)s->call_id, (unsigned long long)peer);
call_end(ctx, s, CALL_REASON_REMOTE_HANGUP);
} else if (s->state == CALL_ST_ENDED) {
/* удалённая сторона уже порвала (DISCONNECT чистит entry без коллбэка) */
s->cm_handle = NULL;
}
}
e = next;
}
}
/* ─────────────────── обработчики приёма ─────────────────── */
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);
/* повторная доставка INVITE (тот же call_id) — идемпотентно RINGING */
struct call_session* s = call_session_find(ctx, call_id);
if (s && s->state != CALL_ST_ENDED) {
DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "%s: INVITE retransmit id=%016llx -> re-RINGING", 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);
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);
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;
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) return;
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: ACCEPT id=%016llx -> conn_mgr_open", CALL_ID, (unsigned long long)call_id);
s->state = CALL_ST_CONNECTING;
if (s->ring_timer) { uasync_cancel_timeout(ctx->inst->ua, s->ring_timer); s->ring_timer = NULL; }
int rc = conn_mgr_open(ctx->inst, s->group_id, s->peer_node_id, call_conn_cb, s, &s->cm_handle);
if (rc != 0) {
/* либо уже получили TIMEOUT (сессия ENDED), либо ошибка без коллбэка */
if (s->state != CALL_ST_ENDED) call_end(ctx, s, CALL_REASON_CONNECT_FAIL);
}
}
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) 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;
}
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));
/* релей 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;
}
etcp_add_conn_status_cbk(inst, call_on_conn_status, inst);
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->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);
etcp_remove_conn_status_cbk(inst, call_on_conn_status, inst);
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++;
}
}
int call_send_media(struct UTUN_INSTANCE* inst, uint64_t call_id, const uint8_t* opus, int len) {
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;
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);
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++; }
}
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 = 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++;
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 = out_call_id;
inv->group_id = 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;
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);
if (call_send(ctx, group_id, peer_node_id, buf, sizeof(buf), ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT) != 0) {
call_end(ctx, s, CALL_REASON_CONNECT_FAIL);
return;
}
s->ring_timer = uasync_set_timeout(inst->ua, CALL_RING_TIMEOUT_TB, s, call_ring_timeout_cb, "call_ring");
}
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) return;
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: call_accept id=%016llx -> ACCEPT + conn_mgr_open",
CALL_ID, (unsigned long long)call_id);
s->state = CALL_ST_CONNECTING;
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);
int rc = conn_mgr_open(inst, s->group_id, s->peer_node_id, call_conn_cb, s, &s->cm_handle);
if (rc != 0 && s->state != CALL_ST_ENDED) {
call_end(ctx, s, CALL_REASON_CONNECT_FAIL);
}
}
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);
u_free(a);
}
+96
View File
@@ -0,0 +1,96 @@
// call.h — P2P аудио-звонок (сигналинг + медиа), per-instance
//
// Модуль оперирует (group_id, peer_node_id): звонок можно инициировать из
// CHAT-группы или из DM (group_id — группа маршрутизации, как в dm_core).
// Сигналинг идёт через etcp_router (ROUTE_CRYPTO_SIGN|ENCRYPT), соединение
// устанавливается через conn_mgr (3-фазное DIRECT→REVERSE→INDIRECT).
//
// Все функции — в uasync-потоке. GUI/headless вызывают через trampoline
// (gui_bridge_post_uasync_fn / uasync_post).
#ifndef CALL_H
#define CALL_H
#include <stdint.h>
struct UTUN_INSTANCE;
#ifdef __cplusplus
extern "C" {
#endif
/* ── жизненный цикл ── */
int call_init(struct UTUN_INSTANCE* inst); /* из utun_instance.c после dm_core_init */
void call_destroy(struct UTUN_INSTANCE* inst); /* перед dm_core_destroy */
/* ── сигналинг (uasync-поток) ── */
/* Caller: начать звонок. out_call_id — случайный uint64, генерит вызывающий. */
void call_start (struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t peer_node_id, uint64_t out_call_id);
/* Callee: ответить/отклонить. hangup — любая сторона в любой момент. */
void call_accept (struct UTUN_INSTANCE* inst, uint64_t call_id);
void call_decline(struct UTUN_INSTANCE* inst, uint64_t call_id);
void call_hangup (struct UTUN_INSTANCE* inst, uint64_t call_id);
/* Read-only онлайн-проверка (GUI может звать из любого потока). */
int call_is_online(struct UTUN_INSTANCE* inst, const char* ch_id, uint64_t peer_node_id);
/* ── медиа (релей Opus-кадров, uasync-поток) ── */
/* Отправить закодированный Opus-кадр пиру (оборачивает в CALL_MEDIA).
* Только когда звонок ACTIVE. seq/ts выставляются автоматически. */
int call_send_media(struct UTUN_INSTANCE* inst, uint64_t call_id, const uint8_t* opus, int len);
/* Коллбэк приёма медиа: вызывается в uasync-потоке на каждый Opus-кадр пира.
* Используется headless-аудио-сокетом (и в фазе 3 — декодером GUI). */
typedef void (*call_media_recv_fn)(struct UTUN_INSTANCE* inst, uint64_t call_id,
const uint8_t* opus, int len, void* arg);
void call_set_media_recv_cb(struct UTUN_INSTANCE* inst, call_media_recv_fn fn, void* arg);
/* ── аудио-движок (опционально, GUI-сборки) ──
* Сервисный аудио-контур (codec + джиттер-буфер + тоны) регистрируется через
* call_set_audio_ops. call.c зовёт on_media на каждый принятый медиа-кадр (с seq/ts)
* и get_stats для события CALL_STATS. Демон без аудио (headless) — ops не ставит. */
struct call_audio_ops {
/* uasync-поток: принятый Opus-кадр пира (декод + jitter-буфер). */
void (*on_media)(struct UTUN_INSTANCE* inst, uint64_t call_id,
uint16_t seq, uint32_t ts_ms, const uint8_t* opus, int len);
/* uasync-поток: текущие метрики джиттер-буфера (buffer_ms, tempo_x100). */
int (*get_stats)(uint64_t call_id, int* buffer_ms, int* tempo_x100);
};
void call_set_audio_ops(struct UTUN_INSTANCE* inst, const struct call_audio_ops* ops);
/* Отправить Opus-кадр из ДРУГОГО потока (GUI-аудио): post в uasync через trampoline. */
struct call_media_arg {
struct UTUN_INSTANCE* inst;
uint64_t call_id;
uint16_t len;
uint8_t opus[256];
};
void call_send_media_trampoline(void* arg);
/* ── trampoline-структуры для gui_bridge_post_uasync_fn / uasync_post ── */
struct call_start_arg {
struct UTUN_INSTANCE* inst;
char ch_id[64];
uint64_t peer_node_id;
uint64_t call_id;
};
struct call_id_arg {
struct UTUN_INSTANCE* inst;
uint64_t call_id;
};
void call_start_trampoline (void* arg);
void call_accept_trampoline (void* arg);
void call_decline_trampoline(void* arg);
void call_hangup_trampoline (void* arg);
#ifdef __cplusplus
}
#endif
#endif /* CALL_H */
+283
View File
@@ -0,0 +1,283 @@
// call_audio.c — сервисный аудио-движок P2P-звонка (Opus + джиттер + тоны).
//
// Один активный звонок: encoder (аудио-поток), decoder (uasync-поток),
// адаптивный джиттер-буфер ajb (SoundTouch) и генератор тонов.
//
// RX: call.c → call_audio_on_media (uasync) → decode → ajb_push;
// аудио-поток call_audio_pull_pcm → ajb_pull (+ глитч/завершение).
// TX: аудио-поток call_audio_feed_pcm → encode → uasync_post(call_send_media).
#include "call_audio.h"
#include "call.h"
#include "audio_jitter.h"
#include "call_tones.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 <pthread.h>
#include <string.h>
#define CALL_AUDIO_ID "call_audio"
#define CALL_AUDIO_BITRATE 32000
#define CALL_AJ_MAX_FRAMES 50 /* ~1000мс @20мс */
#define CALL_STALL_TB 5000 /* 500мс без медиа → глитч-тон */
static pthread_mutex_t g_mtx = PTHREAD_MUTEX_INITIALIZER;
static struct UTUN_INSTANCE* g_inst = NULL;
static int g_active = 0;
static int g_ending = 0;
static int g_ended_started = 0; /* только аудио-поток: тон завершения запущен */
static uint64_t g_call_id = 0;
static uint64_t g_last_media_tb = 0;
static opus_codec_encoder_t* g_encoder = NULL;
static opus_codec_decoder_t* g_decoder = NULL;
static struct ajb* g_ajb = NULL;
static struct call_tones* g_tones = NULL;
static const struct call_audio_ops g_ops = {
.on_media = call_audio_on_media,
.get_stats = call_audio_get_stats,
};
int call_audio_init(struct UTUN_INSTANCE* inst) {
if (!inst) return -1;
if (g_inst == inst) return 0;
g_inst = inst;
call_set_audio_ops(inst, &g_ops);
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: initialized", CALL_AUDIO_ID);
return 0;
}
void call_audio_destroy(struct UTUN_INSTANCE* inst) {
if (!inst || g_inst != inst) return;
call_set_audio_ops(inst, NULL);
g_inst = NULL;
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: destroyed", CALL_AUDIO_ID);
}
/* ── uasync-поток: приём медиа (декод + ajb_push) ── */
void call_audio_on_media(struct UTUN_INSTANCE* inst, uint64_t call_id,
uint16_t seq, uint32_t ts_ms, const uint8_t* opus, int len) {
(void)inst; (void)seq; (void)ts_ms;
int16_t pcm[CALL_AUDIO_FRAME_SAMPLES];
int n = 0;
pthread_mutex_lock(&g_mtx);
if (g_active && call_id == g_call_id && g_decoder && opus && len > 0) {
n = opus_codec_decode(g_decoder, opus, len, pcm, CALL_AUDIO_FRAME_SAMPLES);
if (n > 0 && g_ajb) {
ajb_push(g_ajb, pcm, n);
g_last_media_tb = get_time_tb();
}
}
pthread_mutex_unlock(&g_mtx);
}
/* ── GUI-поток: lifecycle ── */
int call_audio_start(uint64_t call_id) {
if (call_id == 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: start bad call_id=0", CALL_AUDIO_ID);
return -1;
}
pthread_mutex_lock(&g_mtx);
if (g_active && g_call_id == call_id) {
pthread_mutex_unlock(&g_mtx);
return 0;
}
if (g_active) {
pthread_mutex_unlock(&g_mtx);
call_audio_release(g_call_id);
pthread_mutex_lock(&g_mtx);
}
opus_codec_encoder_t* enc = opus_codec_encoder_create(CALL_AUDIO_SAMPLE_RATE, 1);
if (!enc) {
pthread_mutex_unlock(&g_mtx);
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: encoder create failed", CALL_AUDIO_ID);
return -1;
}
opus_codec_encoder_bitrate_set(enc, CALL_AUDIO_BITRATE);
opus_codec_decoder_t* dec = opus_codec_decoder_create(CALL_AUDIO_SAMPLE_RATE, 1);
if (!dec) {
opus_codec_encoder_destroy(enc);
pthread_mutex_unlock(&g_mtx);
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: decoder create failed", CALL_AUDIO_ID);
return -1;
}
struct ajb* j = ajb_create(CALL_AUDIO_SAMPLE_RATE, 1, CALL_AUDIO_FRAME_SAMPLES, CALL_AUDIO_FRAME_MS, CALL_AJ_MAX_FRAMES);
if (!j) {
opus_codec_decoder_destroy(dec);
opus_codec_encoder_destroy(enc);
pthread_mutex_unlock(&g_mtx);
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: ajb create failed", CALL_AUDIO_ID);
return -1;
}
struct call_tones* tones = call_tones_create();
if (!tones) {
ajb_destroy(j);
opus_codec_decoder_destroy(dec);
opus_codec_encoder_destroy(enc);
pthread_mutex_unlock(&g_mtx);
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: tones create failed", CALL_AUDIO_ID);
return -1;
}
g_encoder = enc;
g_decoder = dec;
g_ajb = j;
g_tones = tones;
g_call_id = call_id;
g_active = 1;
g_ending = 0;
g_ended_started = 0;
g_last_media_tb = 0;
pthread_mutex_unlock(&g_mtx);
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: started call=0x%016llx", CALL_AUDIO_ID, (unsigned long long)call_id);
return 0;
}
void call_audio_begin_end(uint64_t call_id) {
pthread_mutex_lock(&g_mtx);
if (g_active && call_id == g_call_id && !g_ending) {
g_ending = 1;
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: begin_end call=0x%016llx (ended tone)", CALL_AUDIO_ID, (unsigned long long)call_id);
}
pthread_mutex_unlock(&g_mtx);
}
void call_audio_release(uint64_t call_id) {
opus_codec_encoder_t* enc = NULL;
opus_codec_decoder_t* dec = NULL;
struct ajb* j = NULL;
struct call_tones* tones = NULL;
int was_active = 0;
pthread_mutex_lock(&g_mtx);
if (g_active && g_call_id == call_id) {
was_active = 1;
g_active = 0;
g_ending = 0;
g_ended_started = 0;
g_call_id = 0;
g_last_media_tb = 0;
enc = g_encoder; g_encoder = NULL;
dec = g_decoder; g_decoder = NULL;
j = g_ajb; g_ajb = NULL;
tones = g_tones; g_tones = NULL;
}
pthread_mutex_unlock(&g_mtx);
if (!was_active) return;
/* дальше on_media не тронет объекты (active=0, указатели NULL) */
if (enc) opus_codec_encoder_destroy(enc);
if (dec) opus_codec_decoder_destroy(dec);
if (j) ajb_destroy(j);
if (tones) call_tones_destroy(tones);
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: released call=0x%016llx", CALL_AUDIO_ID, (unsigned long long)call_id);
}
/* ── аудио-поток: TX / RX ── */
int call_audio_feed_pcm(uint64_t call_id, const int16_t* pcm, int count) {
if (!pcm || count != CALL_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 || call_id != g_call_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 || !ua || !inst) return -1;
struct call_media_arg* a = (struct call_media_arg*)u_malloc(sizeof(*a));
if (!a) return -1;
a->inst = inst;
a->call_id = call_id;
a->len = (uint16_t)len;
memcpy(a->opus, opus, (size_t)len);
uasync_post(ua, call_send_media_trampoline, a);
return 0;
}
int call_audio_pull_pcm(uint64_t call_id, int16_t* out, int max_samples) {
if (!out || max_samples <= 0) return 0;
int active = 0, ending = 0;
struct ajb* j = NULL;
struct call_tones* tones = NULL;
uint64_t last_media = 0;
pthread_mutex_lock(&g_mtx);
if (g_active && call_id == g_call_id) {
active = 1;
ending = g_ending;
j = g_ajb;
tones = g_tones;
last_media = g_last_media_tb;
}
pthread_mutex_unlock(&g_mtx);
if (!active || !j || !tones) return 0;
/* завершение: доигрываем тон (450мс), затем тишина. Тон стартует лениво
* здесь (аудио-поток) — begin_end лишь выставляет флаг, чтобы не гонять
* генератор тонов с GUI-потоком. */
if (ending) {
if (!g_ended_started) {
call_tones_ended_start(tones);
g_ended_started = 1;
}
if (call_tones_ended_pending(tones))
return call_tones_render(tones, out, max_samples);
return 0;
}
int n = ajb_pull(j, out, max_samples);
if (n >= max_samples) return n;
/* недобор: либо сталь (глитч-тон), либо тишина */
uint64_t now = get_time_tb();
int stalled = (last_media != 0) && (now - last_media > CALL_STALL_TB);
if (stalled) {
call_tones_glitch_start(tones);
int m = call_tones_render(tones, out + n, max_samples - n);
n += m;
} else {
call_tones_glitch_stop(tones);
}
for (int i = n; i < max_samples; i++) out[i] = 0;
return max_samples;
}
/* ── uasync-поток: метрики ── */
int call_audio_get_stats(uint64_t call_id, int* buffer_ms, int* tempo_x100) {
int ok = 0;
pthread_mutex_lock(&g_mtx);
if (g_active && call_id == g_call_id && g_ajb) {
ajb_get_stats(g_ajb, buffer_ms, tempo_x100);
ok = 1;
}
pthread_mutex_unlock(&g_mtx);
return ok ? 0 : -1;
}
+61
View File
@@ -0,0 +1,61 @@
#ifndef CALL_AUDIO_H
#define CALL_AUDIO_H
// call_audio.h — сервисный аудио-движок P2P-звонка (GUI-сборки: desktop/Android).
//
// Владеет Opus codec, адаптивным джиттер-буфером (SoundTouch time-stretch) и
// тонами (глитч при сталле / завершение). Потребитель (desktop miniaudio или
// Android AudioTrack) получает финальный PCM через call_audio_pull_pcm и отдаёт
// захваченный PCM через call_audio_feed_pcm.
//
// Потоки:
// - call_audio_start / begin_end / release — GUI-поток (по ACCEPTED/ENDED);
// - call_audio_feed_pcm / pull_pcm — аудио-поток (single-writer);
// - call_audio_on_media / get_stats — uasync-поток (из call.c).
// Общие указатели (encoder/decoder/ajb/tones) защищены мьютексом; encoder —
// только аудио-поток, decoder+ajb_push — только uasync-поток.
#include <stdint.h>
struct UTUN_INSTANCE;
#ifdef __cplusplus
extern "C" {
#endif
#define CALL_AUDIO_SAMPLE_RATE 48000
#define CALL_AUDIO_FRAME_MS 20
#define CALL_AUDIO_FRAME_SAMPLES (CALL_AUDIO_SAMPLE_RATE * CALL_AUDIO_FRAME_MS / 1000) /* 960 */
/* Зарегистрировать аудио-движок в call.c (идемпотентно). */
int call_audio_init(struct UTUN_INSTANCE* inst);
void call_audio_destroy(struct UTUN_INSTANCE* inst);
/* GUI-поток: создать codec + jitter + tones для звонка. 0 = успех. */
int call_audio_start(uint64_t call_id);
/* GUI-поток: по CALL_ENDED — начать тон завершения (pull_pcm выдаст 450мс тона). */
void call_audio_begin_end(uint64_t call_id);
/* GUI-поток (после остановки аудиоустройства): освободить все объекты. */
void call_audio_release(uint64_t call_id);
/* аудио-поток: закодировать PCM-кадр (ровно CALL_AUDIO_FRAME_SAMPLES) и отправить пиру. */
int call_audio_feed_pcm(uint64_t call_id, const int16_t* pcm, int count);
/* аудио-поток: выдать до max_samples финального PCM (jitter + stretch + тоны).
* Возвращает число сэмплов (может быть < max — андерфлоу/конец тона). */
int call_audio_pull_pcm(uint64_t call_id, int16_t* out, int max_samples);
/* uasync-поток (из call.c): принятый Opus-кадр — декод → ajb_push. */
void call_audio_on_media(struct UTUN_INSTANCE* inst, uint64_t call_id,
uint16_t seq, uint32_t ts_ms, const uint8_t* opus, int len);
/* uasync-поток: метрики джиттер-буфера для CALL_STATS. */
int call_audio_get_stats(uint64_t call_id, int* buffer_ms, int* tempo_x100);
#ifdef __cplusplus
}
#endif
#endif /* CALL_AUDIO_H */
+278
View File
@@ -0,0 +1,278 @@
// call_headless.c — headless-аудио-сокет для P2P-звонков
//
// TCP-сокет с бинарным фреймингом [type:1][call_id:8][len:2][data:len].
// Клиент привязывается к звонку (HELLO) и шлёт/принимает Opus-кадры (FRAME).
// Сигналинг — на control-сокете (chat_headless_control), медиа — здесь.
#include "call_headless.h"
#include "call.h"
#include "../utun_instance.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 CH_ID "call_audio"
#define AUDIO_RECV_BUF_SIZE 65536
#define AUDIO_SEND_BUF_SIZE 65536
#define AUDIO_MAX_CLIENTS 16
#define AUDIO_MAX_FRAME 2048
struct call_audio_client {
socket_t fd;
void* socket_id;
uint64_t call_id; /* 0 = не привязан */
uint8_t recv_buf[AUDIO_RECV_BUF_SIZE];
int recv_len;
uint8_t send_buf[AUDIO_SEND_BUF_SIZE];
int send_len;
int send_offset;
int closing;
struct call_audio* ca;
struct call_audio_client* next;
};
struct call_audio {
struct UASYNC* ua;
struct UTUN_INSTANCE* inst;
struct call_audio_client* clients;
int client_count;
socket_t listen_fd;
void* listen_sock_id;
int running;
};
static struct call_audio* ca_of(struct UTUN_INSTANCE* inst) {
return inst ? (struct call_audio*)inst->call_audio : NULL;
}
/* ── отправка фрейма клиенту (с буферизацией как в control-сокете) ── */
static void ca_send_raw(struct call_audio_client* cli, const uint8_t* data, size_t len) {
struct call_audio* ca = cli->ca;
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 (ca && ca->ua) uasync_set_socket_write(ca->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 (ca && ca->ua) uasync_set_socket_write(ca->ua, cli->socket_id, 1);
}
}
static void ca_write_frame(struct call_audio_client* cli, uint8_t type, uint64_t call_id,
const uint8_t* data, uint16_t len) {
uint8_t hdr[CALL_AUDIO_HDR_SIZE];
hdr[0] = type;
memcpy(hdr + 1, &call_id, 8);
hdr[9] = (uint8_t)(len >> 8);
hdr[10] = (uint8_t)(len & 0xFF);
ca_send_raw(cli, hdr, sizeof(hdr));
if (len) ca_send_raw(cli, data, len);
}
/* ── приём медиа от пира → клиенту ── */
static void ca_media_recv(struct UTUN_INSTANCE* inst, uint64_t call_id,
const uint8_t* opus, int len, void* arg) {
struct call_audio* ca = (struct call_audio*)arg;
(void)inst;
if (!ca) return;
for (struct call_audio_client* c = ca->clients; c; c = c->next) {
if (c->call_id == call_id && !c->closing) {
ca_write_frame(c, CALL_AUDIO_FRAME, call_id, opus, (uint16_t)len);
break;
}
}
}
/* ── разбор входящих фреймов ── */
static void ca_close_client(struct call_audio_client* cli);
static void ca_parse_frames(struct call_audio_client* cli) {
struct call_audio* ca = cli->ca;
uint8_t* p = cli->recv_buf;
int avail = cli->recv_len;
int consumed = 0;
while (avail - consumed >= CALL_AUDIO_HDR_SIZE) {
uint8_t type = p[consumed];
uint64_t call_id; memcpy(&call_id, p + consumed + 1, 8);
uint16_t flen = (uint16_t)((p[consumed + 9] << 8) | p[consumed + 10]);
if (flen > AUDIO_MAX_FRAME) {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: frame too big %u — closing", CH_ID, flen);
ca_close_client(cli);
return;
}
if (avail - consumed < (int)(CALL_AUDIO_HDR_SIZE + flen)) break; /* нужен ещё кусок */
const uint8_t* data = p + consumed + CALL_AUDIO_HDR_SIZE;
if (type == CALL_AUDIO_HELLO) {
cli->call_id = call_id;
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: audio client bound to call_id=%016llx", CH_ID, (unsigned long long)call_id);
} else if (type == CALL_AUDIO_FRAME) {
if (call_send_media(ca->inst, call_id, data, flen) != 0)
DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: send_media failed call_id=%016llx len=%u", CH_ID, (unsigned long long)call_id, flen);
} else {
DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: unknown frame type=0x%02x — closing", CH_ID, type);
ca_close_client(cli);
return;
}
consumed += CALL_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 ca_write_callback(socket_t fd, void* arg) {
struct call_audio_client* cli = (struct call_audio_client*)arg;
struct call_audio* ca = cli ? cli->ca : NULL;
if (!cli || cli->closing || fd != cli->fd) return;
if (cli->send_len <= 0) { if (ca && ca->ua) uasync_set_socket_write(ca->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) { cli->send_len = 0; cli->send_offset = 0; if (ca && ca->ua) uasync_set_socket_write(ca->ua, cli->socket_id, 0); return; }
cli->send_offset += (int)r;
if (cli->send_offset >= cli->send_len) { cli->send_len = 0; cli->send_offset = 0; if (ca && ca->ua) uasync_set_socket_write(ca->ua, cli->socket_id, 0); }
}
static void ca_close_client(struct call_audio_client* cli) {
struct call_audio* ca = cli ? cli->ca : NULL;
if (!cli || cli->closing) return;
cli->closing = 1;
if (ca && ca->ua && cli->socket_id) uasync_remove_socket_t(ca->ua, cli->fd);
if (cli->fd != SOCKET_INVALID) { socket_close_wrapper(cli->fd); cli->fd = SOCKET_INVALID; }
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: audio client closed call_id=%016llx", CH_ID, (unsigned long long)cli->call_id);
}
static void ca_read_callback(socket_t fd, void* arg) {
struct call_audio_client* cli = (struct call_audio_client*)arg;
if (!cli || cli->closing) return;
uint8_t tmp[4096];
ssize_t r = recv(fd, tmp, (int)sizeof(tmp), 0);
if (r <= 0) { ca_close_client(cli); return; }
if (cli->recv_len + (int)r > AUDIO_RECV_BUF_SIZE) {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: recv buffer overflow — closing", CH_ID);
ca_close_client(cli);
return;
}
memcpy(cli->recv_buf + cli->recv_len, tmp, (size_t)r);
cli->recv_len += (int)r;
ca_parse_frames(cli);
}
static void ca_accept_callback(socket_t fd, void* arg) {
struct call_audio* ca = (struct call_audio*)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_CALL, "%s: accept failed: %s", CH_ID, socket_strerror(e));
return;
}
socket_set_nonblocking(cfd);
if (ca->client_count >= AUDIO_MAX_CLIENTS) { socket_close_wrapper(cfd); DEBUG_WARN(DEBUG_CATEGORY_CALL, "%s: max clients reached", CH_ID); return; }
struct call_audio_client* cli = u_calloc(1, sizeof(*cli));
if (!cli) { socket_close_wrapper(cfd); return; }
cli->fd = cfd;
cli->ca = ca;
cli->socket_id = uasync_add_socket_t(ca->ua, cfd, ca_read_callback, ca_write_callback, NULL, cli);
if (!cli->socket_id) { u_free(cli); socket_close_wrapper(cfd); DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: failed to register client fd", CH_ID); return; }
cli->next = ca->clients; ca->clients = cli; ca->client_count++;
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: audio client connected fd=%d total=%d", CH_ID, (int)cfd, ca->client_count);
}
/* ── периодическая очистка закрытых клиентов ── */
static void ca_cleanup(void* arg) {
struct call_audio* ca = (struct call_audio*)arg;
struct call_audio_client** prev = &ca->clients;
while (*prev) {
if ((*prev)->closing) {
struct call_audio_client* dead = *prev; *prev = dead->next;
if (dead->fd != SOCKET_INVALID) socket_close_wrapper(dead->fd);
u_free(dead); ca->client_count--;
} else prev = &(*prev)->next;
}
if (ca->ua) uasync_set_timeout(ca->ua, 5000, ca, ca_cleanup, "ca_cleanup");
}
/* ── init / destroy ── */
int call_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 call_audio* ca = ca_of(inst);
if (ca && ca->running) return 0;
if (!ca) {
ca = u_calloc(1, sizeof(*ca));
if (!ca) return -1;
inst->call_audio = ca;
}
ca->ua = ua; ca->inst = inst;
ca->listen_fd = socket(AF_INET, SOCK_STREAM, 0);
if (ca->listen_fd == SOCKET_INVALID) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: socket() failed: %s", CH_ID, socket_strerror(socket_get_error())); return -1; }
socket_set_reuseaddr(ca->listen_fd, 1);
socket_set_nonblocking(ca->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(ca->listen_fd); ca->listen_fd = SOCKET_INVALID; return -1; }
if (bind(ca->listen_fd, (struct sockaddr*)&sin, sizeof(sin)) < 0) {
DEBUG_ERROR(DEBUG_CATEGORY_CALL, "%s: bind(%s:%d) failed: %s", CH_ID, bind_ip, port, socket_strerror(socket_get_error()));
socket_close_wrapper(ca->listen_fd); ca->listen_fd = SOCKET_INVALID; return -1;
}
if (listen(ca->listen_fd, 5) < 0) { socket_close_wrapper(ca->listen_fd); ca->listen_fd = SOCKET_INVALID; return -1; }
ca->listen_sock_id = uasync_add_socket_t(ua, ca->listen_fd, ca_accept_callback, NULL, NULL, ca);
if (!ca->listen_sock_id) { socket_close_wrapper(ca->listen_fd); ca->listen_fd = SOCKET_INVALID; return -1; }
call_set_media_recv_cb(inst, ca_media_recv, ca);
uasync_set_timeout(ua, 5000, ca, ca_cleanup, "ca_cleanup");
ca->running = 1;
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: audio socket listening on %s:%d", CH_ID, bind_ip, port);
return 0;
}
void call_headless_destroy(struct UTUN_INSTANCE* inst) {
struct call_audio* ca = ca_of(inst);
if (!ca || !ca->running) return;
ca->running = 0;
call_set_media_recv_cb(inst, NULL, NULL);
if (ca->listen_sock_id && ca->ua) uasync_remove_socket_t(ca->ua, ca->listen_fd);
if (ca->listen_fd != SOCKET_INVALID) { socket_close_wrapper(ca->listen_fd); ca->listen_fd = SOCKET_INVALID; }
struct call_audio_client* c = ca->clients;
while (c) {
struct call_audio_client* n = c->next;
if (c->fd != SOCKET_INVALID) socket_close_wrapper(c->fd);
u_free(c);
c = n;
}
ca->clients = NULL; ca->client_count = 0;
DEBUG_INFO(DEBUG_CATEGORY_CALL, "%s: audio socket destroyed", CH_ID);
}
+34
View File
@@ -0,0 +1,34 @@
// call_headless.h — headless-аудио-сокет для P2P-звонков
//
// Отдельный TCP-сокет, через который headless-клиент (CLI/скрипт) обменивается
// Opus-кадрами с ядром звонка. Сигналинг идёт по control-сокету (chat_headless_control),
// аудио — по этому сокету.
//
// Бинарный фрейм (обе стороны):
// [type:1][call_id:8][len:2][data:len]
// type = 0x01 HELLO (клиент → ядро: привязать соединение к call_id, data пусто)
// type = 0x02 FRAME (Opus-кадр: клиент → ядро = отправить пиру; ядро → клиент = кадр от пира)
#ifndef CALL_HEADLESS_H
#define CALL_HEADLESS_H
#include <stdint.h>
struct UASYNC;
struct UTUN_INSTANCE;
#define CALL_AUDIO_HELLO 0x01
#define CALL_AUDIO_FRAME 0x02
#define CALL_AUDIO_HDR_SIZE 11 /* type(1) + call_id(8) + len(2) */
#ifdef __cplusplus
extern "C" {
#endif
int call_headless_init(struct UASYNC* ua, struct UTUN_INSTANCE* inst,
const char* bind_ip, int port);
void call_headless_destroy(struct UTUN_INSTANCE* inst);
#ifdef __cplusplus
}
#endif
#endif /* CALL_HEADLESS_H */
+95
View File
@@ -0,0 +1,95 @@
// call_proto.h — wire-формат и константы P2P аудио-звонка (svc_id ETCP_RT_ID_CALL 0x36)
//
// Приём/отправка через etcp_router. На проводе первый байт payload = subcmd.
// Сигналинг идёт с ROUTE_CRYPTO_SIGN|ROUTE_CRYPTO_ENCRYPT, медиа — без подписи
// (уже внутри шифрованного канала etcp_router).
#ifndef CALL_PROTO_H
#define CALL_PROTO_H
#include <stdint.h>
#ifdef __cplusplus
extern "C" {
#endif
/* ── подкоманды (первый байт payload после svc_id) ── */
enum {
CALL_SUBCMD_INVITE = 0x01, /* {call_id:8, group_id:8, tracks:[]} */
CALL_SUBCMD_RINGING = 0x02, /* {call_id:8} */
CALL_SUBCMD_ACCEPT = 0x03, /* {call_id:8} */
CALL_SUBCMD_DECLINE = 0x04, /* {call_id:8, reason:1} */
CALL_SUBCMD_BUSY = 0x05, /* {call_id:8} */
CALL_SUBCMD_CANCEL = 0x06, /* {call_id:8} */
CALL_SUBCMD_HANGUP = 0x07, /* {call_id:8, reason:1} */
CALL_SUBCMD_MEDIA = 0x10, /* {call_id:8, track_id:1, seq:2, ts_ms:4, payload[]} */
};
/* ── треки и кодеки (модель расширяема под видео) ── */
enum { CALL_TRACK_AUDIO = 0x01, CALL_TRACK_VIDEO = 0x02 };
enum { CALL_CODEC_OPUS = 0x01 };
/* ── причины завершения (ENDED) ── */
enum {
CALL_REASON_LOCAL_HANGUP = 0,
CALL_REASON_REMOTE_HANGUP = 1,
CALL_REASON_DECLINE = 2,
CALL_REASON_BUSY = 3,
CALL_REASON_CANCEL = 4,
CALL_REASON_NO_TRAFFIC = 5,
CALL_REASON_RING_TIMEOUT = 6,
CALL_REASON_CONNECT_FAIL = 7,
};
#pragma pack(push, 1)
struct call_pkt_track {
uint8_t type; /* CALL_TRACK_AUDIO / CALL_TRACK_VIDEO */
uint8_t codec; /* CALL_CODEC_OPUS */
uint32_t sample_rate;
uint8_t channels;
uint32_t bitrate;
};
struct call_pkt_invite {
uint8_t subcmd; /* CALL_SUBCMD_INVITE */
uint64_t call_id;
uint64_t group_id;
uint8_t num_tracks;
struct call_pkt_track tracks[1]; /* tracks[num_tracks] */
};
struct call_pkt_ringing { uint8_t subcmd; uint64_t call_id; };
struct call_pkt_accept { uint8_t subcmd; uint64_t call_id; };
struct call_pkt_decline { uint8_t subcmd; uint64_t call_id; uint8_t reason; };
struct call_pkt_busy { uint8_t subcmd; uint64_t call_id; };
struct call_pkt_cancel { uint8_t subcmd; uint64_t call_id; };
struct call_pkt_hangup { uint8_t subcmd; uint64_t call_id; uint8_t reason; };
struct call_pkt_media {
uint8_t subcmd; /* CALL_SUBCMD_MEDIA */
uint64_t call_id;
uint8_t track_id;
uint16_t seq;
uint32_t ts_ms;
/* payload[] */
};
#pragma pack(pop)
#define CALL_MEDIA_HDR_SIZE sizeof(struct call_pkt_media) /* 16 */
/* ── параметры аудио (Opus 48kHz mono, 20ms) ── */
#define CALL_OPUS_SAMPLE_RATE 48000
#define CALL_OPUS_FRAME_MS 20
#define CALL_OPUS_FRAME_SAMPLES (CALL_OPUS_SAMPLE_RATE * CALL_OPUS_FRAME_MS / 1000) /* 960 */
/* ── таймауты и ёмкости (timebase 0.1ms) ── */
#define CALL_RING_TIMEOUT_TB 450000 /* 45с — нет ответа → CANCEL */
#define CALL_NO_TRAFFIC_TB 200000 /* 20с — нет входящих медиа → HANGUP */
#define CALL_TEARDOWN_DELAY_TB 5000 /* 500мс — задержка разрыва conn_mgr после завершения */
#define CALL_MAX_TX_FRAMES 50 /* ~1с буфер отправки (50 кадров × 20мс) */
#ifdef __cplusplus
}
#endif
#endif /* CALL_PROTO_H */
+165
View File
@@ -0,0 +1,165 @@
// call_tones.c — генератор тонов звонка (48kHz mono int16).
//
// Глитч (сталл >500мс): 1000Гц, паттерн из doc/chat_call_arch.md §8 —
// пи-пи (2×150мс, gap 100мс) … тишина до 1.0с … пи(200мс) пауза(200мс) пи(200мс),
// цикл 2с, повторяется пока длится сталл.
// Завершение: 3 нисходящих тона 700/500/400 Гц × 150мс (один раз, 450мс).
//
// Генерация — синус с коротким fade in/out (без кликов). Только аудио-поток.
#include "call_tones.h"
#include <math.h>
#include <string.h>
#include <stdlib.h>
#define TONES_SAMPLE_RATE 48000
#define TONES_AMPLITUDE 0.25f /* от полной шкалы int16 */
#define TONES_FADE_SAMPLES (TONES_SAMPLE_RATE * 5 / 1000) /* 5мс */
enum tones_mode { TONES_NONE = 0, TONES_GLITCH, TONES_ENDED };
struct tone_seg {
int tone; /* 1 = звук, 0 = тишина */
int dur_ms;
int freq_hz;
};
static const struct tone_seg GLITCH_SEGS[] = {
{1, 150, 1000}, {0, 100, 0}, {1, 150, 1000}, {0, 600, 0},
{1, 200, 1000}, {0, 200, 0}, {1, 200, 1000}, {0, 400, 0},
};
#define GLITCH_SEG_COUNT ((int)(sizeof(GLITCH_SEGS) / sizeof(GLITCH_SEGS[0])))
static const struct tone_seg ENDED_SEGS[] = {
{1, 150, 700}, {1, 150, 500}, {1, 150, 400},
};
#define ENDED_SEG_COUNT ((int)(sizeof(ENDED_SEGS) / sizeof(ENDED_SEGS[0])))
struct call_tones {
enum tones_mode mode;
const struct tone_seg* segs;
int seg_count;
int seg_idx;
int seg_pos; /* сэмплов отыграно в текущем сегменте */
int ended_done;
double phase; /* фаза синуса (радианы) */
};
struct call_tones* call_tones_create(void) {
struct call_tones* t = (struct call_tones*)calloc(1, sizeof(*t));
if (!t) return NULL;
t->mode = TONES_NONE;
t->segs = NULL;
t->seg_count = 0;
t->seg_idx = 0;
t->seg_pos = 0;
t->ended_done = 0;
t->phase = 0.0;
return t;
}
void call_tones_destroy(struct call_tones* t) {
free(t);
}
void call_tones_reset(struct call_tones* t) {
if (!t) return;
t->mode = TONES_NONE;
t->segs = NULL;
t->seg_count = 0;
t->seg_idx = 0;
t->seg_pos = 0;
t->ended_done = 0;
t->phase = 0.0;
}
void call_tones_glitch_start(struct call_tones* t) {
if (!t || t->mode == TONES_GLITCH) return;
t->mode = TONES_GLITCH;
t->segs = GLITCH_SEGS;
t->seg_count = GLITCH_SEG_COUNT;
t->seg_idx = 0;
t->seg_pos = 0;
t->phase = 0.0;
}
void call_tones_glitch_stop(struct call_tones* t) {
if (!t || t->mode != TONES_GLITCH) return;
t->mode = TONES_NONE;
t->segs = NULL;
t->seg_count = 0;
t->seg_idx = 0;
t->seg_pos = 0;
t->phase = 0.0;
}
void call_tones_ended_start(struct call_tones* t) {
if (!t) return;
t->mode = TONES_ENDED;
t->segs = ENDED_SEGS;
t->seg_count = ENDED_SEG_COUNT;
t->seg_idx = 0;
t->seg_pos = 0;
t->ended_done = 0;
t->phase = 0.0;
}
int call_tones_ended_pending(struct call_tones* t) {
return t && t->mode == TONES_ENDED && !t->ended_done;
}
/* Перейти к следующему сегменту (или завершить тон). */
static void tones_advance(struct call_tones* t) {
t->seg_idx++;
t->seg_pos = 0;
t->phase = 0.0;
if (t->seg_idx >= t->seg_count) {
if (t->mode == TONES_GLITCH) {
t->seg_idx = 0; /* цикл глитча */
} else if (t->mode == TONES_ENDED) {
t->ended_done = 1;
t->mode = TONES_NONE;
t->segs = NULL;
}
}
}
int call_tones_render(struct call_tones* t, int16_t* out, int max_samples) {
if (!t || !out || max_samples <= 0) return 0;
int written = 0;
while (written < max_samples) {
if (t->mode == TONES_NONE) {
out[written++] = 0; /* тишина до конца запроса */
continue;
}
const struct tone_seg* s = &t->segs[t->seg_idx];
int seg_len = s->dur_ms * TONES_SAMPLE_RATE / 1000;
int remain = seg_len - t->seg_pos;
int n = max_samples - written;
if (n > remain) n = remain;
for (int i = 0; i < n; i++) {
int pos = t->seg_pos + i;
float v = 0.0f;
if (s->tone) {
v = (float)sin(t->phase) * TONES_AMPLITUDE * 32767.0f;
t->phase += 2.0 * M_PI * (double)s->freq_hz / TONES_SAMPLE_RATE;
/* fade in/out по краям сегмента (без кликов) */
int fade = pos < TONES_FADE_SAMPLES ? pos
: (seg_len - pos - 1) < TONES_FADE_SAMPLES ? (seg_len - pos - 1)
: TONES_FADE_SAMPLES;
if (fade < TONES_FADE_SAMPLES)
v *= (float)fade / (float)TONES_FADE_SAMPLES;
}
int iv = (int)v;
out[written + i] = (int16_t)(iv > 32767 ? 32767 : (iv < -32768 ? -32768 : iv));
}
written += n;
t->seg_pos += n;
if (t->seg_pos >= seg_len) tones_advance(t);
}
return written;
}
+32
View File
@@ -0,0 +1,32 @@
#ifndef CALL_TONES_H
#define CALL_TONES_H
// call_tones.h — генератор тонов звонка (48kHz mono int16): глитч при сталле
// и тон завершения. Используется только аудио-потоком (single-writer).
#include <stdint.h>
#ifdef __cplusplus
extern "C" {
#endif
struct call_tones;
struct call_tones* call_tones_create(void);
void call_tones_destroy(struct call_tones* t);
void call_tones_reset(struct call_tones* t); /* в тишину */
void call_tones_glitch_start(struct call_tones* t); /* запустить/продолжить цикл глитча */
void call_tones_glitch_stop(struct call_tones* t); /* прекратить глитч (тишина) */
void call_tones_ended_start(struct call_tones* t); /* 3 нисходящих тона, один раз */
int call_tones_ended_pending(struct call_tones* t); /* 1 = тон завершения ещё звучит */
/* Выдать до max_samples сэмплов (тон или тишина). Возвращает записанное число. */
int call_tones_render(struct call_tones* t, int16_t* out, int max_samples);
#ifdef __cplusplus
}
#endif
#endif /* CALL_TONES_H */
+8
View File
@@ -54,6 +54,14 @@ extern "C" {
#define CHAT_EVT_CHANNEL_DELETED 31 /* [ch_id_len:1][ch_id:var] — канал удалён локально */
#define CHAT_EVT_DM_MSG_RECEIVED 32 /* [conv_id_len:1][conv_id:var][author_node_id:8] */
#define CHAT_EVT_DM_CONV_UPDATED 33 /* [conv_id_len:1][conv_id:var] — беседа создана/изменена */
#define CHAT_EVT_CALL_INCOMING 34 /* [ch_id_len:1][ch_id:var][caller_node_id:8][call_id:8] */
#define CHAT_EVT_CALL_RINGING 35 /* [call_id:8] */
#define CHAT_EVT_CALL_ACCEPTED 36 /* [call_id:8] */
#define CHAT_EVT_CALL_ENDED 37 /* [call_id:8][reason:1] */
#define CHAT_EVT_CALL_DECLINED 38 /* [call_id:8][reason:1] */
#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]* — расшифрованные сообщения беседы */
typedef void (*chat_event_handler_fn)(struct UTUN_INSTANCE* inst, int type, const uint8_t* data, int len);
+71 -2
View File
@@ -19,6 +19,8 @@
#include "invite_build.h"
#include "../dm/dm_core.h"
#include "../dm/dm_crypto.h"
#include "../call/call.h"
#include "../call/call_proto.h"
#include "../transport_layer/secure_channel.h"
#include "../transport_layer/etcp.h"
#include "../transport_layer/etcp_connections.h"
@@ -201,6 +203,27 @@ static void hc_on_chat_event(struct UTUN_INSTANCE* inst, int type, const uint8_t
} else if (type == 33 && data && len >= 2) { /* DM_CONV_UPDATED */
uint8_t cl = data[0];
snprintf(json, sizeof(json), "{\"event\":\"dm_conv\",\"conv\":\"%.*s\"}", (int)cl, data + 1);
} else if (type == CHAT_EVT_CALL_INCOMING && data && len >= 1 + 8 + 8) { /* CALL_INCOMING */
uint8_t cl = data[0];
if (len >= 1 + (int)cl + 8 + 8) {
uint64_t caller; memcpy(&caller, data + 1 + cl, 8);
uint64_t cid; memcpy(&cid, data + 1 + cl + 8, 8);
snprintf(json, sizeof(json),
"{\"event\":\"call_incoming\",\"ch\":\"%.*s\",\"caller_id\":\"0x%016llx\",\"call_id\":\"0x%016llx\"}",
(int)cl, data + 1, (unsigned long long)caller, (unsigned long long)cid);
}
} else if (type == CHAT_EVT_CALL_RINGING && data && len >= 8) { /* CALL_RINGING */
uint64_t cid; memcpy(&cid, data, 8);
snprintf(json, sizeof(json), "{\"event\":\"call_ringing\",\"call_id\":\"0x%016llx\"}", (unsigned long long)cid);
} else if (type == CHAT_EVT_CALL_ACCEPTED && data && len >= 8) { /* CALL_ACCEPTED */
uint64_t cid; memcpy(&cid, data, 8);
snprintf(json, sizeof(json), "{\"event\":\"call_accepted\",\"call_id\":\"0x%016llx\"}", (unsigned long long)cid);
} else if (type == CHAT_EVT_CALL_ENDED && data && len >= 9) { /* CALL_ENDED */
uint64_t cid; memcpy(&cid, data, 8);
snprintf(json, sizeof(json), "{\"event\":\"call_ended\",\"call_id\":\"0x%016llx\",\"reason\":%d}", (unsigned long long)cid, data[8]);
} else if (type == CHAT_EVT_CALL_DECLINED && data && len >= 9) { /* CALL_DECLINED */
uint64_t cid; memcpy(&cid, data, 8);
snprintf(json, sizeof(json), "{\"event\":\"call_declined\",\"call_id\":\"0x%016llx\",\"reason\":%d}", (unsigned long long)cid, data[8]);
}
if (json[0]) hc_broadcast_event(hc, json);
}
@@ -650,12 +673,58 @@ static void hc_handle_dm_send(struct headless_client* cli, int id, const char* j
send_response(cli, id, "{\"sent\":true}", NULL);
}
/* ── Звонки (P2P аудио) ── */
static uint64_t hc_gen_call_id(void) {
uint64_t id = ((uint64_t)rand() << 32) ^ (uint64_t)rand();
return id ? id : 1;
}
static void hc_handle_call_start(struct headless_client* cli, int id, const char* json) {
char ch[64], node_id_str[32], call_id_str[32];
if (json_get_str(json, "ch", ch, sizeof(ch)) < 0) { send_response(cli, id, NULL, "missing 'ch' param"); return; }
if (json_get_str(json, "node_id", node_id_str, sizeof(node_id_str)) < 0) { send_response(cli, id, NULL, "missing 'node_id' param"); return; }
uint64_t peer = strtoull(node_id_str, NULL, 16);
if (!peer || !cli->hc->inst) { send_response(cli, id, NULL, "invalid node_id"); return; }
uint64_t call_id = 0;
if (json_get_str(json, "call_id", call_id_str, sizeof(call_id_str)) == 0) call_id = strtoull(call_id_str, NULL, 16);
if (!call_id) call_id = hc_gen_call_id();
call_start(cli->hc->inst, ch, peer, call_id);
char resp[256]; snprintf(resp, sizeof(resp), "{\"call_id\":\"0x%016llx\",\"calling\":true}", (unsigned long long)call_id);
send_response(cli, id, resp, NULL);
}
static void hc_handle_call_accept(struct headless_client* cli, int id, const char* json) {
char call_id_str[32];
if (json_get_str(json, "call_id", call_id_str, sizeof(call_id_str)) < 0) { send_response(cli, id, NULL, "missing 'call_id' param"); return; }
uint64_t call_id = strtoull(call_id_str, NULL, 16);
call_accept(cli->hc->inst, call_id);
send_response(cli, id, "{\"accepted\":true}", NULL);
}
static void hc_handle_call_decline(struct headless_client* cli, int id, const char* json) {
char call_id_str[32];
if (json_get_str(json, "call_id", call_id_str, sizeof(call_id_str)) < 0) { send_response(cli, id, NULL, "missing 'call_id' param"); return; }
uint64_t call_id = strtoull(call_id_str, NULL, 16);
call_decline(cli->hc->inst, call_id);
send_response(cli, id, "{\"declined\":true}", NULL);
}
static void hc_handle_call_hangup(struct headless_client* cli, int id, const char* json) {
char call_id_str[32];
if (json_get_str(json, "call_id", call_id_str, sizeof(call_id_str)) < 0) { send_response(cli, id, NULL, "missing 'call_id' param"); return; }
uint64_t call_id = strtoull(call_id_str, NULL, 16);
call_hangup(cli->hc->inst, call_id);
send_response(cli, id, "{\"hungup\":true}", 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", "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_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", "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 void hc_handle_command(struct headless_client* cli, const char* json) {
if (!cli || !json) return;
-4
View File
@@ -43,10 +43,6 @@ static uint8_t parse_adm_tags_flags(const char* adm_tags) {
if (json_flat_get(adm_tags, "supernode", buf, sizeof(buf)) == 0
&& strcmp(buf, "yes") == 0) flags |= CHAT_MEMBER_FLAG_SUPERNODE;
/* dm_storage: узел хранит offline-сообщения прямого чата (DM) */
if (json_flat_get(adm_tags, "dm_storage", buf, sizeof(buf)) == 0
&& strcmp(buf, "yes") == 0) flags |= CHAT_MEMBER_FLAG_DM_STORAGE;
/* admin: значение — накопительная строка ed...ed..., последний символ e/d */
if (json_flat_get(adm_tags, "admin", buf, sizeof(buf)) == 0) {
const char* last = buf;
-1
View File
@@ -30,7 +30,6 @@ extern "C" {
#define CHAT_MEMBER_FLAG_ADMIN 0x02
#define CHAT_MEMBER_FLAG_MODER 0x04
#define CHAT_MEMBER_FLAG_DELETED 0x08
#define CHAT_MEMBER_FLAG_DM_STORAGE 0x10
#define CHAT_MEMBER_DISPLAY_SIZE 79
+5
View File
@@ -833,6 +833,11 @@ static int parse_chatserver(const char *key, const char *value, struct global_co
global->headless_control_bind[sizeof(global->headless_control_bind) - 1] = '\0';
return 0;
}
if (strcmp(key, "call_audio_bind") == 0) {
strncpy(global->call_audio_bind, value, sizeof(global->call_audio_bind) - 1);
global->call_audio_bind[sizeof(global->call_audio_bind) - 1] = '\0';
return 0;
}
return -1;
}
+1
View File
@@ -222,6 +222,7 @@ struct global_config {
uint64_t chatserver_storage_total_size; // bytes, общий лимит хранилища (0 = безлимит)
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 аудио-сокета звонков (пусто = отключено)
struct chat_setting_state chat_settings; // chat-настройки из [chatserver] (per-instance)
// Reality-камуфляж ([reality] section)
+295 -46
View File
@@ -124,6 +124,33 @@ static int dm_share_group(struct dm_state* dm, uint64_t peer_id) {
return shared;
}
/* Собрать node_id всех мемберов общих групп (без self и дублей). До max штук.
* Используется для offline-pull: потенциальные отправители почты = мемберы общих групп. */
static int dm_shared_peers(struct dm_state* dm, uint64_t* out, int max) {
if (!dm || !out || max <= 0) return 0;
uint64_t* ch_ids = NULL;
int ch_count = 0;
if (topo_node_sqlite_get_member_channels(dm->db, dm->inst->node_id, &ch_ids, &ch_count) != 0)
return 0;
int found = 0;
for (int c = 0; c < ch_count && found < max; c++) {
char tbl[80]; snprintf(tbl, sizeof(tbl), "peers_%llu", (unsigned long long)ch_ids[c]);
char sql[256]; snprintf(sql, sizeof(sql), "SELECT node_id FROM \"%s\"", tbl);
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(dm->db, sql, -1, &st, NULL) != SQLITE_OK) continue;
while (sqlite3_step(st) == SQLITE_ROW && found < max) {
uint64_t nid = (uint64_t)sqlite3_column_int64(st, 0);
if (nid == dm->inst->node_id) continue;
int dup = 0;
for (int i = 0; i < found; i++) if (out[i] == nid) { dup = 1; break; }
if (!dup) out[found++] = nid;
}
sqlite3_finalize(st);
}
if (ch_ids) u_free(ch_ids);
return found;
}
/* Доступен ли пир для прямой доставки: живой ETCP-линк или присутствие в BGP-группе. */
static int dm_peer_reachable(struct dm_state* dm, uint64_t peer_id) {
struct ETCP_CONN* conn = instance_find_conn(dm->inst, peer_id);
@@ -315,6 +342,15 @@ static void dm_on_msg(struct dm_state* dm, const uint8_t* msg, size_t mlen) {
dm_conv_save(dm, &c);
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: auto-created conversation conv=%s peer=0x%016llx",
DM_ID, conv_id, (unsigned long long)author);
/* событие GUI: появилась новая беседа (чтобы GUI добавил её в список) */
{
size_t cl = strlen(conv_id);
uint8_t evt[1 + 64];
evt[0] = (uint8_t)cl;
memcpy(evt + 1, conv_id, cl);
chat_event_post(dm->inst, CHAT_EVT_DM_CONV_UPDATED, evt, 1 + (int)cl);
}
}
/* проверка подписи автора по всем полям до sig */
@@ -415,7 +451,51 @@ static void dm_on_hello(struct dm_state* dm, uint64_t conv_num, uint64_t from_pe
snprintf(conv_id, sizeof(conv_id), "%llu", (unsigned long long)conv_num);
struct dm_conv c;
if (dm_conv_load(dm, conv_id, &c) != 0) return;
if (dm_conv_load(dm, conv_id, &c) != 0) {
/* Беседы нет локально (первое сообщение ушло, пока мы были offline):
* автосоздать её (anti-spam: conv_id + общая группа) и ответить HELLO,
* чтобы отправитель дослал пропущенные сообщения. */
if (dm_derive_conv_id(dm->inst->node_id, from_peer) != conv_num) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: hello conv_id mismatch from 0x%016llx", DM_ID, (unsigned long long)from_peer);
return;
}
if (!dm_share_group(dm, from_peer)) {
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: hello no shared group with 0x%016llx — ignore", DM_ID, (unsigned long long)from_peer);
return;
}
memset(&c, 0, sizeof(c));
snprintf(c.conv_id, sizeof(c.conv_id), "%s", conv_id);
c.peer_node_id = from_peer;
if (dm_node_pubkeys(dm, from_peer, c.peer_x25519, c.peer_ed25519) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: hello no pubkeys for 0x%016llx", DM_ID, (unsigned long long)from_peer);
return;
}
dm_derive_content_key(dm->inst->my_keys.private_key, c.peer_x25519, c.content_key);
dm_conv_save(dm, &c);
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: hello auto-created conversation conv=%s peer=0x%016llx",
DM_ID, conv_id, (unsigned long long)from_peer);
{
size_t cl = strlen(conv_id);
uint8_t evt[1 + 64];
evt[0] = (uint8_t)cl;
memcpy(evt + 1, conv_id, cl);
chat_event_post(dm->inst, CHAT_EVT_DM_CONV_UPDATED, evt, 1 + (int)cl);
}
/* ответить HELLO (out=0,in=0) — отправитель увидит peer_in_seq < last_out_seq и дослал сообщения */
uint8_t hello[24];
uint64_t z = 0;
memcpy(hello, &conv_num, 8);
memcpy(hello + 8, &z, 8);
memcpy(hello + 16, &z, 8);
if (dm_route_send(dm, 0, from_peer, DM_SUBCMD_HELLO, hello, sizeof(hello)) != 0)
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: hello reply send failed peer=0x%016llx", DM_ID, (unsigned long long)from_peer);
else
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: hello reply → 0x%016llx conv=%s (auto-created)",
DM_ID, (unsigned long long)from_peer, conv_id);
return;
}
if (peer_in_seq < c.last_out_seq) {
sqlite3_stmt* st = NULL;
@@ -506,33 +586,60 @@ static void dm_on_conn_status(struct ETCP_CONN* conn, int status, void* arg) {
/* На подъём ЛЮБОГО соединения: HELLO для беседы с подключившимся пиром +
* PULL mailbox для бесед, чей пир не подключён напрямую (offline-доставка). */
uint64_t conv_peers[64];
int nconv = 0;
sqlite3_stmt* st = NULL;
if (sqlite3_prepare_v2(dm->db,
"SELECT conv_id, peer_node_id, last_out_seq, last_in_seq FROM dm_conversations",
-1, &st, NULL) != SQLITE_OK) return;
while (sqlite3_step(st) == SQLITE_ROW) {
const char* conv_id = (const char*)sqlite3_column_text(st, 0);
uint64_t conv_peer = (uint64_t)sqlite3_column_int64(st, 1);
uint64_t out_seq = (uint64_t)sqlite3_column_int64(st, 2);
uint64_t in_seq = (uint64_t)sqlite3_column_int64(st, 3);
uint64_t conv_num = strtoull(conv_id, NULL, 10);
-1, &st, NULL) == SQLITE_OK) {
while (sqlite3_step(st) == SQLITE_ROW) {
const char* conv_id = (const char*)sqlite3_column_text(st, 0);
uint64_t conv_peer = (uint64_t)sqlite3_column_int64(st, 1);
uint64_t out_seq = (uint64_t)sqlite3_column_int64(st, 2);
uint64_t in_seq = (uint64_t)sqlite3_column_int64(st, 3);
uint64_t conv_num = strtoull(conv_id, NULL, 10);
if (conv_peer == peer) {
uint8_t hello[24];
memcpy(hello, &conv_num, 8);
memcpy(hello + 8, &out_seq, 8);
memcpy(hello + 16, &in_seq, 8);
/* group_id берём из строки беседы — для простоты HELLO по 0, маршрутизация глобальная */
dm_route_send(dm, 0, peer, DM_SUBCMD_HELLO, hello, sizeof(hello));
}
if (nconv < (int)(sizeof(conv_peers) / sizeof(conv_peers[0])))
conv_peers[nconv++] = conv_peer;
/* пир не подключён напрямую — тянем mailbox со storage-узлов */
struct ETCP_CONN* pc = instance_find_conn(inst, conv_peer);
if (!pc || !pc->links_up || !pc->initialized) {
dm_mailbox_pull(inst, conv_peer, in_seq);
if (conv_peer == peer) {
uint8_t hello[24];
memcpy(hello, &conv_num, 8);
memcpy(hello + 8, &out_seq, 8);
memcpy(hello + 16, &in_seq, 8);
/* group_id берём из строки беседы — для простоты HELLO по 0, маршрутизация глобальная */
if (dm_route_send(dm, 0, peer, DM_SUBCMD_HELLO, hello, sizeof(hello)) != 0)
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: hello send failed conv=%s peer=0x%016llx",
DM_ID, conv_id, (unsigned long long)peer);
else
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: hello → 0x%016llx conv=%s out=%llu in=%llu",
DM_ID, (unsigned long long)peer, conv_id,
(unsigned long long)out_seq, (unsigned long long)in_seq);
}
/* пир не подключён напрямую — тянем mailbox со storage-узлов */
struct ETCP_CONN* pc = instance_find_conn(inst, conv_peer);
if (!pc || !pc->links_up || !pc->initialized) {
dm_mailbox_pull(inst, conv_peer, in_seq);
}
}
sqlite3_finalize(st);
}
/* Первое сообщение могло прийти, пока мы были offline: мемберы общих групп,
* с которыми ещё нет беседы, тоже опрашиваем mailbox (отправитель мог уйти
* offline, тогда HELLO-восстановление недоступно). */
uint64_t peers[128];
int np = dm_shared_peers(dm, peers, (int)(sizeof(peers) / sizeof(peers[0])));
for (int i = 0; i < np; i++) {
uint64_t p = peers[i];
struct ETCP_CONN* pc = instance_find_conn(inst, p);
if (pc && pc->links_up && pc->initialized) continue; /* online — HELLO-путь */
int has_conv = 0;
for (int j = 0; j < nconv; j++) if (conv_peers[j] == p) { has_conv = 1; break; }
if (has_conv) continue; /* уже обработан циклом выше */
dm_mailbox_pull(inst, p, 0);
}
sqlite3_finalize(st);
}
/* ── публичный API ── */
@@ -554,27 +661,26 @@ int dm_start(struct UTUN_INSTANCE* inst, uint64_t peer_node_id,
struct dm_conv c;
if (dm_conv_load(dm, conv_id, &c) == 0) {
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: conversation already exists conv=%s", DM_ID, conv_id);
return 0;
} else {
memset(&c, 0, sizeof(c));
snprintf(c.conv_id, sizeof(c.conv_id), "%s", conv_id);
c.peer_node_id = peer_node_id;
memcpy(c.peer_x25519, peer_x25519, 32);
memcpy(c.peer_ed25519, peer_ed25519, 32);
if (peer_name) snprintf(c.peer_name, sizeof(c.peer_name), "%s", peer_name);
c.group_id = source_ch_id ? strtoull(source_ch_id, NULL, 10) : 0;
dm_derive_content_key(inst->my_keys.private_key, c.peer_x25519, c.content_key);
if (dm_conv_save(dm, &c) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conv_save failed conv=%s", DM_ID, conv_id);
return -1;
}
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: started conv=%s peer=0x%016llx name=%s group=%llu",
DM_ID, conv_id, (unsigned long long)peer_node_id, c.peer_name,
(unsigned long long)c.group_id);
}
memset(&c, 0, sizeof(c));
snprintf(c.conv_id, sizeof(c.conv_id), "%s", conv_id);
c.peer_node_id = peer_node_id;
memcpy(c.peer_x25519, peer_x25519, 32);
memcpy(c.peer_ed25519, peer_ed25519, 32);
if (peer_name) snprintf(c.peer_name, sizeof(c.peer_name), "%s", peer_name);
c.group_id = source_ch_id ? strtoull(source_ch_id, NULL, 10) : 0;
dm_derive_content_key(inst->my_keys.private_key, c.peer_x25519, c.content_key);
if (dm_conv_save(dm, &c) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: conv_save failed conv=%s", DM_ID, conv_id);
return -1;
}
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: started conv=%s peer=0x%016llx name=%s group=%llu",
DM_ID, conv_id, (unsigned long long)peer_node_id, c.peer_name,
(unsigned long long)c.group_id);
/* событие GUI */
/* событие GUI (всегда, чтобы GUI мог открыть/выделить беседу) */
{
size_t cl = strlen(conv_id);
uint8_t evt[1 + 64];
@@ -634,15 +740,32 @@ int dm_send(struct UTUN_INSTANCE* inst, const char* conv_id, const char* content
/* доставка: доступен → прямой send, иначе → mailbox */
if (dm_peer_reachable(dm, c.peer_node_id)) {
dm_route_send(dm, c.group_id, c.peer_node_id, DM_SUBCMD_MSG, body, body_len);
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: sent direct conv=%s seq=%llu → 0x%016llx",
DM_ID, conv_id, (unsigned long long)seq, (unsigned long long)c.peer_node_id);
if (dm_route_send(dm, c.group_id, c.peer_node_id, DM_SUBCMD_MSG, body, body_len) != 0)
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: direct send failed conv=%s seq=%llu → 0x%016llx",
DM_ID, conv_id, (unsigned long long)seq, (unsigned long long)c.peer_node_id);
else
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: sent direct conv=%s seq=%llu → 0x%016llx",
DM_ID, conv_id, (unsigned long long)seq, (unsigned long long)c.peer_node_id);
} else {
dm_mailbox_put(inst, c.peer_node_id, inst->node_id, body, body_len);
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: peer offline conv=%s seq=%llu → mailbox",
DM_ID, conv_id, (unsigned long long)seq);
if (dm_mailbox_put(inst, c.peer_node_id, inst->node_id, body, body_len) != 0)
DEBUG_WARN(DEBUG_CATEGORY_DM, "%s: mailbox put failed conv=%s seq=%llu (peer offline, no storage)",
DM_ID, conv_id, (unsigned long long)seq);
else
DEBUG_INFO(DEBUG_CATEGORY_DM, "%s: peer offline conv=%s seq=%llu → mailbox",
DM_ID, conv_id, (unsigned long long)seq);
}
u_free(body);
/* событие GUI (своё сообщение тоже, чтобы GUI обновил список) */
{
size_t cl = strlen(conv_id);
uint8_t evt[1 + 64 + 8];
evt[0] = (uint8_t)cl;
memcpy(evt + 1, conv_id, cl);
uint64_t me = inst->node_id;
memcpy(evt + 1 + cl, &me, 8);
chat_event_post(inst, CHAT_EVT_DM_MSG_RECEIVED, evt, 1 + (int)cl + 8);
}
return 0;
}
@@ -783,3 +906,129 @@ int dm_list_messages_json(struct UTUN_INSTANCE* inst, const char* conv_id, int c
*out_len = (size_t)off;
return 0;
}
/* ── трамплины для gui_bridge (GUI → uasync) ── */
/* Начать беседу с узлом из общей группы: находим pubkey пира в nodes, зовём dm_start. */
void dm_start_trampoline(void* arg) {
struct dm_start_req* req = (struct dm_start_req*)arg;
if (!req) return;
struct UTUN_INSTANCE* inst = req->inst;
uint64_t peer = req->peer_node_id;
char src[64]; strncpy(src, req->source_ch_id, sizeof(src) - 1); src[sizeof(src) - 1] = '\0';
char name[128]; strncpy(name, req->peer_name, sizeof(name) - 1); name[sizeof(name) - 1] = '\0';
u_free(arg);
struct dm_state* dm = dm_of(inst);
if (!dm || !dm->initialized || !inst || !peer) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: dm_start_trampoline — not initialized", DM_ID);
return;
}
if (peer == inst->node_id) return;
uint8_t x25519[32], ed25519[32];
if (dm_node_pubkeys(dm, peer, x25519, ed25519) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: dm_start_trampoline — no pubkeys for peer=0x%016llx",
DM_ID, (unsigned long long)peer);
return;
}
if (dm_start(inst, peer, x25519, ed25519, name[0] ? name : NULL, src[0] ? src : NULL) != 0)
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: dm_start_trampoline — dm_start failed peer=0x%016llx",
DM_ID, (unsigned long long)peer);
}
/* Отправить текстовое сообщение в беседу. */
void dm_send_trampoline(void* arg) {
struct dm_send_req* req = (struct dm_send_req*)arg;
if (!req) return;
struct UTUN_INSTANCE* inst = req->inst;
char conv[64]; strncpy(conv, req->conv_id, sizeof(conv) - 1); conv[sizeof(conv) - 1] = '\0';
int rc = dm_send(inst, conv, "text", req->data, req->data_len);
u_free(arg);
if (rc != 0)
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: dm_send_trampoline — dm_send failed conv=%s",
DM_ID, conv);
}
/* Собрать расшифрованные сообщения беседы и отправить в GUI бинарным событием:
* [conv_id_len:1][conv_id:var][count:2]([dir:1][seq:8][ts:8][author:8][ct_len:1][ct:var][data_len:2][data:var])* */
void dm_messages_trampoline(void* arg) {
struct dm_messages_req* req = (struct dm_messages_req*)arg;
if (!req) return;
struct UTUN_INSTANCE* inst = req->inst;
char conv_id[64]; strncpy(conv_id, req->conv_id, sizeof(conv_id) - 1); conv_id[sizeof(conv_id) - 1] = '\0';
u_free(arg);
struct dm_state* dm = dm_of(inst);
if (!dm || !dm->initialized) return;
struct dm_conv c;
if (dm_conv_load(dm, conv_id, &c) != 0) {
DEBUG_ERROR(DEBUG_CATEGORY_DM, "%s: dm_messages_trampoline — no conversation %s", DM_ID, conv_id);
return;
}
uint64_t conv_num = (uint64_t)strtoull(conv_id, NULL, 10);
sqlite3_stmt* st = NULL;
uint8_t* msgs = NULL;
size_t msgs_len = 0, msgs_cap = 0;
int count = 0;
if (sqlite3_prepare_v2(dm->db,
"SELECT dir,seq,ts,author,ct,data FROM dm_messages WHERE conv_id=? ORDER BY seq",
-1, &st, NULL) == SQLITE_OK) {
sqlite3_bind_text(st, 1, conv_id, -1, SQLITE_STATIC);
while (sqlite3_step(st) == SQLITE_ROW) {
uint8_t dir = (uint8_t)sqlite3_column_int(st, 0);
uint64_t seq = (uint64_t)sqlite3_column_int64(st, 1);
uint64_t ts = (uint64_t)sqlite3_column_int64(st, 2);
uint64_t author = (uint64_t)sqlite3_column_int64(st, 3);
const char* ct = (const char*)sqlite3_column_text(st, 4);
const uint8_t* enc = (const uint8_t*)sqlite3_column_blob(st, 5);
int enc_len = sqlite3_column_bytes(st, 5);
size_t ct_len = ct ? strnlen(ct, 255) : 0;
uint8_t plain[DM_DATA_MAX];
size_t plen = 0;
if (enc && enc_len >= (int)DM_TAG_SIZE) {
uint8_t nonce[DM_NONCE_SIZE];
dm_build_nonce(conv_num, author, seq, nonce);
if (dm_decrypt(c.content_key, nonce, enc, (size_t)enc_len, plain, &plen) != 0)
plen = 0;
}
size_t need = 1 + 8 + 8 + 8 + 1 + ct_len + 2 + plen;
if (msgs_len + need > msgs_cap) {
msgs_cap = msgs_cap ? msgs_cap * 2 : 4096;
while (msgs_cap < msgs_len + need) msgs_cap *= 2;
uint8_t* tmp = u_realloc(msgs, msgs_cap);
if (!tmp) break;
msgs = tmp;
}
uint16_t plen16 = (uint16_t)plen;
msgs[msgs_len] = dir; msgs_len += 1;
memcpy(msgs + msgs_len, &seq, 8); msgs_len += 8;
memcpy(msgs + msgs_len, &ts, 8); msgs_len += 8;
memcpy(msgs + msgs_len, &author, 8); msgs_len += 8;
msgs[msgs_len] = (uint8_t)ct_len; msgs_len += 1;
if (ct_len) { memcpy(msgs + msgs_len, ct, ct_len); msgs_len += ct_len; }
memcpy(msgs + msgs_len, &plen16, 2); msgs_len += 2;
if (plen) { memcpy(msgs + msgs_len, plain, plen); msgs_len += plen; }
count++;
}
sqlite3_finalize(st);
}
size_t cl = strlen(conv_id);
size_t evt_sz = 1 + cl + 2 + msgs_len;
uint8_t* evt = u_malloc(evt_sz);
if (!evt) { if (msgs) u_free(msgs); return; }
uint16_t u16cnt = (uint16_t)count;
evt[0] = (uint8_t)cl;
memcpy(evt + 1, conv_id, cl);
memcpy(evt + 1 + cl, &u16cnt, 2);
if (msgs_len) memcpy(evt + 1 + cl + 2, msgs, msgs_len);
chat_event_post(inst, CHAT_EVT_DM_MESSAGES, evt, (int)evt_sz);
u_free(evt);
if (msgs) u_free(msgs);
}
+30
View File
@@ -70,6 +70,36 @@ int dm_list_conversations_json(struct UTUN_INSTANCE* inst, char* buf, size_t buf
int dm_list_messages_json(struct UTUN_INSTANCE* inst, const char* conv_id, int count, int offset,
char* buf, size_t buf_size, size_t* out_len);
/* ── Трамплины для gui_bridge (GUI → uasync) ── */
/* Начать беседу с узлом из общей группы. Трамплин сам находит x25519/ed25519
* пира в таблице nodes (как headless hc_handle_dm_start), после чего вызывает
* dm_start(). Результат — событие CHAT_EVT_DM_CONV_UPDATED. */
struct dm_start_req {
struct UTUN_INSTANCE* inst;
uint64_t peer_node_id;
char source_ch_id[64];
char peer_name[128];
};
void dm_start_trampoline(void* arg);
/* Отправить текстовое сообщение в беседу (content_type="text"). */
struct dm_send_req {
struct UTUN_INSTANCE* inst;
char conv_id[64];
uint32_t data_len;
uint8_t* data; /* data_len байт (выделяется в хвосте структуры) */
};
void dm_send_trampoline(void* arg);
/* Запросить расшифрованные сообщения беседы. Результат — CHAT_EVT_DM_MESSAGES:
* [conv_id_len:1][conv_id:var][count:2]([dir:1][seq:8][ts:8][author:8][ct_len:1][ct:var][data_len:2][data:var])* */
struct dm_messages_req {
struct UTUN_INSTANCE* inst;
char conv_id[64];
};
void dm_messages_trampoline(void* arg);
#ifdef __cplusplus
}
#endif
+2 -2
View File
@@ -269,7 +269,7 @@ void dm_mailbox_destroy(struct UTUN_INSTANCE* inst) {
/* ── Публичные операции ── */
/* Собрать storage-узлы (dm_storage=yes) из общих групп без дублей. */
/* Собрать storage-узлы (storage=yes в adm_tags) из общих групп без дублей. */
int dm_mailbox_get_storage_nodes(struct UTUN_INSTANCE* inst, uint64_t* out, int max) {
struct dm_mb_state* mb = mb_of(inst);
if (!inst || !out || max <= 0 || !mb || !mb->db) return 0;
@@ -291,7 +291,7 @@ int dm_mailbox_get_storage_nodes(struct UTUN_INSTANCE* inst, uint64_t* out, int
uint64_t nid = (uint64_t)sqlite3_column_int64(st, 0);
if (nid == my_id) continue;
const char* tags = (const char*)sqlite3_column_text(st, 1);
if (tags && json_flat_get(tags, "dm_storage", buf, sizeof(buf)) == 0
if (tags && json_flat_get(tags, "storage", buf, sizeof(buf)) == 0
&& strcmp(buf, "yes") == 0) {
int dup = 0;
for (int i = 0; i < found; i++) if (out[i] == nid) { dup = 1; break; }
+3 -3
View File
@@ -2,7 +2,7 @@
* dm_mailbox.h — offline-хранение DM-сообщений на storage-узлах
*
* Когда пир недоступен (нет прямой связи), отправитель кладёт зашифрованное
* сообщение на storage-узлы (мемберы общих групп с флагом dm_storage=yes).
* сообщение на storage-узлы (мемберы общих групп с флагом storage=yes).
* Получатель при выходе в online вытягивает накопившуюся почту, проверяет
* подпись, расшифровывает и вставляет в dm_messages (dedup по seq).
*
@@ -39,8 +39,8 @@ int dm_mailbox_put(struct UTUN_INSTANCE* inst, uint64_t recipient, uint64_t send
* всех известных storage-узлов. Вызывается из uasync-потока. */
void dm_mailbox_pull(struct UTUN_INSTANCE* inst, uint64_t peer_node_id, uint64_t since_seq);
/* Собрать node_id storage-узлов (dm_storage=yes) из общих групп. До max штук.
* Возвращает количество найденных. out — буфер на max*8 байт. */
/* Собрать node_id storage-узлов (storage=yes в adm_tags) из общих групп. До max
* штук. Возвращает количество найденных. out — буфер на max*8 байт. */
int dm_mailbox_get_storage_nodes(struct UTUN_INSTANCE* inst, uint64_t* out, int max);
#ifdef __cplusplus
+1
View File
@@ -49,6 +49,7 @@ extern "C" {
#define ETCP_RT_ID_ADMIN_KEY 0x32 // chat_admin — передача канального приватного ключа (encrypted+signed)
#define ETCP_RT_ID_DM 0x34 // dm_core — прямой p2p чат (msg/ack/hello)
#define ETCP_RT_ID_DM_MAILBOX 0x35 // dm_mailbox — offline-хранение на storage-узлах (put/pull/ack)
#define ETCP_RT_ID_CALL 0x36 // call — P2P аудио-звонок (сигналинг + медиа)
// Connection status events (instance-level callback)
#define ETCP_CONN_STATUS_NEW 0 // соединение создано
+19
View File
@@ -20,6 +20,8 @@
#include "chat/chat_headless_control.h"
#include "dm/dm_core.h"
#include "dm/dm_mailbox.h"
#include "call/call.h"
#include "call/call_headless.h"
#include "broadcast.h"
#include "stcp_server.h"
#include "control_server.h"
@@ -490,6 +492,8 @@ void utun_instance_destroy(struct UTUN_INSTANCE *instance) {
chat_media_startup_backfill_destroy(instance);
dm_mailbox_destroy(instance);
dm_core_destroy(instance);
call_headless_destroy(instance);
call_destroy(instance);
chat_sync_destroy(instance);
chat_core_destroy(instance);
DEBUG_INFO(DEBUG_CATEGORY_SYS, "[DESTROY] H done — chat");
@@ -686,6 +690,7 @@ int utun_instance_init(struct UTUN_INSTANCE *instance) {
/* dm_mailbox до dm_core: dm_core_init ставит deliver-cb в mailbox */
dm_mailbox_init(instance);
dm_core_init(instance);
call_init(instance);
}
}
@@ -758,6 +763,20 @@ int utun_instance_init(struct UTUN_INSTANCE *instance) {
if (chat_headless_control_init(instance->ua, instance, ip_str, port) != 0)
DEBUG_WARN(DEBUG_CATEGORY_GENERAL, "Failed to init headless chat control (non-fatal)");
}
// Initialize headless call audio socket if configured in [chatserver]
if (instance->config->global.call_audio_bind[0] != '\0') {
char ip_str[64] = "127.0.0.1"; int port = 10001;
const char* colon = strrchr(instance->config->global.call_audio_bind, ':');
if (colon) {
size_t ip_len = (size_t)(colon - instance->config->global.call_audio_bind);
if (ip_len < sizeof(ip_str)) { memcpy(ip_str, instance->config->global.call_audio_bind, ip_len); ip_str[ip_len] = '\0'; }
port = atoi(colon + 1);
if (port <= 0 || port > 65535) port = 10001;
}
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)");
}
// Start the main loop
instance->running = 1;
+3
View File
@@ -39,6 +39,7 @@ struct chat_sync;
struct join_key_entry; /* chat_join.c */
struct dm_state;
struct dm_mb_state;
struct call_ctx; /* call/call.c */
struct uasync_s;
struct ROUTE_TABLE;
struct ETCP_CONN;
@@ -199,6 +200,8 @@ struct UTUN_INSTANCE {
struct join_key_entry* join_keys; // chat_join.c (was g_keys; собственный init/destroy)
struct dm_state* dm; // dm/dm_core.c (was g_dm)
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 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)