DriverPlatform-V3.1/LogicDevice--mqtt/LogicDevice.cpp

// LogicDevice.cpp : 定义 DLL 应用程序的导出函数。
//

#include <cstring>
#include <iostream>
#include <chrono>
#include <codecvt>
#include "LogicDevice.h"
#include <stddef.h>  // 用于 offsetof
#include "PacketAnalizer.h"
#include "MessageInfo.h"
#include "common_api.h"
#include "LocalConfig.h"
#include "Base64.h"
#include "SystemLogger.hpp"
#include "AsyncMsgHandler.h"
#include "mqttclient.h"
#include "LinuxEvent.h"
#include "LogLocalHelper.h" 
#include "Log4CPP.h"

//Log4CPP::Logger* gLogger = nullptr;

void msgarrivd(void* client, message_data_t* message);

std::string CurrentDateTime();

//-------------Logic Device SysIF--------------------------
LogicDeviceSysIF::LogicDeviceSysIF(void)
{

}
LogicDeviceSysIF::~LogicDeviceSysIF(void)
{

}
//init 
void LogicDeviceSysIF::SetLogicDevice(LogicDevice *p)
{
	m_pLogicDev = p;
	//p->SetSysLogicDevice(this);
};

LogicDevice* LogicDeviceSysIF::GetLogicDevice()
{
	return m_pLogicDev;
}


//Command In and Out
//notify from lower layer
RET_STATUS HW_ACTION LogicDeviceSysIF::CmdFromLogicDev(ResDataObject PARAM_IN *pCmd)
{
	return RET_FAILED;
};

//notify to lower layer
RET_STATUS SYSTEM_CALL LogicDeviceSysIF::CmdToLogicDev(ResDataObject PARAM_IN *pCmd)
{
	if (m_pLogicDev)
	{
		return m_pLogicDev->CmdToLogicDev(pCmd);
	}

	return RET_NOSUPPORT;

};

const std::string SERVER_ADDRESS("127.0.0.1");
/*
string DEVICE_ID;  //AS CLIENT_ID
//const std::string CLIENT_ID("paho_cpp_async_subcribe");
//const std::string TOPIC("hello");
mqtt::async_client* m_pMqttClient = NULL;		//MQTT 对象

const int	QOS = 1;
const int	N_RETRY_ATTEMPTS = 5;*/

using namespace std;

LogicDevice* g_pDPCDeviceObject = NULL;
//string DEVICE_ID;

//-------------Data Logic Device--------------------------
LogicDevice::LogicDevice(void)
{
	m_EvtNotify = LinuxEvent::CreateEvent(LinuxEvent::AUTO_RESET, false);
	uuid_t uuid;
	char uuid_str[37];
	uuid_generate_random(uuid);
	uuid_unparse(uuid, uuid_str);
	m_pDevInstance = new char[40];
	strncpy(m_pDevInstance, uuid_str, 40);

	m_pResErrorList = new ResDataObject();
	sem_init(&m_SemphRequest, 0, 0);
	sem_init(&m_SemphPublish, 0, 0);

	m_pDrvDPC = NULL;

	g_pDPCDeviceObject = this;

	m_pMqttConntion = nullptr;

	//m_strServer = "tcp://localhost:1883";
	m_strServer = SERVER_ADDRESS;// "192.168.2.225";
	m_strServerPort = "1883";
	m_bMqttUseSSL = false;

	m_strMqttUser = "";
	m_strMqttPassword = "";
	m_strEBusRoot = "";
	m_strCCOSDevicePath = "";

	m_pParent = nullptr;

	m_topicFilter = nullptr;

	m_pPacketReceivedQue = new MsgQueue<ResDataObject>();
	m_pPacketSendingQue = new MsgQueue<ResDataObject>();
	int x = 0;
	for ( x = 0; x < sizeof(szPad); x++)
		szPad[x] = 'A' + x % 26;
	szPad[x-1] = 0;

	memset(szPad2, 0, sizeof(szPad2));
	m_dwLastPacket = GetTickCount();
	string strLogPath = GetProcessDirectory() + R"(/Conf/log_config.xml)";
	string LogHost = (string)getRootpath();
	std::string moduleName = "LogicDevice";
	bool ret = initLogModule(
		LogHost,       // 主机名（用于日志路径中的{host}占位符）
		moduleName,        // 唯一模块名
		strLogPath,  // 配置文件路径
		true           // 是否输出到控制台（可选）
	);
	if (!ret) {
		std::cerr << "Log init failed!" << std::endl;
	}
	LogicDevice_SetLocalModuleName(moduleName);
}

LogicDevice::~LogicDevice(void)
{
	delete []m_pDevInstance;
	delete m_pResErrorList;
	sem_destroy(&m_SemphRequest);
	sem_destroy(&m_SemphPublish);
	//m_EvtNotify = NULL;
	delete m_pPacketReceivedQue;
	delete m_pPacketSendingQue;
}

void LogicDevice::SetClientRootID(const char* pszEBusRoot, const char* pszCCOSRoot) {

	if (m_strClientID.length() > 0)
	{
		FINFO("Aready set RootID EBUS: [{$}] CCOS :  [{$}] result m_strClientID [{$}]", pszEBusRoot, pszCCOSRoot, m_strClientID);
		return;

	}
	FINFO("Set RootID EBUS: [{$}] CCOS :  [{$}]", pszEBusRoot, pszCCOSRoot);
	if (pszEBusRoot[0] == '/') {
		m_strEBusRoot = pszEBusRoot + 1;
	}
	else {
		m_strEBusRoot = pszEBusRoot;
	}

	char szKeys[256];
	strcpy(szKeys, pszEBusRoot);
	char* pt = szKeys;
	while (*pt != 0)
	{
		if (*pt == '/' || *pt == '{' || *pt == '}'|| *pt == '-')
			*pt = '_';
		pt++;
	}	
	m_strClientID = CCOS_CLIENT_ID_PREFIX;
	if (szKeys[0] == '_')
		m_strClientID += (szKeys + 1);
	else
		m_strClientID += szKeys;

	if (pszCCOSRoot != nullptr)
	{
		std::cout << "LogicDevice::SetClientRootID [ Set CCOS path is" << pszCCOSRoot << "]" << std::endl;
		FINFO("Set CCOS path: {$}", pszCCOSRoot);
		m_strCCOSDevicePath = pszCCOSRoot;
		int nPos = m_strCCOSDevicePath.find('/');
		if (nPos != string::npos)
		{
			//CCOS/
			nPos = m_strCCOSDevicePath.find('/', nPos + 1);
			//CCOS/DEVICE/
			if (nPos != string::npos)
			{
				m_strCCOSRoot = m_strCCOSDevicePath.substr(0, nPos);
				//CCOS/DEVICE/Generator
				nPos = m_strCCOSDevicePath.find('/', nPos + 1);
				if (nPos != string::npos)
				{
					m_strAbstractPath = m_strCCOSDevicePath.substr(0, nPos);
				}
			}
		}
		FINFO("Set CCOS path: CCOSROOT {$} AbstractPath {$} ", m_strCCOSRoot, m_strAbstractPath);
	}

	FINFO("Set MQTT ClientName {$} @CCOSRoot {$}", m_strClientID, m_strCCOSDevicePath);
	SetName(m_strClientID.c_str());
	OnSetClientID();
}


void LogicDevice::OnSetClientID() {


}

void LogicDevice::SubscribeSelf() {

	if (m_strDevicePath.length() > 0 && m_strDevicePath[0] != '/') {
		m_strDevicePath = "/" + m_strDevicePath;
	}

	FINFO("{$} try Subscribe {$} use conn {$} ", m_strClientID, m_strEBusRoot, (UINT64)m_pMqttConntion);
	std::cout << "----***** " << m_strClientID << "  [" << m_strEBusRoot << "] use conn " << (UINT64)m_pMqttConntion << endl;
	if (m_strEBusRoot.length() > 0) {
		SubscribeTopic(m_pMqttConntion, (m_strEBusRoot).c_str());
	}
	if (m_strCCOSDevicePath.length() > 0)
	{
		FINFO("CCOSDevice [{$}] ", m_strCCOSDevicePath + m_strDevicePath);
		SubscribeTopic(m_pMqttConntion, (m_strCCOSDevicePath + m_strDevicePath).c_str());

		//订阅
	}
	FINFO("AbstractPath [{$}] ", m_strAbstractPath);
	if (m_strAbstractPath.length() > 0)
	{
		SubscribeTopic(m_pMqttConntion, (m_strAbstractPath).c_str());

		if (m_strDevicePath.length() > 0)
			SubscribeTopic(m_pMqttConntion, (m_strAbstractPath + m_strDevicePath).c_str());
	}
	FINFO("CCOSRoot  [{$}] ", m_strCCOSRoot);
	if (m_strCCOSRoot.length() > 0)
	{
		SubscribeTopic(m_pMqttConntion, (m_strCCOSRoot).c_str());

		//订阅
	}

	// 等待初始订阅完成（EBusRoot、CCOSDevice、Abstract等）
	FINFO("{$} Waiting for initial subscriptions to complete...", m_strClientID);
	const DWORD initialSubTimeoutMs = 5000;
	if (!WaitForSubscriptionComplete(m_pMqttConntion, initialSubTimeoutMs)) {
		FERROR("{$} Some initial subscriptions did not complete within timeout", m_strClientID);
	}

	// SUBACK收到后，等待Broker完成路由表更新
	// 延时可通过环境变量MQTT_BROKER_ROUTING_DELAY_MS配置，默认200ms
	WaitForBrokerRouting(m_strClientID.c_str());

	// 订阅Actions（内部也会等待SUBACK + Broker延时）
	SubscribeActions();
}

void LogicDevice::SubScribeTopic(const char* pszTopic, bool bSubscribe)
{
	cout << "SubScribeTopic IN" << endl;
	if (bSubscribe)
		SubscribeTopic(m_pMqttConntion, pszTopic);
	else
		UnSubscribe(m_pMqttConntion, pszTopic);
}

void LogicDevice::NotifyDrvThread()
{
		m_EvtNotify->SetEvent();
}

std::shared_ptr<LinuxEvent> LogicDevice::GetEvtHandle()
{
	return m_EvtNotify;
}

//1. init part
//void LogicDevice::SetSysLogicDevice(LogicDeviceSysIF *pLogic)
//{
//	m_pSysLogic = pLogic;
//}

//void LogicDevice::SetLogHandle(Logger PARAM_IN *pLogger)
//{
//	m_pLogger = pLogger;
//}
//
//Logger *LogicDevice::GetLogHandle()
//{
//	return m_pLogger;
//}

void LogicDevice::SetDrvDPC(DriverDPC *pDPC)
{
	m_pDrvDPC = pDPC;
}

DriverDPC *LogicDevice::GetDrvDPC()
{
	return m_pDrvDPC;
}

RET_STATUS LogicDevice::AddEbusChildren(LogicDevice* pChild, const char* szEbusDevPath)
{
	if (szEbusDevPath == nullptr)
		return RET_FAILED;

	for (auto dev : m_subDevices) {
		if (dev->GetRootPath() == szEbusDevPath)
			return RET_SUCCEED;
	}
	m_subDevices.push_back(pChild);
	return RET_SUCCEED;
}

RET_STATUS LogicDevice::AddCcosChildren(LogicDevice* pChild, const char* szCcosDevPath)
{
	if (szCcosDevPath == nullptr)
		return RET_FAILED;

	for (auto dev : m_subCcosDevices) {
		if (dev->GetCcosRootPath() == szCcosDevPath)
			return RET_SUCCEED;
	}
	m_subCcosDevices.push_back(pChild);
	return RET_SUCCEED;
}

LogicDevice* LogicDevice::GetEbusChild(const char* szEbusDevPath)
{
	for (auto dev : m_subDevices) {
		if (dev->GetRootPath() == szEbusDevPath)
			return dev;
	}
	return nullptr;
}

LogicDevice* LogicDevice::GetCcosChild(const char* szCcosDevPath)
{
	for (auto dev : m_subCcosDevices) {
		if (dev->GetCcosRootPath() == szCcosDevPath)
			return dev;
	}
	return nullptr;
}

void SYSTEM_CALL LogicDevice::CompleteInit()
{
	FINFO("\n===============log begin : version:1.0.0.0 ===================\n");
	FINFO("{$}Connect MQTT Server  {$}:{$}", m_strServer, m_strServerPort, m_strClientID);

	if (m_strClientID.length() <= 0)
	{
		FWARN("No Client name ......" );
		std::cout << "No Client name ......." << endl;
		return;
	}
	std::cout << "CompleteInit->NewConnection ......." << endl;
	m_pMqttConntion = NewConnection(m_strServer.c_str(), m_strServerPort.c_str(), m_strMqttUser.c_str(), m_strMqttPassword.c_str(), m_strClientID.c_str(), 
		[this](ResDataObject* req, const  char* topic, void* conn) {

			//这里只处理当前层次设备的请求，下一层的直接靠URI来路由
			//首先根据topic判断是请求我的，还是我请求的，请求我的topic 是 以 ROOT开头的URI
			// 发送给我的，如果需要应答，则需要调用Request返回Resp，否则直接调用，不返回应答
			// 消息处理如果耗时较长，则需要开线程来处理
			if (strncmp(topic, m_strEBusRoot.c_str(), m_strEBusRoot.length()) == 0 ||
				strncmp(topic, m_strCCOSDevicePath.c_str(), m_strCCOSDevicePath.length()) == 0)
			{
				//主题以ebus或者ccos开头，给我发的消息，进行处理
				ProcessSubscribeRequest(req);
			}
			else
			{
				//我主动订阅的外部模块的消息
				ProcessSubscribeMsg(req);
			}

			//CmdToLogicDev(req);
		});

	if (m_pMqttConntion != nullptr)
	{
		// 使用异步启动避免阻塞：Sync=false表示不等待线程启动完成
		// 如果Sync=true，会等待300秒直到线程设置m_RunFlag，可能导致启动卡顿
		StartThread(false); //启动Request线程（异步）

		SubscribeSelf();

		//DWORD dwThreadID = 0;
		//CreateThread(0, 0, Thread_Publish_Thread, this, 0, &dwThreadID);
		//FINFO("[{$}] Publish Thread id [{$}]", m_strClientID, dwThreadID);
	}

}

RET_STATUS LogicDevice::ProcessSubscribeRequest(ResDataObject* req) {
	
	PACKET_TYPE type = PacketAnalizer::GetPacketType(req);
	PACKET_CMD cmd = PacketAnalizer::GetPacketCmd(req);

	switch (type) {
	case PACKET_TYPE_REQ:
		PacketAnalizer::GetPacketTransaction(req, m_strCurTransaction);
		ProcessRequest(req, cmd);//请求
		break;
	case PACKET_TYPE_RES:
		ProcessResponse(req, cmd);//应答
		break;
	case PACKET_TYPE_NOTIFY:
		ProcessNotify(req, cmd);
		//通知
		break;
	}

	return RET_FAILED;
}

RET_STATUS LogicDevice::ProcessSubscribeMsg(ResDataObject* pCmd)
{
	ProcessSubscribeRequest(pCmd);
	return RET_SUCCEED;
}

RET_STATUS LogicDevice::ProcessRequest(ResDataObject* pCmd, PACKET_CMD cmd)
{
	//Open命令
	if (cmd == PACKET_CMD_OPEN) {

		/* {
			"IDX": "40",
				"TYPE" : "0",
				"CMD" : "0",
				"HANDLE" :
				{
				"ROUTE": "1",
					"FLAGS" : "63",
					"LANG" : "en-US",
					"HANDLEID" : "0",
					"OWNERID" :
					{
						"EBUSID": "ImageSave",
						"MACHINEID" : "DESKTOP-FVD53H8",
						"PROCID" : "35408",
						"ADDR" : "2631366957696"

					},
					"DEVID":
					{
						"EBUSID": "ccosChannel",
							"MACHINEID" : "",
							"PROCID" : "0",
							"ADDR" : "0"
					}
				},
				"KEY": "\/ccosChannel"

					ResDataObject resRes, resResponse;
					ResDataObject resTopic;
					PacketAnalizer::GetContextTopic(pCmd, resTopic);
					if (cmd == PACKET_CMD_OPEN )
					{

						//std::cout << "publis " << (const char*)resTopic << "msg body" << resResponse.encode() << endl;

						PublishAction(&resResponse, (const char*)resTopic, m_pMqttConntion);
						return RET_SUCCEED;
					}
					if (cmd == PACKET_CMD_CLOSE)
					{
						//CLOSE 客户端主动断开
					}

		} */
		PacketArrived(pCmd);


		return RET_SUCCEED;
	}
	else if (cmd == PACKET_CMD_CLOSE)
	{
		ResDataObject resp;
		InnerOpenClose(*pCmd, resp, false);

	}
	else {

		PacketArrived(pCmd);

	}
	return RET_SUCCEED;

}

RET_STATUS LogicDevice::ProcessResponse(ResDataObject* pCmd, PACKET_CMD cmd)
{
	return RET_SUCCEED;

}

RET_STATUS LogicDevice::ProcessNotify(ResDataObject* pCmd, PACKET_CMD cmd)
{
	PacketArrived(pCmd);
	return RET_SUCCEED;
}

void LogicDevice::PacketArrived(ResDataObject* pRequest)
{
	//m_pPacketReceivedQue->Lock();
	FINFO("PacketArrived  msg [id: {$} ]: cmd {$} key: [{$}]", m_strClientID, (int)PacketAnalizer::GetPacketCmd(pRequest), PacketAnalizer::GetPacketKey(pRequest));
	//if (Thread_Lock(5000) != WAIT_OBJECT_0)
	//{
	//	FERROR("PacketArrived Lock Timeout for msg [id: {$} ]: cmd {$} key: ", m_strClientID, (int)PacketAnalizer::GetPacketCmd(pRequest), PacketAnalizer::GetPacketKey(pRequest));
	//	//GPRINTA_ERROR("OpenDev Lock Timeout for Dev[flag:%d]:%s", flags, pPath);
	//	return ;
	//}
	m_pPacketReceivedQue->InQueue(*pRequest);

	FINFO("PacketArrived  msg INTO QUEUE [id: {$} ]: cmd {$} key: [{$}]", m_strClientID, (int)PacketAnalizer::GetPacketCmd(pRequest), PacketAnalizer::GetPacketKey(pRequest));
	//SetEvent(m_EvtNotify);//notify to user
	long nLast = 0;
	int released = sem_post(&m_SemphRequest);  // 释放信号量，通知等待的线程

	if (released <= 0)
	{
		FERROR("PacketArrived: sem_post failed, error code: {$}, description: {$}",
			errno, strerror(errno));
	}
}


