Skip to content

Commit

Permalink
fix: modified code to use flush() when internal queue is not loaded
Browse files Browse the repository at this point in the history
  • Loading branch information
i-chvets committed Jul 12, 2024
1 parent 1d2e943 commit d75c90f
Showing 1 changed file with 11 additions and 2 deletions.
13 changes: 11 additions & 2 deletions src/framework/KafkaManager.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -107,7 +107,16 @@ namespace OpenWifi {
NewMessage.partition(0);
NewMessage.payload(Msg->Payload());
Producer.produce(NewMessage);
Producer.flush();
if (Queue_.size() < 100) {
// use flush when internal queue is lightly loaded, i.e. flush after each
// message
Producer.flush();
}
else {
// use poll when internal queue is loaded to allow messages to be sent in
// batches
Producer.poll((std::chrono::milliseconds) 0);
}
}
} catch (const cppkafka::HandleException &E) {
poco_warning(Logger_,
Expand Down Expand Up @@ -324,4 +333,4 @@ namespace OpenWifi {
partitions.front().get_partition()));
}

} // namespace OpenWifi
} // namespace OpenWifi

0 comments on commit d75c90f

Please sign in to comment.