From 126aa9df1028d673732c06440b4774bfd22386af Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 13 Oct 2026 22:09:46 +0200 Subject: [PATCH] Read media on demand through indexed cursors over the RS ring --- src/stream/stream_media_buffer.c | 385 +++++++++++++++++++++++++++++++ src/stream/stream_media_buffer.h | 56 +++++ 2 files changed, 441 insertions(+) create mode 100644 src/stream/stream_media_buffer.c create mode 100644 src/stream/stream_media_buffer.h diff --git a/src/stream/stream_media_buffer.c b/src/stream/stream_media_buffer.c new file mode 100644 index 00000000..bafed38c --- /dev/null +++ b/src/stream/stream_media_buffer.c @@ -0,0 +1,385 @@ +#include +#include + +#include "../../lib/debug_config.h" +#include "../../lib/mem.h" +#include "stream_media_buffer.h" + +struct media_position { + uint64_t first, last, key_seq; + int64_t video_us, audio_us, key_us; + uint32_t generation; + int gap; +}; +struct stream_media_store { + struct stream_buffer *buffer; + struct stream_descriptor descriptor; + struct media_position *positions; + unsigned capacity, head, count; + uint32_t generation; + int configured, waiting_key, finished, failed; + struct stream_media_format format; + struct stream_media_video pending; + uint64_t pending_seq, key_seq, last_frame_id; + uint32_t pending_used; + int64_t video_us, audio_us, key_us, last_audio_us; + uint64_t pressure_report_us; +}; +struct cursor_track { uint64_t seq; unsigned entry; int ended; }; +struct stream_media_cursor { + struct stream_media_store *store; + struct cursor_track video, audio; + uint32_t generation; + int initialized, key_delivered; + int64_t start_us; + uint8_t *scratch[2], *frame; + size_t scratch_capacity, frame_capacity; + uint64_t scratch_seq[2]; + int scratch_valid[2]; + unsigned scratch_next; + uint8_t opus[STREAM_MEDIA_MAX_OPUS]; + struct stream_media_video pending; + struct stream_media_format pending_format; + uint32_t pending_used; +}; + +/* Логический порядок сохранённых диапазонов не зависит от переиспользования массива. */ +static struct media_position *media_at(struct stream_media_store *s, unsigned index) { + return &s->positions[(s->head + index) % s->capacity]; +} + +/* Диапазоны DATA/GAP идут по seq; бинарный поиск не сканирует всё окно на каждом аудиопакете. */ +static struct media_position *media_find(struct stream_media_store *s, uint64_t seq) { + unsigned low = 0, high = s->count; + while (low < high) { + unsigned middle = low + (high - low) / 2; + struct media_position *p = media_at(s, middle); + if (p->last < seq) low = middle + 1; else high = middle; + } + if (low == s->count) return NULL; + struct media_position *p = media_at(s, low); + return p->first <= seq ? p : NULL; +} + +/* Индекс не удерживает старые части; его предел памяти также не зависит от продолжительности просмотра. */ +void stream_media_store_trim(struct stream_media_store *s) { + if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media store trim: null store"); return; } + struct stream_buffer_stats stats; + if (stream_buffer_get_stats(s->buffer, &stats) != STREAM_OK) return; + while (s->count && media_at(s, 0)->last < stats.expired_before) { + s->head = (s->head + 1) % s->capacity; s->count--; + } +} + +/* Один элемент на подтверждённую пачку/пропуск, без копирования H264/Opus. */ +static void media_append(struct stream_media_store *s, struct media_position position) { + stream_media_store_trim(s); + if (s->count == s->capacity) { + uint64_t now = get_time_us(); + if (!s->pressure_report_us || now - s->pressure_report_us >= 1000000) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media index pressure source=%llx oldest=%llu incoming=%llu capacity=%u", + (unsigned long long)s->descriptor.source_node_id, (unsigned long long)media_at(s, 0)->first, + (unsigned long long)position.first, s->capacity); + s->pressure_report_us = now; + } + s->head = (s->head + 1) % s->capacity; s->count--; + } + *media_at(s, s->count++) = position; +} + +/* Геометрия индекса ограничена вместимостью RS и отдельным лимитом метаданных. */ +int stream_media_store_create(struct stream_media_store **out, struct stream_buffer *buffer, const struct stream_descriptor *d) { + if (!out || *out || !buffer || stream_descriptor_validate(d) != STREAM_OK) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media store create invalid arguments"); return STREAM_ERR_ARGUMENT; + } + struct stream_buffer_stats stats; + if (stream_buffer_get_stats(buffer, &stats) != STREAM_OK) return STREAM_ERR_ARGUMENT; + struct stream_media_store *s = u_calloc(1, sizeof(*s)); + if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media store allocation failed"); return STREAM_ERR_MEMORY; } + size_t slots = stats.capacity_bytes / stream_block_capacity(d); + s->capacity = (unsigned)(2 * (slots / (d->n - d->r)) + 1); + unsigned maximum = STREAM_MEDIA_INDEX_BYTES / sizeof(*s->positions); + if (s->capacity > maximum) s->capacity = maximum; + s->positions = u_calloc(s->capacity, sizeof(*s->positions)); + if (!s->positions) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media index allocation failed slots=%u", s->capacity); u_free(s); return STREAM_ERR_MEMORY; + } + s->buffer = buffer; s->descriptor = *d; s->generation = 1; s->waiting_key = 1; + s->video_us = s->audio_us = s->key_us = s->last_audio_us = -1; s->key_seq = UINT64_MAX; + *out = s; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "media index created source=%llx records=%u bytes=%zu shared_RS=1", + (unsigned long long)d->source_node_id, s->capacity, s->capacity * sizeof(*s->positions)); + return STREAM_OK; +} + +/* Store освобождается после прекращения обращений reader; cursor destroy не разыменовывает store. */ +void stream_media_store_destroy(struct stream_media_store *s) { + if (!s) return; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "media index destroy source=%llx records=%u generation=%u", + (unsigned long long)s->descriptor.source_node_id, s->count, s->generation); + u_free(s->positions); u_free(s); +} + +/* Окончательный пропуск завершает старый непрерывный участок; следующая цепочка начинается с KEY. */ +void stream_media_store_gap(struct stream_media_store *s, uint64_t first, uint64_t last) { + if (!s || first > last || last == UINT64_MAX || s->generation == UINT32_MAX) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media index gap invalid first=%llu last=%llu", (unsigned long long)first, (unsigned long long)last); + if (s) s->failed = 1; + return; + } + media_append(s, (struct media_position){.first = first, .last = last, .generation = s->generation, .gap = 1, + .video_us = s->video_us, .audio_us = s->audio_us, .key_seq = s->key_seq, .key_us = s->key_us}); + s->generation++; s->pending_used = 0; s->waiting_key = 1; + s->video_us = s->audio_us = s->key_us = s->last_audio_us = -1; s->key_seq = UINT64_MAX; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "media index GAP source=%llx first=%llu last=%llu generation=%u", + (unsigned long long)s->descriptor.source_node_id, (unsigned long long)first, (unsigned long long)last, s->generation); +} + +/* Метаданные полного кадра считаются готовыми только после всех последовательных подписанных фрагментов. */ +int stream_media_store_feed(struct stream_media_store *s, uint64_t seq, const uint8_t *data, size_t length) { + if (!s || !data || seq == UINT64_MAX) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media index feed invalid seq=%llu", (unsigned long long)seq); return STREAM_ERR_ARGUMENT; + } + if (s->failed) return STREAM_ERR_FORMAT; + if (s->count && seq <= media_at(s, s->count - 1)->last) return STREAM_DUPLICATE; + struct stream_media_block b; + int result = stream_media_parse(data, length, &b); + if (result != STREAM_OK) { s->failed = 1; return result; } + if (s->configured && (b.format.audio_codec != s->format.audio_codec || b.format.revision < s->format.revision)) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media index conflicting format source=%llx revision=%u/%u audio=%u/%u", + (unsigned long long)s->descriptor.source_node_id, s->format.revision, b.format.revision, + s->format.audio_codec, b.format.audio_codec); s->failed = 1; return STREAM_ERR_CONFLICT; + } + if (s->configured && b.format.revision != s->format.revision) { s->pending_used = 0; s->waiting_key = 1; } + s->format = b.format; s->configured = 1; + for (unsigned i = 0; i < b.audio_count; i++) { + const struct stream_media_audio *a = &b.audio[i]; + if (s->last_audio_us >= 0 && (int64_t)a->pts_us <= s->last_audio_us) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media index stale audio seq=%llu pts=%llu", (unsigned long long)seq, (unsigned long long)a->pts_us); + continue; + } + s->last_audio_us = (int64_t)a->pts_us; s->audio_us = s->last_audio_us + 20000; + } + for (unsigned i = 0; i < b.video_count; i++) { + const struct stream_media_video *v = &b.video[i]; + if (!v->offset) { + if (s->pending_used) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media index incomplete frame=%llu before frame=%llu; require KEY", + (unsigned long long)s->pending.frame_id, (unsigned long long)v->frame_id); + s->pending_used = 0; s->waiting_key = 1; + } + if (s->waiting_key && !(v->flags & STREAM_MEDIA_KEY)) continue; + s->pending = *v; s->pending.data = NULL; s->pending_seq = seq; + } + if (s->waiting_key && !(v->flags & STREAM_MEDIA_KEY)) continue; + if (v->frame_id != s->pending.frame_id || v->offset != s->pending_used || v->pts_us != s->pending.pts_us + || v->frame_size != s->pending.frame_size || v->flags != s->pending.flags) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media index frame discontinuity seq=%llu frame=%llu offset=%u expected=%u", + (unsigned long long)seq, (unsigned long long)v->frame_id, v->offset, s->pending_used); + s->pending_used = 0; s->waiting_key = 1; continue; + } + s->pending_used += v->length; + if (s->pending_used != s->pending.frame_size) continue; + s->video_us = (int64_t)v->pts_us; s->last_frame_id = v->frame_id; s->waiting_key = 0; s->pending_used = 0; + if (v->flags & STREAM_MEDIA_KEY) { s->key_seq = s->pending_seq; s->key_us = (int64_t)v->pts_us; } + } + media_append(s, (struct media_position){.first = seq, .last = seq, .generation = s->generation, + .video_us = s->video_us, .audio_us = s->audio_us, .key_seq = s->key_seq, .key_us = s->key_us}); + return STREAM_OK; +} + +void stream_media_store_finish(struct stream_media_store *s) { + if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media index finish: null store"); return; } + s->finished = 1; +} + +/* Курсор хранит только собственные рабочие буферы, не копию накопленного медиапотока. */ +int stream_media_cursor_create(struct stream_media_cursor **out, struct stream_media_store *s) { + if (!out || *out || !s) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media cursor create invalid arguments"); return STREAM_ERR_ARGUMENT; } + struct stream_media_cursor *c = u_calloc(1, sizeof(*c)); + if (!c) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media cursor allocation failed"); return STREAM_ERR_MEMORY; } + c->store = s; c->scratch_capacity = stream_input_size(&s->descriptor); + if (c->scratch_capacity > STREAM_MEDIA_MAX_BLOCK) c->scratch_capacity = STREAM_MEDIA_MAX_BLOCK; + for (unsigned i = 0; i < 2; i++) c->scratch[i] = u_malloc((uint32_t)c->scratch_capacity); + if (!c->scratch[0] || !c->scratch[1]) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media cursor scratch allocation failed bytes=%zu", c->scratch_capacity); + stream_media_cursor_destroy(c); return STREAM_ERR_MEMORY; + } + *out = c; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "media cursor created source=%llx scratch=%zu", (unsigned long long)s->descriptor.source_node_id, + 2 * c->scratch_capacity); + return STREAM_OK; +} + +void stream_media_cursor_destroy(struct stream_media_cursor *c) { + if (!c) return; + u_free(c->scratch[0]); u_free(c->scratch[1]); u_free(c->frame); u_free(c); +} + +void stream_media_cursor_reset(struct stream_media_cursor *c) { + if (!c) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media cursor reset: null cursor"); return; } + c->initialized = c->key_delivered = 0; c->pending_used = 0; + c->scratch_valid[0] = c->scratch_valid[1] = 0; + c->video = (struct cursor_track){0}; c->audio = (struct cursor_track){0}; +} + +/* Выбирается последний полный KEY; тёплая история аудио перед ним остаётся в том же RS-кольце. */ +static int cursor_start(struct stream_media_cursor *c) { + struct stream_media_store *s = c->store; + stream_media_store_trim(s); + if (s->failed) return STREAM_ERR_FORMAT; + if (c->initialized) return STREAM_OK; + if (!s->count || s->key_seq == UINT64_MAX || !media_find(s, s->key_seq)) return s->finished ? STREAM_GAP : STREAM_WAIT; + c->generation = s->generation; c->start_us = s->key_us; + c->video.seq = s->key_seq; c->audio.seq = s->key_seq; + for (unsigned i = 0; i < s->count; i++) { + struct media_position *p = media_at(s, i); + if (p->generation == c->generation && !p->gap && p->audio_us > c->start_us - STREAM_MEDIA_PREROLL_US) { + c->audio.seq = p->first; break; + } + } + c->initialized = 1; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "media cursor start source=%llx key_seq=%llu pts=%lld audio_seq=%llu generation=%u", + (unsigned long long)s->descriptor.source_node_id, (unsigned long long)c->video.seq, + (long long)c->start_us, (unsigned long long)c->audio.seq, c->generation); + return STREAM_OK; +} + +/* Граница относится к участку этого reader, а не к данным после ещё не пройденного GAP. */ +int stream_media_cursor_status(struct stream_media_cursor *c, struct stream_media_status *out) { + if (!c || !out) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media cursor status invalid arguments"); return STREAM_ERR_ARGUMENT; } + struct stream_media_store *s = c->store; + int result = cursor_start(c); + struct stream_media_status status = {.ready_us = -1, .video_us = -1, .audio_us = -1, .start_us = -1, + .audio_enabled = s->configured && s->format.audio_codec, .finished = s->finished}; + if (result < 0) return result; + if (c->initialized) { + for (unsigned i = s->count; i-- > 0;) { + struct media_position *p = media_at(s, i); + if (p->generation != c->generation) continue; + status.video_us = p->video_us; status.audio_us = p->audio_us; break; + } + status.ready_us = status.video_us; + if (status.audio_enabled && status.audio_us < status.ready_us) status.ready_us = status.audio_us; + status.start_us = c->start_us; status.generation = c->generation; status.initialized = 1; + } + *out = status; + return STREAM_OK; +} + +/* Два scratch сохраняют соседние audio/video пачки; память кольца никогда не покидает uasync. */ +static int cursor_block(struct stream_media_cursor *c, uint64_t seq, struct stream_media_block *block) { + struct stream_media_store *s = c->store; + if (s->count && seq < media_at(s, 0)->first) return STREAM_EXPIRED; + struct media_position *p = media_find(s, seq); + if (!p) return s->finished ? STREAM_GAP : STREAM_WAIT; + if (p->gap || p->generation != c->generation) return STREAM_GAP; + unsigned slot = 0; + for (; slot < 2; slot++) if (c->scratch_valid[slot] && c->scratch_seq[slot] == seq) break; + size_t length = 0; + if (slot == 2) { + slot = c->scratch_next++ % 2; + int result = stream_buffer_decode(s->buffer, seq, c->scratch[slot], c->scratch_capacity, &length); + if (result != STREAM_OK) return result == STREAM_REJECTED ? STREAM_GAP : result; + c->scratch_seq[slot] = seq; c->scratch_valid[slot] = 1; + } else { + /* length сохраняется в первом поле зарезервированного хвоста не нужна: длину даёт metadata RS. */ + const uint8_t *wire; uint64_t actual; size_t wire_length; + int found = STREAM_GAP; + for (unsigned sub = 0; sub < s->descriptor.n; sub++) { + found = stream_buffer_read(s->buffer, seq, (uint8_t)sub, &wire, &wire_length, &actual); + if (found == STREAM_OK) { length = (size_t)stream_read_be(wire + STREAM_BLOCK_LENGTH_OFF, 4); break; } + } + if (found != STREAM_OK) return found; + } + return stream_media_parse(c->scratch[slot], length, block); +} + +/* Позиции треков независимы: чтение видео не вытаскивает дальний звук из общего кольца. */ +static int cursor_peek(struct stream_media_cursor *c, struct cursor_track *track, int video, + struct stream_media_block *block, int64_t *pts) { + if (track->ended) return STREAM_GAP; + for (unsigned work = 0; work < 8; work++) { + int result = cursor_block(c, track->seq, block); + if (result != STREAM_OK) { + struct stream_media_store *s = c->store; + if (s->finished && s->count && track->seq > media_at(s, s->count - 1)->last) { track->ended = 1; return STREAM_GAP; } + return result; + } + unsigned count = video ? block->video_count : block->audio_count; + if (track->entry < count) { + *pts = (int64_t)(video ? block->video[track->entry].pts_us : block->audio[track->entry].pts_us); + return STREAM_OK; + } + track->seq++; track->entry = 0; + } + return STREAM_WAIT; +} + +/* Сборка одного ближайшего H264 кадра; остальные кадры продолжают храниться только в RS. */ +static int cursor_video(struct stream_media_cursor *c, struct stream_media_packet *packet) { + for (unsigned work = 0; work < 16; work++) { + struct stream_media_block b; int64_t pts; + int result = cursor_peek(c, &c->video, 1, &b, &pts); + if (result != STREAM_OK) return result; + const struct stream_media_video *v = &b.video[c->video.entry++]; + if (!v->offset) { + if (!c->key_delivered && (!(v->flags & STREAM_MEDIA_KEY) || pts != c->start_us)) continue; + if (v->frame_size > c->frame_capacity) { + uint8_t *frame = u_realloc(c->frame, v->frame_size); + if (!frame) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media cursor frame allocation failed bytes=%u", v->frame_size); return STREAM_ERR_MEMORY; } + c->frame = frame; c->frame_capacity = v->frame_size; + } + c->pending = *v; c->pending.data = NULL; c->pending_used = 0; c->pending_format = b.format; + } + if (!c->pending.frame_size || v->frame_id != c->pending.frame_id || v->offset != c->pending_used + || v->pts_us != c->pending.pts_us || v->frame_size != c->pending.frame_size) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media cursor fragment conflict seq=%llu frame=%llu offset=%u used=%u", + (unsigned long long)c->video.seq, (unsigned long long)v->frame_id, v->offset, c->pending_used); + return STREAM_GAP; + } + memcpy(c->frame + c->pending_used, v->data, v->length); c->pending_used += v->length; + if (c->pending_used != c->pending.frame_size) continue; + struct stream_media_packet out = {.kind = STREAM_MEDIA_READ_VIDEO, .format = c->pending_format, .video = c->pending}; + out.video.offset = 0; out.video.length = out.video.frame_size; out.video.data = c->frame; + c->pending_used = 0; c->pending.frame_size = 0; c->key_delivered = 1; *packet = out; + return STREAM_OK; + } + return STREAM_WAIT; +} + +/* Выдача по времени потребителя оставляет накопленный запас в общем кольце. */ +int stream_media_cursor_read(struct stream_media_cursor *c, int64_t position, int64_t lookahead, struct stream_media_packet *packet) { + if (!c || !packet || lookahead < 0 || lookahead > 1000000) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media cursor read invalid arguments ahead=%lld", (long long)lookahead); return STREAM_ERR_ARGUMENT; + } + int result = cursor_start(c); + if (result != STREAM_OK) { + if (result == STREAM_GAP && c->store->finished && !c->initialized) { + *packet = (struct stream_media_packet){.kind = STREAM_MEDIA_READ_END}; return STREAM_OK; + } + return result; + } + struct stream_media_block audio, video; int64_t audio_pts = INT64_MAX, video_pts = INT64_MAX; + int ar = c->store->format.audio_codec ? cursor_peek(c, &c->audio, 0, &audio, &audio_pts) : STREAM_GAP; + if (!c->store->format.audio_codec) c->audio.ended = 1; + int vr = cursor_peek(c, &c->video, 1, &video, &video_pts); + if ((ar == STREAM_GAP && !c->audio.ended) || (vr == STREAM_GAP && !c->video.ended)) return STREAM_GAP; + if (ar == STREAM_EXPIRED || vr == STREAM_EXPIRED) return STREAM_EXPIRED; + if (ar < 0 || vr < 0) return ar < 0 ? ar : vr; + if (c->audio.ended && c->video.ended) { + *packet = (struct stream_media_packet){.kind = STREAM_MEDIA_READ_END}; return STREAM_OK; + } + int64_t limit = position < 0 ? c->start_us : position + lookahead; + if (ar == STREAM_OK && audio_pts <= limit && (audio_pts < video_pts || vr != STREAM_OK)) { + /* video peek мог переиспользовать scratch аудио: заново взять ту же пачку из двухэлементного cache. */ + result = cursor_block(c, c->audio.seq, &audio); + if (result != STREAM_OK) return result; + struct stream_media_audio a = audio.audio[c->audio.entry++]; + memcpy(c->opus, a.data, a.length); a.data = c->opus; + *packet = (struct stream_media_packet){.kind = STREAM_MEDIA_READ_AUDIO, .format = audio.format, .audio = a}; + return STREAM_OK; + } + if (vr == STREAM_OK && video_pts <= limit) return cursor_video(c, packet); + return STREAM_WAIT; +} diff --git a/src/stream/stream_media_buffer.h b/src/stream/stream_media_buffer.h new file mode 100644 index 00000000..0e22681c --- /dev/null +++ b/src/stream/stream_media_buffer.h @@ -0,0 +1,56 @@ +/* Индекс медиапотока поверх общего RS-кольца: хранит только позиции/PTS, без копии кодированного потока. + * Один uasync-поток; store принадлежит stream-session, cursor — конкретному локальному reader. + * feed получает уже проверенные последовательные пачки; gap обозначает окончательно потерянный диапазон. + * Независимые audio/video позиции читают оригинал из кольца. Рабочая память cursor — один H264 кадр, + * один Opus-пакет и два scratch медиаблока. Курсоры ничего не закрепляют против вытеснения. + * read возвращает заимствованный packet до следующего read/destroy; frontend копирует только ближайшие записи. + * WAIT — материал ещё не готов/дальше lookahead. GAP/EXPIRED требуют новой точки KEY; END — весь материал выдан. */ +#ifndef STREAM_MEDIA_BUFFER_H +#define STREAM_MEDIA_BUFFER_H + +#include "stream_buffer.h" +#include "stream_media.h" + +#ifdef __cplusplus +extern "C" { +#endif + +#define STREAM_MEDIA_INDEX_BYTES (4u * 1024u * 1024u) +#define STREAM_MEDIA_READ_AHEAD_US 80000 +#define STREAM_MEDIA_PREROLL_US 500000 + +struct stream_media_store; +struct stream_media_cursor; +enum stream_media_read_kind { STREAM_MEDIA_READ_VIDEO, STREAM_MEDIA_READ_AUDIO, STREAM_MEDIA_READ_END }; +struct stream_media_packet { + enum stream_media_read_kind kind; + struct stream_media_format format; + struct stream_media_video video; + struct stream_media_audio audio; +}; +struct stream_media_status { + int64_t ready_us, video_us, audio_us, start_us; + uint64_t generation; + int initialized, audio_enabled, finished; +}; + +int stream_media_store_create(struct stream_media_store **out, struct stream_buffer *buffer, const struct stream_descriptor *descriptor); +void stream_media_store_destroy(struct stream_media_store *store); +int stream_media_store_feed(struct stream_media_store *store, uint64_t seq, const uint8_t *data, size_t length); +void stream_media_store_gap(struct stream_media_store *store, uint64_t first, uint64_t last); +void stream_media_store_finish(struct stream_media_store *store); +void stream_media_store_trim(struct stream_media_store *store); +int stream_media_cursor_create(struct stream_media_cursor **out, struct stream_media_store *store); +void stream_media_cursor_destroy(struct stream_media_cursor *cursor); +void stream_media_cursor_reset(struct stream_media_cursor *cursor); +int stream_media_cursor_status(struct stream_media_cursor *cursor, struct stream_media_status *status); +/* position_us<0 — старт: прогреть аудио перед выбранным KEY. Затем отдавать только до position_us+lookahead_us. + * При GAP/EXPIRED cursor сбрасывают отдельно: это позволяет frontend сначала обработать разрыв. + * Бюджет одного вызова ограничен; WAIT при сборке длинного кадра допускает повтор сразу/по следующему событию. */ +int stream_media_cursor_read(struct stream_media_cursor *cursor, int64_t position_us, int64_t lookahead_us, + struct stream_media_packet *packet); + +#ifdef __cplusplus +} +#endif +#endif