bool LogicDevice::OnStartThread()
{
	return true;
}
bool LogicDevice::OnEndThread()
{
	return true;
}

//单一发送线程，暂时未用
DWORD  LogicDevice::Thread_Publish_Thread(void* pPara)
{
	INT ret = 0;
	int waitevent = 0;
	LogicDevice* handle = (LogicDevice*)pPara;
	//prev work usleep(30000); 

	FINFO("Thread Start [{$}]", handle->m_strClientID);

	while (!handle->m_ExitFlag->Wait(1))
	{
		//do work


		ResDataObject resSend;
		DWORD wait = handle->m_pPacketSendingQue->WaitForInQue(100);
		if (wait != WAIT_OBJECT_0)
			continue;

		bool bHasPacket = handle->m_pPacketSendingQue->DeQueue(resSend);
		if (bHasPacket)
		{
			PACKET_CMD cmd = PacketAnalizer::GetPacketCmd(&resSend);
			PACKET_TYPE type = PacketAnalizer::GetPacketType(&resSend);
			FINFO(" [{$}] Publish key [{$}] type [{$}] cmd [{$}] ", handle->m_strClientID, PacketAnalizer::GetPacketKey(&resSend), (int)type, (int)cmd);
			if (PACKET_CMD_NONE == cmd)
			{
				FDEBUG(" [{$}] Publish what packet {$} ", handle->m_strClientID, resSend.encode());
			}
			if (type == PACKET_TYPE_NOTIFY)
			{
				CcosDevFileHandle* pHandle = new CcosDevFileHandle;
				PacketAnalizer::UpdateNotifyHandle(resSend, *pHandle);
				FINFO("Notify Transaction: {$}", handle->m_strCurTransaction);
				PacketAnalizer::UpdatePacketTransaction(resSend, handle->m_strCurTransaction);

				PacketAnalizer::UpdateDeviceNotifyResponse(resSend, getLocalMachineId(), getLocalEbusId(), (UINT64)pthread_self(), (UINT64)handle->m_pMqttConntion);
				FDEBUG("Notify: {$}", resSend.encode());
				//printf("-->  %s \n", pCmd->encode());
				PublishAction(&resSend, (handle->m_strEBusRoot + "/Notify").c_str(), handle->m_pMqttConntion);
				PublishAction(&resSend, (handle->m_strCCOSRoot + "/Notify/" + PacketAnalizer::GetPacketKey(&resSend)).c_str(), handle->m_pMqttConntion);

				delete pHandle;
			}
			else
			{
				ResDataObject resTopic;
				PacketAnalizer::GetPacketTopic(&resSend, resTopic);
				//PacketAnalizer::GetPacketTopicGetPacketTopic
				//PacketAnalizer::UpdatePacketTopic(&resResponse, resTopic);
				FINFO(" [{$}] Publish [{$}] type [{$}] cmd [{$}] to [{$}]", handle->m_strClientID, PacketAnalizer::GetPacketKey(&resSend), (int)type, (int)cmd, (const char*)resTopic);

				PublishAction(&resSend, (const char*)resTopic, handle->m_pMqttConntion);
			}

		}


	}


	return 0;

}

RET_STATUS  LogicDevice::DevOpen(const char* pszDevUri, const char* pszGroup, ResDataObject& resRespons)
{
	cout << "LogicDevice::DevOpen - Entering. pszDevUri: " << (pszDevUri ? pszDevUri : "nullptr")
		<< ", pszGroup: " << (pszGroup ? pszGroup : "nullptr") << endl;

	ResDataObject req, reqParam;
	reqParam = pszGroup;
	cout << "LogicDevice::DevOpen - Creating OPEN request. Command: PACKET_CMD_OPEN" << endl;
	PacketAnalizer::MakeRequest(req, pszDevUri, PACKET_CMD_OPEN, &reqParam);

	RET_STATUS ret = InnerOpenClose(req, resRespons, true);
	cout << "LogicDevice::DevOpen - InnerOpenClose returned: " << ret << endl;
	return ret;
}


RET_STATUS  LogicDevice::DevClose(const char* pszDevUri, const char* pszGroup, ResDataObject& resRespons)
{
	cout << "LogicDevice::DevClose - Entering. pszDevUri: " << (pszDevUri ? pszDevUri : "nullptr")
		<< ", pszGroup: " << (pszGroup ? pszGroup : "nullptr") << endl;

	ResDataObject req, reqParam;
	reqParam = pszGroup;
	cout << "LogicDevice::DevClose - Creating CLOSE request. Command: PACKET_CMD_CLOSE" << endl;
	PacketAnalizer::MakeRequest(req, pszDevUri, PACKET_CMD_CLOSE, &reqParam);

	RET_STATUS ret = InnerOpenClose(req, resRespons, false);
	cout << "LogicDevice::DevClose - InnerOpenClose returned: " << ret << endl;
	return ret;
}

/// <summary>
/// Get device property
/// </summary>
RET_STATUS  LogicDevice::DevGet(const char* pszDevUri, const char* pszProperty, ResDataObject& resRespons)
{
	cout << "LogicDevice::DevGet - Entering. pszDevUri: " << (pszDevUri ? pszDevUri : "nullptr")
		<< ", pszProperty: " << (pszProperty ? pszProperty : "nullptr") << endl;

	ResDataObject req, reqParam;
	cout << "LogicDevice::DevGet - Creating GET request. Command: PACKET_CMD_GET, Property: " << (pszProperty ? pszProperty : "nullptr") << endl;
	PacketAnalizer::MakeRequest(req, pszProperty, PACKET_CMD_GET, &reqParam);

	RET_STATUS ret = Request(&req, &resRespons);
	cout << "LogicDevice::DevGet - Request returned: " << ret << endl;
	return ret;
}

/// <summary>
/// Set device property
/// </summary>
RET_STATUS  LogicDevice::DevSet(const char* pszDevUri, const char* pszProperty, const char* pszValueSet, ResDataObject& resRespons)
{
	cout << "LogicDevice::DevSet - Entering. pszDevUri: " << (pszDevUri ? pszDevUri : "nullptr")
		<< ", pszProperty: " << (pszProperty ? pszProperty : "nullptr")
		<< ", pszValueSet: " << (pszValueSet ? pszValueSet : "nullptr") << endl;

	ResDataObject req, reqParam;
	if (pszValueSet != nullptr && pszValueSet[0] != 0) {
		if (pszValueSet[0] == '{') {
			cout << "LogicDevice::DevSet - Decoding JSON value for property" << endl;
			reqParam.decode(pszValueSet);
		}
		else {
			cout << "LogicDevice::DevSet - Using raw value for property" << endl;
			reqParam = pszValueSet;
		}
	}
	else {
		cout << "LogicDevice::DevSet - No value provided for property" << endl;
	}

	cout << "LogicDevice::DevSet - Creating SET request. Command: PACKET_CMD_SET, Property: " << (pszProperty ? pszProperty : "nullptr") << endl;
	PacketAnalizer::MakeRequest(req, pszProperty, PACKET_CMD_SET, &reqParam);

	RET_STATUS ret = Request(&req, &resRespons);
	cout << "LogicDevice::DevSet - Request returned: " << ret << endl;
	return ret;
}

/// <summary>
/// Update device property
/// </summary>
RET_STATUS  LogicDevice::DevUpdate(const char* pszDevUri, const char* pszProperty, const char* pszValueUpdate, ResDataObject& resRespons)
{
	cout << "LogicDevice::DevUpdate - Entering. pszDevUri: " << (pszDevUri ? pszDevUri : "nullptr")
		<< ", pszProperty: " << (pszProperty ? pszProperty : "nullptr")
		<< ", pszValueUpdate: " << (pszValueUpdate ? pszValueUpdate : "nullptr") << endl;

	ResDataObject req, reqParam;
	if (pszValueUpdate != nullptr && pszValueUpdate[0] != 0) {
		if (pszValueUpdate[0] == '{') {
			cout << "LogicDevice::DevUpdate - Decoding JSON value for update" << endl;
			reqParam.decode(pszValueUpdate);
		}
		else {
			cout << "LogicDevice::DevUpdate - Using raw value for update" << endl;
			reqParam = pszValueUpdate;
		}
	}
	else {
		cout << "LogicDevice::DevUpdate - No update value provided" << endl;
	}

	cout << "LogicDevice::DevUpdate - Creating UPDATE request. Command: PACKET_CMD_UPDATE, Property: " << (pszProperty ? pszProperty : "nullptr") << endl;
	PacketAnalizer::MakeRequest(req, pszProperty, PACKET_CMD_UPDATE, &reqParam);

	RET_STATUS ret = Request(&req, &resRespons);
	cout << "LogicDevice::DevUpdate - Request returned: " << ret << endl;
	return ret;
}

/// <summary>
/// Add device property
/// </summary>
RET_STATUS  LogicDevice::DevAdd(const char* pszDevUri, const char* pszProperty, const char* pszValueAdd, ResDataObject& resRespons)
{
	cout << "LogicDevice::DevAdd - Entering. pszDevUri: " << (pszDevUri ? pszDevUri : "nullptr")
		<< ", pszProperty: " << (pszProperty ? pszProperty : "nullptr")
		<< ", pszValueAdd: " << (pszValueAdd ? pszValueAdd : "nullptr") << endl;

	ResDataObject req, reqParam;
	if (pszValueAdd != nullptr && pszValueAdd[0] != 0) {
		if (pszValueAdd[0] == '{') {
			cout << "LogicDevice::DevAdd - Decoding JSON value for addition" << endl;
			reqParam.decode(pszValueAdd);
		}
		else {
			cout << "LogicDevice::DevAdd - Using raw value for addition" << endl;
			reqParam = pszValueAdd;
		}
	}
	else {
		cout << "LogicDevice::DevAdd - No value provided for addition" << endl;
	}

	cout << "LogicDevice::DevAdd - Creating ADD request. Command: PACKET_CMD_ADD, Property: " << (pszProperty ? pszProperty : "nullptr") << endl;
	PacketAnalizer::MakeRequest(req, pszProperty, PACKET_CMD_ADD, &reqParam);

	RET_STATUS ret = Request(&req, &resRespons);
	cout << "LogicDevice::DevAdd - Request returned: " << ret << endl;
	return ret;
}

/// <summary>
/// Delete device property
/// </summary>
RET_STATUS  LogicDevice::DevDel(const char* pszDevUri, const char* pszProperty, const char* pszValueDel, ResDataObject& resRespons)
{
	cout << "LogicDevice::DevDel - Entering. pszDevUri: " << (pszDevUri ? pszDevUri : "nullptr")
		<< ", pszProperty: " << (pszProperty ? pszProperty : "nullptr")
		<< ", pszValueDel: " << (pszValueDel ? pszValueDel : "nullptr") << endl;

	ResDataObject req, reqParam;
	if (pszValueDel != nullptr && pszValueDel[0] != 0) {
		if (pszValueDel[0] == '{') {
			cout << "LogicDevice::DevDel - Decoding JSON value for deletion" << endl;
			reqParam.decode(pszValueDel);
		}
		else {
			cout << "LogicDevice::DevDel - Using raw value for deletion" << endl;
			reqParam = pszValueDel;
		}
	}
	else {
		cout << "LogicDevice::DevDel - No value provided for deletion" << endl;
	}

	cout << "LogicDevice::DevDel - Creating DEL request. Command: PACKET_CMD_DEL, Property: " << (pszProperty ? pszProperty : "nullptr") << endl;
	PacketAnalizer::MakeRequest(req, pszProperty, PACKET_CMD_DEL, &reqParam);

	RET_STATUS ret = Request(&req, &resRespons);
	cout << "LogicDevice::DevDel - Request returned: " << ret << endl;
	return ret;
}

/// <summary>
/// Execute device action
/// </summary>
RET_STATUS  LogicDevice::DevAction(const char* pszDevUri, const char* pszActionName, const char* pszParams, ResDataObject& resRespons)
{
	cout << "LogicDevice::DevAction - Entering. pszDevUri: " << (pszDevUri ? pszDevUri : "nullptr")
		<< ", pszActionName: " << (pszActionName ? pszActionName : "nullptr")
		<< ", pszParams: " << (pszParams ? pszParams : "nullptr") << endl;

	ResDataObject req, reqParam;
	if (pszParams != nullptr && pszParams[0] != 0) {
		if (pszParams[0] == '{') {
			cout << "LogicDevice::DevAction - Decoding JSON parameters for action" << endl;
			reqParam.decode(pszParams);
		}
		else {
			cout << "LogicDevice::DevAction - Using raw parameters for action" << endl;
			reqParam = pszParams;
		}
	}
	else {
		cout << "LogicDevice::DevAction - No parameters provided for action" << endl;
	}

	cout << "LogicDevice::DevAction - Creating EXE request. Command: PACKET_CMD_EXE, Action: " << (pszActionName ? pszActionName : "nullptr") << endl;
	PacketAnalizer::MakeRequest(req, pszActionName, PACKET_CMD_EXE, &reqParam);

	ResDataObject resResult;
	RET_STATUS ret = Request(&req, &resResult);
	cout << "LogicDevice::DevAction - Request returned: " << ret << endl;

	PacketAnalizer::GetPacketContext(&resResult, resRespons);
	cout << "LogicDevice::DevAction - Extracted packet context to response" << endl;

	return ret;
}

/// <summary>
/// Send message to device
/// </summary>
RET_STATUS  LogicDevice::DevMessage(const char* pszDevUri, const char* pszTopic, const char* pszMessageValue, ResDataObject& resRespons)
{
	cout << "LogicDevice::DevMessage - Entering. pszDevUri: " << (pszDevUri ? pszDevUri : "nullptr")
		<< ", pszTopic: " << (pszTopic ? pszTopic : "nullptr")
		<< ", pszMessageValue: " << (pszMessageValue ? pszMessageValue : "nullptr") << endl;

	ResDataObject req, reqParam;
	if (pszMessageValue != nullptr && pszMessageValue[0] != 0) {
		if (pszMessageValue[0] == '{') {
			cout << "LogicDevice::DevMessage - Decoding JSON message value" << endl;
			reqParam.decode(pszMessageValue);
		}
		else {
			cout << "LogicDevice::DevMessage - Using raw message value" << endl;
			reqParam = pszMessageValue;
		}
	}
	else {
		cout << "LogicDevice::DevMessage - No message value provided" << endl;
	}

	cout << "LogicDevice::DevMessage - Creating MSG request. Command: PACKET_CMD_MSG, Topic: " << (pszTopic ? pszTopic : "nullptr") << endl;
	PacketAnalizer::MakeRequest(req, pszTopic, PACKET_CMD_MSG, &reqParam);

	RET_STATUS ret = Request(&req, &resRespons);
	cout << "LogicDevice::DevMessage - Request returned: " << ret << endl;
	return ret;
}
RET_STATUS LogicDevice::InnerOpenClose(ResDataObject& req, ResDataObject& resp, bool openOrClose)
{
	if (openOrClose)
	{
		ResDataObject resContext;
		GetDeviceResource(&resp);
		LogicDevice::GetDeviceResource(&resp);
		PacketAnalizer::GetPacketTransaction(&req, m_strCurTransaction);

		if (PacketAnalizer::GetPacketContext(&req, resContext))
		{
			//如果有上线参数
			if (resContext.GetFirstOf("Online") >= 0)
			{
				FINFO("Got Online Request {$}", resContext.encode());
				int idx = resContext.GetFirstOf("Online");
				//如果有上线参数
				if (idx >= 0)
				{
					do
					{
						string wsName = resContext[idx].encode();
						FINFO("SubscribeGroupActions [{$}] ", wsName);
						SubscribeGroupActions(wsName, true);
						idx = resContext.GetNextOf("Online", idx);
					} while (idx >= 0);
				}
				resp.update("Online", m_rsOnlineGroup);
			}
		}
	}
	else
	{
		ResDataObject context;
		if (PacketAnalizer::GetPacketContext(&req, context))
		{
			int idx = context.GetFirstOf("Offline");
			//如果有上线参数
			if (idx >= 0)
			{
				do
				{
					string wsName = context[idx].encode();
					SubscribeGroupActions(wsName, false);
					idx = context.GetNextOf("Offline", idx);
				} while (idx >= 0);
			}
			resp.update("Online", m_rsOnlineGroup);
		}
	}
	
	return RET_SUCCEED;
}

