98 lines
2.8 KiB
C++
98 lines
2.8 KiB
C++
#pragma once
|
|
|
|
#include <string>
|
|
#include <functional>
|
|
#include <memory>
|
|
#include <thread>
|
|
#include <atomic>
|
|
#include <queue>
|
|
#include <mutex>
|
|
#include <condition_variable>
|
|
#include <nlohmann/json.hpp>
|
|
#include <mqtt/async_client.h>
|
|
|
|
namespace custom {
|
|
|
|
class MqttClient {
|
|
public:
|
|
using MessageCallback = std::function<void(const std::string& topic, const std::string& payload)>;
|
|
using ConnectionCallback = std::function<void(bool connected)>;
|
|
|
|
MqttClient(const std::string& broker, int port, const std::string& clientId);
|
|
~MqttClient();
|
|
|
|
bool connect(const std::string& username = "", const std::string& password = "");
|
|
void disconnect();
|
|
bool isConnected() const;
|
|
|
|
bool subscribe(const std::string& topic, int qos = 1);
|
|
bool unsubscribe(const std::string& topic);
|
|
|
|
bool publish(const std::string& topic, const std::string& payload, int qos = 1, bool retain = false);
|
|
bool publishJson(const std::string& topic, const nlohmann::json& json, int qos = 1, bool retain = false);
|
|
|
|
void setMessageCallback(MessageCallback callback);
|
|
void setConnectionCallback(ConnectionCallback callback);
|
|
|
|
// Message queue for async processing
|
|
void startMessageProcessor();
|
|
void stopMessageProcessor();
|
|
|
|
private:
|
|
class Callback : public virtual mqtt::callback,
|
|
public virtual mqtt::iaction_listener {
|
|
public:
|
|
Callback(MqttClient* client) : client_(client) {}
|
|
|
|
void connected(const std::string& cause) override;
|
|
void connection_lost(const std::string& cause) override;
|
|
void message_arrived(mqtt::const_message_ptr msg) override;
|
|
void delivery_complete(mqtt::delivery_token_ptr token) override;
|
|
|
|
void on_failure(const mqtt::token& tok) override;
|
|
void on_success(const mqtt::token& tok) override;
|
|
|
|
private:
|
|
MqttClient* client_;
|
|
};
|
|
|
|
struct QueuedMessage {
|
|
std::string topic;
|
|
std::string payload;
|
|
};
|
|
|
|
void processMessageQueue();
|
|
void handleConnectionLost();
|
|
void handleConnectionEstablished();
|
|
|
|
std::unique_ptr<mqtt::async_client> client_;
|
|
std::unique_ptr<Callback> callback_;
|
|
|
|
MessageCallback messageCallback_;
|
|
ConnectionCallback connectionCallback_;
|
|
|
|
std::atomic<bool> connected_;
|
|
std::atomic<bool> reconnecting_;
|
|
|
|
// Message processing
|
|
std::queue<QueuedMessage> messageQueue_;
|
|
std::mutex queueMutex_;
|
|
std::condition_variable queueCondition_;
|
|
std::thread messageProcessor_;
|
|
std::atomic<bool> processorRunning_;
|
|
|
|
// Connection parameters
|
|
std::string broker_;
|
|
int port_;
|
|
std::string clientId_;
|
|
std::string username_;
|
|
std::string password_;
|
|
|
|
// Reconnection
|
|
std::thread reconnectThread_;
|
|
std::atomic<bool> shouldReconnect_;
|
|
int reconnectDelay_ = 5; // seconds
|
|
};
|
|
|
|
} // namespace custom
|