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

Fix tracker promise from throwing when set multiple times

Browse Source
This commit is contained in:
Alexander Damian
2020-02-03 16:46:28 -05:00
parent f1de729d4e
commit bda2f4156d
1 changed files with 16 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -771,7 +771,16 @@ void BufferedProducer<BufferType, Allocator>::clear() {
template <typename BufferType, typename Allocator>
size_t BufferedProducer<BufferType, Allocator>::get_buffer_size() const {
return messages_.size() + retry_messages_.size();
int size = 0;
{
std::lock_guard<std::mutex> lock(mutex_);
size += messages_.size();
}
{
std::lock_guard<std::mutex> lock(retry_mutex_);
size += retry_messages_.size();
}
return size;
}
template <typename BufferType, typename Allocator>
@@ -1025,7 +1034,12 @@ void BufferedProducer<BufferType, Allocator>::on_delivery_report(const Message&
}
// Signal producers
if (tracker) {
tracker->should_retry_.set_value(should_retry);
try {
tracker->should_retry_.set_value(should_retry);
}
catch (const std::future_error& ex) {
//This is an async retry and future is not being read
}
}
// Decrement the expected acks and check to prevent underflow
if (pending_acks_ > 0) {