diff --git a/src/stream/stream.c b/src/stream/stream.c index a3da0b4f..acae6656 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -26,6 +26,10 @@ struct stream_session { uint8_t *output; struct stream_metrics *metrics; struct stream_optimizer *optimizer; + struct stream_media_store *media; + void *media_work; + uint64_t media_next, media_seek; + int media_seeking, media_failed; void *delivery; uint64_t next_seq; uint64_t start_seq, last_seq, passed_before; @@ -45,6 +49,7 @@ struct STREAM_HANDLE { int relay; int selection_owned; int reason; + struct stream_media_cursor *media_cursor; }; struct stream_service { @@ -64,6 +69,7 @@ struct stream_service { static void session_deliver(void *arg); static void service_reconcile(void *arg); +static void session_media_schedule(struct stream_session *s); /* Общий каталог не выделяет media-ресурсов: профиль проверяется тем же кодом, что и subscribe. */ int stream_sources_json(struct UTUN_INSTANCE *inst, uint64_t group_id, char **out) { @@ -230,6 +236,8 @@ static void session_free(struct stream_session *s) { while (*link != s) link = &(*link)->next; *link = s->next; if (s->delivery) uasync_call_soon_cancel(s->service->inst->ua, s->delivery); + if (s->media_work) uasync_call_soon_cancel(s->service->inst->ua, s->media_work); + stream_media_store_destroy(s->media); stream_optimizer_destroy(s->optimizer); if (s->buffer) stream_buffer_set_observer(s->buffer, NULL, NULL); if (s->source) stream_source_destroy(s->source); @@ -260,6 +268,7 @@ static void session_close(struct stream_session *s, int reason, int update_profi for (struct STREAM_HANDLE *h = s->service->handles; h; h = h->next) if (h->session == s) { h->session = NULL; h->reason = reason; } if (s->delivery) { uasync_call_soon_cancel(s->service->inst->ua, s->delivery); s->delivery = NULL; } + if (s->media_work) { uasync_call_soon_cancel(s->service->inst->ua, s->media_work); s->media_work = NULL; } struct TOPO_GROUP *group = s->service->inst->topo_groups ? topo_groups_find(s->service->inst->topo_groups, s->descriptor.group_id) : NULL; if (update_profile && group && service_publish_options(s->service, group) != STREAM_OK) @@ -275,6 +284,62 @@ static void session_close(struct stream_session *s, int reason, int update_profi else if (!s->holds) session_free(s); } +/* Поздний reader ищет ближайший сохранённый KEY с ограниченным бюджетом, затем индексирует вперёд. */ +static void session_media_work(void *arg) { + struct stream_session *s = arg; s->media_work = NULL; s->holds++; + int again = 0; + struct stream_buffer_stats stats; stream_buffer_get_stats(s->buffer, &stats); + if (!s->closing && !s->media_failed) { + for (unsigned work = 0; work < 8 && s->media_seeking; work++) { + size_t length = 0; + int result = stream_buffer_decode(s->buffer, s->media_seek, s->output, stream_input_size(&s->descriptor), &length); + if (result == STREAM_OK) { + struct stream_media_block b; + if (stream_media_parse(s->output, length, &b) != STREAM_OK) { s->media_failed = 1; break; } + for (unsigned i = 0; i < b.video_count; i++) { + if (!b.video[i].offset && (b.video[i].flags & STREAM_MEDIA_KEY)) { + s->media_next = s->media_seek; s->media_seeking = 0; break; + } + } + } + if (!s->media_seeking) break; + if (s->media_seek <= stats.expired_before) { s->media_seeking = 0; s->media_next = s->next_seq; break; } + s->media_seek--; + } + if (s->media_seeking) again = 1; + for (unsigned work = 0; work < 8 && !s->media_seeking && !s->media_failed && s->media_next < s->next_seq; work++) { + size_t length = 0; + int result = stream_buffer_decode(s->buffer, s->media_next, s->output, stream_input_size(&s->descriptor), &length); + if (result == STREAM_OK) { + result = stream_media_store_feed(s->media, s->media_next++, s->output, length); + if (result < 0) s->media_failed = 1; + } else if (result == STREAM_EXPIRED || result == STREAM_REJECTED || result == STREAM_WAIT || result == STREAM_GAP) { + uint64_t next = s->next_seq, available; + if (stream_buffer_next_decodable(s->buffer, s->media_next + 1, &available) == STREAM_OK && available < next) next = available; + stream_media_store_gap(s->media, s->media_next, next - 1); s->media_next = next; + } else { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media index decode failed source=%llx seq=%llu result=%d", + (unsigned long long)s->descriptor.source_node_id, (unsigned long long)s->media_next, result); + s->media_failed = 1; + } + } + if (!s->media_seeking && s->end_valid && (s->last_seq == UINT64_MAX || s->media_next > s->last_seq)) + stream_media_store_finish(s->media); + if (!s->media_failed && s->media_next < s->next_seq) again = 1; + } + s->holds--; + if (s->closing) { if (!s->holds && !s->delivery) session_free(s); } + else if (again) session_media_schedule(s); +} + +/* Индексация не зависит от позиции/паузы плеера и не создаёт отдельную очередь медиаданных. */ +static void session_media_schedule(struct stream_session *s) { + if (!s->media || s->media_work || s->closing || s->media_failed || s->service->stopping) return; + s->media_work = uasync_call_soon(s->service->inst->ua, s, session_media_work); + if (!s->media_work) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media index scheduling failed source=%llx", + (unsigned long long)s->descriptor.source_node_id); +} + /* Уведомление использует собственный scratch экземпляра; close внутри callback не освобождает его преждевременно. */ static void session_deliver(void *arg) { struct stream_session *s = arg; @@ -334,6 +399,13 @@ static void session_deliver(void *arg) { s->end_delivered = 1; emit = 1; } if (!s->closing && s->owners == s->relay_owners) emit = 0; + if (!s->closing && s->media) { + if (event.type == STREAM_EVENT_DATA && emit && !s->media_seeking && s->media_next == event.first_seq) { + int result = stream_media_store_feed(s->media, s->media_next++, event.data, event.length); + if (result < 0) s->media_failed = 1; + } + session_media_schedule(s); + } if (emit && service->callback && !service->stopping) { service->dispatching++; service->callback(service->inst, &event, service->arg); @@ -668,6 +740,7 @@ int stream_selection_get_info(const struct STREAM_HANDLE *h, struct stream_selec /* Publisher авторитетно останавливает трансляцию; reader снимает только собственное владение. */ void stream_close(struct STREAM_HANDLE *h) { if (!h) return; + stream_media_cursor_destroy(h->media_cursor); h->media_cursor = NULL; struct stream_service *service = h->service; if (service) { struct STREAM_HANDLE **link = &service->handles; @@ -704,6 +777,53 @@ int stream_get_info(const struct STREAM_HANDLE *h, struct stream_info *info) { return STREAM_OK; } +/* Первый frontend создаёт индекс точной эпохи, остальные владельцы используют его без копии RS-кольца. */ +int stream_media_enable(struct STREAM_HANDLE *h, int reader) { + 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) { + int result = stream_media_store_create(&s->media, s->buffer, &s->descriptor); + if (result != STREAM_OK) return result; + struct stream_buffer_stats stats; stream_buffer_get_stats(s->buffer, &stats); + s->media_next = stats.expired_before; + if (s->anchored && s->next_seq > stats.expired_before) { + 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 (result != STREAM_OK) return result; + } + session_media_schedule(s); + return STREAM_OK; +} + +/* Чтение изменяет только локальный cursor; live ingest/READY/выдачи продолжают двигаться независимо. */ +int stream_media_read(struct STREAM_HANDLE *h, int64_t position, int64_t lookahead, 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); +} + +int stream_media_get_status(struct STREAM_HANDLE *h, struct stream_media_status *status) { + if (!stream_handle_active(h) || !h->media_cursor || !status) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media status: detached/unconfigured reader"); return STREAM_ERR_STATE; + } + if (h->session->media_failed) return STREAM_ERR_FORMAT; + return stream_media_cursor_status(h->media_cursor, status); +} + +int stream_media_resync(struct STREAM_HANDLE *h) { + if (!stream_handle_active(h) || !h->media_cursor) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media resync: detached/unconfigured reader"); return STREAM_ERR_STATE; + } + stream_media_cursor_reset(h->media_cursor); + return STREAM_OK; +} + size_t stream_publish_capacity(const struct STREAM_HANDLE *h) { if (!stream_handle_active(h) || !h->publisher) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream capacity not a publisher"); return 0; } return stream_source_input_capacity(h->session->source); diff --git a/src/stream/stream.h b/src/stream/stream.h index 4625a87a..4ea78179 100644 --- a/src/stream/stream.h +++ b/src/stream/stream.h @@ -8,6 +8,7 @@ #define STREAM_H #include "stream_buffer.h" +#include "stream_media_buffer.h" #ifdef __cplusplus extern "C" { @@ -86,6 +87,15 @@ int stream_handle_active(const struct STREAM_HANDLE *handle); 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. + * Все вызовы — 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_get_status(struct STREAM_HANDLE *handle, struct stream_media_status *status); +int stream_media_resync(struct STREAM_HANDLE *handle); + size_t stream_publish_capacity(const struct STREAM_HANDLE *handle); int stream_publish_write(struct STREAM_HANDLE *handle, const uint8_t *data, size_t length, size_t *accepted); /* Публикует одну непустую FEC-пачку, без склейки/таймера. Не смешивать с publish_write в одном запуске. */