diff --git a/tools/chatgui-android/jni_bridge/android_stream_jni.c b/tools/chatgui-android/jni_bridge/android_stream_jni.c index aef2002d..f4e688b4 100644 --- a/tools/chatgui-android/jni_bridge/android_stream_jni.c +++ b/tools/chatgui-android/jni_bridge/android_stream_jni.c @@ -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( diff --git a/tools/chatgui-android/libutun_lite/stream_bridge.c b/tools/chatgui-android/libutun_lite/stream_bridge.c index d2a5526f..545f82bf 100644 --- a/tools/chatgui-android/libutun_lite/stream_bridge.c +++ b/tools/chatgui-android/libutun_lite/stream_bridge.c @@ -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) { diff --git a/tools/chatgui-android/libutun_lite/stream_bridge.h b/tools/chatgui-android/libutun_lite/stream_bridge.h index fbbcaa00..0963b381 100644 --- a/tools/chatgui-android/libutun_lite/stream_bridge.h +++ b/tools/chatgui-android/libutun_lite/stream_bridge.h @@ -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. */