#include <Broker.h>

Collaboration diagram for iot::mqtt::server::broker::Broker:

Public Member Functions
	Broker (uint8_t maxQoS, const std::string &sessionStoreFileName)

	~Broker ()

void	publish (const std::string &originClientId, const std::string &topic, const std::string &message, uint8_t qoS, bool retain)

Session *	newSession (const std::string &clientId, iot::mqtt::server::Mqtt *mqtt)

Session *	renewSession (const std::string &clientId, iot::mqtt::server::Mqtt *mqtt)

void	restartSession (const std::string &clientId)

void	retainSession (const std::string &clientId)

void	deleteSession (const std::string &clientId)

bool	hasSession (const std::string &clientId)

bool	hasActiveSession (const std::string &clientId)

bool	hasRetainedSession (const std::string &clientId)

bool	isActiveSession (const std::string &clientId, const Mqtt *mqtt)

void	appear (const std::string &clientId, const std::string &topic, uint8_t qoS)

void	release (const std::string &topic)

uint8_t	subscribe (const std::string &clientId, const std::string &topic, uint8_t qoS)

void	unsubscribe (const std::string &clientId)

void	unsubscribe (const std::string &clientId, const std::string &topic)

std::list< std::string >	getSubscriptions (const std::string &clientId) const

std::map< std::string, std::list< std::pair< std::string, uint8_t > > >	getSubscriptionTree () const

std::list< std::pair< std::string, std::string > >	getRetainTree ()

void	sendPublish (const std::string &clientId, Message &message, uint8_t qoS, bool retain)

Static Public Member Functions
static std::shared_ptr< Broker >	instance (uint8_t maxQoS, const std::string &sessionStoreFileName)

Private Attributes
std::string	sessionStoreFileName

uint8_t	maxQoS

iot::mqtt::server::broker::SubscribtionTree	subscribtionTree

iot::mqtt::server::broker::RetainTree	retainTree

std::map< std::string, iot::mqtt::server::broker::Session >	sessionStore

Detailed Description

Definition at line 73 of file Broker.h.

Constructor & Destructor Documentation

◆ Broker()

iot::mqtt::server::broker::Broker::Broker	(	uint8_t	maxQoS,
		const std::string &	sessionStoreFileName
	)

Definition at line 60 of file Broker.cpp.

        : sessionStoreFileName(sessionStoreFileName) // NOLINT
        , maxQoS(maxQoS)
        , subscribtionTree(this)
        , retainTree(this) {
        if (!sessionStoreFileName.empty()) {
            std::ifstream sessionStoreFile(sessionStoreFileName);
 
            if (sessionStoreFile.is_open()) {
                try {
                    nlohmann::json sessionStoreJson;
 
                    sessionStoreFile >> sessionStoreJson;
 
                    for (const auto& [clientId, sessionJson] : sessionStoreJson["session_store"].items()) {
                        sessionStore[clientId].fromJson(sessionJson);
                    }
                    retainTree.fromJson(sessionStoreJson["retain_tree"]);
                    subscribtionTree.fromJson(sessionStoreJson["subscribtion_tree"]);
 
                    LOG(INFO) << "MQTT Broker: Persistent session data loaded successful";
                } catch (const nlohmann::json::exception&) {
                    LOG(INFO) << "MQTT Broker: Starting with empty session: Session store '" << sessionStoreFileName
                              << "' empty or corrupted";
 
                    sessionStore.clear();
                    retainTree.clear();
                    subscribtionTree.clear();
                }
 
                sessionStoreFile.close();
                std::remove(sessionStoreFileName.data()); // NOLINT
 
                LOG(INFO) << "MQTT Broker: Restoring saved session done";
            } else {
                PLOG(WARNING) << "MQTT Broker: Could not read session store '" << sessionStoreFileName << "'";
            }
        } else {
            LOG(INFO) << "MQTT Broker: Session not reloaded: Session store filename empty";
        }
    }

References iot::mqtt::server::broker::RetainTree::clear(), iot::mqtt::server::broker::SubscribtionTree::clear(), iot::mqtt::server::broker::RetainTree::fromJson(), iot::mqtt::server::broker::Session::fromJson(), iot::mqtt::server::broker::SubscribtionTree::fromJson(), maxQoS, retainTree, iot::mqtt::server::broker::RetainTree::RetainTree(), sessionStore, sessionStoreFileName, subscribtionTree, and iot::mqtt::server::broker::SubscribtionTree::SubscribtionTree().

Here is the call graph for this function:

◆ ~Broker()

iot::mqtt::server::broker::Broker::~Broker ( )

