Fix: PublisherZMQ::flush is called after the publisher has been destructed (BehaviorTree#426)

* fix: PublisherZMQ::flush is called after the publisher has been destructed

* style: Adjust code formatting of ~PublisherZMQ

* chore: Install zmq-dev in ubuntu pipeline and exclude gtest_logger_zmq.cpp when zmq is not found.

* chore: Define WIN32_LEAN_AND_MEAN to avoid ambiguity between tinyxml and msxml

Author: Tradias

.github/workflows/cmake.yml

@@ -21,7 +21,7 @@ jobs:
     - uses: actions/checkout@v2
 
     - name: Install Dependencies (Linux)
-      run: sudo apt-get install libboost-dev
+      run: sudo apt-get install libboost-dev libzmq3-dev
       if: matrix.os == 'ubuntu-latest'
 
     - name: Create Build Environment

CMakeLists.txt

@@ -10,7 +10,7 @@ set(CMAKE_CXX_STANDARD 14)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
 
 if(MSVC)
-    add_definitions(-D_CRT_SECURE_NO_WARNINGS)
+    add_definitions(-D_CRT_SECURE_NO_WARNINGS -DWIN32_LEAN_AND_MEAN)
 else()
     add_definitions(-Wpedantic)
 endif()

include/behaviortree_cpp_v3/loggers/bt_zmq_publisher.h

@@ -37,7 +37,7 @@ class PublisherZMQ : public StatusChangeLogger
   TimePoint deadline_;
   std::mutex mutex_;
   std::atomic_bool send_pending_;
-
+  std::condition_variable send_condition_variable_;
   std::future<void> send_future_;
 
   struct Pimpl;

src/loggers/bt_zmq_publisher.cpp

@@ -90,12 +90,13 @@ PublisherZMQ::~PublisherZMQ()
   {
     thread_.join();
   }
-  flush();
-  zmq_->context.shutdown();
-  if (send_future_.valid())
+  if (send_pending_)
   {
-    send_future_.wait();
+    send_condition_variable_.notify_all();
+    send_future_.get();
   }
+  flush();
+  zmq_->context.shutdown();
   delete zmq_;
   ref_count = false;
 }
@@ -116,21 +117,24 @@ void PublisherZMQ::createStatusBuffer()
 void PublisherZMQ::callback(Duration timestamp, const TreeNode& node,
                             NodeStatus prev_status, NodeStatus status)
 {
-  using namespace std::chrono;
-
   SerializedTransition transition =
       SerializeTransition(node.UID(), timestamp, prev_status, status);
   {
     std::unique_lock<std::mutex> lock(mutex_);
     transition_buffer_.push_back(transition);
   }
 
-  if (!send_pending_)
+  if (!send_pending_.exchange(true))
   {
-    send_pending_ = true;
     send_future_ = std::async(std::launch::async, [this]() {
-      std::this_thread::sleep_for(min_time_between_msgs_);
-      flush();
+      std::unique_lock<std::mutex> lock(mutex_);
+      const bool is_server_inactive = send_condition_variable_.wait_for(
+          lock, min_time_between_msgs_, [this]() { return !active_server_; });
+      lock.unlock();
+      if (!is_server_inactive)
+      {
+        flush();
+      }
     });
   }
 }

tests/CMakeLists.txt

@@ -19,9 +19,12 @@ set(BT_TESTS
   gtest_subtree.cpp
   gtest_switch.cpp
   gtest_wakeup.cpp
-  gtest_logger_zmq.cpp
 )
 
+if( ZMQ_FOUND )
+    LIST( APPEND BT_TESTS gtest_logger_zmq.cpp)
+endif()
+
 if( BT_COROUTINES )
     LIST( APPEND BT_TESTS gtest_coroutines.cpp)
 endif()