Switch Android media delivery to demand reads from the shared RS ring

tools/chatgui-android/libutun_lite/stream_bridge.c

@@ -28,9 +28,10 @@ struct media_queue { struct media_record *head, *tail; size_t bytes; unsigned co
 struct android_media_session {
     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, reason;
+    int want_read, waiting, resync, listen_audio;
+    int64_t position, video_ahead, audio_ahead;
 };
 
 static pthread_mutex_t media_lock = PTHREAD_MUTEX_INITIALIZER;
@@ -39,6 +40,8 @@ static struct posted_task *media_post;
 static void *media_timer;
 static struct android_media_session publisher, viewer;
 static struct media_queue video_queue, audio_queue, output_queue;
+static struct stream_media_status viewer_status;
+static uint64_t viewer_status_id;
 static uint64_t watch_group;
 static int watch_pending;
 static void media_process(void *arg);
@@ -129,21 +132,12 @@ static void media_event(struct UTUN_INSTANCE *inst, const struct stream_event *e
     if (inst == media_inst && viewer.handle && event->descriptor.group_id == viewer.group
         && event->descriptor.source_node_id == viewer.source && event->descriptor.source_epoch == viewer.epoch
         && !viewer.stopping && (event->type != STREAM_EVENT_CLOSED || !stream_handle_active(viewer.handle))) {
-        if (event->type == STREAM_EVENT_DATA) {
+        if (event->type == STREAM_EVENT_CLOSED) {
-            int result = stream_media_receiver_feed(viewer.receiver, event->data, event->length);
-            if (result != STREAM_OK) {
-                DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android viewer malformed/unsupported media id=%llu result=%d; closing",
-                            (unsigned long long)viewer.id, result);
-                media_close(&viewer, 6);
-            } else if (viewer.rx_overflow) {
-                stream_media_receiver_gap(viewer.receiver); viewer.rx_overflow = 0;
-                media_output(viewer.id, 3, NULL, 0, 0, NULL, 0);
-            }
-        } 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);
+            media_close(&viewer, 6);
+        } else if (event->type == STREAM_EVENT_AVAILABLE) {
+            viewer.want_read = 1; viewer.waiting = 0;
+            if (media_schedule() < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media ready scheduling failed");
         }
     }
     pthread_mutex_unlock(&media_lock);
@@ -153,7 +147,6 @@ static void media_event(struct UTUN_INSTANCE *inst, const struct stream_event *e
 static void media_close(struct android_media_session *s, int kind) {
     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); }
     memset(s, 0, sizeof(*s));
 }
@@ -225,7 +218,8 @@ static void media_process(void *arg) {
             }
             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);
+            if (result == STREAM_OK) result = stream_media_enable(s->handle,
+                s->source ? STREAM_MEDIA_TRACK_VIDEO | (s->listen_audio ? STREAM_MEDIA_TRACK_AUDIO : 0) : 0);
             struct stream_info info;
             if (result == STREAM_OK) {
                 result = stream_get_info(s->handle, &info);
@@ -242,6 +236,7 @@ static void media_process(void *arg) {
                             (unsigned long long)s->id, (unsigned long long)s->group, (unsigned long long)s->source, result);
                 media_close(s, 6);
             } else {
+                if (s->source) { viewer_status_id = s->id; memset(&viewer_status, 0, sizeof(viewer_status)); s->want_read = 1; }
                 media_output(s->id, 4, NULL, 0, 0, NULL, 0);
                 DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android stream opened id=%llu publisher=%d group=%llx source=%llx",
                            (unsigned long long)s->id, !s->source, (unsigned long long)s->group, (unsigned long long)s->source);
@@ -249,6 +244,32 @@ static void media_process(void *arg) {
         }
     }
     if (viewer.stopping && viewer.id) media_close(&viewer, 5);