Definition at line 102 of file Broker.cpp.

                    {
        if (!sessionStoreFileName.empty()) {
            nlohmann::json sessionStoreJson;
 
            for (auto& [clientId, session] : sessionStore) {
                sessionStoreJson["session_store"][clientId] = session.toJson();
            }
            sessionStoreJson["retain_tree"] = retainTree.toJson();
            sessionStoreJson["subscribtion_tree"] = subscribtionTree.toJson();
 
            if (sessionStoreJson["session_store"].empty()) {
                sessionStoreJson.erase("session_store");
            }
            if (sessionStoreJson["retain_tree"].empty()) {
                sessionStoreJson.erase("retain_tree");
            }
            if (sessionStoreJson["subscribtion_tree"].empty()) {
                sessionStoreJson.erase("subscribtion_tree");
            }
 
            std::ofstream sessionStoreFile(sessionStoreFileName);
 
            if (sessionStoreFile.is_open()) {
                if (!sessionStoreJson.empty()) {
                    sessionStoreFile << sessionStoreJson;
                }
 
                sessionStoreFile.close();
 
                LOG(INFO) << "MQTT Broker: Session store written '" << sessionStoreFileName << "'";
            } else {
                PLOG(ERROR) << "MQTT Broker: Could not write session store '" << sessionStoreFileName << "'";
            }
        } else {
            LOG(INFO) << "MQTT Broker: Session not saved: Session store filename empty";
        }
    }

References retainTree, sessionStore, sessionStoreFileName, subscribtionTree, iot::mqtt::server::broker::RetainTree::toJson(), iot::mqtt::server::broker::SubscribtionTree::toJson(), and iot::mqtt::server::broker::Session::toJson().

Here is the call graph for this function:

Member Function Documentation

◆ appear()

void iot::mqtt::server::broker::Broker::appear	(	const std::string &	clientId,
		const std::string &	topic,
		uint8_t	qoS
	)

Definition at line 198 of file Broker.cpp.

                                                                                      {
        retainTree.appear(clientId, topic, qoS);
    }

References iot::mqtt::server::broker::RetainTree::appear(), and retainTree.

Referenced by iot::mqtt::server::broker::SubscribtionTree::TopicLevel::appear().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ deleteSession()

void iot::mqtt::server::broker::Broker::deleteSession ( const std::string & clientId )

Definition at line 177 of file Broker.cpp.

                                                        {
        subscribtionTree.unsubscribe(clientId);
        sessionStore.erase(clientId);
    }

References sessionStore, subscribtionTree, and iot::mqtt::server::broker::SubscribtionTree::unsubscribe().

Referenced by iot::mqtt::server::Mqtt::releaseSession().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ getRetainTree()

std::list< std::pair< std::string, std::string > > iot::mqtt::server::broker::Broker::getRetainTree ( )

Definition at line 237 of file Broker.cpp.

                                                                 {
        return retainTree.getRetainTree();
    }

References iot::mqtt::server::broker::RetainTree::getRetainTree(), and retainTree.

Here is the call graph for this function:

◆ getSubscriptions()

std::list< std::string > iot::mqtt::server::broker::Broker::getSubscriptions ( const std::string & clientId ) const

Definition at line 229 of file Broker.cpp.

                                                                               {
        return subscribtionTree.getSubscriptions(clientId);
    }

References iot::mqtt::server::broker::SubscribtionTree::getSubscriptions(), and subscribtionTree.

Referenced by iot::mqtt::server::Mqtt::getSubscriptions().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ getSubscriptionTree()

std::map< std::string, std::list< std::pair< std::string, uint8_t > > > iot::mqtt::server::broker::Broker::getSubscriptionTree ( ) const

Definition at line 233 of file Broker.cpp.

                                                                                              {
        return subscribtionTree.getSubscriptionTree();
    }

References iot::mqtt::server::broker::SubscribtionTree::getSubscriptionTree(), and subscribtionTree.

Here is the call graph for this function:

◆ hasActiveSession()

bool iot::mqtt::server::broker::Broker::hasActiveSession ( const std::string & clientId )

Definition at line 186 of file Broker.cpp.

                                                           {
        return hasSession(clientId) && sessionStore[clientId].isActive();
    }

References hasSession(), iot::mqtt::server::broker::Session::isActive(), and sessionStore.

Referenced by iot::mqtt::server::Mqtt::initSession().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ hasRetainedSession()

bool iot::mqtt::server::broker::Broker::hasRetainedSession ( const std::string & clientId )

Definition at line 190 of file Broker.cpp.

                                                             {
        return hasSession(clientId) && !sessionStore[clientId].isActive();
    }

References hasSession(), iot::mqtt::server::broker::Session::isActive(), and sessionStore.

Referenced by iot::mqtt::server::Mqtt::initSession().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ hasSession()

bool iot::mqtt::server::broker::Broker::hasSession ( const std::string & clientId )

