Using single mutex version

Author: accelerated

CMakeLists.txt

@@ -51,7 +51,7 @@ find_package(Boost REQUIRED ${FIND_PACKAGE_QUIET})
 find_package(RdKafka REQUIRED ${FIND_PACKAGE_QUIET})
 
 if (Boost_FOUND)
-    find_package(Boost COMPONENTS thread program_options ${FIND_PACKAGE_QUIET})
+    find_package(Boost COMPONENTS program_options ${FIND_PACKAGE_QUIET})
     set(Boost_USE_STATIC_LIBS ${CPPKAFKA_BOOST_STATIC_LIBS})
     set(Boost_USE_MULTITHREADED ${CPPKAFKA_BOOST_USE_MULTITHREADED})
     include_directories(${Boost_INCLUDE_DIRS})
@@ -91,7 +91,7 @@ if(DOXYGEN_FOUND)
     )
 endif(DOXYGEN_FOUND)
 
-if(NOT CPPKAFKA_DISABLE_TESTS AND Boost_THREAD_FOUND)
+if(NOT CPPKAFKA_DISABLE_TESTS)
     set(CATCH_ROOT ${CMAKE_SOURCE_DIR}/third_party/Catch2)
     if(EXISTS ${CATCH_ROOT}/CMakeLists.txt)
         set(CATCH_INCLUDE ${CATCH_ROOT}/single_include)

include/cppkafka/utils/buffered_producer.h

@@ -37,8 +37,7 @@
 #include <unordered_set>
 #include <unordered_map>
 #include <map>
-#include <boost/thread/mutex.hpp>
-#include <boost/thread/shared_mutex.hpp>
+#include <mutex>
 #include <atomic>
 #include <boost/optional.hpp>
 #include "../producer.h"
@@ -190,8 +189,7 @@ class CPPKAFKA_API BufferedProducer {
 
     Producer producer_;
     QueueType messages_;
-    mutable boost::mutex exclusive_access_;
-    mutable boost::shared_mutex shared_access_;
+    mutable std::mutex mutex_;
     ProduceFailureCallback produce_failure_callback_;
     std::atomic_ulong expected_acks_{0};
     std::atomic_ullong total_messages_acked_{0};
@@ -221,14 +219,14 @@ void BufferedProducer<BufferType>::produce(const MessageBuilder& builder) {
 
 template <typename BufferType>
 void BufferedProducer<BufferType>::flush() {
-    {
-        boost::shared_lock<boost::shared_mutex> grant(shared_access_);
-        size_t num_messages = messages_.size();
-        while (num_messages--) {
-            produce_message(messages_.front());
-            boost::lock_guard<boost::mutex> require(exclusive_access_);
-            messages_.pop();
+    size_t num_messages = messages_.size();
+    while (num_messages--) {
+        std::lock_guard<std::mutex> lock(mutex_);
+        if (messages_.empty()) {
+            break; //perhaps clear() was called
         }
+        produce_message(messages_.front());
+        messages_.pop();
     }
     wait_for_acks();
 }
@@ -253,16 +251,15 @@ void BufferedProducer<BufferType>::wait_for_acks() {
 
 template <typename BufferType>
 void BufferedProducer<BufferType>::clear() {
-    boost::unique_lock<boost::shared_mutex> restrict(shared_access_);
+    std::lock_guard<std::mutex> lock(mutex_);
     QueueType tmp;
     std::swap(tmp, messages_);
 }
 
 template <typename BufferType>
 template <typename BuilderType>
 void BufferedProducer<BufferType>::do_add_message(BuilderType&& builder) {
-    boost::shared_lock<boost::shared_mutex> grant(shared_access_);
-    boost::lock_guard<boost::mutex> require(exclusive_access_);
+    std::lock_guard<std::mutex> lock(mutex_);
     messages_.push(std::move(builder));
 }
 

tests/CMakeLists.txt

@@ -8,7 +8,7 @@ add_custom_target(tests)
 
 include_directories(${CMAKE_CURRENT_SOURCE_DIR})
 add_library(cppkafka-test EXCLUDE_FROM_ALL test_utils.cpp)
-target_link_libraries(cppkafka-test cppkafka ${RDKAFKA_LIBRARY} ${Boost_LIBRARIES} pthread)
+target_link_libraries(cppkafka-test cppkafka ${RDKAFKA_LIBRARY} pthread)
 
 add_definitions("-DKAFKA_TEST_INSTANCE=\"${KAFKA_TEST_INSTANCE}\"")