#include <Mqtt.h>

Inheritance diagram for mqtt::mqttintegrator::lib::Mqtt:

Collaboration diagram for mqtt::mqttintegrator::lib::Mqtt:

Classes
class	DelayedQueue
struct	ScheduledPublish

Public Member Functions
	Mqtt (const std::string &connectionName, std::shared_ptr< mqtt::lib::MqttMapper > mqttMapper, const std::string &sessionStoreFileName)
	~Mqtt () override

Static Public Member Functions
static mqtt::lib::admin::ReloadResult	updateSubscriptions (bool mustReconnect)

Private Types
using	Super = iot::mqtt::client::Mqtt

Private Member Functions
void	onConnected () final
bool	onSignal (int signum) final
void	onConnack (const iot::mqtt::packets::Connack &connack) final
void	onPublish (const iot::mqtt::packets::Publish &publish) final
std::pair< std::size_t, std::size_t >	resubscribe ()

Private Attributes
std::shared_ptr< mqtt::lib::MqttMapper >	mqttMapper
std::list< iot::mqtt::Topic >	currentSubscriptions
class mqtt::mqttintegrator::lib::Mqtt::DelayedQueue	delayedQueue

Static Private Attributes
static std::set< Mqtt * >	mqttInstances

Detailed Description

Definition at line 78 of file Mqtt.h.

Member Typedef Documentation

◆ Super

using mqtt::mqttintegrator::lib::Mqtt::Super = iot::mqtt::client::Mqtt

private

Definition at line 88 of file Mqtt.h.

Constructor & Destructor Documentation

◆ Mqtt()

mqtt::mqttintegrator::lib::Mqtt::Mqtt	(	const std::string &	connectionName,
		std::shared_ptr< mqtt::lib::MqttMapper >	mqttMapper,
		const std::string &	sessionStoreFileName )

explicit

Definition at line 72 of file Mqtt.cpp.

        : iot::mqtt::client::Mqtt(connectionName, //
                                  mqttMapper->getConnection()["client_id"],
                                  mqttMapper->getConnection()["keep_alive"],
                                  sessionStoreFileName)
        , mqttMapper(mqttMapper)
        , currentSubscriptions(mqttMapper->extractSubscriptions())
        , delayedQueue(this) {
        mqttInstances.insert(this);
    }

References currentSubscriptions, mqtt::mqttintegrator::lib::Mqtt::DelayedQueue::DelayedQueue(), mqtt::lib::MqttMapper::extractSubscriptions(), mqtt::lib::MqttMapper::getConnection(), mqttInstances, and mqttMapper.

Here is the call graph for this function:

◆ ~Mqtt()

mqtt::mqttintegrator::lib::Mqtt::~Mqtt ( )

override

Definition at line 85 of file Mqtt.cpp.

                {
        mqttInstances.erase(this);
    }

References mqttInstances.

Member Function Documentation

◆ onConnack()

void mqtt::mqttintegrator::lib::Mqtt::onConnack ( const iot::mqtt::packets::Connack & connack )

finalprivate

Definition at line 130 of file Mqtt.cpp.

                                                             {
        if (connack.getReturnCode() == 0 && !connack.getSessionPresent()) {
            sendSubscribe(currentSubscriptions);
        }
    }

References currentSubscriptions.

◆ onConnected()

void mqtt::mqttintegrator::lib::Mqtt::onConnected ( )

finalprivate

Definition at line 113 of file Mqtt.cpp.

                           {
        const nlohmann::json& connection = mqttMapper->getConnection();
 
        sendConnect(connection["clean_session"],
                    connection["will_topic"],
                    connection["will_message"],
                    connection["will_qos"],
                    connection["will_retain"],
                    connection["username"],
                    connection["password"]);
    }

References mqtt::lib::MqttMapper::getConnection(), and mqttMapper.

Here is the call graph for this function:

◆ onPublish()

void mqtt::mqttintegrator::lib::Mqtt::onPublish ( const iot::mqtt::packets::Publish & publish )

finalprivate

Definition at line 136 of file Mqtt.cpp.

                                                             {
        auto [immediatePublishes, scheduledPublishes] = mqttMapper->getMappings(publish);
 
        for (const mqtt::lib::MqttMapper::ScheduledPublish& delayedPublish : scheduledPublishes) {
            delayedQueue.delayPublish(delayedPublish.delay, delayedPublish.publish);
        }
 
        for (const iot::mqtt::packets::Publish& mappedPublish : immediatePublishes) {
            sendPublish(mappedPublish.getTopic(), mappedPublish.getMessage(), mappedPublish.getQoS(), mappedPublish.getRetain());
 
            onPublish(mappedPublish);
        }
    }

References mqtt::lib::MqttMapper::ScheduledPublish::delay, mqtt::mqttintegrator::lib::Mqtt::DelayedQueue::delayPublish(), mqtt::lib::MqttMapper::getMappings(), mqttMapper, onPublish(), and mqtt::lib::MqttMapper::ScheduledPublish::publish.

Referenced by onPublish(), and mqtt::mqttintegrator::lib::Mqtt::DelayedQueue::processDue().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ onSignal()

