From 47caca82a19072336119df0a0c112e4ebb75ca98 Mon Sep 17 00:00:00 2001 From: evgeny Date: Mon, 12 Oct 2026 01:43:31 +0200 Subject: [PATCH] Expose atomic publication of stream batches --- src/stream/stream.c | 13 +++++++++++++ src/stream/stream.h | 2 ++ 2 files changed, 15 insertions(+) diff --git a/src/stream/stream.c b/src/stream/stream.c index 13647aeb..2565c0b2 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -547,6 +547,19 @@ int stream_publish_write(struct STREAM_HANDLE *h, const uint8_t *data, size_t le return result; } +/* Блочная публикация использует те же holds и обработку отказов, что и источник непрерывных байтов. */ +int stream_publish_block(struct STREAM_HANDLE *h, const uint8_t *data, size_t length) { + if (!stream_handle_active(h) || !h->publisher) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream block write not an active publisher"); return STREAM_ERR_STATE; + } + struct stream_session *s = h->session; s->holds++; + int result = stream_source_write_block(s->source, data, length); + if (result != STREAM_OK && result != STREAM_ERR_ARGUMENT) session_close(s, STREAM_CLOSE_FAILED, 1, 1); + s->holds--; + if (s->closing && !s->holds && !s->delivery) session_free(s); + return result; +} + int stream_publish_finish(struct STREAM_HANDLE *h) { if (!stream_handle_active(h) || !h->publisher) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream finish not a publisher"); return STREAM_ERR_STATE; } struct stream_session *s = h->session; diff --git a/src/stream/stream.h b/src/stream/stream.h index 37803e0f..6f531ab5 100644 --- a/src/stream/stream.h +++ b/src/stream/stream.h @@ -82,6 +82,8 @@ int stream_get_info(const struct STREAM_HANDLE *handle, struct stream_info *info 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 в одном запуске. */ +int stream_publish_block(struct STREAM_HANDLE *handle, const uint8_t *data, size_t length); /* Сбросить остаток, снять статус публикации; DATA перед END. Буфер живёт до close. */ int stream_publish_finish(struct STREAM_HANDLE *handle);