Definition at line 182 of file Broker.cpp.

                                                     {
        return sessionStore.contains(clientId);
    }

References sessionStore.

Referenced by hasActiveSession(), hasRetainedSession(), and isActiveSession().

Here is the caller graph for this function:

◆ instance()

std::shared_ptr< Broker > iot::mqtt::server::broker::Broker::instance	(	uint8_t	maxQoS,
		const std::string &	sessionStoreFileName
	)

static

Definition at line 140 of file Broker.cpp.

                                                                                                {
        static const std::shared_ptr<Broker> broker = std::make_shared<Broker>(maxQoS, sessionStoreFileName);
 
        return broker;
    }

Referenced by iot::mqtt::server::SharedSocketContextFactory::create().

Here is the caller graph for this function:

◆ isActiveSession()

bool iot::mqtt::server::broker::Broker::isActiveSession	(	const std::string &	clientId,
		const Mqtt *	mqtt
	)

Definition at line 194 of file Broker.cpp.

                                                                                         {
        return hasSession(clientId) && sessionStore[clientId].isOwnedBy(mqtt);
    }

References hasSession(), iot::mqtt::server::broker::Session::isOwnedBy(), and sessionStore.

Referenced by iot::mqtt::server::Mqtt::releaseSession().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ newSession()

Session * iot::mqtt::server::broker::Broker::newSession	(	const std::string &	clientId,
		iot::mqtt::server::Mqtt *	mqtt
	)

Definition at line 155 of file Broker.cpp.

                                                                                  {
        sessionStore[clientId] = iot::mqtt::server::broker::Session(mqtt);
 
        return &sessionStore[clientId];
    }

References iot::mqtt::server::broker::Session::Session(), and sessionStore.

Referenced by iot::mqtt::server::Mqtt::initSession().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ publish()

void iot::mqtt::server::broker::Broker::publish	(	const std::string &	originClientId,
		const std::string &	topic,
		const std::string &	message,
		uint8_t	qoS,
		bool	retain
	)

Definition at line 147 of file Broker.cpp.

                                                                                                                               {
        subscribtionTree.publish(Message(originClientId, topic, message, qoS, retain));
 
        if (retain) {
            retainTree.retain(Message(originClientId, topic, message, qoS, retain));
        }
    }

References iot::mqtt::server::broker::Message::Message(), iot::mqtt::server::broker::SubscribtionTree::publish(), iot::mqtt::server::broker::RetainTree::retain(), retainTree, and subscribtionTree.

Referenced by iot::mqtt::server::Mqtt::_onPublish(), and iot::mqtt::server::Mqtt::~Mqtt().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ release()

void iot::mqtt::server::broker::Broker::release ( const std::string & topic )

Definition at line 202 of file Broker.cpp.

                                               {
        retainTree.release(topic);
    }

References iot::mqtt::server::broker::RetainTree::release(), and retainTree.

Here is the call graph for this function:

◆ renewSession()

Session * iot::mqtt::server::broker::Broker::renewSession	(	const std::string &	clientId,
		iot::mqtt::server::Mqtt *	mqtt
	)

Definition at line 161 of file Broker.cpp.

                                                                                    {
        return sessionStore[clientId].renew(mqtt);
    }

References iot::mqtt::server::broker::Session::renew(), and sessionStore.

Referenced by iot::mqtt::server::Mqtt::initSession().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ restartSession()

void iot::mqtt::server::broker::Broker::restartSession ( const std::string & clientId )

Definition at line 165 of file Broker.cpp.

                                                         {
        LOG(INFO) << "MQTT Broker:   Retained: Send PUBLISH: " << clientId;
        subscribtionTree.appear(clientId);
 
        LOG(INFO) << "MQTT Broker:   Queued: Send PUBLISH: " << clientId;
        sessionStore[clientId].publishQueued();
    }

References iot::mqtt::server::broker::SubscribtionTree::appear(), iot::mqtt::server::broker::Session::publishQueued(), sessionStore, and subscribtionTree.

Referenced by iot::mqtt::server::Mqtt::initSession().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ retainSession()

void iot::mqtt::server::broker::Broker::retainSession ( const std::string & clientId )

Definition at line 173 of file Broker.cpp.

                                                        {
        sessionStore[clientId].retain();
    }

References iot::mqtt::server::broker::Session::retain(), and sessionStore.

Referenced by iot::mqtt::server::Mqtt::releaseSession().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ sendPublish()

void iot::mqtt::server::broker::Broker::sendPublish	(	const std::string &	clientId,
		Message &	message,
		uint8_t	qoS,
		bool	retain
	)