bool LogicDevice::Exec(void) 
{
	struct timespec ts;
	clock_gettime(CLOCK_REALTIME, &ts);
	ts.tv_sec += 3;

	// 等待退出信号（3秒超时）
	DWORD dwRet = 0;
	dwRet = m_ExitFlag->Wait(3);
	if (dwRet == WAIT_OBJECT_0)
	{
		return false;
	}

	// 等待请求信号（3秒超时）
	int ret = sem_timedwait(&m_SemphRequest, &ts);
	if (ret == 0) {
		FDEBUG("[{$}] Got Packet Event ", m_strClientID);
		ResDataObject req;
		bool got = true;
		do {
			got = m_pPacketReceivedQue->DeQueue(req);
			if (got) {
				m_dwLastPacket = GetTickCount();
				PACKET_CMD cmd = PacketAnalizer::GetPacketCmd(&req);
				std::string keystr = PacketAnalizer::GetPacketKey(&req);

				FINFO(" {$}  Got Request key {$} transaction {$}",
					m_strClientID, keystr, m_strCurTransaction);

				if (cmd == PACKET_CMD_OPEN) {
					ResDataObject resRes, resResponse;
					InnerOpenClose(req, resRes, true);

					PacketAnalizer::MakeOpenResponse(req, resResponse, resRes);
					resResponse.update("Online", m_rsOnlineGroup);
					PacketAnalizer::UpdatePacketTransaction(resResponse, m_strCurTransaction);

					ResDataObject resTopic;
					PacketAnalizer::GetContextTopic(&req, resTopic);
					PacketAnalizer::UpdatePacketTopic(&resResponse, resTopic, m_strClientID.c_str());

					if (keystr.substr(0, 4) == "CCOS") {
						PacketAnalizer::UpdatePacketKey(&resResponse, m_strCCOSDevicePath.c_str());
					}
					else {
						PacketAnalizer::UpdateDeviceNotifyResponse(resResponse,
							getLocalMachineId(),
							getLocalEbusId(),
							(uint64_t)getpid(),
							(uint64_t)m_pMqttConntion);
					}

					PublishAction(&resResponse, (const char*)resTopic, m_pMqttConntion);

					// 发送连接状态通知
					ResDataObject NotifyData;
					PacketAnalizer::MakeNotify(NotifyData, PACKET_CMD_UPDATE,
						"ConnectionStatus", "1");
					PacketAnalizer::UpdatePacketTransaction(NotifyData, m_strCurTransaction);
					CmdFromLogicDev(&NotifyData);
				}
				else {
					PacketAnalizer::GetPacketTransaction(&req, m_strCurTransaction);
					ResDataObject resPacket;
					RET_STATUS retStatus = RET_FAILED;

					if (cmd == PACKET_CMD_EXE && keystr == "UpdateDeviceResource") {
						ResDataObject devRes;
						if ((retStatus = GetDeviceResource(&devRes)) == RET_SUCCEED) {
							LogicDevice::GetDeviceResource(&devRes);
							PacketAnalizer::UpdatePacketContext(resPacket, devRes);
						}
					}
					else {
						retStatus = Request(&req, &resPacket);
					}

					PacketAnalizer::MakeRetCode(retStatus, &resPacket);
					PacketAnalizer::CloneTransaction(&req, &resPacket);

					ResDataObject resTopic;
					PacketAnalizer::GetContextTopic(&req, resTopic);
					PacketAnalizer::UpdatePacketTopic(&resPacket, resTopic, m_strClientID.c_str());
					PublishAction(&resPacket, (const char*)resTopic, m_pMqttConntion);
				}
			}
		} while (got);

		// 检查心跳（10分钟无数据发送心跳）
		if (GetTickCount() - m_dwLastPacket > 600000) {
			m_dwLastPacket = GetTickCount();
			ResDataObject heartBeat;
			PacketAnalizer::MakeNotify(heartBeat, PACKET_CMD_ONLINE,
				"HeartBeat", m_strClientID.c_str());
			PublishAction(&heartBeat, "CCOS/DEVICE/HeartBeat", m_pMqttConntion);
		}
		return true;
	}

	// 检查心跳（即使没有请求）
	if (GetTickCount() - m_dwLastPacket > 600000) {
		m_dwLastPacket = GetTickCount();
		ResDataObject heartBeat;
		PacketAnalizer::MakeNotify(heartBeat, PACKET_CMD_ONLINE,
			"HeartBeat", m_strClientID.c_str());
		PublishAction(&heartBeat, "CCOS/DEVICE/HeartBeat", m_pMqttConntion);
	}

	return true;
}



void SYSTEM_CALL LogicDevice::CompleteUnInit()
{
	StopThread();
}


int LogicDevice::GetDevice_Thread_Priority()
{
	return THREAD_PRIORITY_NONE;
}
RET_STATUS LogicDevice::GetDeviceResource(ResDataObject PARAM_OUT *pDeviceResource)
{
	//pDeviceResource->update("ClientType", DPC_UnitClient);
	GUID guid;
	string name;
	GetDeviceType(guid);
	guid_2_string(guid, name);
	pDeviceResource->update("DeviceType", name.c_str());

	//Get Unit Type (Unit GUID)
	if (pDeviceResource->GetFirstOf("LogicDevInstance")<0)
	{
		pDeviceResource->add("LogicDevInstance", m_pDevInstance);
	}
	////
	size_t idx = (*pDeviceResource)["Attribute"].size();
	if (idx > 0)
	{
		int erroridx = (*pDeviceResource)["Attribute"].GetFirstOf("ErrorList");
		if (erroridx < 0)
		{
			(*pDeviceResource)["Attribute"].add("ErrorList", *m_pResErrorList);
		}
		else
		{
			(*pDeviceResource)["Attribute"]["ErrorList"] = *m_pResErrorList;
		}
	}
	else
	{
		ResDataObject Attribute;
		Attribute.add("ErrorList", *m_pResErrorList);
		pDeviceResource->add("Attribute", Attribute);
	}
	// (*pDeviceResource)["Action"].size();
	if (pDeviceResource->GetFirstOf("Action") < 0)
	{
		pDeviceResource->add("Action", m_Actions);
	}
	return RET_SUCCEED;
}
//notify from lower layer
RET_STATUS LogicDevice::CmdFromLogicDev(ResDataObject *pCmd)
{
	PACKET_CMD cmd = PacketAnalizer::GetPacketCmd(pCmd);
	PACKET_TYPE type = PacketAnalizer::GetPacketType(pCmd);
	if (type == PACKET_TYPE_NOTIFY)
	{
		CcosDevFileHandle* pHandle = new CcosDevFileHandle;
		PacketAnalizer::UpdateNotifyHandle(*pCmd, *pHandle);
		FDEBUG("Notify Transaction: {$}", m_strCurTransaction);
		PacketAnalizer::UpdatePacketTransaction(*pCmd, m_strCurTransaction);
		;
		//if (m_strEBusRoot.length() <= 0)
		//{
		//	//FWARN("EBusRoot is null");
		//}
		PacketAnalizer::UpdateDeviceNotifyResponse(
			*pCmd,
			getLocalMachineId(),
			getLocalEbusId(),
			static_cast<uint64_t>(getpid()),  // Linux 上获取进程 ID
			reinterpret_cast<uint64_t>(m_pMqttConntion)  // 假设 m_pMqttConntion 在 Linux 上是指针类型
		);

		FDEBUG("[{$}] Notify: {$}", m_strClientID, pCmd->encode());
		//printf("-->  %s \n", pCmd->encode());
		PacketAnalizer::UpdatePacketTopic(pCmd, (m_strEBusRoot + "/Notify").c_str(), m_strClientID.c_str());
		PublishAction(pCmd, (m_strEBusRoot + "/Notify").c_str(), m_pMqttConntion);

		string strNotifyPath = "/Notify/" + PacketAnalizer::GetPacketKey(pCmd);
		PacketAnalizer::UpdatePacketTopic(pCmd, (m_strCCOSRoot + strNotifyPath).c_str(), m_strClientID.c_str());
		PublishAction(pCmd, (m_strCCOSDevicePath + strNotifyPath).c_str(), m_pMqttConntion);
		if (m_strAbstractPath.length() > 0)
		{
			PublishAction(pCmd, (m_strAbstractPath + strNotifyPath).c_str(), m_pMqttConntion);
			string realPath,devPath;
			devPath = m_strAbstractPath.substr(((string)"CCOS/DEVICE").length());
			for (int x = 0; x < m_rsOnlineGroup.size(); x++)
			{
				if ((int)m_rsOnlineGroup[x] == 1)
				{
					realPath = "CCOS/"  +(string)m_rsOnlineGroup.GetKey(x) + devPath + strNotifyPath;
					PublishAction(pCmd, realPath.c_str(), m_pMqttConntion);
				}
			}
		}		//m_pPacketSendingQue->InQueue(*pCmd);

		delete pHandle;
	}
	return RET_SUCCEED;
	/*
	if (pCmd && m_pSysLogic)
	{
		return m_pSysLogic->CmdFromLogicDev(pCmd);
	}
	//put log here

	return RET_FAILED;*/

}

std::wstring mb2wc_a(const char* mbstr) {
	if (!mbstr) return L"";

	try {
		// 使用C++11的codecvt进行UTF-8到wstring的转换
		std::wstring_convert<std::codecvt_utf8<wchar_t>> converter;
		return converter.from_bytes(mbstr);
	}
	catch (...) {
		// 如果转换失败，返回空宽字符串
		return L"";
	}
}

RET_STATUS HW_ACTION LogicDevice::AddErrorMessageUnicode(const char* DevInstance, const char* Code, int &Level, const wchar_t* ResInfo, const wchar_t* Description, int nMessageType)
{
	string ResBase64, DesBase64;

	wstring wResUTF = ResInfo;
	wstring wDesUTF = Description;

	CBase64::Encode((const unsigned char *)wResUTF.c_str(), (unsigned long)wResUTF.size() * sizeof(wchar_t), ResBase64);

	CBase64::Encode((const unsigned char *)wDesUTF.c_str(), (unsigned long)wDesUTF.size() * sizeof(wchar_t), DesBase64);


	return AddErrorMessage(DevInstance, Code, Level, ResBase64.c_str(), DesBase64.c_str(), nMessageType);
}


RET_STATUS HW_ACTION LogicDevice::AddErrorMessage(const char* DevInstance, const char* Code, int &Level, const char* ResInfo, const char* Description, int nMessageType, const char* pAppId)
{
	string ResBase64,DesBase64;
	cout << "ResInfo:" << ResInfo << endl;
	wstring wResUTF = mb2wc_a(ResInfo);
	wstring wDesUTF = mb2wc_a(Description);

	const unsigned char* resData = reinterpret_cast<const unsigned char*>(wResUTF.c_str());
	unsigned long resDataLen = wResUTF.size() * sizeof(wchar_t);

	const unsigned char* desData = reinterpret_cast<const unsigned char*>(wDesUTF.c_str());
	unsigned long desDataLen = wDesUTF.size() * sizeof(wchar_t);

	// 进行Base64编码
	CBase64::Encode(resData, resDataLen, ResBase64);
	CBase64::Encode(desData, desDataLen, DesBase64);

	cout << "ResBase64:" << ResBase64 << endl;
	return AddErrorMessageBase(DevInstance, Code, Level, ResInfo, Description, nMessageType, pAppId);
}


RET_STATUS LogicDevice::AddErrorMessageBase(const char* DevInstance, const char* Code, int &Level, const char* ResInfo, const char* Description, int nMessageType, const char* pAppId)
{
	//int ret = 1;
	if (Code == 0 || (string)ResInfo == "")
	{
		FERROR("Code or ResInfo is empty");
		return RET_FAILED;
	}
	MessageInfo info;
	info.CodeID = Code;
	info.Type = nMessageType;
	info.Level = Level;
	info.Resouceinfo = ResInfo;
	info.Description = Description;

	string strDevInstanceCode = DevInstance;
	strDevInstanceCode += Code;

	for (size_t i = 0; i < m_pResErrorList->size(); i++)
	{
		string strInstancekey = m_pResErrorList->GetKey(i);
		if (strInstancekey == strDevInstanceCode)
		{
			//for (size_t j = 0; j < (*m_pResErrorList)[DevInstance].size(); j++)
			//{
			//	string strCodekey = (*m_pResErrorList)[DevInstance].GetKey(j);
			//	string strtype = (*m_pResErrorList)[DevInstance][strCodekey.c_str()]["Type"];
			//	if (strCodekey == (string)Code && atoi(strtype.c_str()) == nMessageType)
			//	{
					//ret = 0;
					FWARN("Same Code:%s with MessageType:%d already Exist", Code,nMessageType);
					return RET_SUCCEED;
			//	}
			//}
			//ret = 2;
			//break;
		}
	}

	//if (ret==1)
	{
		
		ResDataObject NotifyData, ResNotify, ErrorInfo/*, tempInfo*/;
		//info.GetResDataObject(tempInfo);
		//ErrorInfo.update(Code, tempInfo);
		info.GetResDataObject(ErrorInfo);

		struct timeval tv;
		struct tm* tm_info;
		char TimeTag[30];

		// 获取当前时间
		gettimeofday(&tv, NULL);

		// 转换为本地时间
		tm_info = localtime(&tv.tv_sec);

		// 格式化时间为字符串
		snprintf(TimeTag, sizeof(TimeTag), "%04d-%02d-%02d %02d:%02d:%02d.%03ld",
			tm_info->tm_year + 1900, tm_info->tm_mon + 1, tm_info->tm_mday,
			tm_info->tm_hour, tm_info->tm_min, tm_info->tm_sec, tv.tv_usec / 1000);

		ErrorInfo.add("CreationTime", TimeTag);
		ErrorInfo.add("AppId", pAppId);
		ErrorInfo.add("InstanceId", DevInstance);
		
		if (nMessageType == ERRORTYPE)//只有错误会增加到错误列表中，警告通知上层即可
		{
			m_pResErrorList->update(strDevInstanceCode.c_str(), ErrorInfo);
		}
		ResNotify.update(strDevInstanceCode.c_str(), ErrorInfo);


		PacketAnalizer::MakeNotify(NotifyData, PACKET_CMD_ADD, "ErrorList", ResNotify);

		FWARN( "preposterror ErrorType:{$} {$}", nMessageType,NotifyData.encode());

		CmdFromLogicDev(&NotifyData);
		//PublishAction(&NotifyData, (m_strEBusRoot + "/Notify").c_str(), m_pMqttConntion);
	}
	//else if (ret == 2)
	//{
	//	ResDataObject NotifyData, ResNotify, ErrorInfo, tempInfo;
	//	info.GetResDataObject(tempInfo);
	//	ErrorInfo.update(Code, tempInfo);
	//	if (nMessageType == ERRORTYPE)//只有错误会增加到错误列表中，警告通知上层即可
	//	{
	//		(*m_pResErrorList)[DevInstance].update(Code, tempInfo);
	//	}

	//	ResNotify.update(DevInstance, ErrorInfo);
	//	PacketAnalizer::MakeNotify(NotifyData, PACKET_CMD_ADD, "ErrorList", ResNotify);

	//	RES_FDEBUG(NotifyData, "preposterror RET:%d,ErrorType:%u", ret, nMessageType);

	//	CmdFromLogicDev(&NotifyData);
	//}
	//put log here
	return RET_SUCCEED;

}
RET_STATUS LogicDevice::AddErrorMessage(const char* Code, int &Level, const char* ResInfo, int nMessageType, const char* pAppId)
{
	AddErrorMessage(m_pDevInstance, Code, Level, ResInfo, "",nMessageType, pAppId);
	//put log here
	return RET_FAILED;

}
RET_STATUS LogicDevice::DelErrorMessage(const char* DevInstance, const char* Code, int &Level, const char* ResInfo, const char* Description, int nMessageType)
{
	int index = -1;
	bool m_bClearAll = false;

	//trim empty code string
	string CodeStr = Code;
	if (CodeStr.size() == 0 || CodeStr == "0" || CodeStr == "")
	{
		CodeStr = "";
		Code = CodeStr.c_str();
		m_bClearAll = true;
	}

	

	//if ((string)Code == "0" || (string)Code == "")
	//{
	//	m_bClearAll = true;
	//}

	string strDevInstanceCode = DevInstance;
	strDevInstanceCode += Code;

	if (m_bClearAll)
	{
		m_pResErrorList->clear();
	}
	else
	{
		MessageInfo info;
		info.CodeID = Code;
		info.Type = nMessageType;
		info.Level = Level;
		info.Resouceinfo = ResInfo;
		info.Description = Description;
		
		for (size_t i = 0; i < m_pResErrorList->size(); i++)
		{
			string strInstancekey = m_pResErrorList->GetKey(i);
			if (strInstancekey == strDevInstanceCode)
			{
				//for (size_t j = 0; j < (*m_pResErrorList)[DevInstance].size(); j++)
				//{
				//	string strCodekey = (*m_pResErrorList)[DevInstance].GetKey(j);
				//	string strtype = (*m_pResErrorList)[DevInstance][strCodekey.c_str()]["Type"];
				//	if (strCodekey == (string)Code && atoi(strtype.c_str()) == nMessageType)
				//	{
				//		index = (INT)j;
				//		break;
				//	}
				//}
				index = (int)i;
				break;
			}
		}

	}
	MessageInfo info;
	info.CodeID = Code;
	info.Type = nMessageType;
	info.Level = Level;
	info.Resouceinfo = ResInfo;
	info.Description = Description;
	ResDataObject NotifyData, ResNotify, ErrorInfo/*, tempInfo*/;
	//info.GetResDataObject(tempInfo);
	//ErrorInfo.add(Code, tempInfo);
	info.GetResDataObject(ErrorInfo);
	ErrorInfo.add("InstanceId", DevInstance);


	bool result = false;
	if (index>=0)
	{
		//result = (*m_pResErrorList)[DevInstance].eraseOneOf(Code, index);
		result = (*m_pResErrorList).eraseAllOf(strDevInstanceCode.c_str());
		
	}
	if (index >= 0 || m_bClearAll || nMessageType == WARNTYPE)
	{
		ResNotify.add(strDevInstanceCode.c_str(), ErrorInfo);
		PacketAnalizer::MakeNotify(NotifyData, PACKET_CMD_DEL, "ErrorList", ResNotify);
		FWARN( "pre Del error ErrorCode:{$} {$}", Code ,NotifyData.encode());
		CmdFromLogicDev(&NotifyData);
		//PublishAction(&NotifyData, (m_strEBusRoot + "/Notify").c_str(), m_pMqttConntion);
	}
	return RET_SUCCEED;

}
RET_STATUS LogicDevice::DelErrorMessage(const char* Code, int &Level, const char* ResInfo, int nMessageType)
{
	DelErrorMessage(m_pDevInstance, Code, Level, ResInfo, "", nMessageType);
	//put log here
	return RET_FAILED;

}

RET_STATUS LogicDevice::EvtProcedure()
{
	return RET_FAILED;

}

bool LogicDevice::CheckFeatureLicense(const char *pszFeatureId)
{
	FERROR("NOT FINISHED YET");

	return false;
}


RET_STATUS LogicDevice::IoSystemLog(int Level, const char* pCode, const char* pContext, size_t ContextSize, const char* pAppId)
{
	std::string strResult = "";

	//组Context包
	//if (m_pLogger)
	{
		wstring wContect = mb2wc_a(pContext);
		CBase64::Encode((const unsigned char*)wContect.c_str(), (unsigned long)wContect.size() * sizeof(wchar_t), strResult);
	}

	//Thread_Lock();
	//if (NULL != fmt)
	//{
	//	va_list marker = NULL;
	//	va_start(marker, fmt);
	//	size_t nLength = _vscprintf(fmt, marker) + 1;
	//	std::vector<char> vBuffer(nLength, '\0');
	//	int nWritten = vsnprintf_s(&vBuffer[0], vBuffer.size(), nLength, fmt, marker);
	//	if (nWritten > 0)
	//	{
	//		strResult = &vBuffer[0];
	//	}
	//	va_end(marker);
	//}
	//Thread_UnLock();

	ResDataObject SysLogNode;

	string guidstr;
	GUID DeviceGuid;
	//组Log包
	if (GetDeviceType(DeviceGuid))
	{
		guid_2_string(DeviceGuid, guidstr);
		SysLogNode.add("Module", guidstr.c_str());
		SysLogNode.add("AppId", pAppId);
		SysLogNode.add("ThreadId", GetCurrentThreadId());
		if (pCode)
		{
			SysLogNode.add("BusinessKey", pCode);
		}
		else
		{
			SysLogNode.add("BusinessKey", "");
		}

		SysLogNode.add("IP", (const char*)getLocalIpAddress());

		struct timeval tv;
		struct tm* tm_info;
		char TimeTag[30];

		// 获取当前时间
		gettimeofday(&tv, NULL);

		// 转换为本地时间
		tm_info = localtime(&tv.tv_sec);

		// 格式化时间为字符串
		snprintf(TimeTag, sizeof(TimeTag), "%04d-%02d-%02d %02d:%02d:%02d.%03ld",
			tm_info->tm_year + 1900, tm_info->tm_mon + 1, tm_info->tm_mday,
			tm_info->tm_hour, tm_info->tm_min, tm_info->tm_sec, tv.tv_usec / 1000);

		SysLogNode.add("CreationTime", TimeTag);

		string strLevel = SysLogLevel2str(Level);
		SysLogNode.add("Level", strLevel.c_str());

		SysLogNode.add("HostName", (const char*)getLocalMachineId());

		SysLogNode.add("ProcessName", (const char*)GetModuleTitle());

		SysLogNode.add("FreeText", strResult.c_str());

		SysLogNode.add("Context", pCode);

		ResDataObject NotifyData;
		PacketAnalizer::MakeNotify(NotifyData, PACKET_CMD_MSG, "Syslog", SysLogNode);
		CmdFromLogicDev(&NotifyData);
		//PublishAction(&NotifyData, (m_strEBusRoot + "/Notify").c_str(), m_pMqttConntion);

	}
	else
	{
		FWARN("no Guid??");

		return RET_FAILED;
	}

	//打印LOG
	switch (Level)
	{
	case Syslog_Debug:
		//RES_PRINTA_DEBUG(m_pLogger, SysLogNode, "SysLog");
		FDEBUG("SysLog {$} ", SysLogNode.encode());
		break;
	case Syslog_Information:
		///RES_PRINTA_INFO(m_pLogger, SysLogNode, "SysLog");
		FINFO("SysLog {$} ", SysLogNode.encode());
		break;
	case Syslog_Warning:
		//RES_PRINTA_WARN(m_pLogger, SysLogNode, "SysLog");
		FWARN("SysLog {$} ", SysLogNode.encode());

		break;
	case Syslog_Error:
		//RES_PRINTA_ERROR(m_pLogger, SysLogNode, "SysLog");
		FERROR("SysLog {$} ", SysLogNode.encode());

		break;
	case Syslog_Fatal:
		//RES_PRINTA_FATAL(m_pLogger, SysLogNode, "SysLog");
		FINFO("SysLog {$} ", SysLogNode.encode());

		break;
	default:
		FINFO("SysLog {$} " ,SysLogNode.encode() );
		break;
	}

	return RET_SUCCEED;
}

RET_STATUS LogicDevice::SystemLog(SYSLOGLEVEL Level,const char *pCode, const char* fmt, ...)
{
	std::string strResult = "";

	//组Context包
	//if (m_pLogger)
	//{
	//	m_pLogger->Thread_Lock();

	//	if (NULL != fmt)
	//	{
	//		va_list marker = NULL;
	//		va_start(marker, fmt);
	//		size_t nLength = _vscprintf(fmt, marker) + 1;
	//		std::vector<char> vBuffer(nLength, '\0');
	//		int nWritten = vsnprintf_s(&vBuffer[0], vBuffer.size(), nLength, fmt, marker);
	//		if (nWritten > 0)
	//		{
	//			strResult = &vBuffer[0];
	//		}
	//		va_end(marker);
	//	}

	//	m_pLogger->Thread_UnLock();
	//}
	//else
	{
		Thread_Lock();
		if (fmt != NULL) {
			va_list marker;
			va_start(marker, fmt);

			// 获取格式化字符串所需的大小（不包括终止符）
			int nLength = vsnprintf(NULL, 0, fmt, marker);

			if (nLength >= 0) {
				std::vector<char> vBuffer(nLength + 1, '\0'); // +1 为终止符
				vsnprintf(&vBuffer[0], vBuffer.size(), fmt, marker);
				strResult = &vBuffer[0];  // 将缓冲区内容赋值给结果字符串
			}

			va_end(marker);
		}
		Thread_UnLock();
	}



	ResDataObject SysLogNode;

	string guidstr;
	GUID DeviceGuid;
	//组Log包
	if (GetDeviceType(DeviceGuid))
	{
		guid_2_string(DeviceGuid, guidstr);
		SysLogNode.add("Module", guidstr.c_str());
		SysLogNode.add("AppId", "");
		SysLogNode.add("ThreadId", GetCurrentThreadId());
		if (pCode)
		{
			SysLogNode.add("BusinessKey", pCode);
		}
		else
		{
			SysLogNode.add("BusinessKey", "");
		}

		SysLogNode.add("IP", (const char *)getLocalIpAddress());

		struct timeval tv;
		struct tm* tm_info;
		char TimeTag[30];

		// 获取当前时间
		gettimeofday(&tv, NULL);

		// 转换为本地时间
		tm_info = localtime(&tv.tv_sec);

		// 格式化时间为字符串
		snprintf(TimeTag, sizeof(TimeTag), "%04d-%02d-%02d %02d:%02d:%02d.%03ld",
			tm_info->tm_year + 1900, tm_info->tm_mon + 1, tm_info->tm_mday,
			tm_info->tm_hour, tm_info->tm_min, tm_info->tm_sec, tv.tv_usec / 1000);

		SysLogNode.add("CreationTime", TimeTag);

		SysLogNode.add("Level", Level);

		SysLogNode.add("HostName", (const char *)getLocalMachineId());

		SysLogNode.add("ProcessName", (const char *)GetModuleTitle());

		SysLogNode.add("FreeText", strResult.c_str());

		ResDataObject NotifyData;
		PacketAnalizer::MakeNotify(NotifyData, PACKET_CMD_MSG, "Syslog", SysLogNode);
		CmdFromLogicDev(&NotifyData);
		//PublishAction(&NotifyData, (m_strEBusRoot + "/Notify").c_str(), m_pMqttConntion);
		if(m_pParent != nullptr)
			NotifyParent(&NotifyData, "/Notify");
	}
	else
	{
		FERROR("no Guid??");

		return RET_FAILED;
	}

	//打印LOG
	switch (Level)
	{
	case Syslog_Debug:
		//RES_PRINTA_DEBUG(m_pLogger, SysLogNode, "SysLog");
		FDEBUG("SysLog {$}", SysLogNode.encode());
		break;
	case Syslog_Information:
		//RES_PRINTA_INFO(m_pLogger, SysLogNode, "SysLog");
		FINFO("SysLog {$}", SysLogNode.encode());
		break;
	case Syslog_Warning:
		//RES_PRINTA_WARN(m_pLogger, SysLogNode, "SysLog");
		FWARN("SysLog {$}", SysLogNode.encode());

		break;
	case Syslog_Error:
		//RES_PRINTA_ERROR(m_pLogger, SysLogNode, "SysLog");
		FERROR("SysLog {$}", SysLogNode.encode());

		break;
	case Syslog_Fatal:
		//RES_PRINTA_FATAL(m_pLogger, SysLogNode, "SysLog");
		FINFO("SysLog {$}", SysLogNode.encode());

		break;
	default:
		FINFO( "SysLog {$}", SysLogNode.encode());
		break;
	}

	return RET_SUCCEED;
}

/////////////////////////////////////////////////////////////////////////////


ccos_mqtt_connection* LogicDevice::CreateConnection(const char* pszClientID, ccos_mqtt_callback onmsg)
{

	return nullptr;
}

std::string CurrentDateTime()
{
	auto now = std::chrono::system_clock::now();
	auto now_us = std::chrono::duration_cast<std::chrono::microseconds>(now.time_since_epoch());
	auto now_time_t = std::chrono::system_clock::to_time_t(now);

	std::tm now_tm{};
	localtime_r(&now_time_t, &now_tm);  // 线程安全版本

	char buf[64];
	std::strftime(buf, sizeof(buf), "%Y-%m-%d %H:%M:%S", &now_tm);
	std::string now_time_str = buf;

	// 格式化微秒部分并追加
	snprintf(buf, sizeof(buf), ".%06lld", static_cast<long long>(now_us.count() % 1000000));
	now_time_str += buf;

	return now_time_str;
}

static string CurrentDateTime2()
{
	std::time_t now = std::chrono::system_clock::to_time_t(std::chrono::system_clock::now());

	char buf[100] = { 0 };
	std::strftime(buf, sizeof(buf), " %Y-%m-%d %H:%M:%S ", std::localtime(&now));
	return buf;
}

bool LogicDevice::GetActions(ResDataObject& resAction)
{
	for (int x = 0; x < m_Actions.size(); x++)
	{
		resAction.update(m_Actions.GetKey(x), "");
	}
	return true;
}

void LogicDevice::NotifyParent(ResDataObject* NotifyData, const char* pszTopic, ccos_mqtt_connection* hConnection)
{

	if (m_pParent != nullptr)
	{
		if (hConnection == nullptr)
		{
			hConnection = m_pParent->m_pMqttConntion;
		}		
		PublishAction(NotifyData, (m_pParent->GetRootPath() + pszTopic).c_str(), hConnection);
	}
}

const mqtt_qos_t MQTT_QOS = QOS2;

/// <summary>
/// 工作位设备组 上线或者下线，设备路径为 CCOS/Table/Detector
/// CCOS/Demo/Detector
/// </summary>
/// <param name="bOnline"></param>
void LogicDevice::SubscribeGroupActions(string wsName, int bOnline)
{
	if (m_rsOnlineGroup.GetFirstOf(wsName.c_str()) >= 0)
	{
		int lastOnline = (int)m_rsOnlineGroup[wsName.c_str()];
		if (bOnline == lastOnline)
		{
			FWARN("Group Abstract is already ? Online {$}", bOnline);
			return;
		}
	}


	if ( m_strAbstractPath.length() > 0)
	{
		std::string pszAction, gpPath;
		int ret = 0;
		int num = m_Actions.size();

		FINFO("Group Device onLine or offLine {$}", bOnline);
		gpPath = m_strAbstractPath;
		gpPath.replace(0, string("CCOS/DEVICE").length(), "CCOS/" + wsName);

		pszAction = gpPath + m_strDevicePath;
		pszAction += "/Action/+";
		SubScribeTopic(pszAction.c_str(), bOnline == 1);

		if (bOnline)
		{
			FINFO("{$} Subscribe Device Group Action {$}  ", m_strClientID, pszAction);
		}
		else
		{
			FINFO("{$} UnSubscribe Device Group Action {$} ", m_strClientID, pszAction);
		}
		//for (size_t i = 0; i < num; i++)
		//{
		//	pszAction = gpPath + m_strDevicePath;
		//	pszAction += "/Action/";
		//	pszAction += (m_Actions.GetKey(i));

		//	SubScribeTopic(pszAction.c_str(), bOnline == 1);

		//	if (bOnline)
		//	{
		//		FINFO("{$} Subscribe Device Group Action {$}  ", m_strClientID, pszAction);
		//	}
		//	else
		//	{
		//		FINFO("{$} UnSubscribe Device Group Action {$} ", m_strClientID, pszAction);
		//	}
		//}
		m_rsOnlineGroup.update(wsName.c_str(), bOnline);

	}
	else
	{
		FWARN("try Online but AbstractPath is none {$}", m_strClientID);
	}

}

void SubscribeModule(ccos_mqtt_connection* pconn, string devuribBase)
{
	string pszAction = devuribBase;

	pszAction += "/Get/+";
	SubscribeTopic(pconn, pszAction.c_str());

	pszAction = devuribBase;
	pszAction += "/Update/+";
	SubscribeTopic(pconn, pszAction.c_str());

	pszAction = devuribBase;
	pszAction += "/Set/+";
	SubscribeTopic(pconn, pszAction.c_str());

	pszAction = devuribBase;
	pszAction += "/Add/+";
	SubscribeTopic(pconn, pszAction.c_str());

	pszAction = devuribBase;
	pszAction += "/Del/+";
	SubscribeTopic(pconn, pszAction.c_str());

	pszAction = devuribBase;
	pszAction += "/Action/+";
	SubscribeTopic(pconn, pszAction.c_str());

	pszAction = devuribBase;
	pszAction += "/Message/+";
	SubscribeTopic(pconn, pszAction.c_str());
}

void  LogicDevice::SubscribeActions()
{
	FINFO("Begain");
	if (nullptr == m_pMqttConntion)
		return;
	int num = m_Actions.size();
	mqtt_client*  pMqttClient = (mqtt_client*)std::get<MQTT_CLT_ID>(*m_pMqttConntion);
	mqtt_topic_list* pTopicList = std::get<MQTT_TIPIC_LIST_ID>(*m_pMqttConntion);


	{
		ResDataObject res,resAction;
		GetDeviceResource(&res);
		std::cout << "LogicDevice::SubscribeActions GetDeviceResource" << res.encode() << endl;
		resAction = res["Action"];
		FINFO("Actions from deviceresource [{$}]", resAction.size());
		for (int x = 0; x < resAction.size(); x++)
		{
			string action = (string)resAction.GetKey(x);
			if (action.length() > 0)
			{
				FINFO("Action from DeviceResource [{$}]", action);
				m_Actions.update(action.c_str(), "");
			}

		}

		num = m_Actions.size();
	}


	std::string pszAction;
	int ret = 0;

	SubscribeModule(m_pMqttConntion, m_strEBusRoot.c_str());
	FINFO("{$} Subscribe EBus Action {$} return {$} topic num {$}", m_strClientID, pszAction, ret, pTopicList->size());

	if (m_strEBusRoot != m_strCCOSDevicePath)
	{
		pszAction = m_strCCOSDevicePath + m_strDevicePath;
		SubscribeModule(m_pMqttConntion, pszAction.c_str());
		FINFO("{$} Subscribe CCOS Action {$} return {$} topic num {$}", m_strClientID, pszAction, ret, pTopicList->size());
	}
	if (m_strAbstractPath.length() > 0)
	{
		pszAction = m_strAbstractPath + m_strDevicePath;
		SubscribeModule(m_pMqttConntion, pszAction.c_str());
		FINFO("{$} Subscribe Abstract Action {$} return {$} topic num {$}", m_strClientID, pszAction, ret, pTopicList->size());
	}

	
	//for (size_t i = 0; i < num; i++)
	//{
	//	//订阅ebus的路径
	//	pszAction = m_strEBusRoot;
	//	pszAction += "/Action/";
	//	pszAction += (m_Actions.GetKey(i));


	//	SubscribeTopic(m_pMqttConntion, pszAction.c_str());
	//	FINFO("{$} Subscribe Action {$} return {$} topic num {$}", m_strClientID, pszAction, ret, pTopicList->size());
	//	

	//	//订阅CCOS设备路径
	//	if (m_strEBusRoot != m_strCCOSDevicePath)
	//	{

	//		pszAction = m_strCCOSDevicePath + m_strDevicePath;
	//		pszAction += "/Action/";
	//		pszAction += (m_Actions.GetKey(i));


	//		SubscribeTopic(m_pMqttConntion, pszAction.c_str());
	//		FINFO("{$} Subscribe Action {$} return {$} topic num {$}", m_strClientID, pszAction, ret, pTopicList->size());
	//	}

	//	//抽象设备路径要一直存在
	//	//订阅抽象设备路径
	//	if ( m_strAbstractPath.length() > 0)
	//	{
	//		pszAction = m_strAbstractPath + m_strDevicePath;
	//		pszAction += "/Action/";
	//		pszAction += (m_Actions.GetKey(i));

	//		SubscribeTopic(m_pMqttConntion, pszAction.c_str());
	//		FINFO("{$} Subscribe Action {$} return {$} topic num {$}", m_strClientID, pszAction, ret, pTopicList->size());
	//	}
	//}

	//pszAction = m_strEBusRoot;
	//pszAction += "/Action/UpdateDeviceResource";
	////std::cout << "=====*******======  【" << m_strClientID << "】 Subscribe Action " << pszAction << endl;
	//pTopicList->push_back(pszAction);
	//mqtt_subscribe(pMqttClient, pszAction.c_str(), MQTT_QOS, msgarrivd);
	//FINFO("{$} Subscribe Action {$} return {$}  topic num {$}", m_strClientID, pszAction, ret, pTopicList->size());
	//*/

	// 等待所有订阅完成：轮询检查ack_list，确保所有SUBACK都已收到
	// 这比固定延时更可靠，能确保订阅真正生效后才继续
	FINFO("{$} Waiting for all subscriptions to complete...", m_strClientID);
	const DWORD subscribeTimeoutMs = 10000; // 10秒超时（可能有多个订阅）
	if (!WaitForSubscriptionComplete(m_pMqttConntion, subscribeTimeoutMs)) {
		FERROR("{$} Some subscriptions did not complete within timeout", m_strClientID);
	} else {
		FINFO("{$} All subscriptions ready", m_strClientID);
	}

	// SUBACK收到后，等待Broker完成内部路由表更新
	// 延时可通过环境变量MQTT_BROKER_ROUTING_DELAY_MS配置，默认200ms
	WaitForBrokerRouting(m_strClientID.c_str());

}
/*
void connlost(void* context, char* cause)
{
	//连接中断
	ccos_mqtt_connection* connection = (ccos_mqtt_connection*)context;
	std::cout << "Connection Lost .....[" << std::get<CLINET_ID_ID>(*connection) <<"] why?" << endl;
	printf("\nConnection lost\n");
	printf("     cause: %s\n", cause);
	std::cout << CurrentDateTime() << "MQTT Server Connection lost...." << endl;
}

void disconnected(void* context, MQTTProperties* props, enum MQTTReasonCodes rc)
{
	ccos_mqtt_connection* connection = (ccos_mqtt_connection*)context;
	std::cout << "Connection disconnected  .....[" << std::get<CLINET_ID_ID>(*connection) << "] why?" << endl;

}
*/

//发送成功
//void delivered(void* context, MQTTAsync_token dt)
//{
//	printf("Message with token value %d delivery confirmed\n", dt);
//	//deliveredtoken = dt;
//}

//void connlost(void* context, char* cause)
//{
//	//printf("Connection 0X%08X Lost ...... ???? %s \n",, cause);
//	//std::cout << "Connection Context [" << (UINT64)context << "] conn lost " << (cause==nullptr?"": cause) << endl;
//
//	//连接中断
//	ccos_mqtt_connection* connection = (ccos_mqtt_connection*)context;
//	mqtt_client* client = (mqtt_client*)std::get<MQTT_CLT_ID>(*connection);
//	std::cout << CurrentDateTime() << "Connection Lost 2 .....[" << std::get<CLINET_ID_ID>(*connection) << "] why? " << (cause == nullptr ? "" : cause) << endl;
//	return;
//
//	MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
//	int rc;
//
//	printf("Reconnecting\n");
//	conn_opts.keepAliveInterval = 20;
//	conn_opts.cleansession = 1;
//	if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
//	{
//		printf("Failed to start connect, return code %d\n", rc);
//		//finished = 1;
//	}
//}

//重连成功
//void onReconnected(void* context, char* cause)
//{
//	//printf("onReconnected 0X%08X Lost ...... ???? %s \n", (UINT64)context, cause);
//	//std::cout << "onReconnected [" << (UINT64)context << "] conn " << (cause == nullptr ? "" : cause) << endl;
//
//	ccos_mqtt_connection* connection = (ccos_mqtt_connection*)context;
//	mqtt_client* client = (mqtt_client*)std::get<MQTT_CLT_ID>(*connection);
//	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
//	mqtt_topic_list* pTopicList = std::get<MQTT_TIPIC_LIST_ID>(*connection);
//	const char* client_id = std::get<CLINET_ID_ID>(*connection);
//	int rc;
//
//	//printf(" %s  \n", std::get<CLINET_ID_ID>(*connection));
//	std::cout << "[" << client_id << "] Successful reconnection  Use context [" << (UINT64)context << "]" << endl;
//	//cout << "Connected ok.. by TID [" << GetCurrentThreadId() << "]" << endl;
//	// 重新订阅，暂不重新订阅，看看
//
//	///*
//	//printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
//	//	"Press Q<Enter> to quit\n\n", TOPIC, CLIENTID, QOS);
//	//opts.onSuccess = onSubscribe;
//	//opts.onFailure = onSubscribeFailure;
//	//opts.context = client;
//	auto it = pTopicList->begin();
//	while(it != pTopicList->end())
//	{
//		rc = MQTTAsync_subscribe(client, (*it).c_str(), 1, &opts);
//		printf("-------- [%s] re subscribe %s, return code %d\n", client_id, (*it).c_str(), rc);
//		it++;
//	}
//	//*/
//
//}
//
////MQTT 连接主动断开连接失败
//void onDisconnectFailure(void* context, MQTTAsync_failureData* response)
//{
//	printf("Disconnect failed, rc %d\n", response->code);
//	//disc_finished = 1;
//}
//
////MQTT 连接主动断开连接成功
//void onDisconnect(void* context, MQTTAsync_successData* response)
//{
//	//printf("Successful disconnection\n");
//	//disc_finished = 1;
//}
//
//void onSubscribe(void* context, MQTTAsync_successData* response)
//{
//	printf("Subscribe succeeded\n");
//	//subscribed = 1;
//}
//
//void onSubscribeFailure(void* context, MQTTAsync_failureData* response)
//{
//	printf("Subscribe failed, rc %d\n", response->code);
//	//finished = 1;
//}
//
//
//void onConnectFailure(void* context, MQTTAsync_failureData* response)
//{
//	printf("Connect failed, rc %d\n", response->code);
//	//finished = 1;
//}

//void onConnect(void* context, MQTTAsync_successData* response)
//{
//
//	ccos_mqtt_connection* connection = (ccos_mqtt_connection*)context;
//	mqtt_client* client = (mqtt_client*)std::get<MQTT_CLT_ID>(*connection);
//
//	std::cout << "[" << std::get<CLINET_ID_ID>(*connection) << "] onConnect Context [" << (UINT64)context << "] " << endl;
//
//	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
//	int rc;
//
//	//printf("[%s] connect MQTT Successful connection\n", std::get< CLINET_ID_ID>(*connection));
//	HANDLE hConnected = std::get< CONNECTED_HANDLE_ID>(*connection);
//	if (hConnected != NULL)
//		SetEvent(hConnected);
//
//	/*
//	printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
//		"Press Q<Enter> to quit\n\n", TOPIC, CLIENTID, QOS);
//	opts.onSuccess = onSubscribe;
//	opts.onFailure = onSubscribeFailure;
//	opts.context = client;
//	if ((rc = MQTTAsync_subscribe(client, TOPIC, QOS, &opts)) != MQTTASYNC_SUCCESS)
//	{
//		printf("Failed to start subscribe, return code %d\n", rc);
//		finished = 1;
//	}*/
//}


//往指定主题发送Action包，携带参数，并指定答复的Topic，异步模式处理Resp，需要在对应连接的回调中处理
int  PublishActionWithoutLock(string  &message, const char* pszTopic, mqtt_client* pMqttClient, std::string client_id, mqtt_qos_t qos = MQTT_QOS);


void resubscribe_topic(void* client, void* reconnect_date)
{
	mqtt_client* pClient = (mqtt_client*)client;
	ccos_mqtt_connection* connection = (ccos_mqtt_connection*)pClient->mqtt_conn_context;

	mqtt_topic_list* pTopicList =  std::get<MQTT_TIPIC_LIST_ID>(*connection);
	string client_id = std::get<CLINET_ID_ID>(*connection);

	FWARN("Connection Reconneted ok. {$}, topic num :{$}", client_id, pTopicList->size());
	CcosLock* pLock =  std::get<CONN_SEND_LOCK_ID>(*connection);
	pLock->Thread_Lock();

	for_each(pTopicList->begin(),pTopicList->end(), [&pClient] (string str)-> void {
		FWARN("Resubscribe {$}", str);
		mqtt_subscribe(pClient, str.c_str(), MQTT_QOS, msgarrivd);
	});


	pLock->Thread_UnLock();

}

//MQTT消息抵达
//int msgarrvd(void* context, char* topicName, int topicLen, MQTTAsync_message* message)
void msgarrivd(void* client, message_data_t* message)
{
	// 日志前缀：时间+线程ID
	const auto logPrefix = CurrentDateTime() + " msgarrivd:: TID [" + std::to_string(GetCurrentThreadId()) + "] ";
	std::cout << logPrefix << std::endl;
	FINFO(logPrefix);

	// 基础参数校验
	if (!client || !message || !message->topic_name) return;

	// 转换并校验客户端对象
	auto* pClient = reinterpret_cast<mqtt_client*>(client);
	if (!pClient || !pClient->mqtt_conn_context) {
		std::cout << logPrefix << "Invalid client or connection context" << std::endl;
		return;
	}

	// 转换并校验连接对象
	ccos_mqtt_connection* connection = nullptr;
	try {
		connection = reinterpret_cast<ccos_mqtt_connection*>(pClient->mqtt_conn_context);
		if (!connection) {
			std::cout << logPrefix << "Connection is null" << std::endl;
			return;
		}
	}
	catch (...) {
		std::cout << logPrefix << "Failed to cast connection context" << std::endl;
		return;
	}

	// 获取并校验锁对象
	CcosLock* pLock = nullptr;
	try {
		pLock = std::get<CONN_SEND_LOCK_ID>(*connection);
		if (!pLock) {
			std::cout << logPrefix << "Lock is null" << std::endl;
			return;
		}
	}
	catch (...) {
		std::cout << logPrefix << "Failed to get lock from connection" << std::endl;
		return;
	}

	// 加锁并二次校验
	pLock->Thread_Lock();
	auto* pLockTemp = std::get<CONN_SEND_LOCK_ID>(*connection);
	if (!pLockTemp) {
		pLock->Thread_UnLock();
		return;
	}
	if (!pClient->mqtt_conn_context) {
		pLock->Thread_UnLock();
		std::cout << logPrefix << "Connection context became null during locking" << std::endl;
		return;
	}

	// 获取客户端ID
	std::string clientId;
	try {
		clientId = std::get<CLINET_ID_ID>(*connection);
	}
	catch (...) {
		pLock->Thread_UnLock();
		std::cout << logPrefix << "Failed to get client ID" << std::endl;
		return;
	}

	// 获取MQTT客户端对象
	void* mqttClient = nullptr;
	try {
		mqttClient = std::get<MQTT_CLT_ID>(*connection);
		if (!mqttClient) {
			pLock->Thread_UnLock();
			std::cout << logPrefix << "MQTT client is null" << std::endl;
			return;
		}
	}
	catch (...) {
		pLock->Thread_UnLock();
		std::cout << logPrefix << "Failed to get MQTT client" << std::endl;
		return;
	}

	// 处理主题名
	std::string topic;
	try {
		const size_t topicLen = strlen(message->topic_name);
		const size_t maxTopicLen = sizeof(message->topic_name) - 1;
		topic = (topicLen > maxTopicLen) ?
			std::string(message->topic_name, maxTopicLen) :
			std::string(message->topic_name);
	}
	catch (...) {
		pLock->Thread_UnLock();
		std::cout << logPrefix << "Failed to process topic name" << std::endl;
		return;
	}

	// 输出消息接收日志
	std::cout << logPrefix << clientId << " Get Msg from " << topic << std::endl;
	FINFO("[msgarrivd] {$} received message from topic: {$}, payload_len: {$}", clientId, topic, message->message->payloadlen);

	// 校验消息数据
	if (!message->message || !message->message->payload) {
		pLock->Thread_UnLock();
		std::cout << logPrefix << "Invalid message data" << std::endl;
		return;
	}

	// 校验消息大小
	const auto payloadLen = message->message->payloadlen;
	if ((topic.length() >= MQTT_TOPIC_LEN_MAX && payloadLen > 16380) || payloadLen > 512 * 1024) {
		pLock->Thread_UnLock();
		std::cout << logPrefix << "Message too large: topic_len=" << topic.length()
			<< ", payload_len=" << payloadLen << std::endl;
		return;
	}

	// 大消息提示
	if (payloadLen > 16380) {
		std::cout << logPrefix << clientId << " Get big Msg from " << topic
			<< " msg Len " << payloadLen << std::endl;
	}

	// 解码消息载荷
	ResDataObject req;
	try {
		std::string payload(static_cast<const char*>(message->message->payload), payloadLen);
		req.decode(payload.c_str());
		FINFO("[msgarrivd] {$} decoded message successfully, KEY: {$}, TYPE: {$}, CMD: {$}",
			clientId,
			PacketAnalizer::GetPacketKey(&req),
			(int)PacketAnalizer::GetPacketType(&req),
			(int)PacketAnalizer::GetPacketCmd(&req));
	}
	catch (...) {
		pLock->Thread_UnLock();
		FERROR("[msgarrivd] {$} failed to decode message from topic: {$}", clientId, topic);
		return;
	}

	// 处理消息钩子
	void* pHook = nullptr;
	try {
		pHook = std::get<MSG_HOOK_ID>(*connection);
	}
	catch (...) {
		pLock->Thread_UnLock();
		return;
	}

	if (pHook) {
		try {
			auto* filter = reinterpret_cast<ccos_mqtt_msg_filter*>(pHook);
			if (!filter) {
				pLock->Thread_UnLock();
				return;
			}

			auto func = std::get<FILTER_FUNC_ID>(*filter);
			std::cout << logPrefix << clientId << " Got Hook Process " << topic << std::endl;
			FINFO("[msgarrivd] {$} processing hook for topic: {$}", clientId, topic);

			auto* pRequests = std::get<FILTER_RES_OBJ_ID>(*filter);
			if (!pRequests) {
				pLock->Thread_UnLock();
				FWARN("[msgarrivd] {$} pRequests is null", clientId);
				return;
			}

			FINFO("[msgarrivd] {$} checking {$} pending requests in hook", clientId, pRequests->size());

			// 遍历请求处理钩子
			for (int x = 0; x < pRequests->size(); x++) {
				std::cout << logPrefix << clientId << " Hook Process Function " << topic << std::endl;

				const std::string keyAction = pRequests->GetKey(x);
				ResDataObject resObj = (*pRequests)[x];
				auto hEvent = std::get<FILTER_HANDLE_ID>(*filter);

				FINFO("[msgarrivd] {$} comparing: pending_key=[{$}] vs received_key=[{$}], msg_type={$}",
					clientId, keyAction, PacketAnalizer::GetPacketKey(&req), (int)PacketAnalizer::GetPacketType(&req));

				// 匹配钩子消息
				if (PacketAnalizer::GetPacketType(&req) == PACKET_TYPE_RES &&
					keyAction == PacketAnalizer::GetPacketKey(&req)) {

					std::cout << logPrefix << clientId << " Got Hook Function Hooked.. " << topic << std::endl;
					FINFO("[msgarrivd] {$} MATCHED! Hook triggered for key: {$}", clientId, keyAction);

					// 打开事件（优先使用resObj，失败则用默认）
					auto event = LinuxEvent::OpenEvent(std::string(resObj).c_str());
					if (!event) event = hEvent;

					// 设置响应对象
					auto* pResponse = std::get<FILTER_RESPONS_OBJ_ID>(*filter);
					if (pResponse) {
						pResponse->add(keyAction.c_str(), req);
						FINFO("[msgarrivd] {$} added response to pResponse for key: {$}", clientId, keyAction);
					} else {
						FWARN("[msgarrivd] {$} pResponse is null, cannot store response", clientId);
					}

					// 触发事件
					if (event) {
						try {
							event->SetEvent();
							FINFO("[msgarrivd] {$} event signaled successfully for key: {$}", clientId, keyAction);
						}
						catch (...) {
							FERROR("[msgarrivd] {$} failed to signal event for key: {$}", clientId, keyAction);
						}
					} else {
						FWARN("[msgarrivd] {$} event is null, cannot signal", clientId);
					}

					// 清理并退出
					pRequests->eraseOneOf(keyAction.c_str(), 0);
					FINFO("[msgarrivd] {$} erased request from pending list, key: {$}", clientId, keyAction);
					pLock->Thread_UnLock();
					return;
				}
			}
		}
		catch (...) {
			FERROR("[msgarrivd] {$} exception in hook processing for topic: {$}", clientId, topic);
		}
	}

	// 未命中钩子，继续处理
	std::cout << logPrefix << clientId << " Go on Process " << topic << std::endl;
	FINFO("[msgarrivd] {$} hook not matched, calling user callback for topic: {$}", clientId, topic);
	pLock->Thread_UnLock();

	// 调用用户回调
	ccos_mqtt_callback onmsg = nullptr;
	try {
		onmsg = std::get<USER_MSG_CAKBCK_ID>(*connection);
	}
	catch (...) {
		return;
	}

	if (onmsg) {
		// 使用异步处理避免阻塞MQTT接收流程
		try {
			AsyncMsgHandler::getInstance().handleMessageAsync(onmsg, req, topic, connection);
			std::cout << logPrefix << clientId << " 消息已提交到异步处理队列: " << topic << std::endl;
		}
		catch (const std::exception& e) {
			std::cout << logPrefix << clientId << " 异步消息处理提交失败: " << e.what()
			         << ", Topic: " << topic << std::endl;
		}
		catch (...) {
			std::cout << logPrefix << clientId << " 异步消息处理提交出现未知错误, Topic: "
			         << topic << std::endl;
		}
	}
	else {
		std::cout << logPrefix << "**** ----- **** USER_MSG_CAKBCK_ID is null "
			<< clientId << " When processing " << topic << std::endl;
	}
}

void* MqttSendThreadFunc(void* pPara)
{
	ccos_mqtt_connection* connection = (ccos_mqtt_connection*)pPara;
	if (!connection) {
		return NULL;
	}

	CcosLock* pLock = std::get<CONN_SEND_LOCK_ID>(*connection);
	mqtt_msg_list* pList = std::get<MSG_LIST_ID>(*connection);
	mqtt_client* pConn = (mqtt_client*)std::get<MQTT_CLT_ID>(*connection);
	std::string client_id = std::get<CLINET_ID_ID>(*connection);
	sem_t* hSemaphore = std::get<SEMAPHORE_HANDLE_ID>(*connection);
	mqtt_topic_list* pTopicList = std::get<MQTT_TIPIC_LIST_ID>(*connection);

	// 线程启动日志
	std::cout << "[" << client_id << "] The MQTT sending thread is started" << std::endl;

	while (std::get<THREAD_RUNNING_ID>(*connection)) {
		struct timespec ts;
		clock_gettime(CLOCK_REALTIME, &ts);
		ts.tv_sec += 1;

		// 等待信号量前的日志
		//std::cout << "[" << client_id << "] Wait for the message semaphore..." << std::endl;

		if (sem_timedwait(hSemaphore, &ts) == 0) {
			std::cout << "[" << client_id << "] Received the message semaphore and ready to process the message" << std::endl;
			
			pLock->Thread_Lock();
			std::cout << "[" << client_id << "] Successfully obtained the sending lock" << std::endl;

			if (pList->empty()) {
				std::cout << "[" << client_id << "] The message list is empty, release the lock." << std::endl;
				pLock->Thread_UnLock();
				continue;
			}

			ResDataObject* pMsg = pList->front();
			pList->pop_front();
			pLock->Thread_UnLock();

			std::cout << "[" << client_id << "] Get a message from the message list, the number of remaining messages: " << pList->size() << std::endl;
			std::string m_strSendMessage;
			for (int x = 0; x < pMsg->size(); x++) {
				const char* pTopic = pMsg->GetKey(x);
				m_strSendMessage = (std::string)(*pMsg)[pTopic];

				std::cout<< CurrentDateTime() << " [" << client_id << "] Prepare to publish messages to the topic: " << pTopic
					<< ", Message length: " << m_strSendMessage.length() << "Byte" << std::endl;
				PublishActionWithoutLock(m_strSendMessage, pTopic, pConn, client_id, pTopicList->size() <= 0 ? QOS2 : MQTT_QOS);
			}
			std::cout << "[" << client_id << "] The message processing is completed, and the message object has been released." << std::endl;
			delete pMsg;
			//std::cout << "[" << client_id << "] Release the sending lock" << std::endl;
		}
		else if (errno == ETIMEDOUT) {
			//std::cout << "[" << client_id << "] The semaphore wait timed out, continue waiting." << std::endl;
			continue;
		}
		else {
			// 其他错误情况
			std::cout << "[" << client_id << "] Semaphore wait error, error code: " << errno << std::endl;
			continue;
		}
	}
	return NULL;
}
// 安全遍历MQTT消息处理程序的宏
#define mqtt_list_for_each_entry_safe(pos, n, head, member) \
    for (pos = container_of((head)->next, decltype(*pos), member), \
         n = container_of(pos->member.next, decltype(*pos), member); \
         &pos->member != (head); \
         pos = n, n = container_of(n->member.next, decltype(*n), member))

// 通过成员指针获取包含结构的指针
#define container_of(ptr, type, member) ({ \
    const typeof(((type*)0)->member)* __mptr = (ptr); \
    (type*)((char*)__mptr - offsetof(type, member)); })

// 重连回调函数 - 实现自动重订阅
static void reconnect_handler(void* client, void* reconnect_data) {
	mqtt_client_t* c = (mqtt_client_t*)client;
	message_handlers_t* pos, * n;

	ccos_mqtt_connection* connection = (ccos_mqtt_connection*)c->mqtt_conn_context;

	// 遍历所有已订阅的主题并重新订阅
	CcosLock* pLock = std::get<CONN_SEND_LOCK_ID>(*connection);
	pLock->Thread_Lock();
	/*mqtt_list_for_each_entry_safe(pos, n, &c->mqtt_msg_handler_list, list) {
		mqtt_subscribe(c, pos->topic_filter, pos->qos, pos->handler);
	}*/
	pLock->Thread_UnLock();
}

mqtt_client_t* InnerConnect(ccos_mqtt_connection* connection) {
	
	if (!connection) {
		std::cerr << "[ERROR] Invalid connection context (nullptr)" << std::endl;
		return nullptr;
	}

	//mqtt_log_init();

	mqtt_client_t* client = NULL;
	client = mqtt_lease();
	mqtt_log_init();
	if (!client) {  // 修复：检查客户端指针是否有效
		return nullptr;
	}

	std::get<MQTT_CLT_ID>(*connection) = client;
	client->mqtt_conn_context = connection;
	std::string& pszClientID = std::get<CLINET_ID_ID>(*connection);
	// 使用本地副本存储连接参数
	std::string host = SERVER_ADDRESS;
	std::string port = "1883";
	std::string username = "zskkcc";
	std::string password = "zskk1234";

	std::cout << "[CONFIG] Setting connection parameters:"
		<< "\n\tClient ID: " << pszClientID
		<< "\n\tHost: " << host
		<< "\n\tPort: " << port
		<< std::endl;

	// 设置MQTT连接参数
	mqtt_set_host(client, const_cast<char*>(host.c_str()));
	mqtt_set_port(client, const_cast<char*>(port.c_str()));
	mqtt_set_user_name(client, const_cast<char*>(username.c_str()));
	mqtt_set_password(client, const_cast<char*>(password.c_str()));
	mqtt_set_client_id(client, const_cast<char*>(pszClientID.c_str()));
	mqtt_set_clean_session(client, 0);
	mqtt_set_write_buf_size(client, 8192);
	mqtt_set_read_buf_size(client, 8192);
	mqtt_set_cmd_timeout(client, 2000);  
	mqtt_set_keep_alive_interval(client, 30);  
	//mqtt_subscribe();
	//mqtt_set_reconnect_handler(pMqttClient, reconnect_handler);

	int rc = 0;
	const uint64_t timeout_ms = 10000;  // 10秒超时
	uint64_t start_time = GetTickCount();
	int attempt_count = 0;
	bool connected = false;

	std::cout << "[CONNECT] Initiating connection (timeout: " << timeout_ms << "ms)..." << std::endl;

	try {
		while (true) {
			attempt_count++;
			uint64_t current_time = GetTickCount();
			uint64_t elapsed = current_time - start_time;

			// 检查超时
			if (elapsed > timeout_ms) {
				std::cerr << "[FATAL] Connection timeout after " << timeout_ms << "ms" << std::endl;
				return nullptr;
			}

			std::cout << CurrentDateTime() << " [ATTEMPT #" << attempt_count << "] Trying MQTT connect..." << std::endl;
			rc = mqtt_connect(client);

			if (rc == MQTT_SUCCESS_ERROR) {
				connected = true;
				break;
			}

			std::cerr << "[ERROR] Connection failed (code: " << rc << "), Elapsed: " << elapsed << "ms" << std::endl;

			// 计算等待时间
			uint64_t remaining = timeout_ms - elapsed;
			uint64_t wait_time = (remaining > 2000) ? 2000 : remaining;

			if (wait_time > 0) {
				std::cout << "[RETRY] Waiting " << wait_time << "ms before next attempt..." << std::endl;
				usleep(wait_time * 1000);
			}
		}

		if (connected) {
			uint64_t total_time = GetTickCount() - start_time;
			FINFO("[SUCCESS] Connected in {$}ms after {$} attempts", total_time, attempt_count);

			// 等待CONNACK处理完成：mqtt_connect()是异步的,返回0只表示CONNECT包已发送
			// 需要给MQTT内部线程一点时间处理CONNACK（通常几十毫秒内完成）
			// 实际的订阅完成由WaitForSubscriptionComplete()保证，所以这里只需要短暂等待
			const int connackDelayMs = 100;
			FINFO("[MQTT] Waiting {$}ms for CONNACK processing...", connackDelayMs);
			usleep(connackDelayMs * 1000);
			FINFO("[MQTT] CONNACK wait complete, connection ready for subscriptions");

			// 安全地设置连接上下文
			try {
				return client;
			}
			catch (const std::exception& e) {
				FERROR("[ERROR] Failed to set connection context: {$}", e.what());
				return nullptr;
			}
		}
	}
	catch (const std::exception& e) {
		std::cerr << "[ERROR] Exception during connection: " << e.what() << std::endl;
		return nullptr;
	}
	catch (...) {
		std::cerr << "[ERROR] Unknown exception during connection" << std::endl;
		return nullptr;
	}
	return nullptr;
}

ccos_mqtt_connection* LogicDevice::NewConnection(const char* pszServer,const char* pszServerPort, const char* pszUserName, const char* pszPassword, const char* pszClientID, ccos_mqtt_callback onmsg)
{
	FINFO("TID {$} :  {$} NewConnection  {$}:{$} user: {$} password {$} ", GetCurrentThreadId(), pszClientID, pszServer, pszServerPort, pszUserName, pszPassword);
	std::cout << "TID " << GetCurrentThreadId()
		<< " :  " << pszClientID
		<< " NewConnection  " << pszServer
		<< ":" << pszServerPort
		<< " user: " << pszUserName
		<< " password " << pszPassword
		<< std::endl;
	//连接MQTT broker 
	DWORD dwTick = GetTickCount();

	std::string clientIdStr(pszClientID ? pszClientID : "");

	ccos_mqtt_connection* connection = new ccos_mqtt_connection;
	try {
	// 初始化元组字段
	std::get<MQTT_TIPIC_LIST_ID>(*connection) = new mqtt_topic_list;
	std::get<CLINET_ID_ID>(*connection) = pszClientID;
	std::get<USER_MSG_CAKBCK_ID>(*connection) = onmsg;

	CcosLock* pLock = new CcosLock();
	std::get<CONN_SEND_LOCK_ID>(*connection) = pLock;
	pLock->Thread_Lock();

	ccos_mqtt_msg_filter* pfilter = new ccos_mqtt_msg_filter;
	std::get<FILTER_FUNC_ID>(*pfilter) = nullptr;
	std::get<FILTER_RES_OBJ_ID>(*pfilter) = new ResDataObject;
	std::get<FILTER_RESPONS_OBJ_ID>(*pfilter) = new ResDataObject;

	std::get<MSG_HOOK_ID>(*connection) = pfilter;
	std::cout << "ALLOCATE Connection [" << (UINT64) connection << "] filter [" << (UINT64)pfilter << "] .. for  " << pszClientID << endl;

	mqtt_client* pMqttClient = InnerConnect(connection);
	if (pMqttClient == nullptr)
	{
		pLock->Thread_UnLock();
		throw std::runtime_error("InnerConnect failed");
	}
	std::get<MQTT_CLT_ID>(*connection) = pMqttClient;
	std::get<MSG_LIST_ID>(*connection) = new mqtt_msg_list;

	// 创建POSIX信号量
	sem_t* semaphore = new sem_t;
	if (sem_init(semaphore, 0, 0) != 0) {
		pLock->Thread_UnLock();
		delete semaphore;
		throw std::runtime_error("Failed to initialize semaphore");
	}
	std::get<SEMAPHORE_HANDLE_ID>(*connection) = semaphore;

	std::get<THREAD_RUNNING_ID>(*connection) = true;
	// 创建发送线程
	pthread_t threadId;
	if (pthread_create(&threadId, NULL, MqttSendThreadFunc, connection) != 0) {
		std::cerr << "Failed to create MQTT send thread" << std::endl;
		std::get<THREAD_RUNNING_ID>(*connection) = false;
		// 错误处理...
		sem_destroy(semaphore);
		delete semaphore;
		pLock->Thread_UnLock();
		throw std::runtime_error("Failed to initialize semaphore");
	}

	std::get<CONNECTED_HANDLE_ID>(*connection) = threadId;
	std::cout << "MqttSendThreadFunc thread id [" << threadId << "]" << std::endl;

	pLock->Thread_UnLock();

	uint64_t dwWaitTick = GetTickCount() - dwTick;
	std::cout << "MQTT " << pszClientID << " try ConnMqtt Succeeded. Use Time: "
		<< dwWaitTick << " ms" << std::endl;

	return connection;
	}
	catch (const std::exception& e) {
		std::cerr << "Exception in NewConnection: " << e.what() << std::endl;
		// 清理资源
		if (std::get<SEMAPHORE_HANDLE_ID>(*connection)) {
			sem_destroy(std::get<SEMAPHORE_HANDLE_ID>(*connection));
			delete std::get<SEMAPHORE_HANDLE_ID>(*connection);
		}
		delete std::get<MSG_LIST_ID>(*connection);
		delete std::get<CONN_SEND_LOCK_ID>(*connection);
		delete std::get<MQTT_TIPIC_LIST_ID>(*connection);
		delete connection;
		return nullptr;
	}
}


//创建额外连接，需要提供回调函数
LOGICDEVICE_API ccos_mqtt_connection* NewConnection(const char* pszClientID, ccos_mqtt_callback onmsg, DWORD dwOpenTimeout, bool async)
{
	DWORD dwTick = GetTickCount();

	FINFO("TID {$} :  {$} NewConnection2  {$}:{$} ", GetCurrentThreadId(), pszClientID, SERVER_ADDRESS, 1883);

	//mqtt_client* pMqttClient = nullptr;
	//pMqttClient = mqtt_lease();

	ccos_mqtt_connection* connection = new ccos_mqtt_connection;
	//HANDLE hConneted = CreateEvent(NULL, FALSE, FALSE, NULL);

	//std::get<MQTT_CLT_ID>(*connection) = pMqttClient;

	mqtt_topic_list* pTopicList = new std::list<string>;
	std::get<MQTT_TIPIC_LIST_ID>(*connection) = pTopicList;

	CcosLock* pLock = new CcosLock();
	std::get<CONN_SEND_LOCK_ID>(*connection) = pLock;
	pLock->Thread_Lock();

	ccos_mqtt_msg_filter* pfilter = new ccos_mqtt_msg_filter;
	std::get<FILTER_FUNC_ID>(*pfilter) = nullptr;
	std::get<MSG_HOOK_ID>(*connection) = pfilter;
	std::get<FILTER_RES_OBJ_ID>(*pfilter) = new ResDataObject;
	std::get<FILTER_RESPONS_OBJ_ID>(*pfilter) = new ResDataObject;
	std::cout << "ALLOCATE  Connection 2 [" << (UINT64)connection << "] filter [" << (UINT64)pfilter << "] ..for  " << pszClientID << endl;
	std::get<CLINET_ID_ID>(*connection) = pszClientID;
	std::get< USER_MSG_CAKBCK_ID>(*connection) = onmsg;
	mqtt_client* pMqttClient = InnerConnect(connection);
	if (pMqttClient == nullptr)
	{
		pLock->Thread_UnLock();
		delete std::get<MQTT_TIPIC_LIST_ID>(*connection);  // 删除 pTopicList
		delete std::get<CONN_SEND_LOCK_ID>(*connection);   // 删除 pLock
		delete connection;
		delete pfilter;

		return nullptr;
	}
	std::get<MQTT_CLT_ID>(*connection) = pMqttClient;


	//pMqttClient->mqtt_conn_context = connection;

	//设置回调函数

	//设置MQTT连接参数


	//mqtt_set_resubscribe_handler(pMqttClient, resubscribe_topic);


	/*
	//连接Broker
	int rc = 0;
	dwTick = GetTickCount();
	while (true)
	{
		rc = mqtt_connect(pMqttClient);
		if (rc != MQTT_SUCCESS_ERROR)
		{

			if (GetTickCount() - dwTick > dwOpenTimeout)
			{
				FINFO(" TID {$}  {$} Failed 2 to connect to the MQTT server... {$} return : ", GetCurrentThreadId(), pszClientID, rc);
				pLock->Thread_UnLock();

				delete connection;
				delete pfilter;

				return nullptr;
			}
			else
			{
				FINFO(" TID {$}  {$} Failed 2 to connect to the MQTT server  Try again 20ms later...... {$} return : ", GetCurrentThreadId(), pszClientID, rc);
				//std::cout << CurrentDateTime() << " TID [" << GetCurrentThreadId() << "] Failed to connect to the MQTT server Try again 2s later..." << pszClientID << endl;
				Sleep(20);
			}

		}
		else
		{
			break;
		}
	}*/
	DWORD dwWaitTick = GetTickCount() - dwTick;
	FWARN("MQTT {$} try ConnMqtt Succecced Use Time ****** {$} ms*** wait: {$} ms***  TID ", pszClientID, GetTickCount() - dwTick, dwWaitTick, GetCurrentThreadId());
	std::get<MSG_LIST_ID>(*connection) = new mqtt_msg_list;

	// 创建POSIX信号量
	sem_t* semaphore = new sem_t;
	sem_init(semaphore, 0, 0);
	std::get<SEMAPHORE_HANDLE_ID>(*connection) = semaphore;

	std::get<THREAD_RUNNING_ID>(*connection) = true;  
	// 创建发送线程
	std::cout << "Preapre create MqttSendThreadFunc" << std::endl;
	pthread_t threadId;
	if (pthread_create(&threadId, NULL, MqttSendThreadFunc, connection) != 0) {
		std::cerr << "Failed to create MQTT send thread" << std::endl;
		std::get<THREAD_RUNNING_ID>(*connection) = false;
		// 错误处理...
		sem_destroy(semaphore);
		delete semaphore;
		pLock->Thread_UnLock();
		delete connection;
		return nullptr;
	}

	std::get<CONNECTED_HANDLE_ID>(*connection) = threadId;
	std::cout << "MqttSendThreadFunc thread id [" << threadId << "]" << std::endl;

	pLock->Thread_UnLock();
	std::cout << "MQTT " << pszClientID << " try ConnMqtt Succeeded. Use Time: "
		<< dwWaitTick << " ms" << std::endl;

	return connection;

}

//重置连接
LOGICDEVICE_API void ResetConnection(ccos_mqtt_connection* hConnection)
{
	if (hConnection == nullptr)
	{
		return;
	}
	FWARN(" Reset Mqtt Connection..", std::get<CLINET_ID_ID>(*hConnection));
	std::cout << CurrentDateTime() << std::get<CLINET_ID_ID>(*hConnection) << " Close Mqtt Connection.." << endl;
	mqtt_client* pconn = (mqtt_client*)std::get<MQTT_CLT_ID>(*hConnection);
	if (pconn != nullptr) {
		CcosLock* pLock = std::get<CONN_SEND_LOCK_ID>(*hConnection);
		pLock->Thread_Lock();
		//mqtt_do_reconnect(pconn);
		mqtt_disconnect(pconn);
		/*mqtt_set_resubscribe_handler(pconn, resubscribe_topic);
		mqtt_connect(pconn);*/
		mqtt_release(pconn);
		pconn = InnerConnect(hConnection);
		int rc = MQTT_SUCCESS_ERROR;// mqtt_connect(pconn);
		if (pconn != nullptr)
		{
			//std::get<MQTT_CLT_ID>(*hConnection) = pconn;
			//resubscribe_topic(pconn, pconn->mqtt_conn_context);
			//FWARN("mqtt_connect ret {$}  do mqtt resubscribe {$}", rc, std::get<CLINET_ID_ID>(*hConnection));
		}
		pLock->Thread_UnLock();
	}
}

//关闭并释放连接
LOGICDEVICE_API void CloseConnection(ccos_mqtt_connection* hConnection)
{
	if (!hConnection) return;

	auto clientId = std::get<CLINET_ID_ID>(*hConnection);
	std::cout << CurrentDateTime() << clientId << " Close Mqtt Connection.." << endl;

	std::get<THREAD_RUNNING_ID>(*hConnection) = false;

	sem_t* semaphore = std::get<SEMAPHORE_HANDLE_ID>(*hConnection);
	if (semaphore) {
		sem_post(semaphore);  // 发送信号量唤醒线程
	}

	pthread_t sendThread = std::get<CONNECTED_HANDLE_ID>(*hConnection);
	if (sendThread != 0) {
		// 等待线程退出，超时时间设为1秒
		struct timespec timeout;
		clock_gettime(CLOCK_REALTIME, &timeout);
		timeout.tv_sec += 1;

		int joinResult = pthread_timedjoin_np(sendThread, nullptr, &timeout);
		if (joinResult != 0) {
			std::cerr << CurrentDateTime() << " " << clientId
				<< " Warning: Send thread did not exit in time, force cancel" << std::endl;
			pthread_cancel(sendThread);  // 超时后强制终止（最后的备选方案）
			pthread_join(sendThread, nullptr);
		}
		std::get<CONNECTED_HANDLE_ID>(*hConnection) = 0;  // 重置线程ID
	}

	CcosLock* pLock = std::get<CONN_SEND_LOCK_ID>(*hConnection);	
	if (nullptr != pLock)
		pLock->Thread_Lock();
	else
		return;

	mqtt_client* pMqttClient = static_cast<mqtt_client*>(std::get<MQTT_CLT_ID>(*hConnection));
	if (pMqttClient) {
		mqtt_disconnect(pMqttClient);  // 断开连接
		mqtt_release(pMqttClient);     // 释放客户端实例
		pMqttClient = NULL;
		std::get<MQTT_CLT_ID>(*hConnection) = nullptr;
	}

	ccos_mqtt_msg_filter* pFilter = static_cast<ccos_mqtt_msg_filter*>(std::get<MSG_HOOK_ID>(*hConnection));
	if (pFilter) {
		std::cout << "Free Connection filter [" << (UINT64)pFilter << "] .." << std::endl;

		// 释放filter内部的资源对象
		delete std::get<FILTER_RES_OBJ_ID>(*pFilter);
		delete std::get<FILTER_RESPONS_OBJ_ID>(*pFilter);
		delete pFilter;
		std::get<MSG_HOOK_ID>(*hConnection) = nullptr;
	}

	mqtt_msg_list* pMsgList = std::get<MSG_LIST_ID>(*hConnection);
	if (pMsgList) {
		// 清理剩余未发送的消息
		while (!pMsgList->empty()) {
			ResDataObject* pMsg = pMsgList->front();
			pMsgList->pop_front();
			delete pMsg;  // 释放消息对象
		}
		delete pMsgList;
		std::get<MSG_LIST_ID>(*hConnection) = nullptr;
	}

	if (semaphore) {
		sem_destroy(semaphore);  // 销毁信号量
		delete semaphore;
		std::get<SEMAPHORE_HANDLE_ID>(*hConnection) = nullptr;
	}

	mqtt_topic_list* pTopicList = std::get<MQTT_TIPIC_LIST_ID>(*hConnection);
	if (pTopicList) {
		pTopicList->clear();  // 清空列表
		delete pTopicList;
		std::get<MQTT_TIPIC_LIST_ID>(*hConnection) = nullptr;
	}

	if (pLock) {
		pLock->Thread_UnLock();  // 确保锁被释放
		delete pLock;
		std::get<CONN_SEND_LOCK_ID>(*hConnection) = nullptr;
	}
	
	delete hConnection;

	std::cout << CurrentDateTime() << clientId << " Close Mqtt Connection over.." << endl;
}

//主动订阅主题
LOGICDEVICE_API int  SubscribeTopic(ccos_mqtt_connection* hConnection, const char* pszTopic, bool isShare)
{
	std::cout << "SubscribeTopic called. Topic: " << (pszTopic ? pszTopic : "null")
		<< ", isShare: " << (isShare ? "true" : "false") << std::endl;
	if (hConnection == nullptr)
	{
		std::cout << "SubscribeTopic error: hConnection is nullptr" << std::endl;
		return 0;
	}
	mqtt_client* pMqttClient = (mqtt_client*)std::get<MQTT_CLT_ID>(*hConnection);
	mqtt_topic_list* pTopicList =  std::get<MQTT_TIPIC_LIST_ID>(*hConnection);

	CcosLock* pLock = std::get<CONN_SEND_LOCK_ID>(*hConnection);
	pLock->Thread_Lock();

	pTopicList->push_back(std::string(pszTopic));
	//pMqttClient->subscribe(pszTopic, 1, opts);
	int rc = MQTT_SUCCESS_ERROR;
	DWORD dwTick = GetTickCount();
	int nTryTimes = 0;
	do
	{
		if (pTopicList->size() < 1)
		{
			const char* topicToSubscribe = pTopicList->back().c_str();
			int ret = mqtt_subscribe(pMqttClient, topicToSubscribe, MQTT_QOS, msgarrivd);
			//int ret = mqtt_subscribe(pMqttClient, pszTopic, MQTT_QOS, msgarrivd);
			FWARN("mqtt {$} Subscribe First {$} return {$} topic num {$}", std::get<CLINET_ID_ID>(*hConnection), pszTopic, ret, pTopicList->size());
			std::cout << "mqtt [" << std::get<CLINET_ID_ID>(*hConnection) << " ]Subscribe First " << pszTopic << endl;
		}
		else
		{
			const char* topicToSubscribe = pTopicList->back().c_str();
			int ret = mqtt_subscribe(pMqttClient, topicToSubscribe, MQTT_QOS, msgarrivd);
			//int ret = mqtt_subscribe(pMqttClient, pszTopic, MQTT_QOS, msgarrivd);
			FWARN("mqtt {$} Subscribe ReUse {$}  topic num {$}", std::get<CLINET_ID_ID>(*hConnection), pszTopic, ret, pTopicList->size());
			std::cout << CurrentDateTime() << " mqtt [" << std::get<CLINET_ID_ID>(*hConnection) << " ]Subscribe ReUse " << pszTopic << " ret: "<<ret <<endl;
		}

		if (rc != MQTT_SUCCESS_ERROR)
		{
			FWARN("try do mqtt reconnect ..");
			//mqtt_do_reconnect(pMqttClient);
			mqtt_disconnect(pMqttClient);
			//mqtt_set_resubscribe_handler(pMqttClient, resubscribe_topic);
			mqtt_release(pMqttClient);
			pMqttClient = InnerConnect(hConnection);
			int rc = MQTT_SUCCESS_ERROR;// mqtt_connect(pconn);
			if (pMqttClient != nullptr)
			{
				//std::get<MQTT_CLT_ID>(*hConnection) = pMqttClient;
				//resubscribe_topic(pMqttClient, pMqttClient->mqtt_conn_context);
				FINFO("Reconnected successfully, connection ready for subscriptions");
			}
			// 移除2秒延时：InnerConnect已等待CONNACK，WaitForSubscriptionComplete会确保订阅完成
		}
	} while (rc != MQTT_SUCCESS_ERROR && GetTickCount() - dwTick < 100);

	pLock->Thread_UnLock();

	return 0;
}

// 等待订阅完成（轮询检查ack_list，确保SUBACK已收到）
LOGICDEVICE_API bool WaitForSubscriptionComplete(ccos_mqtt_connection* hConnection, DWORD timeoutMs)
{
	if (hConnection == nullptr) {
		FERROR("WaitForSubscriptionComplete: hConnection is nullptr");
		return false;
	}

	mqtt_client_t* pMqttClient = (mqtt_client_t*)std::get<MQTT_CLT_ID>(*hConnection);
	if (pMqttClient == nullptr) {
		FERROR("WaitForSubscriptionComplete: pMqttClient is nullptr");
		return false;
	}

	const std::string clientId = std::get<CLINET_ID_ID>(*hConnection);
	FINFO("[WaitForSubscriptionComplete] {$} - Waiting for all subscriptions to complete (timeout: {$}ms)",
		clientId, timeoutMs);

	DWORD startTime = GetTickCount();
	const DWORD pollIntervalMs = 10; // 每10ms轮询一次
	int lastAckCount = -1;

	while (GetTickCount() - startTime < timeoutMs) {
		// 检查ack_handler_list中是否还有待处理的SUBACK
		// mqtt_ack_handler_number记录了ack_list中的条目数量
		int currentAckCount = pMqttClient->mqtt_ack_handler_number;

		// 如果ack数量变化，输出日志
		if (currentAckCount != lastAckCount) {
			FINFO("[WaitForSubscriptionComplete] {$} - Pending ack count: {$}", clientId, currentAckCount);
			lastAckCount = currentAckCount;
		}

		// 如果没有待处理的ack，说明所有订阅都已完成
		if (currentAckCount == 0) {
			DWORD elapsed = GetTickCount() - startTime;
			FINFO("[WaitForSubscriptionComplete] {$} - All subscriptions completed in {$}ms",
				clientId, elapsed);
			return true;
		}

		// 短暂休眠后继续轮询
		usleep(pollIntervalMs * 1000);
	}

	// 超时
	int finalAckCount = pMqttClient->mqtt_ack_handler_number;
	FERROR("[WaitForSubscriptionComplete] {$} - Timeout after {$}ms, still have {$} pending acks",
		clientId, timeoutMs, finalAckCount);
	return false;
}

// 等待Broker完成路由表更新 - 可配置延时
// 通过环境变量 MQTT_BROKER_ROUTING_DELAY_MS 配置，默认200ms
LOGICDEVICE_API DWORD WaitForBrokerRouting(const char* clientId)
{
	static int brokerDelayMs = -1; // 静态变量，只读取一次环境变量

	if (brokerDelayMs == -1) {
		const char* envDelay = getenv("MQTT_BROKER_ROUTING_DELAY_MS");
		if (envDelay != nullptr && atoi(envDelay) > 0) {
			brokerDelayMs = atoi(envDelay);
			FINFO("[WaitForBrokerRouting] Using custom delay from environment: {$}ms", brokerDelayMs);
		} else {
			brokerDelayMs = 100; // 默认100ms，足够Broker处理路由表更新
			FINFO("[WaitForBrokerRouting] Using default delay: {$}ms (set MQTT_BROKER_ROUTING_DELAY_MS to customize)", brokerDelayMs);
		}
	}

	if (clientId != nullptr) {
		FINFO("[WaitForBrokerRouting] {$} - Waiting {$}ms for Broker to process subscription routing",
			clientId, brokerDelayMs);
	}

	usleep(brokerDelayMs * 1000);
	return brokerDelayMs;
}

//主题订阅取消
LOGICDEVICE_API int  UnSubscribe(ccos_mqtt_connection* hConnection, const char* pszTopic)
{
	if (hConnection == nullptr)
	{
		return 0;
	}
	mqtt_client* pMqttClient = (mqtt_client*)std::get<MQTT_CLT_ID>(*hConnection);
	//if (!pMqttClient->is_connected()) {
	//	//FERROR( " MQTT connection lost at subscribe %s ", topic);
	//	return 0;
	//}
	//MQTTSubscribe_options opts = MQTTSubscribe_options_initializer;

	//auto ret = pMqttClient->unsubscribe(pszTopic);
	mqtt_topic_list* pTopicList = std::get<MQTT_TIPIC_LIST_ID>(*hConnection);

	CcosLock* pLock = std::get<CONN_SEND_LOCK_ID>(*hConnection);
	pLock->Thread_Lock();

	//MQTTAsync_responseOptions resp;
	pTopicList->remove(pszTopic);
	int rc = MQTT_SUCCESS_ERROR;
	DWORD dwTick = GetTickCount();
	int nTryTimes = 0;
	do
	{
		int ret = mqtt_unsubscribe(pMqttClient, pszTopic);
		FWARN("mqtt {$} Unsubscribe {$} return {$}", std::get<CLINET_ID_ID>(*hConnection), pszTopic, ret);
	} while (rc != MQTT_SUCCESS_ERROR && GetTickCount() - dwTick < 100);

	pLock->Thread_UnLock();
	return 2;
}

//往指定主题发送CCOS协议包整包，使用临时创建连接，仅发送，不接收
LOGICDEVICE_API int  PublishMsg(ResDataObject* pCmd, const char* pszTopic, const char* pszSenderName, DWORD dwTimeout)
{
	char pszClientID[256];
	snprintf(pszClientID, sizeof(pszClientID), "TEMP_%s_%d_0X%08lX",
		(pszSenderName == nullptr) ? "ANONYMOUS" : pszSenderName,
		getpid(), GetCurrentThreadId());

	ccos_mqtt_connection* connObj = NewConnection(pszClientID, [](ResDataObject*, const char*, void* conn) {
		});
	mqtt_client* pConn = (mqtt_client*)std::get<MQTT_CLT_ID>(*connObj);

	PacketAnalizer::UpdatePacketTopic(pCmd, pszTopic, pszClientID);

	//MQTTAsync_responseOptions resp = MQTTAsync_responseOptions_initializer;
	string pLoad = pCmd->encode();
	
	int rc = PublishActionWithoutLock(pLoad, pszTopic, pConn, pszSenderName);
	
	std::cout << "CLT [" << pszClientID << "] at " << CurrentDateTime() << " Publish  to [" << pszTopic << "] result " << rc << endl;
	FDEBUG("CLT {$} Publish  to {$} send  result {$}", pszClientID, pszTopic, rc);

	CloseConnection(connObj);
	return rc;
}

//往指定主题发送CCOS协议包整包，使用已创建的连接发送
LOGICDEVICE_API int  PublishAction(ResDataObject* pAction, const char* pszTopic, ccos_mqtt_connection* hConnection, DWORD dwTimeout)
{
	if (hConnection == nullptr)
	{
		FDEBUG("Who ????? Publish to {$} Action Body: {$}", pszTopic, pAction->encode());
		std::cout << CurrentDateTime() << "Who ????? " << "Publish to [" << pszTopic << "] Action Body: "<< pAction->encode() << endl; //<< pAction->encode() 
		return 0;
	}
	std::cout << CurrentDateTime()<<" " << std::get<CLINET_ID_ID>(*hConnection) << " Publish Action to  [" << pszTopic << "] Action Body: " << endl; //<< pAction->encode() 

	string topic = pszTopic;
	if (topic.length() <= 0)
	{
		FWARN("ignore empty topic packet {$}", pAction->encode());
		return 2;
	}

	mqtt_client* pMqttClient = (mqtt_client*)std::get<MQTT_CLT_ID>(*hConnection);
	//if (!pMqttClient->is_connected()) {
	//	//FERROR( " MQTT connection lost at subscribe %s ", topic);
	//	return 0;
	//}
	std::string client_id = std::get<CLINET_ID_ID>(*hConnection);
	CcosLock* pLock = std::get<CONN_SEND_LOCK_ID>(*hConnection);
	pLock->Thread_Lock();
	FDEBUG("{$} Publish Action to {$} Action Body: {$}", std::get<CLINET_ID_ID>(*hConnection), pszTopic, pAction->encode());

	//string org_publisher = PacketAnalizer::GetPacketPublisher(pAction);
	//if(org_publisher.length() <= 0)
	PacketAnalizer::UpdatePacketTopic(pAction, pszTopic, client_id.c_str() );

	
	ResDataObject* pPacket = new ResDataObject();
	mqtt_msg_list* pList = std::get<MSG_LIST_ID>(*hConnection);
	pPacket->add(pszTopic, pAction->encode());
	//FDEBUG(" {$} Try push packet to Send list: {$}", client_id, pPacket->encode());
	pList->push_back(pPacket);
	sem_post(std::get<SEMAPHORE_HANDLE_ID>(*hConnection));
	std::cout << "try publish " << pAction->encode() << endl;


	//pMqttClient->publish(pszTopic, pAction->encode());

	/*
	//pConn->publish(pszTopic, pCmd->encode());
	std::cout << "try publish " << pAction->encode() << endl;
	const char* pLoad = pAction->encode();
	int len = strlen(pLoad);
	//MQTTAsync_responseOptions resp;
	//MQTTAsync_responseOptions resp = MQTTAsync_responseOptions_initializer;
	mqtt_message_t msg;
	memset(&msg, 0, sizeof(msg));
	msg.payload = (void*)pLoad;
	msg.qos = MQTT_QOS;
	msg.payloadlen = len;

	int rc = MQTT_SUCCESS_ERROR;
	DWORD dwTick = GetTickCount();
	int nTryTimes = 0;
	do
	{
		rc = mqtt_publish(pMqttClient, pszTopic, &msg);
		nTryTimes++;
		if (rc != MQTT_SUCCESS_ERROR)
		{
			FINFO("try mqtt_publish ret {$}  do mqtt reconnect .. {$}",rc, client_id);
			//mqtt_do_reconnect(pMqttClient);
			mqtt_disconnect(pMqttClient);


			rc = mqtt_connect(pMqttClient);
			if (rc == MQTT_SUCCESS_ERROR)
			{
				resubscribe_topic(pMqttClient, pMqttClient->mqtt_conn_context);
				FINFO("mqtt_connect ret {$}  do mqtt resubscribe {$}", rc, client_id);
			}
			Sleep(2);
		}

	} while (rc != MQTT_SUCCESS_ERROR && GetTickCount() - dwTick < dwTimeout);
	//std::cout << "CLT [" << pszClientID << "] at " << CurrentDateTime() << " Publish  to [" << pszTopic << "] result " << rc << endl;
	FINFO("CLT {$} PublishAction  to {$} send Times {$} result {$}", client_id, pszTopic, nTryTimes, rc);

	//int rc = MQTTAsync_send(pMqttClient, pszTopic, len, pLoad, 0, 0, &resp);
	//std::cout << "CLT [" << client_id << "] at " << CurrentDateTime() << " Use mqtt_client " << (UINT64)pMqttClient << " Publish  to [" << pszTopic << "] Send  result " << rc << endl;
	if (rc < 0)
	{
		FERROR("{$} PublishAction failed {$} body: {$}", client_id, pszTopic, rc, pLoad);
		//std::cout << " ErrorCode " << rc << " Send Msg : " << pLoad << endl;
		pLock->Thread_UnLock();
		return rc;
	}*/
	//MQTTResponse resp = MQTTClient_publish5(pMqttClient, pszTopic, strlen(pLoad), pLoad, 0, 0, NULL, NULL);
	//std::cout << "CLT [" << client_id << "] at " << CurrentDateTime()  << " Publish to [" << pszTopic << "] result " << resp.reasonCode << endl;
	FDEBUG("CLT {$} PublishAction  to {$} Send List  has {$} packets ", client_id, pszTopic, pList->size());

	pLock->Thread_UnLock();
	return 2;
}

//往指定主题发送Action包，携带参数，并指定答复的Topic，异步模式处理Resp，需要在对应连接的回调中处理
int  PublishActionWithoutLock(string& message, const char* pszTopic, mqtt_client* pMqttClient, std::string client_id, mqtt_qos_t qos)
{
	const int dwTimeout = 500;
	if (pMqttClient == nullptr)
	{
		FERROR("Who ????? Publish 2 to {$} Action {$} Body: {$}", pszTopic, message);
		return 0;
	}

	FDEBUG("{$} Publish with qos[{$}] Action 2 to {$}  Body: {$} ", client_id, (int)qos, pszTopic, message);

	std::string messageCopy = message;
	const char* pLoad = messageCopy.c_str();
	int len = messageCopy.length();

	mqtt_message_t msg;
	memset(&msg, 0, sizeof(msg));
	msg.payload = (void*)pLoad;
	msg.payloadlen = len;
	msg.qos = qos;
	msg.retained = 0;

	int rc = MQTT_SUCCESS_ERROR;
	DWORD dwTick = GetTickCount();
	int nTryTimes = 0;

	do
	{
		rc = mqtt_publish(pMqttClient, pszTopic, &msg);
		nTryTimes++;
		if (rc != MQTT_SUCCESS_ERROR)
		{
			FWARN("try mqtt_publish ret {$}  do mqtt reconnect .. {$}", rc, client_id);
			
			//mqtt_do_reconnect(pMqttClient);
			ccos_mqtt_connection* hConnection = (ccos_mqtt_connection*)pMqttClient->mqtt_conn_context;
			mqtt_disconnect(pMqttClient);
			/*mqtt_set_resubscribe_handler(pMqttClient, resubscribe_topic);
			mqtt_connect(pMqttClient);*/
			mqtt_release(pMqttClient);
			pMqttClient = nullptr;
			usleep(10000 * 1000);
			pMqttClient = InnerConnect(hConnection);
			//int rc = MQTT_SUCCESS_ERROR;// mqtt_connect(pconn);
			if (pMqttClient != nullptr)
			{
				break;
				//std::get<MQTT_CLT_ID>(*hConnection) = pMqttClient;
				//resubscribe_topic(pMqttClient, hConnection);
				//FWARN("mqtt_connect ret {$}  do mqtt resubscribe {$}", rc, std::get<CLINET_ID_ID>(*hConnection));
			}

		}
	} while (rc != MQTT_SUCCESS_ERROR && (nTryTimes <= 2  || GetTickCount() - dwTick < dwTimeout));

	cout << CurrentDateTime() << " CLT " << client_id.c_str() << " PublishAction  to " << pszTopic << " send Times " << nTryTimes << " result " << rc << endl;
	FDEBUG("CLT {$} PublishAction  to {$} send Times {$} result {$}", client_id, pszTopic, nTryTimes, rc);

	if (rc < 0)
	{
		FERROR("{$} PublishAction {$} failed {$} body: {$}", client_id, pszTopic, rc, pLoad);
		return rc;
	}
	return 2;
}



/*
/// <summary>
/// 往指定主题发送Action包，携带参数，并指定答复的Topic，同步等待resp，
/// 超时没收到应答返回失败，
/// 复用链接时须小心，该函数会接管回调函数，结束后恢复
/// </summary>
/// <param name="pAction">要发送的命令Action名</param>
/// <param name="pContext">命令携带的参数</param>
/// <param name="pszTopic">要发送的目标topic</param>
/// <param name="pszRespTopic">本次请求的应答接收Topic</param>
/// <param name="resObj">应答返回的参数结果</param>
/// <param name="dwWaitTime">等待超时时间</param>
/// <param name="hConnection">复用的MQTT链接句柄</param>
/// <param name="onmsg">复用的链接的消息处理函数</param>
/// <returns>成功返回2，其他返回错误码</returns>
LOGICDEVICE_API int  ActionAndRespWithConnection(ccos_mqtt_connection* hConnection, const char* pAction, ResDataObject& req, ResDataObject* pContext, const char* pszTopic, const char* pszRespTopic,
	ResDataObject& resObj, DWORD dwWaitTime)
{
	std::cout << CurrentDateTime() << std::get<CLINET_ID_ID>(*hConnection) << "\nAction2 : " << pAction << "  to " << pszTopic << endl;// << "\n Action Body: " << " Context " << pContext->encode() << endl << endl; //<< req.encode() 

	if (pszRespTopic != nullptr)
		PacketAnalizer::UpdateContextTopic(pContext, pszRespTopic);

	if (hConnection == nullptr)
	{
		return 0;
	}

	mqtt_client* pMqttClient = (mqtt_client*)std::get<MQTT_CLT_ID>(*hConnection);
	//if (!pMqttClient->is_connected()) {
	//	//FERROR( " MQTT connection lost at subscribe %s ", topic);
	//	return 0;
	//}
	HANDLE hEvent = CreateEvent(NULL, FALSE, FALSE, NULL);

	//SubscribeTopic()
	auto action_func = [&resObj, pszRespTopic, pMqttClient, hConnection, hEvent](void* context, char* topicName, int topicLen, MQTTClient_message* message) {

		string topic = topicName;//msg->get_topic().c_str();

		if (strcmp(pszRespTopic, topic.c_str()) == 0) {
			std::cout << CurrentDateTime() << std::get<CLINET_ID_ID>(*hConnection) << " get Right Resp by " << topic << endl;
			//应答到了，处理
			ResDataObject req;
			req.decode((const char*)message->payload);
			PacketAnalizer::GetPacketContext(&req, resObj);
			SetEvent(hEvent);

			//处理结束 将函数指针换回去
			void* oldFuncPointer = std::get<MQTT_MSG_ARRIVED_ID>(*hConnection);
			//pMqttClient->set_message_callback(*(mqtt::async_client::message_handler*)oldFuncPointer);
			MQTTAsync_setCallbacks(pMqttClient, hConnection, connlost, msgarrvd, NULL);
		}
		else
		{
			//MQTTClient_messageArrived *orgfunc = (MQTTClient_messageArrived*)std::get<MQTT_MSG_ARRIVED_ID>(*hConnection);
			(msgarrvd)(context, topicName, topicLen, message);
		}
	};

	//接管回调
	// pMqttClient->set_message_callback(action_func);
	MQTTClient_setCallbacks(pMqttClient, hConnection, connlost, (MQTTClient_messageArrived*)&action_func, delivered);

	//pMqttClient->subscribe(pszRespTopic, 1);
	MQTTSubscribe_options opts = MQTTSubscribe_options_initializer;
	MQTTProperties prop = MQTTProperties_initializer;

	std::cout << "mqtt 【" << std::get<CLINET_ID_ID>(*hConnection) << " 】Subscribe " << pszTopic << endl;

	//pMqttClient->subscribe(pszTopic, 1, opts);
	MQTTClient_subscribe5(pMqttClient, pszTopic, 1, NULL, NULL);

	PacketAnalizer::UpdatePacketTopic(&req, pszTopic, std::get<CLINET_ID_ID>(*hConnection));

	//pMqttClient->publish(pszTopic, req.encode());
	MQTTClient_message pubmsg = MQTTClient_message_initializer;
	MQTTClient_message* m = NULL;
	MQTTClient_deliveryToken dt;
	MQTTProperty property;

	property.identifier = MQTTPROPERTY_CODE_USER_PROPERTY;
	property.value.data.data = "test user property";
	property.value.data.len = (int)strlen(property.value.data.data);
	property.value.value.data = "test user property value";
	property.value.value.len = (int)strlen(property.value.value.data);

	MQTTProperties properties = MQTTProperties_initializer;
	MQTTProperties_add(&properties, &property);

	//pConn->publish(pszTopic, pCmd->encode());
	const char* pLoad = req.encode();
	MQTTResponse resp = MQTTClient_publish5(pMqttClient, pszTopic, strlen(pLoad), pLoad, 0, 0, &properties, &dt);
	std::cout << "CLT [" << std::get<CLINET_ID_ID>(*hConnection) << "] at " << CurrentDateTime() << " Publish x to [" << pszTopic << "] result " << resp.reasonCode << endl;

	DWORD ret = WaitForSingleObject(hEvent, dwWaitTime);
	if (ret == WAIT_OBJECT_0) {
		//等到应答了
		return 2;
	}
	return 0;


}
*/

/// <summary>
/// 往指定主题发送Action包，携带参数，复用Resp的Topic，同步等待resp，超时没收到应答返回失败，
/// </summary>
/// <param name="hConnection"></param>
/// <param name="pAction"></param>
/// <param name="pContext"></param>
/// <param name="pszTopic"></param>
/// <param name="resObj"></param>
/// <param name="hEvent"></param>
/// <param name="dwWaitTime"></param>
/// <returns></returns>
LOGICDEVICE_API int  ActionAndRespWithConnDefalt(ccos_mqtt_connection* hConnection, const char* pAction, ResDataObject& req, ResDataObject* pContext,
	const char* pszTopic, ResDataObject& resObj, DWORD dwWaitTime)
{
	// QoS=2消息握手需要时间（PUBLISH→PUBREC→PUBREL→PUBCOMP）
	// 连续发送时增加50ms延时，避免握手状态冲突
	const std::string client_id = std::get<CLINET_ID_ID>(*hConnection);
	static std::map<std::string, DWORD> lastSendTime;
	DWORD now = GetTickCount();
	DWORD elapsed = (lastSendTime.count(client_id) > 0) ? (now - lastSendTime[client_id]) : 1000;
	if (elapsed < 50) {
		DWORD sleepMs = 50 - elapsed;
		FINFO("[ActionAndRespWithConnDefalt] {$} - Sleeping {$}ms to avoid QoS handshake conflict", client_id, sleepMs);
		usleep(sleepMs * 1000);
	}
	lastSendTime[client_id] = GetTickCount();

	std::cout << CurrentDateTime() << std::get<CLINET_ID_ID>(*hConnection) << "\nAction : " << pAction << "  to " << pszTopic <<  "\n Action Body: " << " Context " << pContext->encode() << endl << endl; //<< req.encode()

	if (hConnection == nullptr)
	{
		return 0;
	}
	DWORD dwTick = GetTickCount();

	mqtt_client* pMqttClient = (mqtt_client*)std::get<MQTT_CLT_ID>(*hConnection);
	std::shared_ptr<LinuxEvent> hEvent = LinuxEvent::CreateEvent(LinuxEvent::AUTO_RESET, false);

	ResDataObject resContext;
	string strResObject;
	char* pszPad = 0;
	char* pszPad2 = 0;

	auto func = [pAction, hConnection, &pszPad, &resObj, &pszPad2, dwTick, dwWaitTime](ResDataObject* rsp) -> bool {
		FINFO("{$} try check action resp {$} compared with {$}", std::get<CLINET_ID_ID>(*hConnection), pAction, PacketAnalizer::GetPacketKey(rsp));
		std::cout << CurrentDateTime() << std::get<CLINET_ID_ID>(*hConnection) << " try check action resp [" << pAction  << "] compared with " << PacketAnalizer::GetPacketKey(rsp).c_str() << endl;
		if (PacketAnalizer::GetPacketType(rsp) == PACKET_TYPE_RES &&
			strcmp(pAction, PacketAnalizer::GetPacketKey(rsp).c_str()) == 0)
		{
			if (GetTickCount() - dwTick < dwWaitTime)
			{
				FDEBUG("ActionAndRespWithConnDefalt Packet Content {$} content {$}", PacketAnalizer::GetPacketKey(rsp), rsp->encode());
				resObj = *rsp;
				return true;
			}
			else
			{
				FWARN("ActionAndRespWithConnDefalt Packet Content {$} Timeout content {$}", PacketAnalizer::GetPacketKey(rsp), rsp->encode());
				return false;
			}

			std::cout << "  : " << PacketAnalizer::GetPacketKey(rsp) << " content " << rsp->encode() << endl;
		}
		FDEBUG("ActionAndRespWithConnDefalt what ?  {$} content {$}", PacketAnalizer::GetPacketKey(rsp), rsp->encode());
		std::cout << "ActionAndRespWithConnDefalt  what ?  " << PacketAnalizer::GetPacketKey(rsp) << endl;
		return false;
	};

	CcosLock* pLock = std::get<CONN_SEND_LOCK_ID>(*hConnection);
	pLock->Thread_Lock();

	FDEBUG("{$} ActionAndRespWithConnDefalt {$} to Topic {$} body {$} ", std::get<CLINET_ID_ID>(*hConnection), pAction, pszTopic, pContext->encode());


	ccos_mqtt_msg_filter* pfilter = (ccos_mqtt_msg_filter*)std::get<MSG_HOOK_ID>(*hConnection);
	ResDataObject *resActions = std::get<FILTER_RES_OBJ_ID>(*pfilter);

	resActions->add(pAction, pszTopic);

	std::get<FILTER_HANDLE_ID>(*pfilter) = hEvent;
	
	std::get<FILTER_FUNC_ID>(*pfilter) = func;

	PacketAnalizer::UpdatePacketTopic(&req, pszTopic, std::get<CLINET_ID_ID>(*hConnection).c_str());

	const char* pLoad = req.encode();
	ResDataObject* pPacket = new ResDataObject();
	mqtt_msg_list* pList = std::get<MSG_LIST_ID>(*hConnection);
	pPacket->add(pszTopic, pLoad);
	FDEBUG(" {$} Try push packet to Send list: {$}", client_id, pPacket->encode());
	pList->push_back(pPacket);
	sem_post(std::get<SEMAPHORE_HANDLE_ID>(*hConnection));
	/*
	mqtt_message_t msg;
	memset(&msg, 0, sizeof(msg));
	msg.payload = (void*)pLoad;
	msg.qos = MQTT_QOS;
	msg.payloadlen = strlen(pLoad);

	//int rc = mqtt_publish(pMqttClient, pszTopic, &msg);
	int rc = MQTT_SUCCESS_ERROR;
	dwTick = GetTickCount();
	int nTryTimes = 0;
	do
	{
		rc = mqtt_publish(pMqttClient, pszTopic, &msg);
		nTryTimes++;
		if (rc != MQTT_SUCCESS_ERROR)
		{
			FINFO("try mqtt_publish ret {$}  do mqtt reconnect 3.. {$}", rc, client_id);
			//mqtt_do_reconnect(pMqttClient);
			mqtt_disconnect(pMqttClient);

			rc = mqtt_connect(pMqttClient);
			if (rc == MQTT_SUCCESS_ERROR)
			{
				resubscribe_topic(pMqttClient, pMqttClient->mqtt_conn_context);
				FINFO("mqtt_connect ret {$}  do mqtt resubscribe {$}", rc, client_id);
			}
			Sleep(2);
		}
	} while (rc != MQTT_SUCCESS_ERROR && GetTickCount() - dwTick < 3);

	if (nTryTimes >= 1)
		FINFO("CLT {$} Publish  to {$} send Times {$} result {$}", std::get<CLINET_ID_ID>(*hConnection), pszTopic, nTryTimes, rc);

	*/
	pLock->Thread_UnLock();

	dwTick = GetTickCount() - dwTick;
	//FINFO("CLT {$} Publish to {$} Use TID {$} Use Time {$} result {$} ", std::get<CLINET_ID_ID>(*hConnection), pszTopic, GetCurrentThreadId(), dwTick, rc);
	std::cout << "CLT [" << std::get<CLINET_ID_ID>(*hConnection) << "] at " << CurrentDateTime() << " Publish to [" << pszTopic << "] Use TID [" << GetCurrentThreadId()  << "] Use Time[" << dwTick << "]ms" << endl;

	dwTick = GetTickCount();
	DWORD ret = hEvent->Wait(dwWaitTime);
	if (ret) {
		//等到应答了
		dwTick = GetTickCount() - dwTick;

		pLock->Thread_Lock();
		ResDataObject* pResp = std::get<FILTER_RESPONS_OBJ_ID>(*pfilter);
		for (int x = 0; x < pResp->size(); x++)
		{
			ResDataObject r = (*pResp)[x];
			if (string(pAction) == string(pResp->GetKey(x)) && PacketAnalizer::GetPacketIdx(&req) == PacketAnalizer::GetPacketIdx(&r))
			{
				resObj = r;
				pResp->eraseOneOf(pAction, x);
				break;
			}
		}
		pLock->Thread_UnLock();
		std::cout << "CLT [" << std::get<CLINET_ID_ID>(*hConnection) << "] at " << CurrentDateTime() << " try [" << pszTopic << "] getresp ok Use Time[" << dwTick << "]ms" << endl;
		FDEBUG("CLT {$} try {$}  getresp ok Use Time {$}", std::get<CLINET_ID_ID>(*hConnection), pszTopic, dwTick);
		//std::get<FILTER_RES_OBJ_ID>(*pfilter) = nullptr;
		return 2;
	}
	//std::get<FILTER_FUNC_ID>(*pfilter) = nullptr;
	//resObj.decode(strResObject.c_str());
	FERROR("CLT {$} try {$} getresp timeout ", std::get<CLINET_ID_ID>(*hConnection), pszTopic );
	std::cout << "CLT [" << std::get<CLINET_ID_ID>(*hConnection) << "] at " << CurrentDateTime() << "try [" << pszTopic << "] " << endl;
	//std::get<FILTER_RES_OBJ_ID>(*pfilter) = nullptr;
	return 0;

}

/// <summary>
/// 新建MQTT连接发送Ation并等待应答
/// </summary>
/// <param name="pAction"></param>
/// <param name="pContext"></param>
/// <param name="pszTopic"></param>
/// <param name="pszRespTopic"></param>
/// <param name="resObj"></param>
/// <param name="dwWaitTime"></param>
/// <param name="pszSenderName"></param>
/// <returns></returns>
LOGICDEVICE_API int  ActionAndResp(const char* pAction, ResDataObject* pContext, const char* pszTopic, const char* pszRespTopic, ResDataObject& resObj,
	DWORD dwWaitTime, const char* pszSenderName)
{
	ResDataObject req;

	if (pszRespTopic != nullptr)
		PacketAnalizer::UpdateContextTopic(pContext, pszRespTopic);

	//临时创建连接并发送和接收应答
	char pszClientID[256];
	snprintf(pszClientID, sizeof(pszClientID), "TEMP_%s_%d_0X%08lX",
		(pszSenderName == nullptr) ? "ANONYMOUS" : pszSenderName,
		getpid(), GetCurrentThreadId());
	sem_t sem;
	if (sem_init(&sem, 0, 0) != 0) {
		return 0; // 如果初始化失败，返回0
	}

	std::cout << " ActionAndResp->NewConnection " << endl;
	ccos_mqtt_connection* connObj = NewConnection(pszClientID, [&resObj, &sem](ResDataObject* req, const char* topic, void* conn) {
		//应答到了，处理
		PacketAnalizer::GetPacketContext(req, resObj);
		sem_post(&sem);
		});

	////发布消息，并等待应答
	PublishAction(&req, pszTopic, connObj);

	// 等待应答或超时
	struct timespec ts;
	clock_gettime(CLOCK_REALTIME, &ts);
	ts.tv_sec += dwWaitTime / 1000; // 转换为秒
	ts.tv_nsec += (dwWaitTime % 1000) * 1000000; // 转换为纳秒

	// 等待信号量或者超时
	int ret = 0;
	while (ret == 0) {
		ret = sem_timedwait(&sem, &ts); // 超时或者接收到信号量
		if (ret == -1 && errno == ETIMEDOUT) {
			// 超时
			sem_destroy(&sem);
			CloseConnection(connObj);
			return 0; // 超时返回0
		}
	}

	// 等到应答了
	sem_destroy(&sem); // 销毁信号量
	CloseConnection(connObj);
	return 2; // 返回2表示应答成功
}