diff --git a/core-writer/include/ZmqRecvModule.hpp b/core-writer/include/ZmqRecvModule.hpp index 348b821..0d95403 100644 --- a/core-writer/include/ZmqRecvModule.hpp +++ b/core-writer/include/ZmqRecvModule.hpp @@ -11,8 +11,8 @@ class ZmqRecvModule RingBuffer& ring_buffer_; const header_map& header_values_; - const std::atomic_bool& is_writing_; std::atomic_bool is_receiving_; + std::atomic_bool is_writing_; std::vector receiving_threads_; protected: @@ -22,13 +22,15 @@ protected: public: ZmqRecvModule( RingBuffer& ring_buffer, - const header_map& header_values, - const std::atomic_bool& is_writing); + const header_map& header_values); void start_recv(const std::string& connect_address, const uint8_t n_receiving_threads); void stop_recv(); + void start_writing(); + void stop_writing(); + }; diff --git a/core-writer/src/ZmqRecvModule.cpp b/core-writer/src/ZmqRecvModule.cpp index 27be1b1..59f902f 100644 --- a/core-writer/src/ZmqRecvModule.cpp +++ b/core-writer/src/ZmqRecvModule.cpp @@ -8,12 +8,11 @@ using namespace std; ZmqRecvModule::ZmqRecvModule( RingBuffer &ringBuffer, - const header_map &header_values, - const std::atomic_bool& is_writing) : + const header_map &header_values) : ring_buffer_(ring_buffer_), header_values_(header_values), - is_writing_(is_writing), - is_receiving_(false) + is_receiving_(false), + is_writing_(false) {} void ZmqRecvModule::start_recv( @@ -69,6 +68,16 @@ void ZmqRecvModule::stop_recv() receiving_threads_.clear(); } +void ZmqRecvModule::start_writing() +{ + is_writing_ = true; +} + +void ZmqRecvModule::stop_writing() +{ + is_writing_ = false; +} + void ZmqRecvModule::receive_thread(const string& connect_address) { ZmqReceiver receiver(