Definition at line 241 of file Broker.cpp.

                                                                                                  {
        LOG(INFO) << "MQTT Broker: Send PUBLISH: " << clientId;
 
        sessionStore[clientId].sendPublish(message, qoS, retain);
    }

References iot::mqtt::server::broker::Session::sendPublish(), and sessionStore.

Referenced by iot::mqtt::server::broker::RetainTree::TopicLevel::appear(), iot::mqtt::server::broker::RetainTree::TopicLevel::appear(), and iot::mqtt::server::broker::SubscribtionTree::TopicLevel::publish().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ subscribe()

uint8_t iot::mqtt::server::broker::Broker::subscribe	(	const std::string &	clientId,
		const std::string &	topic,
		uint8_t	qoS
	)

Definition at line 206 of file Broker.cpp.

                                                                                            {
        qoS = std::min(maxQoS, qoS);
        uint8_t returnCode = 0;
 
        if (subscribtionTree.subscribe(topic, clientId, qoS)) {
            retainTree.appear(clientId, topic, qoS);
 
            returnCode = SUBSCRIBTION_SUCCESS | qoS;
        } else {
            returnCode = SUBSCRIBTION_FAILURE;
        }
 
        return returnCode;
    }

References iot::mqtt::server::broker::RetainTree::appear(), maxQoS, retainTree, iot::mqtt::server::broker::SubscribtionTree::subscribe(), and subscribtionTree.

Referenced by iot::mqtt::server::Mqtt::_onSubscribe(), and iot::mqtt::server::Mqtt::subscribe().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ unsubscribe() [1/2]

void iot::mqtt::server::broker::Broker::unsubscribe ( const std::string & clientId )

Definition at line 221 of file Broker.cpp.

                                                      {
        subscribtionTree.unsubscribe(clientId);
    }

References subscribtionTree, and iot::mqtt::server::broker::SubscribtionTree::unsubscribe().

Referenced by iot::mqtt::server::Mqtt::initSession().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ unsubscribe() [2/2]

void iot::mqtt::server::broker::Broker::unsubscribe	(	const std::string &	clientId,
		const std::string &	topic
	)

Definition at line 225 of file Broker.cpp.

                                                                              {
        subscribtionTree.unsubscribe(topic, clientId);
    }

References subscribtionTree, and iot::mqtt::server::broker::SubscribtionTree::unsubscribe().

Referenced by iot::mqtt::server::Mqtt::_onUnsubscribe(), and iot::mqtt::server::Mqtt::unsubscribe().

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

◆ maxQoS

uint8_t iot::mqtt::server::broker::Broker::maxQoS

private

Definition at line 110 of file Broker.h.

Referenced by Broker(), and subscribe().

◆ retainTree

iot::mqtt::server::broker::RetainTree iot::mqtt::server::broker::Broker::retainTree

private

Definition at line 113 of file Broker.h.

Referenced by appear(), Broker(), getRetainTree(), publish(), release(), subscribe(), and ~Broker().

◆ sessionStore

std::map<std::string, iot::mqtt::server::broker::Session> iot::mqtt::server::broker::Broker::sessionStore

private

Definition at line 115 of file Broker.h.

Referenced by Broker(), deleteSession(), hasActiveSession(), hasRetainedSession(), hasSession(), isActiveSession(), newSession(), renewSession(), restartSession(), retainSession(), sendPublish(), and ~Broker().

◆ sessionStoreFileName

std::string iot::mqtt::server::broker::Broker::sessionStoreFileName

private

Definition at line 109 of file Broker.h.

Referenced by Broker(), and ~Broker().

◆ subscribtionTree

iot::mqtt::server::broker::SubscribtionTree iot::mqtt::server::broker::Broker::subscribtionTree

private

Definition at line 112 of file Broker.h.

Referenced by Broker(), deleteSession(), getSubscriptions(), getSubscriptionTree(), publish(), restartSession(), subscribe(), unsubscribe(), unsubscribe(), and ~Broker().

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

iot/mqtt/server/broker/Broker.h
iot/mqtt/server/broker/Broker.cpp

Public Member Functions

Static Public Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ Broker()

◆ ~Broker()

Member Function Documentation

◆ appear()

◆ deleteSession()

◆ getRetainTree()

◆ getSubscriptions()

◆ getSubscriptionTree()

◆ hasActiveSession()

◆ hasRetainedSession()

◆ hasSession()

◆ instance()

◆ isActiveSession()

◆ newSession()

◆ publish()

◆ release()

◆ renewSession()

◆ restartSession()

◆ retainSession()

◆ sendPublish()

◆ subscribe()

◆ unsubscribe() [1/2]

◆ unsubscribe() [2/2]

Member Data Documentation

◆ maxQoS

◆ retainTree

◆ sessionStore

◆ sessionStoreFileName

◆ subscribtionTree