zhipuzi_pos_windows/lewaimai_dispatch/network/CMessagePush.cpp

#include "../pch/pch.h"
#include "CMessagePush.h"

#include "../tool/CPosPrinter.h"
#include "../order/CWaimaiOrder.h"

#include "mmsystem.h"

#include "PosMessage.h"

CMessagePush::~CMessagePush()
{

}

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

	m_nStopNum = 0;

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

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

    //处理声音提醒
    std::thread(&CMessagePush::HandleVoice, this).detach();

    //处理订单确认
    std::thread(&CMessagePush::HandleConfirm, this).detach();

	//处理打印
	std::thread(&CMessagePush::HandlePrinter, this).detach();

    //处理收银打印
    std::thread(&CMessagePush::HandleShouyinPrinter, this).detach();

    //处理厨房打印
    std::thread(&CMessagePush::HandleChufangPrinter, this).detach();
}

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

	socket_.close();

	m_voice_mutex.lock();
	while (!m_voice_queue.empty())
	{
		m_voice_queue.pop();
	}
	m_voice_mutex.unlock();

	m_confirm_mutex.lock();
	while (!m_confirm_queue.empty())
	{
		m_confirm_queue.pop();
	}
	m_confirm_mutex.unlock();

	m_printer_mutex.lock();
	while (!m_printer_queue.empty())
	{
		m_printer_queue.pop();
	}
	m_printer_mutex.unlock();

	m_shouyin_printer_mutex.lock();
	while (!m_shouyin_printer_queue.empty())
	{
		m_shouyin_printer_queue.pop();
	}
	m_shouyin_printer_mutex.unlock();

	m_chufang_printer_mutex.lock();
	while (!m_chufang_printer_queue.empty())
	{
		m_chufang_printer_queue.pop();
	}
	m_chufang_printer_mutex.unlock();
}

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

		std::string username = CSetting::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[] = "127.0.0.1";
				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::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();

				//10秒后再重试
				CSystem::my_sleep(10);

				continue;
			}
		}

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

		//休眠10秒钟，之后再发心跳包
		CSystem::my_sleep(10);
	}

	AddStopNum();
}

void CMessagePush::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;

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

			int type = document["msg_type"].GetInt();

			if (type == 1)
			{
				std::string order_id = document["waimai_order_id"].GetString();
				std::string order_no = document["waimai_order_no"].GetString();

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

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

				//判断是否右下角弹框提醒
				if (CSetting::GetParam("setting_is_new_waimai_dialog") == "1")
				{

				}

				AddPinter(order_id, order_no);
			}

			//处理完了，接着处理下一条
			continue;
		}
		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 CMessagePush::AddVoice(std::string order_id)
{
    m_voice_mutex.lock();

    m_voice_queue.push(order_id);

    m_voice_mutex.unlock();
}

void CMessagePush::AddConfirm(std::string order_id)
{
    m_confirm_mutex.lock();

    m_confirm_queue.push(order_id);

    m_confirm_mutex.unlock();
}

void CMessagePush::AddPinter(std::string order_id, std::string order_no)
{
    m_printer_mutex.lock();

    WaimaiPinterInfo newPrinter;
    newPrinter.order_id = order_id;
    newPrinter.order_no = order_no;

    m_printer_queue.push(newPrinter);

    m_printer_mutex.unlock();
}

void CMessagePush::AddShouyinPrinter(CWaimaiOrder order)
{
    m_shouyin_printer_mutex.lock();

    m_shouyin_printer_queue.push(order);

    m_shouyin_printer_mutex.unlock();
}

void CMessagePush::AddChufangPrinter(CWaimaiOrder order)
{
    m_chufang_printer_mutex.lock();

    m_chufang_printer_queue.push(order);

    m_chufang_printer_mutex.unlock();
}

void CMessagePush::HandleVoice()
{
    while(m_is_work)
    {
        m_voice_mutex.lock();

        if(m_voice_queue.empty())
        {
            m_voice_mutex.unlock();

            CSystem::my_sleep(1);
            continue;
        }

        std::string order_id = m_voice_queue.front();
        m_voice_queue.pop();

        m_voice_mutex.unlock();

        wstring path = CSystem::GetProgramDir() + L"\\music\\new_wamai_order.wav";
        PlaySound(path.c_str(), NULL, SND_FILENAME | SND_ASYNC);

        //8秒内最多播放一次
        CSystem::my_sleep(8);
    }

	AddStopNum();
}

void CMessagePush::HandleConfirm()
{
    while(m_is_work)
    {
        m_confirm_mutex.lock();

        if(m_confirm_queue.empty())
        {
            m_confirm_mutex.unlock();

            CSystem::my_sleep(1);
            continue;
        }

        std::string order_id = m_confirm_queue.front();
        m_confirm_queue.pop();

        m_confirm_mutex.unlock();

        CWaimaiOrder newOrder;
        newOrder.ConfirmeOrder(order_id);
    }

	AddStopNum();
}

void CMessagePush::HandlePrinter()
{
    while(m_is_work)
    {
        m_printer_mutex.lock();

        if(m_printer_queue.empty())
        {
            m_printer_mutex.unlock();

            CSystem::my_sleep(1);
            continue;
        }

        WaimaiPinterInfo printerInfo = m_printer_queue.front();
        std::string order_id = printerInfo.order_id;
        std::string order_no = printerInfo.order_no;

        m_printer_queue.pop();

        m_printer_mutex.unlock();

        CWaimaiOrder order;
        if(CSetting::GetParam("setting_is_new_waimai_printer") == "1" || CSetting::GetParam("setting_is_new_waimai_chufang_printer") == "1")
        {
            order.InitData(order_id, order_no);
        }

        //判断是否自动打印收银小票
        if(CSetting::GetParam("setting_is_new_waimai_printer") == "1")
        {
            AddShouyinPrinter(order);
        }

        //判断是否进行自动的厨房打印
        if(CSetting::GetParam("setting_is_new_waimai_chufang_printer") == "1")
        {
            AddChufangPrinter(order);
        }
    }

	AddStopNum();
}

void CMessagePush::HandleShouyinPrinter()
{
    while(m_is_work)
    {
        m_shouyin_printer_mutex.lock();

        if(m_shouyin_printer_queue.empty())
        {
            m_shouyin_printer_mutex.unlock();

            CSystem::my_sleep(1);
            continue;
        }

        CWaimaiOrder order = m_shouyin_printer_queue.front();

        m_shouyin_printer_queue.pop();

        m_shouyin_printer_mutex.unlock();

        CPosPrinter printer;
        printer.PrintWaimaiOrderShouyin(order);
    }

	AddStopNum();
}

void CMessagePush::HandleChufangPrinter()
{
    while(m_is_work)
    {
        m_chufang_printer_mutex.lock();

        if(m_chufang_printer_queue.empty())
        {
            m_chufang_printer_mutex.unlock();

            CSystem::my_sleep(1);
            continue;
        }

        CWaimaiOrder order = m_chufang_printer_queue.front();

        m_chufang_printer_queue.pop();

        m_chufang_printer_mutex.unlock();

        CPosPrinter printer;
        printer.PrintWaimaiOrderChufang(order);
    }

	AddStopNum();
}

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

	m_nStopNum++;

	m_nStopNumMutex.unlock();

	if (m_nStopNum == 7)
	{
		//确认所有子线程都退出了，再删除自己
		delete this;
	}
}