midware_ws/NetLib/wsClient/NetClient.cpp

#include "stdafx.h"
#include <boost\bind.hpp>
#include <chrono>
#include <thread>
#include <boost/asio.hpp>
#include "NetClient.h"
#include "PduEntity.h"
#include "PduEntityHead.h"
#include "PduDataFormat.h"
#include "Logger.h"


CNetClient::CNetClient(): thread_(nullptr), m_pSchedulerHeart(nullptr), m_pSchedulerAutoConn(nullptr), m_IsConnSuccess(false)
{
}
CNetClient::~CNetClient()
{
}
void CNetClient::setLinkStateChanged(LinkStateChangedFuncCB funcCB)
{
	if (funcCB != nullptr&&this->m_LinkStateChangedFuncCB == nullptr) {
		this->m_LinkStateChangedFuncCB = funcCB;
	}
}

void CNetClient::setRecvCommand(RecvComdFuncCB funcCB)
{
	if (funcCB != nullptr&&this->m_RecvComdFuncCB == nullptr) {
		this->m_RecvComdFuncCB = funcCB;
	}
}

bool CNetClient::CreatePduClient(CString a_strFarIp, int a_nFarPort, PDU_DEV_TYPE a_nLocalType, int a_nLocalId, bool IsAutoReconnect)
{
	char logName[128];
	snprintf(logName, 128, "wsClient_%s", devTypeToStr(a_nLocalType).c_str());
	if (Logger::GetInstance()->Init("./log/", "wsClient_" + devTypeToStr(a_nLocalType) + "_log")) {
		Logger::GetInstance()->Start();
	}
	if (!CPduDataFormat::getInstance()->Load("./PDUFormat.ini")) {
		LOG_DEBUG("PDU配置文件加载失败");
	}
	init();
	m_IsAutoReconnect = IsAutoReconnect;
	return ConnectToServer(a_strFarIp, a_nFarPort, a_nLocalType, a_nLocalId);
}

bool CNetClient::ClosePduClient(PDU_DEV_TYPE a_nFarType, int a_nFarId)
{
	m_IsAutoReconnect = false;
	KillTimerListen();
	KillTimerConn();
	close();
	terminate();
	LOG_DEBUG("Pdu Client关闭完成");
	return true;
}

bool CNetClient::Send(CPduEntity * a_pCmd, PDU_DEV_TYPE a_nDestType, int a_nDestId)
{
	a_pCmd->SetLocalDevInfo(m_DevType, m_DevId);    // 填充发送方信息
	a_pCmd->SetPeerDevInfo(a_nDestType, a_nDestId);  // 填充接收方信息
	// 生成要发送的数据缓冲区
	UCHAR szBuf[PDU_HEAD_LEN + PDU_MAX_DATA_LEN];
	ZeroMemory(szBuf, PDU_HEAD_LEN + PDU_MAX_DATA_LEN);
	int nLen = a_pCmd->CreatePackge(szBuf);
	return __SendBuf(szBuf, nLen);
}

void CNetClient::SetLinkContent(PDU_LINK_STATE a_nLinkInfo)
{
	PduLinkContent linkContent;
	memset(&linkContent, 0, sizeof(PduLinkContent));
	linkContent.nLocalType = m_DevType;
	linkContent.nLocalId = m_DevId;
	linkContent.nLinkState = a_nLinkInfo;
	lstrcpy(linkContent.szFarIp, uri.c_str());

	if (this->m_LinkStateChangedFuncCB != nullptr) {
		this->m_LinkStateChangedFuncCB(linkContent);
	}
}

void CNetClient::SetTimerListen()
{
	// 注册成功启用心跳
	if (m_pSchedulerHeart == nullptr) {
		m_pSchedulerHeart = std::make_unique<CScheduler>(ios);
		m_pSchedulerHeart->setOnTimeFuncCB(std::bind(&this_type::onTime, this, std::placeholders::_1));
	}
	m_pSchedulerHeart->setTimer(PDU_LISTEN_TIMER, VAL_LISTEN_INTERVAL);
	LOG_DEBUG("ws开启心跳");
}

void CNetClient::KillTimerListen()
{
	if (m_pSchedulerHeart != nullptr) {
		m_pSchedulerHeart->killTimer();
	}
	LOG_DEBUG("ws关闭心跳");
}

