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

Expose demand based audio and video reads on local stream handles

master
evgeny 6 days ago
parent
866ca0923d
commit
9f37c54a3c
3 changed files with 18 additions and 16 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 11
      src/stream/stream.c
  2. 7
      src/stream/stream.h
  3. 16
      tests/test_stream_playback.c

11
src/stream/stream.c
Unescape View File

@ -784,7 +784,7 @@ int stream_get_info(const struct STREAM_HANDLE *h, struct stream_info *info) {
} }
/* Первый frontend создаёт индекс точной эпохи, остальные владельцы используют его без копии RS-кольца. */ /* Первый 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; } if (!stream_handle_active(h)) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media enable: detached handle"); return STREAM_ERR_STATE; }
struct stream_session *s = h->session; struct stream_session *s = h->session;
if (!s->media) { 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; s->media_seek = s->next_seq - 1; s->media_seeking = 1;
} }
} }
if (reader && !h->media_cursor) { if (tracks && !h->media_cursor) {
int result = stream_media_cursor_create(&h->media_cursor, s->media); int result = stream_media_cursor_create(&h->media_cursor, s->media, tracks);
if (result != STREAM_OK) return result; if (result != STREAM_OK) return result;
} }
session_media_schedule(s); session_media_schedule(s);
@ -805,13 +805,14 @@ int stream_media_enable(struct STREAM_HANDLE *h, int reader) {
} }
/* Чтение изменяет только локальный cursor; live ingest/READY/выдачи продолжают двигаться независимо. */ /* Чтение изменяет только локальный 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) { if (!stream_handle_active(h) || !h->media_cursor) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media read: detached/unconfigured reader"); return STREAM_ERR_STATE; 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_failed) return STREAM_ERR_FORMAT;
if (h->session->media_seeking || h->session->media_next < h->session->next_seq) return STREAM_WAIT; 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) { int stream_media_get_status(struct STREAM_HANDLE *h, struct stream_media_status *status) {

7
src/stream/stream.h
Unescape View File

@ -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); int stream_get_info(const struct STREAM_HANDLE *handle, struct stream_info *info);
/* Медиаплеер использует тот же RS-buffer, но собственные позиции чтения. /* Медиаплеер использует тот же RS-buffer, но собственные позиции чтения.
* reader=0 включает только общий индекс (publisher), reader=1 создаёт cursor этого handle. * tracks=0 включает только общий индекс; VIDEO или VIDEO|AUDIO создаёт cursor этого handle.
* Все вызовы — uasync. Заимствованный packet действует до следующего read/close. * Все вызовы — uasync. Заимствованный packet действует до следующего read/close.
* LIVE DATA/GAP/END callbacks описывают сетевой прогресс; конец отложенного просмотра получает media_read. */ * LIVE DATA/GAP/END callbacks описывают сетевой прогресс; конец отложенного просмотра получает media_read. */
int stream_media_enable(struct STREAM_HANDLE *handle, int reader); int stream_media_enable(struct STREAM_HANDLE *handle, unsigned tracks);
int stream_media_read(struct STREAM_HANDLE *handle, int64_t position_us, int64_t lookahead_us, struct stream_media_packet *packet); 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_get_status(struct STREAM_HANDLE *handle, struct stream_media_status *status);
int stream_media_resync(struct STREAM_HANDLE *handle); int stream_media_resync(struct STREAM_HANDLE *handle);

16
tests/test_stream_playback.c
Unescape View File

@ -106,7 +106,7 @@ static void test_shared_media_ring(void) {
CHECK(stream_encoder_create(&encoder, &descriptor, key) == STREAM_OK); 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_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_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); append_media(encoder, buffer, store, 0, 0, STREAM_MEDIA_KEY, 1);
struct stream_media_status status; struct stream_media_status status;
CHECK(stream_media_cursor_status(fast, &status) == STREAM_OK && status.initialized && status.ready_us == 0); 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, 1, 40000, 0, 0);
append_media(encoder, buffer, store, 2, 80000, 0, 0); append_media(encoder, buffer, store, 2, 80000, 0, 0);
struct stream_media_packet packet; 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))); CHECK(packet.video.pts_us == 0 && !memcmp(packet.video.data, video_bytes, sizeof(video_bytes)));
unsigned videos = 1, audio = 0; unsigned videos = 1, audio = 0;
for (unsigned step = 0; step < 20; step++) { 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; if (result == STREAM_WAIT) break;
CHECK(result == STREAM_OK); CHECK(result == STREAM_OK);
if (packet.kind == STREAM_MEDIA_READ_VIDEO) videos++; if (packet.kind == STREAM_MEDIA_READ_VIDEO) videos++;
else { CHECK(packet.kind == STREAM_MEDIA_READ_AUDIO && packet.audio.pts_us <= 80000); audio++; } else { CHECK(packet.kind == STREAM_MEDIA_READ_AUDIO && packet.audio.pts_us <= 80000); audio++; }
} }
CHECK(videos == 3 && audio == 5); 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))); 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); 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); CHECK(stream_media_cursor_status(slow, &status) == STREAM_OK && status.start_us == 0 && status.ready_us == 120000);
stream_media_store_gap(store, 4, 4); stream_media_store_gap(store, 4, 4);
append_media(encoder, buffer, store, 5, 200000, STREAM_MEDIA_KEY, 0); 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; int gap = 0;
for (unsigned step = 0; step < 20; step++) { 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; } if (result == STREAM_GAP) { gap = 1; break; }
CHECK(result == STREAM_OK || result == STREAM_WAIT); CHECK(result == STREAM_OK || result == STREAM_WAIT);
} }
CHECK(gap); CHECK(gap);
stream_media_cursor_reset(fast); 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); CHECK(packet.video.pts_us == 200000);
stream_media_store_finish(store); stream_media_store_finish(store);
int ended = 0; int ended = 0;
for (unsigned step = 0; step < 20; step++) { 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); CHECK(result == STREAM_OK || result == STREAM_WAIT);
if (result == STREAM_OK && packet.kind == STREAM_MEDIA_READ_END) { ended = 1; break; } if (result == STREAM_OK && packet.kind == STREAM_MEDIA_READ_END) { ended = 1; break; }
} }

Write Preview
Loading…
Cancel
Save