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