145 lines
3.4 KiB
C++
145 lines
3.4 KiB
C++
#include "CVrZeroMQPubSub.h"
|
|
#include "VrError.h"
|
|
#include <thread>
|
|
#include <string>
|
|
#include <cstring>
|
|
#include <exception>
|
|
#include "VrLog.h"
|
|
|
|
// ========== 发布者实现 ==========
|
|
bool IVrZeroMQPublisher::CreateObject(IVrZeroMQPublisher** ppPublisher)
|
|
{
|
|
*ppPublisher = new CVrZeroMQPublisher();
|
|
return true;
|
|
}
|
|
|
|
CVrZeroMQPublisher::CVrZeroMQPublisher()
|
|
{
|
|
m_pContext = new zmq::context_t;
|
|
m_pSocket = new zmq::socket_t(*m_pContext, ZMQ_PUB);
|
|
}
|
|
|
|
CVrZeroMQPublisher::~CVrZeroMQPublisher() {}
|
|
|
|
int CVrZeroMQPublisher::Init(int nPort)
|
|
{
|
|
try {
|
|
std::string addr = "tcp://*:" + std::to_string(nPort);
|
|
LOG_DEBUG("zeroMQ: %s \n", addr.c_str());
|
|
m_pSocket->bind(addr);
|
|
return 0;
|
|
}
|
|
catch (const zmq::error_t& e) {
|
|
LOG_ERROR("ZeroMQ bind failed: %s\n", e.what());
|
|
return ERR_CODE(NET_ERR_CREAT_BIND);
|
|
}
|
|
}
|
|
|
|
int CVrZeroMQPublisher::Publish(const char* pTopic, const char* pData, const size_t nLen)
|
|
{
|
|
if (!m_pSocket || !pTopic || (!pData && nLen > 0))
|
|
{
|
|
LOG_ERROR("ZeroMQ publish invalid args socket=%p topic=%p data=%p len=%zu\n",
|
|
m_pSocket, pTopic, pData, nLen);
|
|
return ERR_CODE(NET_ERR_ARG);
|
|
}
|
|
|
|
try
|
|
{
|
|
size_t topicLen = strlen(pTopic);
|
|
zmq::message_t topicMsg(topicLen);
|
|
memcpy(topicMsg.data(), pTopic, topicLen);
|
|
m_pSocket->send(topicMsg, zmq::send_flags::sndmore);
|
|
|
|
zmq::message_t dataMsg(pData, nLen);
|
|
m_pSocket->send(dataMsg, zmq::send_flags::none);
|
|
return SUCCESS;
|
|
}
|
|
catch (const zmq::error_t& e)
|
|
{
|
|
LOG_ERROR("ZeroMQ publish failed topic=%s len=%zu: %s\n", pTopic, nLen, e.what());
|
|
return ERR_CODE(NET_ERR_SEND_DATA);
|
|
}
|
|
catch (const std::exception& e)
|
|
{
|
|
LOG_ERROR("ZeroMQ publish exception topic=%s len=%zu: %s\n", pTopic, nLen, e.what());
|
|
return ERR_CODE(NET_ERR_SEND_DATA);
|
|
}
|
|
catch (...)
|
|
{
|
|
LOG_ERROR("ZeroMQ publish unknown exception topic=%s len=%zu\n", pTopic, nLen);
|
|
return ERR_CODE(NET_ERR_SEND_DATA);
|
|
}
|
|
}
|
|
|
|
int CVrZeroMQPublisher::UnInit()
|
|
{
|
|
delete m_pSocket;
|
|
delete m_pContext;
|
|
m_pSocket = nullptr;
|
|
m_pContext = nullptr;
|
|
return 0;
|
|
}
|
|
|
|
// ========== 订阅者实现 ==========
|
|
bool IVrZeroMQSubscriber::CreateObject(IVrZeroMQSubscriber** ppSubscriber)
|
|
{
|
|
*ppSubscriber = new CVrZeroMQSubscriber();
|
|
return true;
|
|
}
|
|
|
|
CVrZeroMQSubscriber::CVrZeroMQSubscriber()
|
|
: m_bRecv(false), m_funRecv(nullptr)
|
|
{
|
|
m_pContext = new zmq::context_t;
|
|
m_pSocket = new zmq::socket_t(*m_pContext, ZMQ_SUB);
|
|
}
|
|
|
|
CVrZeroMQSubscriber::~CVrZeroMQSubscriber() {}
|
|
|
|
int CVrZeroMQSubscriber::Init(const char* sIP, int nPort, const char* pTopic, FunSubRecv funRecv)
|
|
{
|
|
std::string addr = "tcp://" + std::string(sIP) + ":" + std::to_string(nPort);
|
|
m_pSocket->connect(addr);
|
|
m_pSocket->set(zmq::sockopt::subscribe, pTopic);
|
|
|
|
m_funRecv = funRecv;
|
|
m_bRecv = true;
|
|
std::thread recvThread(&CVrZeroMQSubscriber::_RecvData, this);
|
|
recvThread.detach();
|
|
|
|
return 0;
|
|
}
|
|
|
|
int CVrZeroMQSubscriber::UnInit()
|
|
{
|
|
m_bRecv = false;
|
|
delete m_pSocket;
|
|
delete m_pContext;
|
|
m_pSocket = nullptr;
|
|
m_pContext = nullptr;
|
|
return 0;
|
|
}
|
|
|
|
void CVrZeroMQSubscriber::_RecvData()
|
|
{
|
|
while (m_bRecv)
|
|
{
|
|
zmq::message_t topicMsg;
|
|
zmq::recv_result_t ret1 = m_pSocket->recv(topicMsg, zmq::recv_flags::dontwait);
|
|
if (!ret1.has_value()) continue;
|
|
|
|
zmq::message_t dataMsg;
|
|
zmq::recv_result_t ret2 = m_pSocket->recv(dataMsg, zmq::recv_flags::dontwait);
|
|
if (!ret2.has_value()) continue;
|
|
|
|
if (m_funRecv)
|
|
{
|
|
m_funRecv(static_cast<char*>(topicMsg.data()),
|
|
topicMsg.size(),
|
|
static_cast<char*>(dataMsg.data()),
|
|
dataMsg.size());
|
|
}
|
|
}
|
|
}
|