zhipuzi_pos_windows/zhipuzi_pos_windows/worker/CMqttClientWorker.cpp

#include "pch.h"
#include "CMqttClientWorker.h"

#include <openssl/hmac.h>
#include <openssl/bio.h>

#include "mmsystem.h"

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

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

#include "../tool/CAppEnv.h"

CMqttClientWorker::CMqttClientWorker()
{

}

CMqttClientWorker::~CMqttClientWorker()
{
	m_is_closing = true;
	m_is_running = false;
	m_stopCv.notify_all();
	m_async_client.reset();
}

void CMqttClientWorker::Start()
{
	m_is_closing = false;
	m_is_running = true;

	//处理Mqtt消息接收
	std::thread(&CMqttClientWorker::Run, this).detach();
	CAppEnv::GetInstance()->AddWorkerNum();
}

void CMqttClientWorker::Stop()
{
	m_is_closing = true;
	m_is_running = false;
	m_stopCv.notify_all();
}

void CMqttClientWorker::Run()
{
	m_client_id = "GID_ZHIPUZI_WINDOWS_POS@@@" + CSetting::GetInstance()->getUsername();
	 m_async_client = std::make_shared<mqtt::async_client>(m_server_address, m_client_id);

	// A subscriber often wants the server to remember its messages when its
	// disconnected. In that case, it needs a unique ClientID and a
	// non-clean session.
	m_connOpts.set_clean_session(false);

	//计算用户名和密码
	this->CalUserInfo();

	m_connOpts.set_user_name(m_UserName.c_str());
	m_connOpts.set_password(m_Password.c_str());

	// Install the callback(s) before connecting.
	m_async_client->set_callback(*this);

	// Start the connection.
	// When completed, the callback will subscribe to topic.

	try
	{
		LOG_INFO("Connecting to the MQTT server...");
		m_async_client->connect(m_connOpts, nullptr, *this);
	}
	catch (const mqtt::exception & exc)
	{
		std::string info = "ERROR: Unable to connect to MQTT server:" + m_server_address + exc.get_message();

		LOG_INFO(info.c_str());
		return;
	}

	while (m_is_running && !m_is_closing)
	{
		std::unique_lock<std::mutex> lock(m_stopMutex);
		m_stopCv.wait_for(lock, std::chrono::milliseconds(100), [this]()
			{
				return !m_is_running || m_is_closing;
			});
	}

	CLewaimaiLog::OutputDebugMessage("准备退出");

	//准备退出，禁止回调里继续重连/处理
	m_is_closing = true;
	m_stopCv.notify_all();

	try
	{
		CLewaimaiLog::OutputDebugMessage("Disconnecting from the MQTT server...");

		auto client = m_async_client;
		if (client && client->is_connected())
		{
			auto tok = client->disconnect();

			if (!tok->wait_for(std::chrono::milliseconds(200)))
			{
				CLewaimaiLog::OutputDebugMessage("MQTT disconnect timeout, wait for pending callbacks to finish.");
			}
			else
			{
				CLewaimaiLog::OutputDebugMessage("MQTT disconnected.");
			}
		}
		else
		{
			CLewaimaiLog::OutputDebugMessage("MQTT client is already disconnected.");
		}
	}
	catch (const mqtt::exception& exc)
	{
		// is_connected() 与 disconnect() 之间连接仍可能被远端断开。
		if (exc.get_return_code() == MQTTASYNC_DISCONNECTED)
		{
			CLewaimaiLog::OutputDebugMessage("MQTT client was already disconnected.");
		}
		else
		{
			const std::string msg =
				"MQTT disconnect error [" +
				std::to_string(exc.get_return_code()) +
				"]: " + exc.get_message();

			CLewaimaiLog::OutputDebugMessage(msg.c_str());
		}
	}

	CLewaimaiLog::OutputDebugMessage("MQTT client cleanup complete.");

	//不要 delete，改成释放智能指针
	m_async_client.reset();

	CLewaimaiLog::OutputDebugMessage("MQTT client worker thread exiting.");

	CAppEnv::GetInstance()->DelWorkerNum(L"CMqttClientWorker::Run");

	return;
}

void CMqttClientWorker::reconnect()
{
	if (m_is_closing)
		return;

	auto client = m_async_client;
	if (!client)
		return;

	//暂停10秒后重连
	{
		std::unique_lock<std::mutex> lock(m_stopMutex);
		if (m_stopCv.wait_for(lock, std::chrono::milliseconds(10000), [this]()
			{
				return m_is_closing.load();
			}))
		{
			return;
		}
	}

	if (m_is_closing)
		return;

	try
	{
		m_async_client->connect(m_connOpts, nullptr, *this);
	}
	catch (const mqtt::exception & exc)
	{
		//重连异常了
		std::string err = "Error: ";
		err += exc.what();
		LOG_INFO(err.c_str());
	}
}

