diff --git a/src/call/call_video_transport.c b/src/call/call_video_transport.c new file mode 100644 index 00000000..64055bfd --- /dev/null +++ b/src/call/call_video_transport.c @@ -0,0 +1,230 @@ +#include "call_video_transport.h" +#include "call_priv.h" +#include "../utun_instance.h" +#include "../routing_layer/etcp_router.h" +#include "../routing_layer/route_crypto.h" +#include "../../lib/mem.h" +#include "../../lib/u_async.h" +#include "../../lib/debug_config.h" +#include + +enum { VIDEO_CAPS = 1, VIDEO_KEY_REQUEST = 2 }; +struct call_video_transport { + struct call_session *session; + struct call_video_caps local, peer; + struct call_video_ops ops; + void *arg, *timer; + struct call_video_receiver *receiver; + struct call_video_frame tx; + uint8_t *data; + uint64_t tx_started_tb, tx_last_id, key_request_tb; + int64_t audio_end_pts, tx_last_pts; + int have_peer, need_key; +}; + +static struct call_session *active(struct UTUN_INSTANCE *inst, uint64_t id) { + struct call_session *s = inst ? call_session_find(inst->call, id) : NULL; + return s && s->state == CALL_ST_ACTIVE ? s : NULL; +} +static void tick(void *arg); +static struct call_video_transport *ensure(struct call_session *s) { + if (s->video) return s->video; + struct call_video_transport *v = u_calloc(1, sizeof(*v)); + if (!v) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video transport allocation failed"); return NULL; } + v->receiver = call_video_receiver_create(); + if (!v->receiver) { u_free(v); return NULL; } + v->session = s; v->need_key = 1; v->audio_end_pts = v->tx_last_pts = -1; + v->timer = uasync_set_timeout(s->ctx->inst->ua, 500, v, tick, "call_video"); + if (!v->timer) { + DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video timer allocation failed"); + call_video_receiver_destroy(v->receiver); u_free(v); return NULL; + } + s->video = v; return v; +} +static int control_send(struct call_video_transport *v, uint8_t type) { + struct call_session *s = v->session; + uint8_t data[9 + CALL_VIDEO_CAPS_BYTES]; data[0] = type; memcpy(data + 1, &s->call_id, 8); + size_t length = 9; + if (type == VIDEO_CAPS) { + if (call_video_caps_encode(&v->local, data + 9)) return -1; + length += CALL_VIDEO_CAPS_BYTES; + } + struct ETCP_ROUTER_CONN *conn = etcp_router_conn_get(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO_CONTROL); + if (!conn || etcp_router_conn_send(conn, data, length, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT)) { + DEBUG_WARN(DEBUG_CATEGORY_CALL, "video control send failed call=%llu type=%u", (unsigned long long)s->call_id, type); return -1; + } + return 0; +} +static void discard_tx(struct call_video_transport *v, const char *reason) { + DEBUG_WARN(DEBUG_CATEGORY_CALL, "video TX discard call=%llu frame=%llu offset=%u total=%u reason=%s", + (unsigned long long)v->session->call_id, (unsigned long long)v->tx.frame_id, v->tx.offset, v->tx.size, reason); + u_free(v->data); v->data = NULL; v->need_key = 1; + if (v->ops.need_key) v->ops.need_key(v->arg); +} +static void flush(struct call_video_transport *v) { + if (!v->data) return; + struct call_session *s = v->session; + if (get_time_tb() - v->tx_started_tb > 20000) { discard_tx(v, "deadline"); return; } + // Видео не обгоняет соответствующий аудиокадр. Отказ камеры не задерживает звук. + if (v->tx.pts_us > v->audio_end_pts) return; + struct ETCP_ROUTER_CONN *conn = etcp_router_conn_get(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO); + if (!conn) { discard_tx(v, "router_allocation"); return; } + uint8_t wire[CALL_VIDEO_HEADER_BYTES + CALL_VIDEO_FRAGMENT_BYTES]; + // Ограниченный burst одного вызова; повторяется по audio send и таймеру 50мс. + for (unsigned budget = 0; v->data && budget < 16; budget++) { + if (!etcp_router_send_q_has_room(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO)) return; + uint32_t remaining = v->tx.size - v->tx.offset; + v->tx.length = remaining < CALL_VIDEO_FRAGMENT_BYTES ? remaining : CALL_VIDEO_FRAGMENT_BYTES; + v->tx.data = v->data + v->tx.offset; + size_t length; + if (call_video_fragment_encode(&v->tx, wire, sizeof(wire), &length) + || etcp_router_conn_send(conn, wire, length, 0)) { discard_tx(v, "router_send"); return; } + v->tx.offset += v->tx.length; + if (v->tx.offset == v->tx.size) { u_free(v->data); v->data = NULL; } + } +} +static void tick(void *arg) { + struct call_video_transport *v = arg; + v->timer = NULL; + flush(v); + v->timer = uasync_set_timeout(v->session->ctx->inst->ua, 500, v, tick, "call_video"); + if (!v->timer) DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video timer rearm failed call=%llu", (unsigned long long)v->session->call_id); +} +void call_video_audio_sent(struct call_session *s, int64_t end_pts_us) { + if (!s->video || end_pts_us < 0) return; + s->video->audio_end_pts = end_pts_us; + flush(s->video); +} +void call_video_transport_stop(struct call_session *s) { + struct call_video_transport *v = s->video; + if (!v) return; + s->video = NULL; + if (v->timer) uasync_cancel_timeout(s->ctx->inst->ua, v->timer); + call_video_receiver_destroy(v->receiver); u_free(v->data); u_free(v); + // Уже принятые роутером пакеты содержат call_id и не принадлежат следующему звонку. + etcp_router_resume_receive(s->ctx->inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO); +} +int call_video_configure(struct UTUN_INSTANCE *inst, uint64_t id, const struct call_video_caps *caps, + const struct call_video_ops *ops, void *arg) { + struct call_session *s = active(inst, id); + if (!s || !call_video_caps_valid(caps) || (ops && (!ops->frame || !ops->receive_ready))) { + DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video configure invalid call/capabilities/callbacks id=%llu", (unsigned long long)id); return -1; + } + struct call_video_transport *v = ensure(s); + if (!v) return -1; + v->local = *caps; v->ops = ops ? *ops : (struct call_video_ops){0}; v->arg = arg; + if (!ops) { + memset(&v->local, 0, sizeof(v->local)); + if (v->data) discard_tx(v, "platform_detached"); + call_video_receiver_reset(v->receiver); + } + if (v->have_peer && v->ops.capabilities) v->ops.capabilities(v->arg, &v->peer); + call_video_resume_receive(inst, id); + DEBUG_INFO(DEBUG_CATEGORY_CALL, "video configured call=%llu encode=%u/%u decode=%u/%u", (unsigned long long)id, + v->local.encode[0].width, v->local.encode[1].width, v->local.decode[0].width, v->local.decode[1].width); + return control_send(v, VIDEO_CAPS); +} +int call_video_send_frame(struct UTUN_INSTANCE *inst, const struct call_video_frame *f) { + struct call_session *s = f ? active(inst, f->call_id) : NULL; + struct call_video_transport *v = s ? s->video : NULL; + if (!v || !call_video_format_valid(&f->format) || !f->data || !f->size || f->size > CALL_VIDEO_MAX_FRAME + || f->offset || f->length != f->size || !f->frame_id || f->pts_us < 0 || f->pts_us > INT64_MAX / 2 + || (f->flags & ~CALL_VIDEO_KEY) || f->frame_id <= v->tx_last_id || f->pts_us <= v->tx_last_pts) { + DEBUG_WARN(DEBUG_CATEGORY_CALL, "video TX invalid frame/session"); return -1; + } + if (!v->have_peer || call_video_choose(&v->local, &v->peer, f->format.codec, f->format.width, f->format.height, f->format.fps) + != f->format.codec) { DEBUG_WARN(DEBUG_CATEGORY_CALL, "video TX unsupported peer format"); return -1; } + if (v->data) { v->need_key = 1; return 0; } + int changed = !call_video_same_format(&f->format, &v->tx.format); + if (f->format.generation < v->tx.format.generation || (changed && f->format.generation == v->tx.format.generation)) { + DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video TX format changed without new generation"); return -1; + } + if (changed || (v->tx_last_id && f->frame_id != v->tx_last_id + 1)) v->need_key = 1; + if (v->need_key && !(f->flags & CALL_VIDEO_KEY)) return 0; + v->data = u_malloc(f->size); + if (!v->data) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video TX allocation failed bytes=%u", f->size); return -1; } + memcpy(v->data, f->data, f->size); v->tx = *f; v->tx.data = v->data; + v->tx_started_tb = get_time_tb(); v->tx_last_id = f->frame_id; v->tx_last_pts = f->pts_us; v->need_key = 0; + flush(v); return 1; +} +void call_video_resume_receive(struct UTUN_INSTANCE *inst, uint64_t id) { + struct call_session *s = active(inst, id); + if (s) etcp_router_resume_receive(inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO); +} +void call_video_request_key(struct UTUN_INSTANCE *inst, uint64_t id) { + struct call_session *s = active(inst, id); + struct call_video_transport *v = s ? s->video : NULL; + if (!v) return; + uint64_t now = get_time_tb(); + if (v->key_request_tb && now - v->key_request_tb < 5000) return; + if (!control_send(v, VIDEO_KEY_REQUEST)) v->key_request_tb = now; +} +static int ready(struct UTUN_INSTANCE *inst, uint64_t group, uint64_t peer, const uint8_t *data, size_t length) { + struct call_video_frame f; + if (call_video_fragment_parse(data, length, &f)) return 1; + struct call_session *s = active(inst, f.call_id); + struct call_video_transport *v = s ? s->video : NULL; + if (!v || s->group_id != group || s->peer_node_id != peer || !v->ops.frame || f.offset) return 1; + return v->ops.receive_ready(v->arg, f.size); +} +static void receive(struct ETCP_CONN *conn, struct ll_entry *entry) { + if (!entry || !entry->dgram || entry->len < ROUTER_SVC_PAYLOAD_OFF || !conn) { + DEBUG_WARN(DEBUG_CATEGORY_CALL, "video RX invalid router envelope"); + if (entry) { queue_dgram_free(entry); queue_entry_free(entry); } return; + } + const uint8_t *data = entry->dgram + ROUTER_SVC_PAYLOAD_OFF; + size_t length = entry->len - ROUTER_SVC_PAYLOAD_OFF; + uint64_t group, peer, id = 0; + memcpy(&group, entry->dgram + ROUTER_SVC_GROUP_OFF, 8); memcpy(&peer, entry->dgram + ROUTER_SVC_SRC_OFF, 8); + int control = entry->dgram[0] == ETCP_RT_ID_CALL_VIDEO_CONTROL; + struct call_video_frame f; + if (control) { + if (length != 9 && length != 9 + CALL_VIDEO_CAPS_BYTES) goto invalid; + memcpy(&id, data + 1, 8); + } else { + if (call_video_fragment_parse(data, length, &f)) goto done; + id = f.call_id; + } + struct call_session *s = active(conn->instance, id); + if (!s) goto done; // Уже завершённый звонок не держит router receive_q. + if (s->group_id != group || s->peer_node_id != peer) goto invalid; + struct call_video_transport *v = ensure(s); + if (!v) goto done; + if (control) { + if (data[0] == VIDEO_CAPS && length == 9 + CALL_VIDEO_CAPS_BYTES) { + struct call_video_caps caps; + if (call_video_caps_parse(data + 9, CALL_VIDEO_CAPS_BYTES, &caps)) goto done; + v->peer = caps; v->have_peer = 1; + DEBUG_INFO(DEBUG_CATEGORY_CALL, "video peer capabilities call=%llu encode=%u/%u decode=%u/%u", (unsigned long long)id, + caps.encode[0].width, caps.encode[1].width, caps.decode[0].width, caps.decode[1].width); + if (v->ops.capabilities) v->ops.capabilities(v->arg, &v->peer); + } else if (data[0] == VIDEO_KEY_REQUEST && length == 9) { + v->need_key = 1; + if (v->ops.need_key) v->ops.need_key(v->arg); + } else goto invalid; + } else if (v->ops.frame) { + if (!v->have_peer || call_video_choose(&v->peer, &v->local, f.format.codec, f.format.width, f.format.height, f.format.fps) + != f.format.codec) goto invalid; + struct call_video_frame complete; + int result = call_video_receiver_push(v->receiver, &f, get_time_us(), &complete); + if (result == 1) v->ops.frame(v->arg, &complete); + else if (call_video_receiver_needs_key(v->receiver)) call_video_request_key(conn->instance, id); + } + goto done; +invalid: + DEBUG_WARN(DEBUG_CATEGORY_CALL, "video RX invalid peer/group/control/format call=%llu peer=%llu bytes=%zu", + (unsigned long long)id, (unsigned long long)peer, length); +done: + queue_dgram_free(entry); queue_entry_free(entry); +} +int call_video_transport_bind(struct UTUN_INSTANCE *inst) { + if (etcp_router_bind(inst, ETCP_RT_ID_CALL_VIDEO_CONTROL, receive)) return -1; + if (etcp_router_bind(inst, ETCP_RT_ID_CALL_VIDEO, receive)) { + etcp_router_unbind(inst, ETCP_RT_ID_CALL_VIDEO_CONTROL); return -1; + } + etcp_router_set_receive_ready(inst, ETCP_RT_ID_CALL_VIDEO, ready); return 0; +} +void call_video_transport_unbind(struct UTUN_INSTANCE *inst) { + etcp_router_set_receive_ready(inst, ETCP_RT_ID_CALL_VIDEO, NULL); + etcp_router_unbind(inst, ETCP_RT_ID_CALL_VIDEO); etcp_router_unbind(inst, ETCP_RT_ID_CALL_VIDEO_CONTROL); +} diff --git a/src/call/call_video_transport.h b/src/call/call_video_transport.h new file mode 100644 index 00000000..b3c5e516 --- /dev/null +++ b/src/call/call_video_transport.h @@ -0,0 +1,39 @@ +/* Видеотранспорт звонка: uasync-поток, отдельные control/audio/video router channels. + * Callback не вызывает изменяющий API звонка реентерабельно; команды worker возвращает через post. + * Callback копирует нужные данные до возврата. При stop callbacks больше не вызываются. */ +#ifndef CALL_VIDEO_TRANSPORT_H +#define CALL_VIDEO_TRANSPORT_H +#include "call_video.h" +#include "call_video_rate.h" +struct UTUN_INSTANCE; +struct call_session; +#ifdef __cplusplus +extern "C" { +#endif +struct call_video_ops { + void (*capabilities)(void *arg, const struct call_video_caps *peer); + int (*receive_ready)(void *arg, uint32_t bytes); + void (*frame)(void *arg, const struct call_video_frame *frame); + void (*need_key)(void *arg); +}; +/* Настройка активного звонка, включая receive-only. NULL ops снимает платформенные callbacks. + * capabilities отражают только реально проверенные профили. 0 — успешно, -1 — ошибка. */ +int call_video_configure(struct UTUN_INSTANCE *inst, uint64_t call_id, const struct call_video_caps *caps, + const struct call_video_ops *ops, void *arg); +/* Копирует один полный кадр. 1 принят, 0 backpressure/ожидание KEY, -1 неверный формат/сессия. + * После 0 producer обязан запросить KEY: пропуск P-кадра разрывает цепочку references. + * Неполная отправка ограничена 2с; следующие кадры не увеличивают очередь. */ +int call_video_send_frame(struct UTUN_INSTANCE *inst, const struct call_video_frame *frame); +void call_video_resume_receive(struct UTUN_INSTANCE *inst, uint64_t call_id); +/* Decoder потерял references. Запрос идемпотентный, отправляется не чаще раза в 500мс. */ +void call_video_request_key(struct UTUN_INSTANCE *inst, uint64_t call_id); + +/* Только call.c: управление общей сессией и граница уже отправленного аудио. */ +int call_video_transport_bind(struct UTUN_INSTANCE *inst); +void call_video_transport_unbind(struct UTUN_INSTANCE *inst); +void call_video_transport_stop(struct call_session *session); +void call_video_audio_sent(struct call_session *session, int64_t end_pts_us); +#ifdef __cplusplus +} +#endif +#endif