void CNetClient::SetTimerConn()
{
	if (m_pSchedulerAutoConn == nullptr) {
		m_pSchedulerAutoConn = std::make_unique<CScheduler>(ios);
		m_pSchedulerAutoConn->setOnTimeFuncCB(std::bind(&this_type::onTime, this, std::placeholders::_1));
	}
	m_pSchedulerAutoConn->setTimer(PDU_RECONN_TIMER, VAL_RECONNECT_INTERVAL);
	LOG_DEBUG("ws开启自动连接");
}

void CNetClient::KillTimerConn()
{
	if (m_pSchedulerAutoConn != nullptr) {
		m_pSchedulerAutoConn->killTimer();
	}
	LOG_DEBUG("ws关闭自动连接");
}

void CNetClient::on_open(websocketpp::connection_hdl hdl)
{
	std::cout << "连接成功" << std::endl;
	m_IsConnSuccess.store(true);
	SetLinkContent(PDU_LINK_STATE_SUCCESSED); // 连接状态通知
	KillTimerConn();    // 关闭自动连接
	// 生成注册命令
	CPduEntity reg(PDU_CMD_REG);
	Send(&reg);
	LOG_DEBUG("ws连接成功并发送注册命令");
}

void CNetClient::on_fail(websocketpp::connection_hdl hdl)
{
	std::cout << "连接失败" << std::endl;
	m_IsConnSuccess.store(false);
	SetLinkContent(PDU_LINK_STATE_FAILED);
	LOG_DEBUG("ws连接失败");
	if (m_IsAutoReconnect) { // 自动尝试重连
		SetTimerConn();
	}
}

void CNetClient::on_close(websocketpp::connection_hdl hdl)
{
	std::cout << "连接关闭" << std::endl;
	m_IsConnSuccess.store(false);
	LOG_DEBUG("ws连接关闭");
	KillTimerListen();  // 定时心跳关闭
	SetLinkContent(PDU_LINK_STATE_DISCONNECTED);
	if (m_IsAutoReconnect) { // 自动尝试重连
		SetTimerConn();
	}

}

void CNetClient::on_message(websocketpp::connection_hdl hdl, message_ptr msg)
{
	auto str = msg->get_payload();
	UCHAR	m_szPdu[PDU_HEAD_LEN + PDU_MAX_DATA_LEN];
	memcpy(m_szPdu, &str[0], str.size());
	CPduEntityHead pduHead;
	// 解析PDU数据包头
	if (!pduHead.Decode(m_szPdu)) {
		memset(m_szPdu, 0, PDU_HEAD_LEN + PDU_MAX_DATA_LEN);
	}
	else {
		// 拷贝当前要处理的数据到临时缓冲区
		UCHAR szTmpBuf[PDU_MAX_DATA_LEN];
		ZeroMemory(szTmpBuf, PDU_MAX_DATA_LEN);
		memcpy_s(szTmpBuf, PDU_MAX_DATA_LEN, m_szPdu, PDU_HEAD_LEN + pduHead.GetDataLen());

		// 对数据进行处理
		CPduEntity pduEntity(&pduHead, &szTmpBuf[PDU_HEAD_LEN]);
		PDU_CMD_TYPE CmdType = pduEntity.GetCmdType();
		if (PDU_CMD_REG == CmdType) {  // 注册
			if (pduEntity.GetIsExecReturn()) {  // 注册返回命令
				if (pduEntity.GetDataBool(0)) {  // 注册成功
					std::cout << "设备注册成功..." << std::endl;
					SetLinkContent(PDU_LINK_STATE_REG_OK);
					SetTimerListen();
					LOG_DEBUG("设备注册成功");
				}
				else {  // 注册失败
					SetLinkContent(PDU_LINK_STATE_REG_FAILED);
					LOG_DEBUG("设备注册失败");
				}
			}
		}
		else {
			auto peerType = devTypeToStr(pduEntity.GetPeerDevType());
			auto localType = devTypeToStr(pduEntity.GetLocalDevType());
			if (PDU_CMD_LISTEN == pduEntity.GetCmdType()) {
				LOG_DEBUG("设备[%s]收到心跳消息", localType.c_str());
			}
			else {
				LOG_DEBUG("设备[%s]收到[%s]的消息[%d]", localType.c_str(), peerType.c_str(), pduEntity.GetCmdType());
			}

			if (this->m_RecvComdFuncCB != nullptr) {
				this->m_RecvComdFuncCB(&pduEntity);
			}
		}
	}
}