bool mqtt::mqttintegrator::lib::Mqtt::onSignal ( int signum )

nodiscardfinalprivate

Definition at line 125 of file Mqtt.cpp.

                                  {
        sendDisconnect();
        return Super::onSignal(signum);
    }

◆ resubscribe()

std::pair< std::size_t, std::size_t > mqtt::mqttintegrator::lib::Mqtt::resubscribe ( )

private

Definition at line 150 of file Mqtt.cpp.

                                                    {
        std::list<iot::mqtt::Topic> newSubscriptions = mqttMapper->extractSubscriptions();
 
        std::list<std::string> topicsToUnsubscribe;
        for (const auto& currentTopic : currentSubscriptions) {
            const bool existsInNew = std::any_of(newSubscriptions.begin(), newSubscriptions.end(), [&](const auto& newTopic) {
                return currentTopic.getName() == newTopic.getName() && currentTopic.getQoS() == newTopic.getQoS();
            });
 
            if (!existsInNew) {
                topicsToUnsubscribe.push_back(currentTopic.getName());
            }
        }
 
        if (!topicsToUnsubscribe.empty()) {
            sendUnsubscribe(topicsToUnsubscribe);
        }
 
        std::list<iot::mqtt::Topic> topicsToSubscribe;
        for (const auto& newTopic : newSubscriptions) {
            const bool existsInOld = std::any_of(currentSubscriptions.begin(), currentSubscriptions.end(), [&](const auto& currentTopic) {
                return currentTopic.getName() == newTopic.getName() && currentTopic.getQoS() == newTopic.getQoS();
            });
 
            if (!existsInOld) {
                topicsToSubscribe.push_back(newTopic);
            }
        }
 
        if (!topicsToSubscribe.empty()) {
            sendSubscribe(topicsToSubscribe);
        }
 
        currentSubscriptions = newSubscriptions;
 
        return {topicsToSubscribe.size(), topicsToUnsubscribe.size()};
    }

References currentSubscriptions, mqtt::lib::MqttMapper::extractSubscriptions(), and mqttMapper.

Referenced by updateSubscriptions().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ updateSubscriptions()

mqtt::lib::admin::ReloadResult mqtt::mqttintegrator::lib::Mqtt::updateSubscriptions ( bool mustReconnect )

static

Definition at line 89 of file Mqtt.cpp.

                                                                         {
        mqtt::lib::admin::ReloadResult reloadResult;
 
        reloadResult.instances = mqttInstances.size();
        if (mustReconnect) {
            reloadResult.mode = "reconnect";
        } else {
            reloadResult.mode = "hot";
        }
 
        for (Mqtt* mqtt : mqttInstances) {
            if (mustReconnect) {
                mqtt->sendDisconnect();
            } else {
                auto [subscribeCount, unsubscribeCount] = mqtt->resubscribe();
 
                reloadResult.subscribed += subscribeCount;
                reloadResult.unsubscribed += unsubscribeCount;
            }
        }
 
        return reloadResult;
    }

References mqtt::lib::admin::ReloadResult::instances, mqtt::lib::admin::ReloadResult::mode, mqttInstances, resubscribe(), mqtt::lib::admin::ReloadResult::subscribed, and mqtt::lib::admin::ReloadResult::unsubscribed.

Referenced by main().

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

◆ currentSubscriptions

std::list<iot::mqtt::Topic> mqtt::mqttintegrator::lib::Mqtt::currentSubscriptions

private

Definition at line 101 of file Mqtt.h.

Referenced by Mqtt(), onConnack(), and resubscribe().

◆ delayedQueue

class mqtt::mqttintegrator::lib::Mqtt::DelayedQueue mqtt::mqttintegrator::lib::Mqtt::delayedQueue

private

◆ mqttInstances

std::set< Mqtt * > mqtt::mqttintegrator::lib::Mqtt::mqttInstances

staticprivate

Definition at line 129 of file Mqtt.h.

Referenced by Mqtt(), updateSubscriptions(), and ~Mqtt().

◆ mqttMapper

std::shared_ptr<mqtt::lib::MqttMapper> mqtt::mqttintegrator::lib::Mqtt::mqttMapper

private

Definition at line 100 of file Mqtt.h.

Referenced by Mqtt(), onConnected(), onPublish(), and resubscribe().

The documentation for this class was generated from the following files:

mqttintegrator/lib/Mqtt.h
mqttintegrator/lib/Mqtt.cpp

MQTTSuite

Classes

Public Member Functions

Static Public Member Functions

Private Types

Private Member Functions

Private Attributes

Static Private Attributes

Detailed Description

Member Typedef Documentation

◆ Super

Constructor & Destructor Documentation

◆ Mqtt()

◆ ~Mqtt()

Member Function Documentation

◆ onConnack()

◆ onConnected()

◆ onPublish()

◆ onSignal()

◆ resubscribe()

◆ updateSubscriptions()

Member Data Documentation

◆ currentSubscriptions

◆ delayedQueue

◆ mqttInstances

◆ mqttMapper