// Re-connection failure
void CMqttClientWorker::on_failure(const mqtt::token & tok)
{
	if (m_is_closing)
		return;

	if (m_is_mqtt_connected == false)
	{
		//连接失败了
		LOG_INFO("Connection attempt failed");
		if (++nretry_ > N_RETRY_ATTEMPTS)
		{
			//重连次数超过了最大重连试错的次数，暂时不做任何处理，还是继续重连
		}

		reconnect();
	}
	else
	{
		//已经连接，那就是订阅失败了
		std::string info = "Subscription failure";
		if (tok.get_message_id() != 0)
		{
			info += " for token: [" + std::to_string(tok.get_message_id()) + "]";
		}

		LOG_INFO(info.c_str());
	}
}

void CMqttClientWorker::on_success(const mqtt::token & tok)
{
	if (m_is_closing)
		return;

	if (m_is_mqtt_connected == false)
	{
		//连接成功了，再connected里面处理
	}
	else
	{
		//订阅成功了
		std::string info = "Subscription success";
		if (tok.get_message_id() != 0)
		{
			info += " for token: [" + std::to_string(tok.get_message_id()) + "]";
		}
		auto top = tok.get_topics();
		if (top && !top->empty())
		{
			info += ",token topic: '" + (*top)[0] + "', ...";
		}

		//LOG_INFO(info.c_str());
	}
}

void CMqttClientWorker::connected(const std::string & cause)
{
	if (m_is_closing)
		return;

	m_is_mqtt_connected = true;

	std::string info = "Connection success, nSubscribing to topic " + m_topic + " for client " + m_client_id + " using QoS" + std::to_string(MQTT_QOS);
	//LOG_INFO(info.c_str());

	//不管是第一次连接，还是重连，都订阅一次
	auto client = m_async_client;
	if (client)
	{
		client->subscribe(m_topic, MQTT_QOS, nullptr, *this);
	}
}

void CMqttClientWorker::connection_lost(const std::string & cause)
{
	if (m_is_closing)
		return;

	m_is_mqtt_connected = false;

	std::string info = "Connection lost";
	if (!cause.empty())
	{
		info += ", cause: " + cause;
	}

	info += ",Reconnecting...";

	LOG_INFO(info.c_str());

	nretry_ = 0;
	reconnect();
}

// Callback for when a message arrives.
void CMqttClientWorker::message_arrived(mqtt::const_message_ptr msg)
{
	if (m_is_closing)
		return;

	std::string info = "Message arrived, ttopic: '" + msg->get_topic() + "', payload: '" + msg->to_string();

	LOG_INFO(info.c_str());

	//要的就是这个，收到消息了
	std::string content = msg->to_string();

	//对接收到的消息进行处理
	this->HandleMessage(content);
}

void CMqttClientWorker::delivery_complete(mqtt::delivery_token_ptr token)
{
}

