zhipuzi_pos_windows/zhipuzi_pos_windows/worker/CMessagePushWorker.cpp

#include "../pch/pch.h"
#include "CMessagePushWorker.h"

#include "mmsystem.h"

#include "../worker/CVoiceWorker.h"
#include "../worker/CCommonWorker.h"

#include "../print/CPosPrinter.h"

void CMessagePushWorker::Start()
{
	m_is_work = true;

	m_nStopNum = 0;

    //心跳包
	std::thread(&CMessagePushWorker::KeepAlive, this).detach();

	//接收消息
	std::thread(&CMessagePushWorker::ReceiveMessage, this).detach();
}

void CMessagePushWorker::Stop()
{
	m_is_work = false;

	socket_.close();
}

void CMessagePushWorker::KeepAlive()
{
	while (m_is_work)
	{
		//生成心跳包
		rapidjson::Document doc;
		doc.SetObject();
		rapidjson::Document::AllocatorType& allocator = doc.GetAllocator(); //获取分配器

		std::string username = CSetting::GetInstance()->getUsername();
		std::string timestamp = to_string(time(NULL));
		doc.AddMember("username", StringRef(username.c_str(), username.length()), allocator);
		doc.AddMember("timestamp", StringRef(timestamp.c_str(), timestamp.length()), allocator);
		doc.AddMember("is_login", "0", allocator);

		rapidjson::StringBuffer buffer;
		rapidjson::Writer<StringBuffer> writer(buffer);
		doc.Accept(writer);

		//返回给接入层的消息
		std::string m_keepalive_msg = buffer.GetString();

		PosMessage new_msg;
		new_msg.m_length = m_keepalive_msg.length();
		memcpy(new_msg.data, m_keepalive_msg.c_str(), m_keepalive_msg.length());
		new_msg.data[new_msg.m_length] = '\0';

		try
		{
			socket_.write_some(boost::asio::buffer(&new_msg, sizeof(PosMessage)));
		}
		catch (const std::exception& e)
		{
			//走到这里来说明心跳包发送失败了，socket失效了
			std::string err = e.what();

			//先把socket关闭掉
			socket_.close();

			try
			{
				//发送失败，重新建立连接
				char host[] = "pospush.zhipuzi.com";
				char port[] = "9001";

				tcp::resolver resolver(m_io_context);
				tcp::resolver::results_type endpoints =
					resolver.resolve(tcp::v4(), host, port);

				boost::asio::connect(socket_, endpoints);

				//刚刚连接上，这是第一次登陆，发送登陆信息
				rapidjson::Document doc_1;
				doc_1.SetObject();
				rapidjson::Document::AllocatorType& allocator_1 = doc_1.GetAllocator(); //获取分配器

				std::string username_1 = CSetting::GetInstance()->getUsername();
				std::string timestamp_1 = to_string(time(NULL));
				doc_1.AddMember("username", StringRef(username_1.c_str(), username_1.length()), allocator_1);
				doc_1.AddMember("timestamp", StringRef(timestamp_1.c_str(), timestamp_1.length()), allocator_1);
				doc_1.AddMember("is_login", "1", allocator_1);

				rapidjson::StringBuffer buffer_1;
				rapidjson::Writer<StringBuffer> writer_1(buffer_1);
				doc_1.Accept(writer_1);

				//返回给接入层的消息
				std::string m_login_message = buffer_1.GetString();

				PosMessage login_msg;
				login_msg.m_length = m_login_message.length();
				memcpy(login_msg.data, m_login_message.c_str(), m_login_message.length());
				login_msg.data[login_msg.m_length] = '\0';

				socket_.write_some(boost::asio::buffer(boost::asio::buffer(&login_msg, sizeof(PosMessage))));
			}
			catch (const std::exception& e)
			{
				//重新连接或者重新发送又失败了，可能是网络断了
				std::string err = e.what();
				LOG_INFO("write err info:" << err.c_str());

				//关闭无效的连接
				socket_.close();

				//30秒后再重试
				if (m_is_work == false)
				{
					continue;
				}

				sleepMutex.lock();
				m_is_sleep = true;
				sleepMutex.unlock();

				CSystem::my_sleep(30);

				sleepMutex.lock();
				m_is_sleep = false;
				sleepMutex.unlock();

				continue;
			}
		}

		//走到这里，说明心跳包发送成功了，socket是连通的

		//休眠30秒钟，之后再发心跳包
		if (m_is_work == false)
		{
			continue;
		}

		sleepMutex.lock();
		m_is_sleep = true;
		sleepMutex.unlock();

		CSystem::my_sleep(30);

		sleepMutex.lock();
		m_is_sleep = false;
		sleepMutex.unlock();
	}

	AddStopNum();
}

