From 9f37c54a3cf4344e554180eb2f1d003f76b8cfa1 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 13 Oct 2026 22:30:47 +0200 Subject: [PATCH] Expose demand based audio and video reads on local stream handles --- src/stream/stream.c | 11 ++++++----- src/stream/stream.h | 7 ++++--- tests/test_stream_playback.c | 16 ++++++++-------- 3 files changed, 18 insertions(+), 16 deletions(-) diff --git a/src/stream/stream.c b/src/stream/stream.c index fe815c78..4b389d97 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -784,7 +784,7 @@ int stream_get_info(const struct STREAM_HANDLE *h, struct stream_info *info) { } /* Первый frontend создаёт индекс точной эпохи, остальные владельцы используют его без копии RS-кольца. */ -int stream_media_enable(struct STREAM_HANDLE *h, int reader) { +int stream_media_enable(struct STREAM_HANDLE *h, unsigned tracks) { if (!stream_handle_active(h)) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media enable: detached handle"); return STREAM_ERR_STATE; } struct stream_session *s = h->session; if (!s->media) { @@ -796,8 +796,8 @@ int stream_media_enable(struct STREAM_HANDLE *h, int reader) { s->media_seek = s->next_seq - 1; s->media_seeking = 1; } } - if (reader && !h->media_cursor) { - int result = stream_media_cursor_create(&h->media_cursor, s->media); + if (tracks && !h->media_cursor) { + int result = stream_media_cursor_create(&h->media_cursor, s->media, tracks); if (result != STREAM_OK) return result; } session_media_schedule(s); @@ -805,13 +805,14 @@ int stream_media_enable(struct STREAM_HANDLE *h, int reader) { } /* Чтение изменяет только локальный cursor; live ingest/READY/выдачи продолжают двигаться независимо. */ -int stream_media_read(struct STREAM_HANDLE *h, int64_t position, int64_t lookahead, struct stream_media_packet *packet) { +int stream_media_read(struct STREAM_HANDLE *h, int64_t position, int64_t video_ahead, int64_t audio_ahead, + struct stream_media_packet *packet) { if (!stream_handle_active(h) || !h->media_cursor) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media read: detached/unconfigured reader"); return STREAM_ERR_STATE; } if (h->session->media_failed) return STREAM_ERR_FORMAT; if (h->session->media_seeking || h->session->media_next < h->session->next_seq) return STREAM_WAIT; - return stream_media_cursor_read(h->media_cursor, position, lookahead, packet); + return stream_media_cursor_read(h->media_cursor, position, video_ahead, audio_ahead, packet); } int stream_media_get_status(struct STREAM_HANDLE *h, struct stream_media_status *status) { diff --git a/src/stream/stream.h b/src/stream/stream.h index d742ba23..05ae3280 100644 --- a/src/stream/stream.h +++ b/src/stream/stream.h @@ -88,11 +88,12 @@ int stream_handle_reason(const struct STREAM_HANDLE *handle); int stream_get_info(const struct STREAM_HANDLE *handle, struct stream_info *info); /* Медиаплеер использует тот же RS-buffer, но собственные позиции чтения. - * reader=0 включает только общий индекс (publisher), reader=1 создаёт cursor этого handle. + * tracks=0 включает только общий индекс; VIDEO или VIDEO|AUDIO создаёт cursor этого handle. * Все вызовы — uasync. Заимствованный packet действует до следующего read/close. * LIVE DATA/GAP/END callbacks описывают сетевой прогресс; конец отложенного просмотра получает media_read. */ -int stream_media_enable(struct STREAM_HANDLE *handle, int reader); -int stream_media_read(struct STREAM_HANDLE *handle, int64_t position_us, int64_t lookahead_us, struct stream_media_packet *packet); +int stream_media_enable(struct STREAM_HANDLE *handle, unsigned tracks); +int stream_media_read(struct STREAM_HANDLE *handle, int64_t position_us, int64_t video_ahead_us, int64_t audio_ahead_us, + struct stream_media_packet *packet); int stream_media_get_status(struct STREAM_HANDLE *handle, struct stream_media_status *status); int stream_media_resync(struct STREAM_HANDLE *handle); diff --git a/tests/test_stream_playback.c b/tests/test_stream_playback.c index 457d04c9..3942b70d 100644 --- a/tests/test_stream_playback.c +++ b/tests/test_stream_playback.c @@ -106,7 +106,7 @@ static void test_shared_media_ring(void) { CHECK(stream_encoder_create(&encoder, &descriptor, key) == STREAM_OK); CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 32 * 1300) == STREAM_OK); CHECK(stream_media_store_create(&store, buffer, &descriptor) == STREAM_OK); - CHECK(stream_media_cursor_create(&fast, store) == STREAM_OK && stream_media_cursor_create(&slow, store) == STREAM_OK); + CHECK(stream_media_cursor_create(&fast, store, 3) == STREAM_OK && stream_media_cursor_create(&slow, store, 3) == STREAM_OK); append_media(encoder, buffer, store, 0, 0, STREAM_MEDIA_KEY, 1); struct stream_media_status status; CHECK(stream_media_cursor_status(fast, &status) == STREAM_OK && status.initialized && status.ready_us == 0); @@ -114,38 +114,38 @@ static void test_shared_media_ring(void) { append_media(encoder, buffer, store, 1, 40000, 0, 0); append_media(encoder, buffer, store, 2, 80000, 0, 0); struct stream_media_packet packet; - CHECK(stream_media_cursor_read(fast, -1, 0, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO); + CHECK(stream_media_cursor_read(fast, -1, 0, 0, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO); CHECK(packet.video.pts_us == 0 && !memcmp(packet.video.data, video_bytes, sizeof(video_bytes))); unsigned videos = 1, audio = 0; for (unsigned step = 0; step < 20; step++) { - int result = stream_media_cursor_read(fast, 80000, 0, &packet); + int result = stream_media_cursor_read(fast, 80000, 0, 0, &packet); if (result == STREAM_WAIT) break; CHECK(result == STREAM_OK); if (packet.kind == STREAM_MEDIA_READ_VIDEO) videos++; else { CHECK(packet.kind == STREAM_MEDIA_READ_AUDIO && packet.audio.pts_us <= 80000); audio++; } } CHECK(videos == 3 && audio == 5); - CHECK(stream_media_cursor_read(slow, -1, 0, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO); + CHECK(stream_media_cursor_read(slow, -1, 0, 0, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO); CHECK(packet.video.pts_us == 0 && !memcmp(packet.video.data, video_bytes, sizeof(video_bytes))); append_media(encoder, buffer, store, 3, 120000, STREAM_MEDIA_KEY, 0); CHECK(stream_media_cursor_status(slow, &status) == STREAM_OK && status.start_us == 0 && status.ready_us == 120000); stream_media_store_gap(store, 4, 4); append_media(encoder, buffer, store, 5, 200000, STREAM_MEDIA_KEY, 0); - CHECK(stream_media_cursor_read(slow, 0, 0, &packet) == STREAM_EXPIRED); + CHECK(stream_media_cursor_read(slow, 0, 0, 0, &packet) == STREAM_EXPIRED); int gap = 0; for (unsigned step = 0; step < 20; step++) { - int result = stream_media_cursor_read(fast, 180000, 0, &packet); + int result = stream_media_cursor_read(fast, 180000, 0, 0, &packet); if (result == STREAM_GAP) { gap = 1; break; } CHECK(result == STREAM_OK || result == STREAM_WAIT); } CHECK(gap); stream_media_cursor_reset(fast); - CHECK(stream_media_cursor_read(fast, -1, 0, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO); + CHECK(stream_media_cursor_read(fast, -1, 0, 0, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO); CHECK(packet.video.pts_us == 200000); stream_media_store_finish(store); int ended = 0; for (unsigned step = 0; step < 20; step++) { - int result = stream_media_cursor_read(fast, 1000000, 0, &packet); + int result = stream_media_cursor_read(fast, 1000000, 0, 0, &packet); CHECK(result == STREAM_OK || result == STREAM_WAIT); if (result == STREAM_OK && packet.kind == STREAM_MEDIA_READ_END) { ended = 1; break; } }