int CNetClient::init()
{
	c.set_access_channels(websocketpp::log::alevel::none ^ websocketpp::log::alevel::none);
	c.set_error_channels(websocketpp::log::alevel::none ^ websocketpp::log::alevel::none);
	c.init_asio(&ios);
	//c.init_asio();
	c.set_message_handler(websocketpp::lib::bind(&this_type::on_message, this, ::_1, ::_2));
	c.set_fail_handler(websocketpp::lib::bind(&this_type::on_fail, this, ::_1));
	c.set_open_handler(websocketpp::lib::bind(&this_type::on_open, this, ::_1));
	c.set_close_handler(websocketpp::lib::bind(&this_type::on_close, this, ::_1));
	c.set_reuse_addr(true);
	c.start_perpetual();
	if (thread_ == nullptr)
		thread_ = websocketpp::lib::make_shared<websocketpp::lib::thread>(boost::bind(&boost::asio::io_service::run, &ios));

	return 1;
}

bool CNetClient::ConnectToServer(const CString & a_strFarIp, int a_nFarPort, PDU_DEV_TYPE a_nLocalType, int a_nLocalId)
{
	CString urlStr;
	urlStr.Format("ws://%s:%d", a_strFarIp, a_nFarPort);
	uri = urlStr.GetBuffer(0);
	urlStr.ReleaseBuffer();
	m_DevType = a_nLocalType;
	LOG_DEBUG("ws连接开始连接[%s]", uri.c_str());
	return connect();
}

bool CNetClient::connect()
{
	websocketpp::lib::error_code ec;
	client::connection_ptr con = c.get_connection(uri, ec);
	if (ec)
	{
		LOG_DEBUG("ws连接失败，失败信息[%s]",ec.message().c_str());
		std::cout << "could not create connection because: " << ec.message() << std::endl;
		return false;
	}

	hdl_ = con->get_handle();
	c.connect(con);
	//thread1_ = websocketpp::lib::make_shared<websocketpp::lib::thread>(boost::bind(&client::run, &c));
	return true;
}

void CNetClient::close()
{
	try
	{
		c.close(hdl_, websocketpp::close::status::normal, "");
	}
	catch (const std::exception& e) {
		std::cout << e.what() << std::endl;
	}
}

void CNetClient::terminate()
{
	c.stop_perpetual();
	if (thread_&&thread_->joinable())
		thread_->join();
}

bool CNetClient::__SendBuf(unsigned char a_szBuf[], std::uint32_t a_nBufLen)
{
	try
	{
		std::string msg((char*)a_szBuf);
		c.send(hdl_, a_szBuf, a_nBufLen, websocketpp::frame::opcode::BINARY);
		return true;
	}
	catch (websocketpp::exception const &)
	{
		return false;
	}
}

void CNetClient::onTime(const std::int32_t& TimerId)
{
	if (TimerId == PDU_LISTEN_TIMER) { // 心跳定时器
		CPduEntity Monitor(PDU_CMD_LISTEN);
		Send(&Monitor, PDU_DEV_TYPE_UNKNOWN, 0);
		static std::int64_t nSecond = 0;
		if (nSecond == 0) {
			std::cout << "定时器开始..." << std::endl;
		}
		else {
			std::cout << "定时器运行中:" << time(NULL) - nSecond << std::endl;
		}
		nSecond = time(NULL);
	}
	else if (PDU_RECONN_TIMER == TimerId && !m_IsConnSuccess.load()) {  // 自动重连
		LOG_DEBUG("ws即将自动重新连接");
		connect();
	}
}

inline std::string CNetClient::devTypeToStr(PDU_DEV_TYPE devType)
{
	std::string devTypeStr = "未知";
	if (PDU_DEV_TYPE_CTI == devType) {
		devTypeStr = "CTI";
	}
	else if (PDU_DEV_TYPE_IVR == devType) {
		devTypeStr = "IVR";
	}
	else if (PDU_DEV_TYPE_ACD == devType) {
		devTypeStr = "ACD";
	}
	else if (PDU_DEV_TYPE_SERVER == devType) {
		devTypeStr = "HpServer";
	}
	return devTypeStr;
}

CNetClient CNetClient::instance;
CNetInterface* CNetInterface::getNetInstance() { return CNetClient::getInstance(); };