iot::mqtt::server::broker::Broker Class Reference

#include <Broker.h>

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

Public Member Functions
	Broker (uint8_t maxQoS, const std::string &sessionStoreFileName)
	~Broker ()
void	appear (const std::string &clientId, const std::string &topic, uint8_t qoS)
void	unsubscribe (const std::string &clientId)
void	publish (const std::string &originClientId, const std::string &topic, const std::string &message, uint8_t qoS, bool retain)
uint8_t	subscribe (const std::string &clientId, const std::string &topic, uint8_t qoS)
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
iot::mqtt::server::broker::RetainTree &	getRetainedTree ()
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)
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)
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 146 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 232 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:

getRetainedTree()

RetainTree & iot::mqtt::server::broker::Broker::getRetainedTree

(

)

Definition at line 190 of file Broker.cpp.

                                        {
        return retainTree;
    }

References retainTree.

getSubscriptions()

std::list< std::string > iot::mqtt::server::broker::Broker::getSubscriptions

(

const std::string &

clientId

)

const

Definition at line 182 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 186 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 198 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 202 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 194 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 206 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 210 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 155 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:

renewSession()

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

Definition at line 216 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 220 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 228 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 237 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 163 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 150 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 178 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 109 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 112 of file Broker.h.

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

sessionStore

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

private

Definition at line 114 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 108 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 111 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