Require selected epochs and verify multiple live publishers

This commit is contained in:
evgeny committed 2026-10-12 22:28:15 +02:00
1 parent e9982a2db6
commit 34349bd663
3 files changed
+15 -5

No files matched your search

+8
View File
@@ -420,6 +420,14 @@ static void test_network(void) {
CHECK(stream_publish_open(nodes[3], GROUP1, &parameters, &parallel_publisher) == STREAM_OK); CHECK(stream_publish_open(nodes[3], GROUP1, &parameters, &parallel_publisher) == STREAM_OK);
struct stream_info parallel_info; CHECK(stream_get_info(parallel_publisher, &parallel_info) == STREAM_OK); struct stream_info parallel_info; CHECK(stream_get_info(parallel_publisher, &parallel_info) == STREAM_OK);
wait_state(1, 3, GROUP1, parallel_info.descriptor.source_epoch, 0); wait_state(1, 3, GROUP1, parallel_info.descriptor.source_epoch, 0);
/* Две независимые публикации одного канала видны через общий каталог; подписчики не становятся источниками. */
char *catalog = NULL, origin_id[64], parallel_id[64], subscriber_id[64];
CHECK(stream_sources_json(nodes[1], GROUP1, &catalog) == STREAM_OK);
snprintf(origin_id, sizeof(origin_id), "\"id\":\"%llu\"", (unsigned long long)nodes[0]->node_id);
snprintf(parallel_id, sizeof(parallel_id), "\"id\":\"%llu\"", (unsigned long long)nodes[3]->node_id);
snprintf(subscriber_id, sizeof(subscriber_id), "\"id\":\"%llu\"", (unsigned long long)nodes[1]->node_id);
CHECK(strstr(catalog, origin_id) && strstr(catalog, parallel_id) && !strstr(catalog, subscriber_id)); u_free(catalog);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] shared catalogue exposes two live publishers in the same channel");
CHECK(stream_subscribe_manual_open(nodes[1], GROUP1, nodes[3]->node_id, &parallel_subscriber) == STREAM_OK); CHECK(stream_subscribe_manual_open(nodes[1], GROUP1, nodes[3]->node_id, &parallel_subscriber) == STREAM_OK);
size_t accepted = 0; size_t accepted = 0;
CHECK(stream_publish_write(parallel_publisher, expected, sizeof(expected), &accepted) == STREAM_OK && accepted == sizeof(expected)); CHECK(stream_publish_write(parallel_publisher, expected, sizeof(expected), &accepted) == STREAM_OK && accepted == sizeof(expected));
@@ -314,8 +314,11 @@ void android_stream_detach(struct UTUN_INSTANCE *inst) {
/* Команда содержит идентификаторы, а не borrowed pointer ядра. */ /* Команда содержит идентификаторы, а не borrowed pointer ядра. */
int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned width, unsigned height, int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned width, unsigned height,
unsigned rotation, int audio, uint64_t expected_epoch) { unsigned rotation, int audio, uint64_t expected_epoch) {
if (!id || !group || (!source && (!width || width > 4096 || !height || height > 4096 || rotation > 270 || rotation % 90))) { if (!id || !group || (source && !expected_epoch)
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open invalid arguments"); return -1; || (!source && (!width || width > 4096 || !height || height > 4096 || rotation > 270 || rotation % 90))) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open invalid id=%llu group=%llx source=%llx epoch=%llx dimensions=%ux%u",
(unsigned long long)id, (unsigned long long)group, (unsigned long long)source,
(unsigned long long)expected_epoch, width, height); return -1;
} }
pthread_mutex_lock(&media_lock); pthread_mutex_lock(&media_lock);
struct android_media_session *s = source ? &viewer : &publisher; struct android_media_session *s = source ? &viewer : &publisher;
@@ -407,4 +410,3 @@ uint8_t *android_stream_pull(uint64_t id, size_t *length) {
} }
pthread_mutex_unlock(&media_lock); return data; pthread_mutex_unlock(&media_lock); return data;
} }
@@ -3,7 +3,7 @@
* Один publisher и один viewer, TX <=2MB/128 records, RX <=4MB/256 records. * Один publisher и один viewer, TX <=2MB/128 records, RX <=4MB/256 records.
* Data принимает полные H.264 Annex B кадры (KEY содержит SPS/PPS/IDR) и Opus 48k mono/20ms. * Data принимает полные H.264 Annex B кадры (KEY содержит SPS/PPS/IDR) и Opus 48k mono/20ms.
* pull передаёт владение u_malloc-буфером: kind:1,flags:1,width:2,height:2,rotation:2,PTS_us:8,length:4,data. * pull передаёт владение u_malloc-буфером: kind:1,flags:1,width:2,height:2,rotation:2,PTS_us:8,length:4,data.
* kinds: video=1,audio=2,gap=3,opened=4,ended=5,error=6. Числа big-endian. * kinds: video=1,audio=2,gap=3,opened=4,ended=5,error=6; error.flags = stream_close_reason. Числа big-endian.
* Медиаданные передаются через pull, не через общий чатовый event bus. */ * Медиаданные передаются через pull, не через общий чатовый event bus. */
#ifndef ANDROID_STREAM_BRIDGE_H #ifndef ANDROID_STREAM_BRIDGE_H
#define ANDROID_STREAM_BRIDGE_H #define ANDROID_STREAM_BRIDGE_H
@@ -15,11 +15,11 @@ struct UTUN_INSTANCE;
/* 0 attached, -1 invalid/busy: нельзя передать singleton другому ядру до detach старого. */ /* 0 attached, -1 invalid/busy: нельзя передать singleton другому ядру до detach старого. */
int android_stream_attach(struct UTUN_INSTANCE *inst); int android_stream_attach(struct UTUN_INSTANCE *inst);
void android_stream_detach(struct UTUN_INSTANCE *inst); void android_stream_detach(struct UTUN_INSTANCE *inst);
/* source=0 — publisher; source!=0 — viewer с обязательным expected_epoch. Команда принята=0, результат через pull. */
int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned width, unsigned height, int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned width, unsigned height,
unsigned rotation, int audio, uint64_t expected_epoch); unsigned rotation, int audio, uint64_t expected_epoch);
/* Наблюдение выбранного канала через CHAT_EVT_STREAM_SOURCES; group=0 снимает его. */ /* Наблюдение выбранного канала через CHAT_EVT_STREAM_SOURCES; group=0 снимает его. */
int android_stream_watch(uint64_t group); int android_stream_watch(uint64_t group);
/* source=0 — publisher; source!=0 — viewer. 0 означает, что команда принята, результат приходит через pull. */
int android_stream_feed(uint64_t id, int video, uint64_t pts, int key, const uint8_t *data, size_t length); int android_stream_feed(uint64_t id, int video, uint64_t pts, int key, const uint8_t *data, size_t length);
/* feed: 1 accepted, 0 dropped (video: запросить KEY), -1 detached/stopped/invalid. */ /* feed: 1 accepted, 0 dropped (video: запросить KEY), -1 detached/stopped/invalid. */
void android_stream_stop(uint64_t id); void android_stream_stop(uint64_t id);