Browse Source

Expose atomic publication of stream batches

master
evgeny 1 week ago
parent
commit
47caca82a1
  1. 13
      src/stream/stream.c
  2. 2
      src/stream/stream.h

13
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;

2
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);

Loading…
Cancel
Save