Allow clearing buffered messages on buffered producer

Author: mfontanini

include/cppkafka/utils/buffered_producer.h

@@ -3,8 +3,8 @@
 
 #include <string>
 #include <queue>
-#include <type_traits>
 #include <cstdint>
+#include <algorithm>
 #include <unordered_set>
 #include <unordered_map>
 #include <map>
@@ -93,6 +93,11 @@ class BufferedProducer {
      */
     void wait_for_acks();
 
+    /**
+     * Clears any buffered messages
+     */
+    void clear();
+
     /**
      * Gets the Producer object
      */
@@ -119,8 +124,7 @@ class BufferedProducer {
      */
     void set_produce_failure_callback(ProduceFailureCallback callback);
 private:
-    // Pick the most appropriate index type depending on the platform we're using
-    using IndexType = std::conditional<sizeof(void*) == 8, uint64_t, uint32_t>::type;
+    using QueueType = std::queue<Builder>;
 
     template <typename BuilderType>
     void do_add_message(BuilderType&& builder);
@@ -129,7 +133,7 @@ class BufferedProducer {
     void on_delivery_report(const Message& message);
 
     Producer producer_;
-    std::queue<Builder> messages_;
+    QueueType messages_;
     ProduceFailureCallback produce_failure_callback_;
     size_t expected_acks_{0};
     size_t messages_acked_{0};
@@ -187,6 +191,12 @@ void BufferedProducer<BufferType>::wait_for_acks() {
     messages_acked_ = 0;
 }
 
+template <typename BufferType>
+void BufferedProducer<BufferType>::clear() {
+    QueueType tmp;
+    std::swap(tmp, messages_);
+}
+
 template <typename BufferType>
 template <typename BuilderType>
 void BufferedProducer<BufferType>::do_add_message(BuilderType&& builder) {

tests/producer_test.cpp

@@ -238,6 +238,9 @@ TEST_F(ProducerTest, BufferedProducer) {
     producer.flush();
     producer.produce(MessageBuilder(KAFKA_TOPIC).partition(partition).payload(payload));
     producer.wait_for_acks();
+    // Add another one but then clear it
+    producer.add_message(producer.make_builder(KAFKA_TOPIC).partition(partition).payload(payload));
+    producer.clear();
     runner.try_join();
 
     const auto& messages = runner.get_messages();