GrabBag/Module/ZeroMQClient/Src/CVrZeroMQClient.cpp

#include "CVrZeroMQClient.h"
#include "VrError.h"
#include "VrLog.h"
#include "IVrUtils.h"
#include "VrStringUtils.h"

// 低延迟配置常量
namespace {
    const int LOW_LATENCY_LINGER = 0;
    const int LOW_LATENCY_HWM = 1;
    const int LOW_LATENCY_BUFFER_SIZE = 4096;  // 减小缓冲区以减少延迟
    const int LOW_LATENCY_CONNECT_TIMEOUT = 500;  // 减少连接超时
    const int LOW_LATENCY_RECONNECT_INTERVAL = 50;  // 更快的重连
    const int LOW_LATENCY_RECV_TIMEOUT = 1000;  // 大幅减少接收超时
}

bool IVrZeroMQClient::CreateObject(IVrZeroMQClient** ppMQClient)
{
    *ppMQClient = new CVrZeroMQClient();
	return true;
}

CVrZeroMQClient::CVrZeroMQClient()
{
	m_pContext = new zmq::context_t;
	m_pSocket = new zmq::socket_t(*m_pContext, ZMQ_REQ);
	
    // 应用低延迟优化配置
    m_pSocket->set(zmq::sockopt::linger, LOW_LATENCY_LINGER);
    m_pSocket->set(zmq::sockopt::rcvhwm, LOW_LATENCY_HWM);
    m_pSocket->set(zmq::sockopt::sndhwm, LOW_LATENCY_HWM);
    // TCP_NODELAY在此ZeroMQ版本中不可用，跳过此选项
    m_pSocket->set(zmq::sockopt::rcvbuf, LOW_LATENCY_BUFFER_SIZE);
    m_pSocket->set(zmq::sockopt::sndbuf, LOW_LATENCY_BUFFER_SIZE);
    // 跨机器时连接可能超过 500ms，这里放宽到 3000ms
    m_pSocket->set(zmq::sockopt::connect_timeout, 3000);
    // ★ 接收超时 3 秒：防止服务端无响应时永久卡死 UI 线程
    m_pSocket->set(zmq::sockopt::rcvtimeo, 3000);
    m_pSocket->set(zmq::sockopt::reconnect_ivl, LOW_LATENCY_RECONNECT_INTERVAL);
	
	// 设置更短的重连间隔以快速恢复
    int reconnect_ivl_max = 200; // 最大重连间隔200ms
    m_pSocket->set(zmq::sockopt::reconnect_ivl_max, reconnect_ivl_max);

}

CVrZeroMQClient::~CVrZeroMQClient()
{
	try {
		if (m_pSocket) m_pSocket->close();
	}
	catch (...) {}

	delete m_pSocket;
	delete m_pContext;
}

int CVrZeroMQClient::Init(const char *sIP, int nPort)
{
    try{
        std::string transferAddr = "tcp://";
        transferAddr.append(sIP);
        transferAddr.append(":");
        transferAddr.append(CVrStringUtils::Int2Str(nPort));
        
        m_pSocket->connect(transferAddr);

    }catch(const std::exception& e){
        LOG_ERROR("Init zmq client error : %s \n", e.what());
        return ERR_CODE(NET_ERR_CONNECT);
    }

    return SUCCESS;
}

// 发送数据
int CVrZeroMQClient::SendAndWaitBack(const char* pdata, const int nLen, char** ppRecvData, size_t& nRecvLen)
{
	std::lock_guard<std::mutex> lck(m_mutexObj);
	zmq::message_t msg(pdata, nLen);

	// 立即发送
	LOG_DEBUG("[ZMQ-REQ] send: %.*s\n", nLen, pdata);
	auto send_result = m_pSocket->send(msg, zmq::send_flags::none);
	if (!send_result.has_value())
	{
		LOG_DEBUG("[ZMQ-REQ] send FAIL\n");
		LOG_ERROR("Failed to send message\n");
		return ERR_CODE(NET_ERR_SEND_DATA);
	}

	// 直接阻塞接收
	LOG_DEBUG("[ZMQ-REQ] waiting for recv...\n");
	zmq::recv_result_t ret = m_pSocket->recv(m_recvmsg, zmq::recv_flags::none);
	if (ret.has_value())
	{
		nRecvLen = m_recvmsg.size();
		*ppRecvData = (char *)m_recvmsg.data();
		LOG_DEBUG("[ZMQ-REQ] recv OK: %.*s\n", (int)nRecvLen, *ppRecvData);
		return SUCCESS;
	}

	LOG_DEBUG("[ZMQ-REQ] recv FAIL\n");
	LOG_ERROR("Failed to receive response\n");
	return ERR_CODE(NET_ERR_RECV_DATA);
}

int CVrZeroMQClient::UnInit()
{
    try {
        if (m_pSocket) m_pSocket->close();
    }
    catch (...) {}

    delete m_pSocket;
    delete m_pContext;

    m_pContext = nullptr;
    m_pSocket = nullptr;

    return SUCCESS;
}