Batch Android viewer records and wake readers on data or stop

tests/test_android_stream.c

@@ -29,6 +29,12 @@ static int producer_started, producer_stop;
 static unsigned source_events;
 static uint64_t source_group;
 static char *source_snapshot;
+struct signal_waiter { struct android_stream_signal *signal; int started, result; };
+static void *wait_signal(void *arg) {
+    struct signal_waiter *w = arg;
+    __atomic_store_n(&w->started, 1, __ATOMIC_RELEASE);
+    w->result = android_stream_signal_wait(w->signal, 50); return NULL;
+}
 
 static void chat_event(struct UTUN_INSTANCE *inst, int type, const uint8_t *data, int length) {
     (void)inst;
@@ -41,7 +47,7 @@ static void chat_event(struct UTUN_INSTANCE *inst, int type, const uint8_t *data
 
 static uint8_t *wait_event(uint64_t id, int kind, size_t *size) {
     for (unsigned i = 0; i < 1000; i++) {
-        if (id == 2 || id == 7) CHECK(android_stream_read_request(id, kind == 5 ? 1000000 : -1, 1, 80000) == 0);
+        if (id == 2 || id == 7) CHECK(android_stream_read_request(id, kind == 5 ? 1000000 : -1, 1, 80000, 1) == 0);
         uint8_t *data = android_stream_pull(id, size);
         if (data && data[0] == kind) return data;
         u_free(data); test_clock_poll(ua, 100);
@@ -98,6 +104,10 @@ int main(void) {
     uint8_t *error = wait_event(5, 6, &size); CHECK(error[1] == STREAM_CLOSE_SOURCE_CHANGED); u_free(error);
     CHECK(android_stream_open(2, group, inst->node_id, 0, 0, 0, 1, member.source.source_epoch, STREAM_MEDIA_H264) == 0);
     u_free(wait_event(2, 4, &size));
+    struct android_stream_signal *signal = android_stream_signal_create(2); CHECK(signal);
+    CHECK(android_stream_signal_wait(signal, 0) == 0);
+    android_stream_signal_notify(signal); android_stream_signal_notify(signal);
+    CHECK(android_stream_signal_wait(signal, 0) == 1 && android_stream_signal_wait(signal, 0) == 0);
     CHECK(android_stream_open(7, group, inst->node_id, 0, 0, 0, 1, member.source.source_epoch, STREAM_MEDIA_H264) == 0);
     u_free(wait_event(7, 4, &size));
     CHECK(android_stream_open(8, group, inst->node_id, 0, 0, 0, 1, member.source.source_epoch, STREAM_MEDIA_H264) == 0);
@@ -109,7 +119,7 @@ int main(void) {
     CHECK(feed(1, 0, 100000, 0, (const uint8_t *)"ops", 3) == 1);
     uint8_t *output = NULL; int video_seen = 0, audio_seen = 0;
     for (unsigned i = 0; i < 1000 && (!video_seen || !audio_seen); i++) {
-        CHECK(android_stream_read_request(2, 100000, 1, 80000) == 0);
+        CHECK(android_stream_read_request(2, 100000, 1, 80000, 1) == 0);
         output = android_stream_pull(2, &size);
         if (output && output[0] == 1) {
             CHECK(size == sizeof(video) + ANDROID_STREAM_HEADER_SIZE && stream_read_be(output + 8, 8) == 100000 && !memcmp(output + ANDROID_STREAM_HEADER_SIZE, video, sizeof(video)));
@@ -137,7 +147,7 @@ int main(void) {
     next_format.video_codec = STREAM_MEDIA_H265;
     video_seen = audio_seen = 0;
     for (unsigned i = 0; i < 1000 && (!video_seen || !audio_seen); i++) {
-        CHECK(android_stream_read_request(2, 300000, 1, 80000) == 0);
+        CHECK(android_stream_read_request(2, 300000, 1, 80000, 1) == 0);
         output = android_stream_pull(2, &size);
         if (output && output[0] == 1) {
             CHECK(stream_read_be(output + 8, 8) == 300000 && stream_read_be(output + 20, 4) == 2 && output[24] == STREAM_MEDIA_H265);
@@ -157,7 +167,7 @@ int main(void) {
         CHECK(feed(1, 0, audio_pts, 0, (const uint8_t *)"ops", 3) == 1);
         video_seen = audio_seen = 0;
         for (unsigned i = 0; i < 1000 && (!video_seen || !audio_seen); i++) {
-            CHECK(android_stream_read_request(2, (int64_t)video_pts, 1, 80000) == 0);
+            CHECK(android_stream_read_request(2, (int64_t)video_pts, 1, 80000, 1) == 0);
             output = android_stream_pull(2, &size);
             if (output && output[0] == 1) {
                 CHECK(stream_read_be(output + 8, 8) == video_pts && stream_read_be(output + 20, 4) == next_format.revision);
@@ -176,7 +186,7 @@ int main(void) {
     debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_DEBUG);
     for (unsigned i = 0; i < 5; i++)
         CHECK(android_stream_feed(1, 1, 800000 + i * 40000, 0, video, sizeof(video), &next_format) == 1);
-    for (unsigned i = 0; i < 50; i++) { CHECK(android_stream_read_request(2, 700000, 1, -1) == 0); test_clock_poll(ua, 100); }
+    for (unsigned i = 0; i < 50; i++) { CHECK(android_stream_read_request(2, 700000, 1, -1, 1) == 0); test_clock_poll(ua, 100); }
     output = android_stream_pull(2, &size); CHECK(output && output[0] == 1 && stream_read_be(output + 8, 8) == 800000); u_free(output);
     /* AVAILABLE между pull и отменой demand раньше выдавал второй кадр при занятом decoder input. */
     CHECK(android_stream_feed(1, 1, 1000000, 0, video, sizeof(video), &next_format) == 1);
@@ -185,37 +195,70 @@ int main(void) {
     DEBUG_INFO(DEBUG_CATEGORY_STREAM, "demand regression: new data after reply, unsolicited_kind=%d", output ? output[0] : 0);
     CHECK(!output);
     test_clock_run_for(ua, 1000); CHECK(!android_stream_pull(2, &size));
-    CHECK(android_stream_read_request(2, 700000, 0, -1) == 0); test_clock_run_for(ua, 1000);
+    CHECK(android_stream_read_request(2, 700000, 0, -1, 1) == 0); test_clock_run_for(ua, 1000);
     CHECK(!android_stream_pull(2, &size));
     for (unsigned frame = 1; frame < 6; frame++) {
         output = NULL;
         for (unsigned i = 0; i < 1000 && !output; i++) {
-            CHECK(android_stream_read_request(2, 700000, 1, -1) == 0); test_clock_poll(ua, 100); output = android_stream_pull(2, &size);
+            CHECK(android_stream_read_request(2, 700000, 1, -1, 1) == 0); test_clock_poll(ua, 100); output = android_stream_pull(2, &size);
         }
         CHECK(output && output[0] == 1 && stream_read_be(output + 8, 8) == 800000 + frame * 40000); u_free(output);
     }
     /* Пустой pull не отменяет WAIT: уже запрошенный кадр приходит после появления данных. */
-    CHECK(android_stream_read_request(2, 1000000, 1, -1) == 0); test_clock_run_for(ua, 1000);
+    CHECK(android_stream_read_request(2, 1000000, 1, -1, 1) == 0); test_clock_run_for(ua, 1000);
     CHECK(!android_stream_pull(2, &size));
     CHECK(android_stream_feed(1, 1, 1040000, 0, video, sizeof(video), &next_format) == 1);
     test_clock_run_for(ua, 1000);
     output = android_stream_pull(2, &size);
     CHECK(output && output[0] == 1 && stream_read_be(output + 8, 8) == 1040000); u_free(output);
     /* Отмена WAIT сохраняет следующий кадр, а запрос только аудио не забирает видео. */
-    CHECK(android_stream_read_request(2, 1040000, 1, -1) == 0); test_clock_run_for(ua, 1000);
-    CHECK(!android_stream_pull(2, &size)); CHECK(android_stream_read_request(2, 1040000, 0, -1) == 0);
+    CHECK(android_stream_read_request(2, 1040000, 1, -1, 1) == 0); test_clock_run_for(ua, 1000);
+    CHECK(!android_stream_pull(2, &size)); CHECK(android_stream_read_request(2, 1040000, 0, -1, 1) == 0);
     CHECK(android_stream_feed(1, 1, 1080000, 0, video, sizeof(video), &next_format) == 1);
     CHECK(feed(1, 0, 220000, 0, (const uint8_t *)"ops", 3) == 1); test_clock_run_for(ua, 1000);
     CHECK(!android_stream_pull(2, &size));
-    CHECK(android_stream_read_request(2, 1040000, 0, 80000) == 0); test_clock_run_for(ua, 1000);
+    CHECK(android_stream_read_request(2, 1040000, 0, 80000, 1) == 0); test_clock_run_for(ua, 1000);
     output = android_stream_pull(2, &size);
     CHECK(output && output[0] == 2 && stream_read_be(output + 8, 8) == 220000); u_free(output);
-    CHECK(android_stream_read_request(2, 1040000, 1, -1) == 0); test_clock_run_for(ua, 1000);
+    CHECK(android_stream_read_request(2, 1040000, 1, -1, 1) == 0); test_clock_run_for(ua, 1000);
     output = android_stream_pull(2, &size);
     CHECK(output && output[0] == 1 && stream_read_be(output + 8, 8) == 1080000); u_free(output);
     DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS demand regression: one reply, WAIT survives empty pull, cancellation and audio-only preserve video");
+    for (unsigned i = 0; i < 3; i++) {
+        CHECK(android_stream_feed(1, 1, 1120000 + i * 40000, 0, video, sizeof(video), &next_format) == 1);
+        CHECK(feed(1, 0, 240000 + i * 20000, 0, (const uint8_t *)"ops", 3) == 1);
+    }
+    test_clock_run_for(ua, 1000);
+    CHECK(android_stream_signal_wait(signal, 0) >= 0);
+    unsigned batch_video = 0, batch_audio = 0, largest = 0;
+    for (unsigned attempt = 0; attempt < 20 && (batch_video < 3 || batch_audio < 3); attempt++) {
+        CHECK(android_stream_read_request(2, 1080000, 3 - batch_video, 80000, 16) == 0);
+        test_clock_run_for(ua, 1000);
+        output = android_stream_pull(2, &size); CHECK(output);
+        CHECK(android_stream_signal_wait(signal, 0) == 1);
+        unsigned records = 0;
+        for (size_t offset = 0; offset < size;) {
+            CHECK(size - offset >= ANDROID_STREAM_HEADER_SIZE);
+            size_t length = (size_t)stream_read_be(output + offset + 16, 4);
+            CHECK(length <= size - offset - ANDROID_STREAM_HEADER_SIZE);
+            uint64_t pts = stream_read_be(output + offset + 8, 8);
+            if (output[offset] == 1) {
+                CHECK(pts == 1120000 + batch_video * 40000 && length == sizeof(video)); batch_video++;
+            } else { CHECK(output[offset] == 2 && pts == 240000 + batch_audio * 20000 && length == 3); batch_audio++; }
+            records++; offset += ANDROID_STREAM_HEADER_SIZE + length;
+        }
+        CHECK(records <= 16); if (records > largest) largest = records; u_free(output);
+    }
+    CHECK(batch_video == 3 && batch_audio == 3 && largest > 1);
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS Android batches: video=%u audio=%u largest=%u notified=1", batch_video, batch_audio, largest);
     debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO);
-    android_stream_stop(1); u_free(wait_event(2, 5, &size)); android_stream_stop(2); u_free(wait_event(1, 5, &size));
+    android_stream_stop(1); u_free(wait_event(2, 5, &size));
+    CHECK(android_stream_signal_wait(signal, 0) >= 0);
+    struct signal_waiter waiter = {.signal = signal}; pthread_t waiter_thread;
+    CHECK(pthread_create(&waiter_thread, NULL, wait_signal, &waiter) == 0);
+    while (!__atomic_load_n(&waiter.started, __ATOMIC_ACQUIRE)) {}
+    android_stream_stop(2); CHECK(pthread_join(waiter_thread, NULL) == 0 && waiter.result == 1);
+    android_stream_signal_destroy(signal); u_free(wait_event(1, 5, &size));
     while ((output = android_stream_pull(2, &size))) u_free(output);
     CHECK(!strcmp(source_snapshot, "[]"));
     current_format.rotation = 0; current_format.audio_codec = 0;

tools/chatgui-android/libutun_lite/stream_bridge.c

@@ -1,4 +1,6 @@
 #include <pthread.h>
+#include <errno.h>
+#include <time.h>
 #include <string.h>
 #include "../../../lib/debug_config.h"
 #include "../../../lib/mem.h"
@@ -126,11 +128,9 @@ void android_stream_stop(uint64_t id) {
     pthread_mutex_unlock(&media_lock);
 }
 
-/* Kotlin получает owned wire-копию только ближайшей записи своего reader. */
-static uint8_t *encode_record(int kind, int flags, const struct stream_media_format *format, uint64_t pts,
-                              const uint8_t *data, size_t bytes, size_t *length) {
-    uint8_t *p = u_calloc(1, ANDROID_STREAM_HEADER_SIZE + bytes);
-    if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android output allocation failed bytes=%zu", bytes); return NULL; }
+/* Запись в заранее выделенный нулевой буфер; записи батча следуют друг за другом без дополнительной копии. */
+static void encode_record(uint8_t *p, int kind, int flags, const struct stream_media_format *format, uint64_t pts,
+                              const uint8_t *data, size_t bytes) {
     p[0] = (uint8_t)kind; p[1] = (uint8_t)flags;
     if (format) {
         stream_write_be(p + 2, format->width, 2); stream_write_be(p + 4, format->height, 2);
@@ -138,7 +138,6 @@ static uint8_t *encode_record(int kind, int flags, const struct stream_media_for
     }
     stream_write_be(p + 8, pts, 8); stream_write_be(p + 16, bytes, 4);
     if (bytes) memcpy(p + ANDROID_STREAM_HEADER_SIZE, data, bytes);
-    *length = ANDROID_STREAM_HEADER_SIZE + bytes; return p;
 }
 
 uint8_t *android_stream_pull(uint64_t id, size_t *length) {
@@ -149,32 +148,49 @@ uint8_t *android_stream_pull(uint64_t id, size_t *length) {
         struct stream_publisher_status status;
         if (stream_publisher_snapshot(transmitter, id, &status) == 0) {
             if (status.state == STREAM_PUBLISHER_ENDED || status.state == STREAM_PUBLISHER_FAILED) {
-                data = encode_record(status.state == STREAM_PUBLISHER_ENDED ? 5 : 6, status.reason, NULL, 0, NULL, 0, length);
+                data = u_calloc(1, ANDROID_STREAM_HEADER_SIZE);
+                if (data) encode_record(data, status.state == STREAM_PUBLISHER_ENDED ? 5 : 6, status.reason, NULL, 0, NULL, 0);
+                else DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android publisher completion allocation failed id=%llu", (unsigned long long)id);
                 if (data) { publisher_id = 0; publisher_opening = 0; }
             } else if (publisher_opening && status.state >= STREAM_PUBLISHER_LIVE && status.state <= STREAM_PUBLISHER_DRAINING) {
-                data = encode_record(4, 0, NULL, 0, NULL, 0, length);
+                data = u_calloc(1, ANDROID_STREAM_HEADER_SIZE);
+                if (data) encode_record(data, 4, 0, NULL, 0, NULL, 0);
+                else DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android publisher open event allocation failed id=%llu", (unsigned long long)id);
                 if (data) publisher_opening = 0;
             }
+            if (data) *length = ANDROID_STREAM_HEADER_SIZE;
         }
     } else {
-        struct stream_viewer_record *r = stream_viewer_pull(viewers, id);
-        if (r) {
-            data = encode_record(r->kind, r->kind == STREAM_VIEWER_CLOSED ? r->reason : r->flags,
-                                 &r->format, r->pts, r->data, r->length, length);
-            if (!data) stream_viewer_resync(viewers, id);
-            u_free(r);
+        struct stream_viewer_record *head = stream_viewer_pull(viewers, id);
+        if (head) {
+            size_t bytes = 0;
+            for (struct stream_viewer_record *r = head; r; r = r->next) bytes += ANDROID_STREAM_HEADER_SIZE + r->length;
+            data = u_calloc(1, bytes);
+            if (data) {
+                uint8_t *p = data;
+                for (struct stream_viewer_record *r = head; r; r = r->next) {
+                    encode_record(p, r->kind, r->kind == STREAM_VIEWER_CLOSED ? r->reason : r->flags,
+                        &r->format, r->pts, r->data, r->length);
+                    p += ANDROID_STREAM_HEADER_SIZE + r->length;
+                }
+                *length = bytes;
+            } else {
+                DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android batch allocation failed id=%llu bytes=%zu", (unsigned long long)id, bytes);
+                stream_viewer_resync(viewers, id);
+            }
+            stream_viewer_records_free(head);
         }
     }
     pthread_mutex_unlock(&media_lock); return data;
 }
 
-int android_stream_read_request(uint64_t id, int64_t position, int want_video, int64_t audio_ahead) {
-    if ((want_video != 0 && want_video != 1) || audio_ahead < -1 || audio_ahead > 60000000) {
-        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media demand invalid video=%d audio_ahead=%lld",
-                    want_video, (long long)audio_ahead); return -1;
+int android_stream_read_request(uint64_t id, int64_t position, unsigned video, int64_t audio_ahead, unsigned records) {
+    if (!records || records > STREAM_VIEWER_BATCH_RECORDS || video > records || audio_ahead < -1 || audio_ahead > 60000000) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android media demand invalid video=%u records=%u audio_ahead=%lld",
+                    video, records, (long long)audio_ahead); return -1;
     }
     pthread_mutex_lock(&media_lock);
-    if (viewers && id != publisher_id) stream_viewer_request(viewers, id, position, want_video, audio_ahead, 1);
+    if (viewers && id != publisher_id) stream_viewer_request(viewers, id, position, video, audio_ahead, records);
     /* CLOSED/END остаются доступны pull даже после закрытия cursor; неизвестный ID даёт пустой pull. */
     pthread_mutex_unlock(&media_lock); return 0;
 }
@@ -191,3 +207,70 @@ int android_stream_resync(uint64_t id) {
     int result = stream_viewer_resync(viewers, id);
     pthread_mutex_unlock(&media_lock); return result;
 }
+
+/* Lifetime принадлежит worker, а не manager: stop/detach снимает callback до освобождения signal. */
+struct android_stream_signal {
+    pthread_mutex_t lock;
+    pthread_cond_t ready;
+    uint64_t id, notifications, waits;
+    int pending;
+};
+
+void android_stream_signal_notify(struct android_stream_signal *s) {
+    if (!s) return;
+    pthread_mutex_lock(&s->lock); s->pending = 1; s->notifications++;
+    pthread_cond_signal(&s->ready); pthread_mutex_unlock(&s->lock);
+}
+
+static void signal_ready(void *arg) { android_stream_signal_notify(arg); }
+
+struct android_stream_signal *android_stream_signal_create(uint64_t id) {
+    struct android_stream_signal *s = u_calloc(1, sizeof(*s));
+    if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android viewer signal allocation failed"); return NULL; }
+    int result = pthread_mutex_init(&s->lock, NULL);
+    if (result) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android viewer signal mutex failed=%d", result); u_free(s); return NULL; }
+    pthread_condattr_t attr;
+    result = pthread_condattr_init(&attr);
+    if (!result) {
+        result = pthread_condattr_setclock(&attr, CLOCK_MONOTONIC);
+        if (!result) result = pthread_cond_init(&s->ready, &attr);
+        pthread_condattr_destroy(&attr);
+    }
+    if (result) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android viewer signal condition failed=%d", result);
+        pthread_mutex_destroy(&s->lock); u_free(s); return NULL;
+    }
+    s->id = id;
+    pthread_mutex_lock(&media_lock);
+    result = stream_viewer_set_notify(viewers, id, signal_ready, s);
+    pthread_mutex_unlock(&media_lock);
+    if (result < 0) { android_stream_signal_destroy(s); return NULL; }
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android viewer signal attached id=%llu", (unsigned long long)id);
+    return s;
+}
+
+int android_stream_signal_wait(struct android_stream_signal *s, unsigned timeout_ms) {
+    if (!s || timeout_ms > 50) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android viewer signal wait invalid timeout=%u", timeout_ms); return -1;
+    }
+    struct timespec until;
+    if (clock_gettime(CLOCK_MONOTONIC, &until) < 0) {
+        DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android viewer signal clock failed errno=%d", errno); return -1;
+    }
+    until.tv_nsec += timeout_ms * 1000000L;
+    until.tv_sec += until.tv_nsec / 1000000000L; until.tv_nsec %= 1000000000L;
+    pthread_mutex_lock(&s->lock); s->waits++;
+    int result = 0;
+    while (!s->pending && !result) result = pthread_cond_timedwait(&s->ready, &s->lock, &until);
+    if (result && result != ETIMEDOUT) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android viewer signal wait failed=%d", result);
+    int signaled = s->pending; s->pending = 0;
+    pthread_mutex_unlock(&s->lock);
+    return result && result != ETIMEDOUT ? -1 : signaled;
+}
+
+void android_stream_signal_destroy(struct android_stream_signal *s) {
+    if (!s) return;
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "Android viewer signal released id=%llu notifications=%llu waits=%llu",
+        (unsigned long long)s->id, (unsigned long long)s->notifications, (unsigned long long)s->waits);
+    pthread_cond_destroy(&s->ready); pthread_mutex_destroy(&s->lock); u_free(s);
+}

tools/chatgui-android/libutun_lite/stream_bridge.h

@@ -1,9 +1,11 @@
 /* Android-очереди live media: attach/detach только в uasync, вне callbacks.
  * UI/кодировщики используют только session_id, никогда не сохраняют inst/ua.
- * Один publisher и независимые viewers, TX <=2MB/128 records, RX — одна ближайшая запись на viewer.
+ * Один publisher и независимые viewers, TX <=2MB/128 records, RX — ограниченный батч на viewer.
  * Data принимает полные access units пяти кодеков uAV2 (KEY: config_length:4/config/frame) и Opus 48k mono/stereo/20ms.
- * pull передаёт u_malloc-буфер: kind:1,flags:1,width:2,height:2,rotation:2,PTS_us:8,length:4,revision:4,codec:1,reserved:3,data.
- * kinds: video=1,audio=2,gap=3,opened=4,ended=5,error=6; error.flags = stream_close_reason. Числа big-endian.
+ * pull передаёт u_malloc-буфер последовательных записей:
+ * kind:1,flags:1,width:2,height:2,rotation:2,PTS_us:8,length:4,revision:4,codec:1,reserved:3,data.
+ * kinds: video=1,audio=2,gap=3,opened=4,ended=5,error=6,video_end=7,audio_end=8; error.flags = stream_close_reason.
+ * Конец трека следует за его media; GAP/error отменяют старые записи. Числа big-endian.
  * Медиаданные передаются через pull, не через общий чатовый event bus. */
 #ifndef ANDROID_STREAM_BRIDGE_H
 #define ANDROID_STREAM_BRIDGE_H
@@ -14,6 +16,7 @@
 struct UTUN_INSTANCE;
 struct stream_media_format;
 struct stream_media_status;
+struct android_stream_signal;
 #define ANDROID_STREAM_HEADER_SIZE 28
 struct android_stream_tx_stats {
     uint64_t id, sample_us, video_bytes, audio_bytes, stream_bytes, transport_bytes, inflight_bytes, acked_1s, total_bytes;
@@ -36,10 +39,18 @@ 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);
-/* Один ответ на запрос: следующий кадр при want_video=1, звук до position_us+audio_ahead_us.
+/* Один батч на запрос: до video кадров и records записей, звук до position_us+audio_ahead_us.
  * Повторные запросы до получения ответа не создают дополнительные записи. Указатели кольца остаются в uasync. */
-int android_stream_read_request(uint64_t id, int64_t position_us, int want_video, int64_t audio_ahead_us);
+int android_stream_read_request(uint64_t id, int64_t position_us, unsigned video, int64_t audio_ahead_us, unsigned records);
 int android_stream_get_status(uint64_t id, struct stream_media_status *status);
 int android_stream_resync(uint64_t id);
+/* Worker владеет signal; create привязывает его к открытому viewer. Готовность/stop/detach будят wait.
+ * signal также будит worker после расходования PCM. Уведомления сливаются, но не теряются до wait.
+ * destroy — только после stream_stop и остановки всех внешних signal/wait callers. */
+struct android_stream_signal *android_stream_signal_create(uint64_t id);
+void android_stream_signal_notify(struct android_stream_signal *signal);
+/* 1 — уведомление, 0 — deadline, -1 — ошибка; timeout_ms 0..50, ожидание не держит mutex ядра/bridge. */
+int android_stream_signal_wait(struct android_stream_signal *signal, unsigned timeout_ms);
+void android_stream_signal_destroy(struct android_stream_signal *signal);
 
 #endif