From e4ce3cacfc0bf6c04c7e78183ce5c448ad218f8a Mon Sep 17 00:00:00 2001 From: Oxan van Leeuwen Date: Tue, 7 Feb 2023 21:46:43 +0100 Subject: [PATCH] Drop data only for clients that can't keep up --- components/stream_server/stream_server.cpp | 30 ++++++++++++++-------- 1 file changed, 19 insertions(+), 11 deletions(-) diff --git a/components/stream_server/stream_server.cpp b/components/stream_server/stream_server.cpp index 0116f04..3630645 100644 --- a/components/stream_server/stream_server.cpp +++ b/components/stream_server/stream_server.cpp @@ -101,23 +101,31 @@ void StreamServerComponent::cleanup() { } void StreamServerComponent::read() { - bool first_iteration = true; + size_t len = 0; int available; while ((available = this->stream_->available()) > 0) { - // Write until the tail is encountered, or wraparound of the ring buffer if that happens before. - size_t max = std::min(this->buf_ahead(this->buf_head_), this->buf_tail_ + this->buf_size_ - this->buf_head_); - if (max == 0) { - // Only warn on the first iteration, the finite buffer size is also used as a throttling mechanism to avoid - // blocking here for too long when a large amount of data comes in. - if (first_iteration) - ESP_LOGW(TAG, "Incoming bytes available in stream, but outgoing buffer is full!"); - break; + size_t free = this->buf_size_ - (this->buf_head_ - this->buf_tail_); + if (free == 0) { + // Only overwrite if nothing has been added yet, otherwise give flush() a chance to empty the buffer first. + if (len > 0) + return; + + ESP_LOGE(TAG, "Incoming bytes available, but outgoing buffer is full: stream will be corrupted!"); + free = std::min(available, this->buf_size_); + this->buf_tail_ += free; + for (Client &client : this->clients_) { + if (client.position < this->buf_tail_) { + ESP_LOGW(TAG, "Dropped %u pending bytes for client %s", this->buf_tail_ - client.position, client.identifier.c_str()); + client.position = this->buf_tail_; + } + } + } - size_t len = std::min(available, max); + // Fill all available contiguous space in the ring buffer. + len = std::min(available, std::min(this->buf_ahead(this->buf_head_), free)); this->stream_->read_array(&this->buf_[this->buf_index(this->buf_head_)], len); this->buf_head_ += len; - first_iteration = false; } }