void CMqttClientWorker::CalUserInfo()
{
	unsigned char tempData[100];
	unsigned int len = 0;

	//username和 Password 签名模式下的设置方法，参考文档 https://help.aliyun.com/document_detail/48271.html?spm=a2c4g.11186623.6.553.217831c3BSFry7
	HMAC(EVP_sha1(), m_secretKey.c_str(), m_secretKey.length(), (unsigned char *)m_client_id.c_str(), m_client_id.length(), tempData, &len);

	char resultData[100];
	int passWordLen = EVP_EncodeBlock((unsigned char *)resultData, tempData, len);
	resultData[passWordLen] = '\0';

	char userNameData[128];
	sprintf_s(userNameData, "Signature|%s|%s", m_accessKey.c_str(), m_instanceId.c_str());

	m_UserName = userNameData;
	m_Password = resultData;
}

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

	if (!document.HasMember("version"))
	{
		return;
	}

	std::string version = CLewaimaiJson::ToString(document["version"]);
	if (version < "1.0")
	{
		return;
	}

	std::string timestamp = CLewaimaiJson::ToString(document["timestamp"]);
	long int int_timesteamp = atol(timestamp.c_str());

	time_t now = time(NULL);

	if (now - int_timesteamp > 60 * 3)
	{
		//大于3分钟的消息直接丢弃
		return;
	}

	rapidjson::Value & data = document["body"];

	std::string type = CLewaimaiJson::ToString(data["order_type"]);

	if (type == "waimai")
	{
		//外卖新订单（非预约单，或者预约单快到时间了）
		std::string order_id = CLewaimaiJson::ToString(data["order_id"]);
		std::string order_no = CLewaimaiJson::ToString(data["order_no"]);

		//新订单来了，首先判断是否要语音提醒
		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 == "waimai_yuyue_notify")
	{
		//外卖新订单（预约单的通知）
		std::string order_id = CLewaimaiJson::ToString(data["order_id"]);
		std::string order_no = CLewaimaiJson::ToString(data["order_no"]);

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

		//判断是否要自动确认
		if (CSetting::GetInstance()->GetParam("setting_is_new_waimai_autoconfirm") == "1")
		{
			CCommonWorker::GetInstance()->AddConfirm(order_id);
		}
	}
	else if (type == "waimai_cancel")
	{
		std::string order_id = CLewaimaiJson::ToString(data["order_id"]);
		std::string order_no = CLewaimaiJson::ToString(data["order_no"]);

		//取消外卖订单
		CVoiceWorker::GetInstance()->AddVoice(3);

		CPosPrinter printer;
		printer.PrintWaimaiOrder(order_id, order_no);
	}
	else if (type == "waimai_order_refund")
	{
		//外卖订单退款
		CVoiceWorker::GetInstance()->AddVoice(4);
	}
	else if (type == "tangshi")
	{
		//快餐扫码下单
		std::string order_id = CLewaimaiJson::ToString(data["order_id"]);

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

		if (CSetting::GetInstance()->GetParam("setting_is_new_diannei_saomadiancan_printer") == "1")
		{
			CPosPrinter printer;
			printer.PrintDiandanOrder(order_id);
		}
	}
	else if (type == "kuaican")
	{
		//新的快餐订单，商家app下单
		std::string order_id = CLewaimaiJson::ToString(data["order_id"]);

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

		if (CSetting::GetInstance()->GetParam("setting_is_new_diannei_saomadiancan_printer") == "1")
		{
			CPosPrinter printer;
			printer.PrintDiandanOrder(order_id);
		}
	}
	else if (type == "zhengcan_jiacai")
	{
		//这个是通过商家app加菜，或者H5加菜被商家app确认，提醒收银系统去打印
		std::string order_id = CLewaimaiJson::ToString(data["order_id"]);
		std::string jiacai_no = CLewaimaiJson::ToString(data["jiacai_no"]);

		if (CSetting::GetInstance()->GetParam("setting_is_new_diannei_saomadiancan_printer") == "1")
		{
			CPosPrinter printer;
			printer.PrintZhengcanOrderXiadan(order_id, jiacai_no);
		}
	}
	else if (type == "zhengcan_jiacai_h5")
	{
		//通过扫码下单加菜，还没确认，提醒收银系统确认
		CVoiceWorker::GetInstance()->AddVoice(6);

		//同时自动刷新桌位
		PostMessage(m_hwnd, WM_ZHENGCAN_SAOMADIANCAI_XIADAN, 0, 0);
	}
	else if (type == "zhengcan_tuicai")
	{
		//正餐退菜
		std::string order_id = CLewaimaiJson::ToString(data["order_id"]);
		std::string tuicai_item_id = CLewaimaiJson::ToString(data["tuicai_item_id"]);

		if (CSetting::GetInstance()->GetParam("setting_is_new_diannei_saomadiancan_printer") == "1")
		{
			CPosPrinter printer;
			printer.PrintZhengcanOrderTuicai(order_id, tuicai_item_id);
		}
	}
	else if (type == "zhengcan_jiesuan")
	{
		//推送正餐收银订单H5上支付成功的消息给收银机，收银打印结算小票
		std::string order_id = CLewaimaiJson::ToString(data["order_id"]);

		if (CSetting::GetInstance()->GetParam("setting_is_new_diannei_saomadiancan_printer") == "1")
		{
			CPosPrinter printer;
			printer.PrintZhengcanOrderJiesuan(order_id);
		}
	}
	else if (type == "shouyintai_order")
	{
		std::string order_id = CLewaimaiJson::ToString(data["order_id"]);
		std::string content = CLewaimaiJson::ToString(data["content"]);

		CVoiceWorker::GetInstance()->AddVoice(content);

		if (CSetting::GetInstance()->GetParam("setting_is_new_diannei_saomadiancan_printer") == "1")
		{
			CPosPrinter printer;
			printer.PrintShoukuanOrder(order_id);
		}
	}
	else if (type == "tangshi_service")
	{
		std::string content = CLewaimaiJson::ToString(data["content"]);

		CVoiceWorker::GetInstance()->AddVoice(content);
	}
	else if (type == "yuyue_order")
	{
		//预约功能的预约单，不是外卖的预约单
		std::string order_id = CLewaimaiJson::ToString(data["order_id"]);
		std::string content = CLewaimaiJson::ToString(data["content"]);

		CVoiceWorker::GetInstance()->AddVoice(content);
	}
}