C++代码建立mqtt客户端发送数据和接受mqtt消息_c++ mqtt

#include "mqttclient.hpp"
#include <Qapplication>
#include <mutex>
/*
1.收到消息的结果回调
2.qtimer，到了时间自动publish
3.异步方式（生产者消费者方式）
3.满足用户不同方式使用客户端（多个构造）
*/
const QHostAddress EXAMPLE_HOST = QHostAddress("127.0.0.1");
const quint16 EXAMPLE_PORT = 1883;
const QString EXAMPLE_TOPIC = "mqtt/lmy";
#if 1
int main(int argc,char **argv)
{
	QApplication app(argc, argv);
	//QHostAddress ip(EXAMPLE_HOST);
	std::shared_ptr<main_test>ptr;
	mqtt_test mtest(EXAMPLE_HOST, EXAMPLE_PORT);
 
	mtest.SetTopic(EXAMPLE_TOPIC);
	mtest.SetCallback( ptr);
 
 
	for (int i = 0; i < 100; ++i)
	{
		std::unique_lock<std::mutex>lock(mtest.m_mtx);
		mtest.m_queue.push(i);
	}
 
	return app.exec();
}
#include "main.moc"
#endif
//#include <filesystem>
//namespace fs = std::filesystem;
//int main()
//{
//	std::cout<<fs::current_path().generic_string();
//	return 0;
//}

#ifndef MQTTCLIENT_HPP
#define MQTTCLIENT_HPP
 
#include <iostream>
#include <Qdebug>
#include <QObject>
#include <qmqtt.h>
#include <queue>
#include <QTimer>
#include <mutex>
#include <memory>
using namespace QMQTT;
const QString EXAMPLE_NAME = "LMY";
 
class Oncallback
{
public:
	virtual void Callback() = 0;
};
 
class main_test :public Oncallback
{
	void Callback()
	{
		std::cout << "recive meassage ";
	}
};
 
 
class mqtt_test :public QObject
{
	Q_OBJECT
public:
	QHostAddress m_ip;
	quint16 m_port;
	QString m_topic;
	QMQTT::Client* m_client;
	QTimer m_timer;
	std::queue<int>m_queue;
	std::mutex m_mtx;
	std::shared_ptr<Oncallback> m_call;
public:
	mqtt_test(QHostAddress ip, quint16 port = 1883)
	{
		m_client = new Client();
		m_client->setHost(ip);
		m_client->setPort(port);
		//m_client->setHostName(EXAMPLE_NAME);
		m_client->connectToHost();
		connect(m_client, &Client::connected,this,
			&mqtt_test::Onconnected);
		connect(&m_timer, &QTimer::timeout, this,
			&mqtt_test::Published);
		
		connect(m_client, &Client::received, this,
			&mqtt_test::Received);
		connect(m_client, &Client::disconnected, this,
			&mqtt_test::OnDisconnected);
	}
	~mqtt_test() { m_client->disconnectFromHost(); }
 
 
public slots:
	void SetTopic(QString topic)
	{
		//m_port = PORT;
		//m_ip = IP;
		m_topic = topic;
		m_client->subscribe(topic, 0);
	}
	
	bool SetCallback(std::shared_ptr<Oncallback>ptr)
	{
		m_call = ptr;
		return true;
	}
 
	void Onconnected()
	{
		m_timer.start(1);
		qDebug() << "connected:";
	}
 
	void Published()
	{
		int x;
		{
			std::unique_lock < std::mutex >lock(m_mtx);
			if (!m_queue.empty())
			{
				x = m_queue.front();
				if (x == 99) 
				{ 
					std::cout << x;
				}
				m_queue.pop();
			}
		}
		if (m_queue.empty())return;
		
		QMQTT::Message message(x, m_topic,QString("publish number %1").arg(x).toUtf8());
		m_client->publish(message);
		qDebug()<< "PUBLISH:" ;
	}
 
	void OnDisconnected()
	{
		qDebug() << "disconnected:";
		m_timer.stop();
	}
 
	void Received(const QMQTT::Message& message)
	{
		//开始回调
		m_call->Callback();
		qDebug() << "publish received: \"" << QString::fromUtf8(message.payload());
	}
 
};
#endif // !MQTTCLIENT_HPP