+    if (viewer.handle) {
+        int result = stream_media_get_status(viewer.handle, &viewer_status);
+        if (result < 0) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media index failed result=%d", result); media_close(&viewer, 6); }
+    }
+    if (viewer.handle && (viewer.resync || viewer.rx_overflow)) {
+        viewer.resync = viewer.rx_overflow = 0; viewer.position = -1; viewer.waiting = 0;
+        if (stream_media_resync(viewer.handle) != STREAM_OK) media_close(&viewer, 6);
+    }
+    unsigned queued_viewer = 0;
+    for (struct media_record *r = output_queue.head; r; r = r->next) if (r->id == viewer.id) queued_viewer++;
+    if (viewer.handle && viewer.want_read && queued_viewer < 2
+        && output_queue.bytes + STREAM_MEDIA_MAX_FRAME + ANDROID_STREAM_HEADER_SIZE <= RX_LIMIT) {
+        viewer.want_read = 0;
+        struct stream_media_packet packet;
+        int result = stream_media_read(viewer.handle, viewer.position, viewer.video_ahead, viewer.audio_ahead, &packet);
+        if (stream_media_get_status(viewer.handle, &viewer_status) < 0) result = STREAM_ERR_FORMAT;
+        if (result == STREAM_OK) {
+            if (packet.kind == STREAM_MEDIA_READ_END) media_close(&viewer, 5);
+            else media_frame(&viewer, &packet.format, packet.kind == STREAM_MEDIA_READ_VIDEO ? &packet.video : NULL,
+                             packet.kind == STREAM_MEDIA_READ_AUDIO ? &packet.audio : NULL);
+        } else if (result == STREAM_WAIT) viewer.waiting = 1;
+        else if (result == STREAM_GAP || result == STREAM_EXPIRED) {
+            DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android playback resync id=%llu result=%d", (unsigned long long)viewer.id, result);
+            stream_media_resync(viewer.handle); viewer.position = -1; media_output(viewer.id, 3, NULL, 0, 0, NULL, 0);
+        } else { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android playback read failed result=%d", result); media_close(&viewer, 6); }
+    }
     if (publisher.handle && !stream_handle_active(publisher.handle)) media_close(&publisher, 6);
     uint64_t now = get_time_us();
     if (video_queue.head && now - video_queue.head->created > MEDIA_AGE_US) {
@@ -328,7 +349,8 @@ int android_stream_open(uint64_t id, uint64_t group, uint64_t source, unsigned w
         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, .expected_epoch = expected_epoch,
-            .opening = 1, .need_key = 1,
+            .opening = 1, .need_key = 1, .listen_audio = !!audio, .position = -1,
+            .video_ahead = STREAM_MEDIA_READ_AHEAD_US, .audio_ahead = STREAM_MEDIA_READ_AHEAD_US,
             .format = {(uint16_t)width, (uint16_t)height, (uint16_t)rotation, STREAM_MEDIA_H264, audio ? STREAM_MEDIA_OPUS_STEREO : 0, 1}};
         result = 0;
     }
@@ -435,5 +457,42 @@ uint8_t *android_stream_pull(uint64_t id, size_t *length) {
             u_free(r);
         } else { previous = r; link = &r->next; }
     }
+    if (viewer.id == id && viewer.handle && (data || !viewer.waiting)) {
+        viewer.want_read = 1;
+        if (media_schedule() < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media read scheduling failed");
+    }
     pthread_mutex_unlock(&media_lock); return data;
 }
+
+int android_stream_read_request(uint64_t id, int64_t position, int64_t video_ahead, int64_t audio_ahead) {
+    if (video_ahead < -1 || video_ahead > 60000000 || audio_ahead < -1 || audio_ahead > 60000000) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media demand invalid ahead=%lld/%lld",
+                    (long long)video_ahead, (long long)audio_ahead); return -1;
+    }
+    pthread_mutex_lock(&media_lock);
+    int result = 0;
+    if (viewer.id == id && (!viewer.waiting || viewer.position != position
+        || viewer.video_ahead != video_ahead || viewer.audio_ahead != audio_ahead)) {
+        viewer.position = position; viewer.video_ahead = video_ahead; viewer.audio_ahead = audio_ahead; viewer.want_read = 1;
+        result = media_schedule();
+    }
+    pthread_mutex_unlock(&media_lock); return result;
+}
+
+int android_stream_get_status(uint64_t id, struct stream_media_status *status) {
+    if (!status) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media status: missing output"); return -1; }
+    pthread_mutex_lock(&media_lock);
+    int result = viewer_status_id == id ? 0 : -1;
+    if (!result) *status = viewer_status;
+    pthread_mutex_unlock(&media_lock); return result;
+}
+
+int android_stream_resync(uint64_t id) {
+    pthread_mutex_lock(&media_lock);
+    int result = -1;
+    if (viewer.id == id && viewer.handle) {
+        media_drop_output(); viewer.resync = viewer.want_read = 1; viewer.waiting = 0; viewer.position = -1;
+        result = media_schedule();
+    }
+    pthread_mutex_unlock(&media_lock); return result;
+}

tools/chatgui-android/libutun_lite/stream_bridge.h

@@ -13,6 +13,7 @@
 
 struct UTUN_INSTANCE;
 struct stream_media_format;
+struct stream_media_status;
 #define ANDROID_STREAM_HEADER_SIZE 24
 /* 0 attached, -1 invalid/busy: нельзя передать singleton другому ядру до detach старого. */
 int android_stream_attach(struct UTUN_INSTANCE *inst);
@@ -28,5 +29,9 @@ int android_stream_feed(uint64_t id, int video, uint64_t pts, int key, const uin
 /* feed: 1 accepted, 0 dropped (video: запросить KEY), -1 detached/stopped/invalid. */
 void android_stream_stop(uint64_t id);
 uint8_t *android_stream_pull(uint64_t id, size_t *length);
+/* Запрос ближайших данных и снимок готовности; UI передаёт PTS, указатели кольца остаются в uasync. */
+int android_stream_read_request(uint64_t id, int64_t position_us, int64_t video_ahead_us, int64_t audio_ahead_us);
+int android_stream_get_status(uint64_t id, struct stream_media_status *status);
+int android_stream_resync(uint64_t id);
 
 #endif