85 lines
2.0 KiB
C++
85 lines
2.0 KiB
C++
#include "CVrZeroMQTask.h"
|
|
#include <thread>
|
|
#include <iostream>
|
|
#include <sstream>
|
|
#include <string>
|
|
|
|
#include "VrError.h"
|
|
#include "IVrUtils.h"
|
|
#include "VrStringUtils.h"
|
|
#include "VrLog.h"
|
|
|
|
CVrZeroMQTask::CVrZeroMQTask()
|
|
: m_pContext(nullptr)
|
|
, m_pSocket(nullptr)
|
|
, m_funMQEvent(nullptr)
|
|
, m_funRecvAck(nullptr)
|
|
, m_bRecv(false)
|
|
, m_bRecvWorking(false)
|
|
{
|
|
}
|
|
|
|
CVrZeroMQTask::~CVrZeroMQTask()
|
|
{
|
|
UnInit();
|
|
}
|
|
|
|
/// 启动服务
|
|
int CVrZeroMQTask::Init(zmq::context_t* pContext, FunMQEvent funMQEvent, FunServerRecvAndAck funRecvAck)
|
|
{
|
|
m_pContext = pContext;
|
|
|
|
m_funMQEvent = funMQEvent;
|
|
m_funRecvAck = funRecvAck;
|
|
|
|
m_bRecv = true;
|
|
m_recvThread = std::thread(&CVrZeroMQTask::_WorkTask, this, pContext);
|
|
m_recvThread.detach();
|
|
|
|
return SUCCESS;
|
|
}
|
|
|
|
/// 销毁服务
|
|
int CVrZeroMQTask::UnInit()
|
|
{
|
|
// Stop receive thread and close socket to interrupt blocking recv
|
|
m_bRecv = false;
|
|
try { if (m_pSocket) m_pSocket->close(); } catch (...) {}
|
|
if (m_recvThread.joinable()) {
|
|
m_recvThread.join();
|
|
}
|
|
|
|
return SUCCESS;
|
|
}
|
|
|
|
|
|
void CVrZeroMQTask::_WorkTask(zmq::context_t *context)
|
|
{
|
|
if (!context) return;
|
|
|
|
m_bRecvWorking = true;
|
|
|
|
m_pSocket = new zmq::socket_t(*context, ZMQ_REP);
|
|
m_pSocket->connect("inproc://workers");
|
|
|
|
try {
|
|
while (m_bRecv) {
|
|
zmq::message_t msg;
|
|
zmq::recv_result_t ret = m_pSocket->recv(msg, zmq::recv_flags::none);
|
|
if (!m_bRecv) break;
|
|
if (ret.has_value()) {
|
|
if (m_funRecvAck) {
|
|
std::string strResponse = m_funRecvAck(static_cast<char*>(msg.data()), msg.size());
|
|
zmq::message_t response(strResponse.c_str(), strResponse.size());
|
|
m_pSocket->send(response, zmq::send_flags::none);
|
|
} else {
|
|
m_pSocket->send(msg, zmq::send_flags::none);
|
|
}
|
|
}
|
|
}
|
|
} catch (const std::exception& e) {
|
|
LOG_DEBUG("Worker recv loop terminated: %s", e.what());
|
|
}
|
|
m_bRecvWorking = false;
|
|
}
|