jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Watch Android stream sources and validate selected epochs

master
evgeny 6 days ago
parent
f91894aa1b
commit
ba3dfa9fcc
3 changed files with 60 additions and 49 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 9
      tools/chatgui-android/jni_bridge/android_stream_jni.c
  2. 96
      tools/chatgui-android/libutun_lite/stream_bridge.c
  3. 4
      tools/chatgui-android/libutun_lite/stream_bridge.h

9
tools/chatgui-android/jni_bridge/android_stream_jni.c
Unescape View File

@ -11,9 +11,14 @@
struct android_stream_opus { opus_codec_encoder_t *encoder; opus_codec_decoder_t *decoder; };
JNIEXPORT jboolean JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamOpen(
JNIEnv *env, jobject obj, jlong id, jlong group, jlong source, jint width, jint height, jint rotation, jboolean audio) {
JNIEnv *env, jobject obj, jlong id, jlong group, jlong source, jint width, jint height, jint rotation, jboolean audio, jlong epoch) {
(void)env; (void)obj;
return android_stream_open((uint64_t)id, (uint64_t)group, (uint64_t)source, width, height, rotation, audio) == 0;
return android_stream_open((uint64_t)id, (uint64_t)group, (uint64_t)source, width, height, rotation, audio, (uint64_t)epoch) == 0;
}
JNIEXPORT jboolean JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamWatch(JNIEnv *env, jobject obj, jlong group) {
(void)env; (void)obj;
return android_stream_watch((uint64_t)group) == 0;
}
JNIEXPORT jint JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamFeed(

96
tools/chatgui-android/libutun_lite/stream_bridge.c
Unescape View File

@ -30,11 +30,11 @@ struct media_record {
};
struct media_queue { struct media_record *head, *tail; size_t bytes; unsigned count; };
struct android_media_session {
uint64_t id, group, source, epoch, next_frame, last_pts[2], drain_until;
uint64_t id, group, source, epoch, expected_epoch, next_frame, last_pts[2], drain_until;
struct STREAM_HANDLE *handle;
struct stream_media_receiver *receiver;
struct stream_media_format format;
int opening, stopping, finished, need_key, seen[2], rx_overflow;
int opening, stopping, finished, need_key, seen[2], rx_overflow, reason;
};
static pthread_mutex_t media_lock = PTHREAD_MUTEX_INITIALIZER;
@ -44,6 +44,8 @@ static void *media_timer;
static struct android_media_session publisher, viewer;
static struct media_queue video_queue, audio_queue, output_queue;
static uint64_t list_group, cached_group, list_updated;
static uint64_t watch_group;
static int watch_pending;
static char *source_list;
static void media_process(void *arg);
static void media_close(struct android_media_session *s, int kind);
@ -146,6 +148,7 @@ static void media_event(struct UTUN_INSTANCE *inst, const struct stream_event *e
} else if (event->type == STREAM_EVENT_GAP) {
stream_media_receiver_gap(viewer.receiver); media_output(viewer.id, 3, NULL, 0, 0, NULL, 0);
} else {
viewer.reason = event->reason;
media_close(&viewer, event->type == STREAM_EVENT_END ? 5 : 6);
}
}
@ -154,7 +157,7 @@ static void media_event(struct UTUN_INSTANCE *inst, const struct stream_event *e
/* Publisher держит конечное окно выдачи после FIN; abrupt detach ничего не обещает получателям. */
static void media_close(struct android_media_session *s, int kind) {
if (kind) media_output(s->id, kind, NULL, 0, 0, NULL, 0);
if (kind) media_output(s->id, kind, NULL, 0, kind == 6 ? s->reason : 0, NULL, 0);
if (s->handle) stream_close(s->handle);
stream_media_receiver_destroy(s->receiver);
if (s == &publisher) { queue_clear(&video_queue); queue_clear(&audio_queue); }
@ -199,44 +202,11 @@ static int media_publish(void) {
/* Список содержит только проверенные публикации живых мемберов; SQL и BGP читаются в uasync. */
static void media_list_sources(void) {
uint64_t group_id = list_group; list_group = 0;
struct TOPO_GROUP *group = media_inst->topo_groups ? topo_groups_find(media_inst->topo_groups, group_id) : NULL;
struct strbuf json = strbuf_new();
sqlite3_stmt *statement = NULL;
int failed = strbuf_addf(&json, "[") < 0;
if (group && !failed) {
char query[256];
for (const char *p = group->channel_id; *p; p++) if (*p < '0' || *p > '9') {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream source namespace invalid group=%llx", (unsigned long long)group_id);
failed = 1; break;
}
snprintf(query, sizeof(query), "SELECT p.node_id,COALESCE(NULLIF(p.local_nick,''),n.name,'') "
"FROM \"peers_%s\" p LEFT JOIN nodes n ON p.node_id=n.node_id", group->channel_id);
sqlite3 *db = chat_core_get_db(media_inst);
int result = failed ? SQLITE_ERROR : sqlite3_prepare_v2(db, query, -1, &statement, NULL), count = 0;
if (result == SQLITE_OK) {
while ((result = sqlite3_step(statement)) == SQLITE_ROW) {
uint64_t node = (uint64_t)sqlite3_column_int64(statement, 0);
if (!stream_member_online(media_inst, group_id, node)) continue;
struct stream_member_state state;
if (stream_member_load(media_inst, group_id, node, &state, NULL) != STREAM_OK || !state.published) continue;
if (count >= 256) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android stream source list truncated group=%llx", (unsigned long long)group_id); break; }
if (strbuf_addf(&json, "%s{\"id\":\"%llu\",\"epoch\":\"%016llx\",\"self\":%d,\"name\":", count ? "," : "",
(unsigned long long)node, (unsigned long long)state.source.source_epoch, node == media_inst->node_id) < 0
|| strbuf_add_json_string(&json, (const char *)sqlite3_column_text(statement, 1)) < 0
|| strbuf_addf(&json, "}") < 0) { failed = 1; break; }
count++;
}
}
if (result != SQLITE_DONE && !(result == SQLITE_ROW && (failed || count == 256))) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream list query failed group=%llx: %s",
(unsigned long long)group_id, sqlite3_errmsg(db)); failed = 1;
}
}
sqlite3_finalize(statement);
if (strbuf_addf(&json, "]") < 0) failed = 1;
u_free(source_list); source_list = failed ? NULL : strbuf_detach(&json);
strbuf_free(&json); cached_group = group_id; list_updated = get_time_us();
if (failed) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream source snapshot failed group=%llx", (unsigned long long)group_id);
u_free(source_list); source_list = NULL;
int result = stream_sources_json(media_inst, group_id, &source_list);
cached_group = group_id; list_updated = get_time_us();
if (result < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream source snapshot failed group=%llx result=%d",
(unsigned long long)group_id, result);
}
/* Один actor обслуживает команды и не больше двух FEC-пачек за вызов; idle не будит мобильное ядро. */
@ -245,6 +215,11 @@ static void media_process(void *arg) {
pthread_mutex_lock(&media_lock);
media_post = NULL; media_timer = NULL;
if (!media_inst) { pthread_mutex_unlock(&media_lock); return; }
if (watch_pending) {
watch_pending = 0;
if (stream_watch_sources(media_inst, watch_group) != STREAM_OK)
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream source watch failed group=%llx", (unsigned long long)watch_group);
}
if (list_group) media_list_sources();
struct android_media_session *sessions[] = {&publisher, &viewer};
for (unsigned i = 0; i < 2; i++) {
@ -252,13 +227,31 @@ static void media_process(void *arg) {
if (s->opening) {
if (s->stopping) { media_close(s, 5); continue; }
s->opening = 0;
int result = s->source ? stream_subscribe_open(media_inst, s->group, s->source, &s->handle)
: stream_publish_open(media_inst, s->group, NULL, &s->handle);
int result = STREAM_OK;
s->reason = STREAM_CLOSE_FAILED;
if (s->source && s->expected_epoch) {
struct stream_member_state state;
result = stream_member_load(media_inst, s->group, s->source, &state, NULL);
if (result == STREAM_OK && (!state.published || state.source.source_epoch != s->expected_epoch)) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android selected source changed group=%llx source=%llx expected=%llx actual=%llx",
(unsigned long long)s->group, (unsigned long long)s->source, (unsigned long long)s->expected_epoch,
(unsigned long long)(state.published ? state.source.source_epoch : 0));
result = STREAM_ERR_CONFLICT; s->reason = STREAM_CLOSE_SOURCE_CHANGED;
}
}
if (result == STREAM_OK) result = s->source ? stream_subscribe_open(media_inst, s->group, s->source, &s->handle)
: stream_publish_open(media_inst, s->group, NULL, &s->handle);
if (result == STREAM_OK && s->source) result = stream_media_receiver_create(&s->receiver, media_frame, s);
struct stream_info info;
if (result == STREAM_OK) {
result = stream_get_info(s->handle, &info);
if (result == STREAM_OK) s->epoch = info.descriptor.source_epoch;
if (result == STREAM_OK) {
s->epoch = info.descriptor.source_epoch;
if (s->expected_epoch && s->epoch != s->expected_epoch) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android viewer opened stale epoch id=%llu", (unsigned long long)s->id);
result = STREAM_ERR_CONFLICT; s->reason = STREAM_CLOSE_SOURCE_CHANGED;
}
}
}
if (result != STREAM_OK) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open failed id=%llu group=%llx source=%llx result=%d",
@ -328,6 +321,7 @@ void android_stream_detach(struct UTUN_INSTANCE *inst) {
if (publisher.id) media_close(&publisher, 0);
if (viewer.id) media_close(&viewer, 0);
media_inst = NULL;
watch_pending = 0; watch_group = 0;
u_free(source_list); source_list = NULL; list_group = cached_group = list_updated = 0;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android stream bridge detached; producers rejected");
}
@ -336,7 +330,7 @@ void android_stream_detach(struct UTUN_INSTANCE *inst) {
/* Команда содержит идентификаторы, а не borrowed pointer ядра. */
int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned width, unsigned height,
unsigned rotation, int audio) {
unsigned rotation, int audio, uint64_t expected_epoch) {
if (!id || !group || (!source && (!width || width > 4096 || !height || height > 4096 || rotation > 270 || rotation % 90))) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open invalid arguments"); return -1;
}
@ -346,13 +340,23 @@ int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned w
if (s->id || id == publisher.id || id == viewer.id || !media_inst || !uasync_is_running(media_inst->ua)) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream open busy/detached id=%llu", (unsigned long long)id);
} else if (media_schedule() == 0) {
*s = (struct android_media_session){.id = id, .group = group, .source = source, .opening = 1, .need_key = 1,
*s = (struct android_media_session){.id = id, .group = group, .source = source, .expected_epoch = expected_epoch,
.opening = 1, .need_key = 1,
.format = {(uint16_t)width, (uint16_t)height, (uint16_t)rotation, STREAM_MEDIA_H264, audio ? STREAM_MEDIA_OPUS : 0}};
result = 0;
}
pthread_mutex_unlock(&media_lock); return result;
}
/* Последняя выбранная группа заменяет ещё не обработанную команду, указатели ядра не покидают mutex. */
int android_stream_watch(uint64_t group) {
pthread_mutex_lock(&media_lock);
int result = media_inst && uasync_is_running(media_inst->ua) ? media_schedule() : -1;
if (result == 0) { watch_group = group; watch_pending = 1; }
else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "Android stream watch rejected group=%llx", (unsigned long long)group);
pthread_mutex_unlock(&media_lock); return result;
}
/* Перегрузка удаляет целые видео-кадры и блокирует P-кадры до нового KEY. */
int android_stream_feed(uint64_t id, int video, uint64_t pts, int key, const uint8_t *data, size_t length) {
if (!data || !length || length > (video ? STREAM_MEDIA_MAX_FRAME : STREAM_MEDIA_MAX_OPUS) || pts > INT64_MAX) {

4
tools/chatgui-android/libutun_lite/stream_bridge.h
Unescape View File

@ -16,7 +16,9 @@ struct UTUN_INSTANCE;
int android_stream_attach(struct UTUN_INSTANCE *inst);
void android_stream_detach(struct UTUN_INSTANCE *inst);
int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned width, unsigned height,
unsigned rotation, int audio);
unsigned rotation, int audio, uint64_t expected_epoch);
/* Наблюдение выбранного канала через CHAT_EVT_STREAM_SOURCES; group=0 снимает его. */
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);
/* feed: 1 accepted, 0 dropped (video: запросить KEY), -1 detached/stopped/invalid. */

Write Preview
Loading…
Cancel
Save