2 changed files with 441 additions and 0 deletions
@ -0,0 +1,385 @@
|
||||
#include <limits.h> |
||||
#include <string.h> |
||||
|
||||
#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; |
||||
} |
||||
@ -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 |
||||
Loading…
Reference in new issue