scottk/wlan-cloud-lib-cppkafka
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-cloud-lib-cppkafka.git synced 2025-11-03 20:18:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #150 from accelerated/flush_bug

Browse Source 
Bug with message leak in BufferedProducer::flush(timeout)
This commit is contained in:
Matias Fontanini
2018-12-16 09:45:35 -08:00
committed by GitHub
parent ccc6738265 8b431c5421
commit f14a4b9e8c
1 changed files with 6 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
include/cppkafka/utils/buffered_producer.h
View File

@@ -633,6 +633,12 @@ bool BufferedProducer<BufferType, Allocator>::flush(std::chrono::milliseconds ti
remaining = timeout - std::chrono::duration_cast<std::chrono::milliseconds> remaining = timeout - std::chrono::duration_cast<std::chrono::milliseconds>
(std::chrono::high_resolution_clock::now() - start_time); (std::chrono::high_resolution_clock::now() - start_time);
} while (!flush_queue.empty() && (remaining.count() > 0)); } while (!flush_queue.empty() && (remaining.count() > 0));
// Re-enqueue remaining messages in original order
if (!flush_queue.empty()) {
std::lock_guard<std::mutex> lock(mutex_);
messages_.insert(messages_.begin(), std::make_move_iterator(flush_queue.begin()), std::make_move_iterator(flush_queue.end()));
}
} }
else { else {
async_flush(); async_flush();