Add connect/close/endpoint method to ZmqClient/ZmqServer.

common/zmqclient.cpp

@@ -22,21 +22,7 @@ ZmqClient::ZmqClient(const std::string& endpoint)
 
 ZmqClient::~ZmqClient()
 {
-    std::lock_guard<std::mutex> lock(m_socketMutex);
-    if (m_socket)
-    {
-        int rc = zmq_close(m_socket);
-        if (rc != 0)
-        {
-            SWSS_LOG_ERROR("failed to close zmq socket, zmqerrno: %d",
-                    zmq_errno());
-        }
-    }
-
-    if (m_context)
-    {
-        zmq_ctx_destroy(m_context);
-    }
+    close();
 }
 
 void ZmqClient::initialize(const std::string& endpoint)
@@ -48,6 +34,11 @@ void ZmqClient::initialize(const std::string& endpoint)
 
     connect();
 }
+
+std::string ZmqClient::endpoint()
+{
+    return m_endpoint;
+}
 
 bool ZmqClient::isConnected()
 {
@@ -102,6 +93,30 @@ void ZmqClient::connect()
     m_connected = true;
 }
 
+void ZmqClient::close()
+{
+    std::lock_guard<std::mutex> lock(m_socketMutex);
+    if (m_socket)
+    {
+        int rc = zmq_close(m_socket);
+        if (rc != 0)
+        {
+            SWSS_LOG_ERROR("failed to close zmq socket, zmqerrno: %d",
+                    zmq_errno());
+        }
+
+        m_socket = nullptr;
+    }
+
+    if (m_context)
+    {
+        zmq_ctx_destroy(m_context);
+        m_context = nullptr;
+    }
+
+    m_connected = false;
+}
+
 void ZmqClient::sendMsg(
         const std::string& dbName,
         const std::string& tableName,

common/zmqclient.h

@@ -17,8 +17,12 @@ class ZmqClient
 
     bool isConnected();
 
+    std::string endpoint();
+
     void connect();
 
+    void close();
+
     void sendMsg(const std::string& dbName,
                  const std::string& tableName,
                  const std::vector<KeyOpFieldsValuesTuple>& kcos,

common/zmqserver.cpp

@@ -15,16 +15,42 @@ ZmqServer::ZmqServer(const std::string& endpoint)
     : m_endpoint(endpoint)
 {
     m_buffer.resize(MQ_RESPONSE_MAX_COUNT);
-    m_runThread = true;
-    m_mqPollThread = std::make_shared<std::thread>(&ZmqServer::mqPollThread, this);
+    connect();
 
     SWSS_LOG_DEBUG("ZmqServer ctor endpoint: %s", endpoint.c_str());
 }
 
 ZmqServer::~ZmqServer()
 {
+    close();
+}
+
+std::string ZmqServer::endpoint()
+{
+    return m_endpoint;
+}
+
+void ZmqServer::connect()
+{
+    if (m_mqPollThread)
+    {
+        return;
+    }
+
+    m_runThread = true;
+    m_mqPollThread = std::make_shared<std::thread>(&ZmqServer::mqPollThread, this);
+}
+
+void ZmqServer::close()
+{
+    if (!m_mqPollThread)
+    {
+        return;
+    }
+
     m_runThread = false;
     m_mqPollThread->join();
+    m_mqPollThread = nullptr;
 }
 
 void ZmqServer::registerMessageHandler(

common/zmqserver.h

@@ -33,6 +33,12 @@ class ZmqServer
     ZmqServer(const std::string& endpoint);
     ~ZmqServer();
 
+    std::string endpoint();
+
+    void connect();
+
+    void close();
+
     void registerMessageHandler(
                                 const std::string dbName,
                                 const std::string tableName,

tests/zmq_state_ut.cpp

@@ -192,6 +192,7 @@ static void producerBatchWorker(string tableName, string endpoint, bool dbPersis
         }
     }
 
+    client.close();
     cout << "Producer thread ended: " << tableName << endl;
 }
 
@@ -268,6 +269,7 @@ static void consumerWorker(string tableName, string endpoint, bool dbPersistence
         }
     }
 
+    server.close();
     cout << "Consumer thread ended: " << tableName << endl;
 }