diff --git a/src/stream/stream_source.c b/src/stream/stream_source.c index e8bddc48..9211b0b4 100644 --- a/src/stream/stream_source.c +++ b/src/stream/stream_source.c @@ -24,6 +24,7 @@ struct stream_source { int has_batch; int failed; int ended; + int write_mode; /* 0 — не выбран, 1 — байты, 2 — целые пачки. */ }; /* Ошибка не скрывает уже принятый префикс: прекращаем таймер и дальнейшие write. */ @@ -143,6 +144,11 @@ int stream_source_write(struct stream_source *source, const uint8_t *data, size_ return STREAM_ERR_STATE; } if (!length) return STREAM_OK; + if (source->write_mode == 2) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source byte write conflicts with block mode length=%zu", length); + return STREAM_ERR_ARGUMENT; + } + source->write_mode = 1; source->busy = 1; int result = STREAM_OK; while (*accepted < length) { @@ -179,3 +185,27 @@ int stream_source_finish(struct stream_source *source) { } return result; } + +/* Сохранить границу медиаблока даже для неполной пачки; общий flush подписывает и сохраняет все подблоки. */ +int stream_source_write_block(struct stream_source *source, const uint8_t *data, size_t length) { + if (!source || !data || !length || length > source->capacity) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source block write invalid length=%zu capacity=%zu", length, source ? source->capacity : 0); + return STREAM_ERR_ARGUMENT; + } + if (source->busy || source->failed || source->ended) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source block write rejected busy=%d failed=%d ended=%d", + source->busy, source->failed, source->ended); + return STREAM_ERR_STATE; + } + if (source->write_mode == 1) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "source block write conflicts with byte mode pending=%zu", source->used); + return STREAM_ERR_ARGUMENT; + } + source->write_mode = 2; + source->busy = 1; + memcpy(source->input, data, length); + source->used = length; + int result = source_flush(source); + source->busy = 0; + return result; +} diff --git a/src/stream/stream_source.h b/src/stream/stream_source.h index 364fcf71..aba33302 100644 --- a/src/stream/stream_source.h +++ b/src/stream/stream_source.h @@ -27,6 +27,9 @@ struct stream_buffer *stream_source_buffer(struct stream_source *source); * accepted получает принятую часть даже при ошибке; нулевой write допустим и не создаёт пустую пачку. * Ошибка публикации/таймера делает источник failed; его уничтожают, не повторяют принятую часть как новую трансляцию. */ int stream_source_write(struct stream_source *source, const uint8_t *data, size_t length, size_t *accepted); +/* Один непустой блок = одна немедленно опубликованная FEC-пачка, length <= input_capacity. + * Режим выбирается первой непустой записью; побайтовый и блочный режимы не смешиваются. */ +int stream_source_write_block(struct stream_source *source, const uint8_t *data, size_t length); /* Сбросить остаток и остановить приём; повторный finish допустим, кольцо доступно до destroy. */ int stream_source_finish(struct stream_source *source);