#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());
		}
	}
}