void CMessagePushWorker::ReceiveMessage()
{
	while (m_is_work)
	{
		try
		{
			PosMessage new_msg;
			
			boost::asio::read(socket_, boost::asio::buffer(&new_msg, sizeof(PosMessage)));

			std::string msg = new_msg.data;

			HandelMessage(msg);
		}
		catch (std::exception& e)
		{
			std::string err = e.what();
			//LOG_INFO("read err:" << err.c_str());

			//如果这里异常了，说明socket失效了，等2秒重新读
			CSystem::my_sleep(2);
			continue;
		}
	}

	AddStopNum();
}

void CMessagePushWorker::AddStopNum()
{
	m_nStopNumMutex.lock();

	m_nStopNum++;

	m_nStopNumMutex.unlock();
}

void CMessagePushWorker::SetHWND(HWND hwnd)
{
	m_hwnd = hwnd;
}

void CMessagePushWorker::HandelMessage(std::string msg)
{
	//收到服务器的消息，对服务器的消息进行处理
	rapidjson::Document document;
	document.Parse(msg.c_str());
	if (!document.IsObject())
	{
		LOG_INFO("message 非法！");
		return;
	}

	int type;
	if (document["msg_type"].IsInt())
	{
		type = document["msg_type"].GetInt();
	}
	else
	{
		type = atoi(document["msg_type"].GetString());
	}

	if (type == MESSAGE_TYPE::LOGIN_OFFLINE)
	{
		//这个表示被人挤下线了
		PostMessage(m_hwnd, WM_LOGIN_AGAIN_OUT, 0, 0);
	}
	else if (type == MESSAGE_TYPE::NEW_WAIMAI_ORDER)
	{
		std::string order_id = document["waimai_order_id"].GetString();
		std::string order_no = document["waimai_order_no"].GetString();

		//新订单来了，首先判断是否要语音提醒
		if (CSetting::GetInstance()->GetParam("setting_is_new_waimai_voice") == "1")
		{
			if (CSetting::GetInstance()->GetParam("setting_is_new_waimai_autoconfirm") == "1")
			{
				CVoiceWorker::GetInstance()->AddVoice(2);
			}
			else
			{
				CVoiceWorker::GetInstance()->AddVoice(1);
			}
		}

		//判断是否要自动确认
		if (CSetting::GetInstance()->GetParam("setting_is_new_waimai_autoconfirm") == "1")
		{
			CCommonWorker::GetInstance()->AddConfirm(order_id);
		}

		CPosPrinter printer;
		printer.PrintWaimaiOrder(order_id, order_no);
	}
	else if (type == MESSAGE_TYPE::QUXIAO_WAIMAI_ORDER)
	{
		CVoiceWorker::GetInstance()->AddVoice(3);
	}
	else if (type == MESSAGE_TYPE::REFUND_WAIMAI_ORDER)
	{
		CVoiceWorker::GetInstance()->AddVoice(4);
	}
	else if (type == MESSAGE_TYPE::NEW_KAUICANTANGSHI_ORDER)
	{
		std::string order_id = document["waimai_order_id"].GetString();

		//新订单来了，首先判断是否要语音提醒
		if (CSetting::GetInstance()->GetParam("setting_is_new_diannei_voice") == "1")
		{
			CVoiceWorker::GetInstance()->AddVoice(5);
		}

		//CPosPrinter printer;
		//printer.PrintDiandanOrder(order_id);
	}
	else if (type == MESSAGE_TYPE::NEW_KUAICANSHANGJIAAPP_ORDER)
	{
		std::string order_id = document["waimai_order_id"].GetString();

		//新订单来了，首先判断是否要语音提醒
		if (CSetting::GetInstance()->GetParam("setting_is_new_diannei_voice") == "1")
		{
			CVoiceWorker::GetInstance()->AddVoice(5);
		}

		CPosPrinter printer;
		printer.PrintDiandanOrder(order_id);
	}
}