Acknowledge video capabilities before sending frame data

src/Makefile.am

@@ -187,6 +187,9 @@ utun_CORE_SOURCES = \
 	call/call_video_rate.h \
 	call/call_video.c \
 	call/call_video.h \
+	call/call_video_transport.c \
+	call/call_video_transport.h \
+	call/call_priv.h \
 	call/call_headless.c \
 	radio/radio.c \
 	radio/radio_headless.c \
@@ -342,6 +345,9 @@ libutun_a_SOURCES = \
 	call/call_video_rate.h \
 	call/call_video.c \
 	call/call_video.h \
+	call/call_video_transport.c \
+	call/call_video_transport.h \
+	call/call_priv.h \
 	call/call_headless.c \
 	radio/radio.c \
 	radio/radio_headless.c \

src/call/call_video_transport.c

@@ -8,7 +8,7 @@
 #include "../../lib/debug_config.h"
 #include <string.h>
 
-enum { VIDEO_CAPS = 1, VIDEO_KEY_REQUEST = 2 };
+enum { VIDEO_CAPS = 1, VIDEO_KEY_REQUEST = 2, VIDEO_CAPS_ACK = 3 };
 struct call_video_transport {
     struct call_session *session;
     struct call_video_caps local, peer;
@@ -18,6 +18,7 @@ struct call_video_transport {
     struct call_video_frame tx;
     uint8_t *data;
     uint64_t tx_started_tb, tx_last_id, key_request_tb;
+    uint64_t local_revision, peer_revision, acknowledged_revision;
     int64_t audio_end_pts, tx_last_pts;
     int have_peer, need_key;
 };
@@ -43,10 +44,12 @@ static struct call_video_transport *ensure(struct call_session *s) {
 }
 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);
+    uint8_t data[17 + CALL_VIDEO_CAPS_BYTES]; data[0] = type; memcpy(data + 1, &s->call_id, 8);
-    size_t length = 9;
+    uint64_t revision = type == VIDEO_CAPS ? v->local_revision : type == VIDEO_CAPS_ACK ? v->peer_revision : 0;
+    memcpy(data + 9, &revision, 8);
+    size_t length = 17;
     if (type == VIDEO_CAPS) {
-        if (call_video_caps_encode(&v->local, data + 9)) return -1;
+        if (call_video_caps_encode(&v->local, data + 17)) 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);
@@ -65,6 +68,7 @@ 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->acknowledged_revision != v->local_revision) 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);
@@ -113,6 +117,9 @@ int call_video_configure(struct UTUN_INSTANCE *inst, uint64_t id, const struct c
     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 (++v->local_revision == 0) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video capabilities revision exhausted"); return -1;
+    }
     if (!ops) {
         memset(&v->local, 0, sizeof(v->local));
         if (v->data) discard_tx(v, "platform_detached");
@@ -132,7 +139,8 @@ int call_video_send_frame(struct UTUN_INSTANCE *inst, const struct call_video_fr
         || (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)
+    if (!v->have_peer || v->acknowledged_revision != v->local_revision) { v->need_key = 1; return 0; }
+    if (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);
@@ -165,6 +173,8 @@ static int ready(struct UTUN_INSTANCE *inst, uint64_t group, uint64_t peer, cons
     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;
+    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) return 1;
     return v->ops.receive_ready(v->arg, f.size);
 }
 static void receive(struct ETCP_CONN *conn, struct ll_entry *entry) {
@@ -179,7 +189,7 @@ static void receive(struct ETCP_CONN *conn, struct ll_entry *entry) {
     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;
+        if (length != 17 && length != 17 + CALL_VIDEO_CAPS_BYTES) goto invalid;
         memcpy(&id, data + 1, 8);
     } else {
         if (call_video_fragment_parse(data, length, &f)) goto done;
@@ -191,14 +201,25 @@ static void receive(struct ETCP_CONN *conn, struct ll_entry *entry) {
     struct call_video_transport *v = ensure(s);
     if (!v) goto done;
     if (control) {
-        if (data[0] == VIDEO_CAPS && length == 9 + CALL_VIDEO_CAPS_BYTES) {
+        uint64_t revision; memcpy(&revision, data + 9, 8);
+        if (data[0] == VIDEO_CAPS && length == 17 + CALL_VIDEO_CAPS_BYTES && revision) {
             struct call_video_caps caps;
-            if (call_video_caps_parse(data + 9, CALL_VIDEO_CAPS_BYTES, &caps)) goto done;
+            if (call_video_caps_parse(data + 17, CALL_VIDEO_CAPS_BYTES, &caps)) goto done;
+            if (revision <= v->peer_revision) {
+                if (revision == v->peer_revision) control_send(v, VIDEO_CAPS_ACK);
+                goto done;
+            }
+            v->peer_revision = revision;
             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) {
+            control_send(v, VIDEO_CAPS_ACK);
+        } else if (data[0] == VIDEO_CAPS_ACK && length == 17 && revision && revision == v->local_revision) {
+            v->acknowledged_revision = revision;
+            DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "video capabilities acknowledged call=%llu revision=%llu",
+                        (unsigned long long)id, (unsigned long long)revision);
+        } else if (data[0] == VIDEO_KEY_REQUEST && length == 17 && !revision) {
             v->need_key = 1;
             if (v->ops.need_key) v->ops.need_key(v->arg);
         } else goto invalid;
@@ -208,7 +229,8 @@ static void receive(struct ETCP_CONN *conn, struct ll_entry *entry) {
         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);
+        else if (result < 0 || (!(f.flags & CALL_VIDEO_KEY) && call_video_receiver_needs_key(v->receiver)))
+            call_video_request_key(conn->instance, id);
     }
     goto done;
 invalid: