resistancemonitor/4.Data/SuperHandler.cpp

#include "stdafx.h"
#include "SuperHandler.h"
#include "LibEventTCPClient.h"
#include <Simplelog.h>
#include "AppService.h"
#include <jsonxx.h>
#include "MonitorObject.h"

using namespace jsonxx;

CSuperHandler::CSuperHandler()
{

}

CSuperHandler::~CSuperHandler()
{

}

BOOL CSuperHandler::HandlePack(LPBYTE pPack, int nPackLen, SOCKADDR* psa, void* pClientContext /*= NULL*/)
{
	auto pClient = (CLibEventTCPClient*)pClientContext;

	//解析
	if (!CHjDataConver::conver_recvpack(pPack, nPackLen))
		return FALSE;

	LPHJDATAHEAD2 lpHead = (LPHJDATAHEAD2)pPack;
	if (lpHead->protocol == E_ZL_PROTOCAL::ZL_KEEP)
		return TRUE;
	//SPDLOG_INFO("recv {}:{} : {}", m_strServerIP, m_nServerPort, CSimpleLog::GetHexString2(pPack, nPackLen));


	BOOL bRet = FALSE;
	switch (lpHead->protocol)
	{
	case E_ZL_PROTOCAL::ZL_MO:
	case E_ZL_PROTOCAL::ZL_MP:
	case E_ZL_PROTOCAL::ZL_RESIST_DATA:
	case E_ZL_PROTOCAL::ZL_RESIST_REAL:
	case E_ZL_PROTOCAL::ZL_TEMP_HUMI:
	case E_ZL_PROTOCAL::ZL_ALARM_SET:
	case E_ZL_PROTOCAL::ZL_ALARM_DATA:
	case E_ZL_PROTOCAL::ZL_ALARM_ACK:
	case E_ZL_PROTOCAL::ZL_ALARM_HANDLE:
	case E_ZL_PROTOCAL::ZL_SVG:
	case E_ZL_PROTOCAL::ZL_REFER:
	case E_ZL_PROTOCAL::ZL_MOVE:
	case E_ZL_PROTOCAL::ZL_ALARM_UNACK:
		bRet = RecvPack(lpHead->packno, lpHead->protocol);  //发出确认包
		break;
	case E_ZL_PROTOCAL::DL_SUB_NOTIFY:
		HandleSubNotify(lpHead, (char*)pPack + sizeof(HJDATAHEAD2), lpHead->len);
		break;
	case E_ZL_PROTOCAL::DL_SYNC:
		CAppService::Instance()->SetSycnSuperTime();
		break;
	case E_ZL_PROTOCAL::DL_ALARM_ACK:
	case E_ZL_PROTOCAL::DL_ALARM_HANDLE:
		break;
	default:
		CSimpleLog::Warn(fmt::format("收到未实现的命令:{}", uint8_t(lpHead->protocol)).c_str());
		break;
	}

	return bRet;
}

BOOL CSuperHandler::RecvPack(uint32_t packno, E_ZL_PROTOCAL protocal)
{
	m_pFather->recvcount++;
	//SPDLOG_INFO("recv 协议号:{} packno:{})", uint8_t(protocal), packno);
	std::lock_guard<std::mutex> lock(m_mtx);
	for (auto it = m_task.begin(); it != m_task.end(); it++)
	{
		auto p = *it;
		if (p->packno == packno && p->protocal == protocal)
		{
			m_task.erase(it);
			p->join();
		}
		return TRUE;
	}
	return FALSE;
}

BOOL CSuperHandler::HandleSubNotify(LPHJDATAHEAD2 lpHead, char* json, int json_len)
{
	Object obj;
	obj.parse(string(json, json_len));
	if (obj.empty()) return FALSE;
	auto momp = obj.get<string>("momp");
	if (momp.empty()) return FALSE;
	string imei; int idx;
	CMonitorObjectMng::Instance()->MOMP2IMEI(momp, imei, idx);
	return CAppService::Instance()->GetHandle()->SendMsgToDevice(imei.c_str());
}

BOOL CSuperHandler::ReSend()
{
	if (m_task.size() == 0) return TRUE;
	std::lock_guard<std::mutex> lock(m_mtx);
	if (m_task.size() > 0)
	{
		auto& it = m_task.front();
		auto ret = m_pClient->Insert(it->pdata, it->nlen);
		if (ret)
		{
			m_pFather->sendlen += it->nlen;
			m_pFather->sendcount++;
		}
		else
			m_pFather->sendfail++;
		return ret;
	}
	return FALSE;
}

bool CSuperHandler::Send(uint8_t* data, uint32_t len)
{
	return m_pClient->Insert(data, len);
}

void CSuperHandler::Insert(LPBYTE pPack, int nPackLen, uint32_t packno, E_ZL_PROTOCAL protocal, bool r)
{
	if (m_task.size() > 1000)
		SPDLOG_WARN("待发送数据队列异常:{}", m_task.size());

	//if (nPackLen > 1500) SPDLOG_INFO("send e packno:{})", packno);
	auto ret = m_pClient->Insert(pPack, nPackLen);
	if (ret == false)
	{
		m_pFather->sendfail++;
		SPDLOG_WARN("send fail. packno:{})", packno);
	}
	else
	{
		m_pFather->sendcount++;
		m_pFather->sendlen += nPackLen;
	}
	//加入队列并发送
	if (r || ret == false)
	{
		CSendTask* pTask = new CSendTask(pPack, nPackLen, packno, protocal, r);
		std::lock_guard<std::mutex> lock(m_mtx);
		m_task.emplace_back(pTask);
		if (m_task.size() > 1500)
		{
			auto it = m_task.front();
			m_task.pop_front();
			it->join();
		}
	}
}