iot::mqtt::Mqtt Class Reference

#include <Mqtt.h>

Inheritance diagram for iot::mqtt::Mqtt:

Collaboration diagram for iot::mqtt::Mqtt:

Public Member Functions
	Mqtt (const std::string &connectionName)
	Mqtt (const std::string &connectionName, const std::string &clientId)
	Mqtt (Mqtt &&)=delete
	Mqtt (const Mqtt &)=delete
Mqtt &	operator= (Mqtt &&)=delete
Mqtt &	operator= (const Mqtt &)=delete
virtual	~Mqtt ()
virtual void	onConnected ()
virtual void	onDisconnected ()
virtual bool	onSignal (int sig)=0
const std::string &	getConnectionName () const
const MqttContext *	getMqttContext () const
void	sendPublish (const std::string &topic, const std::string &message, uint8_t qoS, bool retain)
void	sendPuback (uint16_t packetIdentifier) const
void	sendPubrec (uint16_t packetIdentifier) const
void	sendPubrel (uint16_t packetIdentifier) const
void	sendPubcomp (uint16_t packetIdentifier) const

Static Public Member Functions
static std::string	toHexString (const std::vector< char > &data)
static std::string	toHexString (const std::string &data)

Protected Member Functions
void	initSession (Session *session, utils::Timeval keepAlive)
virtual void	onPublish (const iot::mqtt::packets::Publish &publish)
virtual void	onPuback (const iot::mqtt::packets::Puback &puback)
virtual void	onPubrec (const iot::mqtt::packets::Pubrec &pubrec)
virtual void	onPubrel (const iot::mqtt::packets::Pubrel &pubrel)
virtual void	onPubcomp (const iot::mqtt::packets::Pubcomp &pubcomp)
bool	_onPublish (const iot::mqtt::packets::Publish &publish)
void	_onPuback (const iot::mqtt::packets::Puback &puback)
void	_onPubrec (const iot::mqtt::packets::Pubrec &pubrec)
void	_onPubrel (const iot::mqtt::packets::Pubrel &pubrel)
void	_onPubcomp (const iot::mqtt::packets::Pubcomp &pubcomp)
uint16_t	getPacketIdentifier ()
void	send (const iot::mqtt::ControlPacket &controlPacket) const
void	printVP (const iot::mqtt::ControlPacket &packet) const
void	printFixedHeader (const iot::mqtt::FixedHeader &fixedHeader) const

Protected Attributes
std::string	connectionName
std::string	clientId
MqttContext *	mqttContext = nullptr

Private Member Functions
std::size_t	onReceivedFromPeer ()
void	setMqttContext (MqttContext *mqttContext)
virtual iot::mqtt::ControlPacketDeserializer *	createControlPacketDeserializer (iot::mqtt::FixedHeader &staticHeader)=0
virtual void	deliverPacket (iot::mqtt::ControlPacketDeserializer *controlPacketDeserializer)=0
void	send (const std::vector< char > &data) const

Private Attributes
iot::mqtt::FixedHeader	fixedHeader
iot::mqtt::ControlPacketDeserializer *	controlPacketDeserializer = nullptr
uint16_t	_packetIdentifier = 0
core::timer::Timer	keepAliveTimer
int	state = 0
Session *	session = nullptr

Friends
class	MqttContext

Detailed Description

Definition at line 74 of file Mqtt.h.

Constructor & Destructor Documentation

Mqtt() [1/4]

iot::mqtt::Mqtt::Mqtt

(

const std::string &

connectionName

)

Definition at line 69 of file Mqtt.cpp.

        : connectionName(connectionName) {
    }

Mqtt() [2/4]

iot::mqtt::Mqtt::Mqtt	(	const std::string &	connectionName,
		const std::string &	clientId )

Definition at line 73 of file Mqtt.cpp.

        : connectionName(connectionName)
        , clientId(clientId) {
    }

Mqtt() [3/4]

iot::mqtt::Mqtt::Mqtt ( Mqtt && )

delete

Mqtt() [4/4]

iot::mqtt::Mqtt::Mqtt ( const Mqtt & )

delete

~Mqtt()

iot::mqtt::Mqtt::~Mqtt ( )

virtual

Reimplemented in iot::mqtt::client::Mqtt, and iot::mqtt::server::Mqtt.

Definition at line 78 of file Mqtt.cpp.

                {
        if (controlPacketDeserializer != nullptr) {
            delete controlPacketDeserializer;
            controlPacketDeserializer = nullptr;
        }
 
        keepAliveTimer.cancel();
    }

Member Function Documentation

_onPuback()

void iot::mqtt::Mqtt::_onPuback ( const iot::mqtt::packets::Puback & puback )

protected

Definition at line 308 of file Mqtt.cpp.

                                                           {
        if (puback.getPacketIdentifier() == 0) {
            LOG(ERROR) << connectionName << " MQTT:   PackageIdentifier missing";
            mqttContext->end(true);
        } else {
            LOG(DEBUG) << connectionName << " MQTT:   PacketIdentifier: 0x" << std::hex << std::setfill('0') << std::setw(4)
                       << puback.getPacketIdentifier();
        }
 
        onPuback(puback);
    }

_onPubcomp()

void iot::mqtt::Mqtt::_onPubcomp ( const iot::mqtt::packets::Pubcomp & pubcomp )

protected

Definition at line 353 of file Mqtt.cpp.

                                                              {
        if (pubcomp.getPacketIdentifier() == 0) {
            LOG(ERROR) << connectionName << " MQTT:   PackageIdentifier missing";
            mqttContext->end(true);
        } else {
            LOG(DEBUG) << connectionName << " MQTT:   PacketIdentifier: 0x" << std::hex << std::setfill('0') << std::setw(4)
                       << pubcomp.getPacketIdentifier();
 
            session->publishMap.erase(pubcomp.getPacketIdentifier());
            session->pubrelPacketIdentifierSet.erase(pubcomp.getPacketIdentifier());
        }
 
        onPubcomp(pubcomp);
    }

_onPublish()

bool iot::mqtt::Mqtt::_onPublish ( const iot::mqtt::packets::Publish & publish )

protected

Definition at line 268 of file Mqtt.cpp.

                                                       {
        bool deliver = true;
 
        LOG(INFO) << connectionName << " MQTT:   Topic: " << publish.getTopic();
        LOG(INFO) << connectionName << " MQTT:   Message:\n" << toHexString(publish.getMessage());
        LOG(DEBUG) << connectionName << " MQTT:   QoS: " << static_cast<uint16_t>(publish.getQoS());
        LOG(DEBUG) << connectionName << " MQTT:   PacketIdentifier: " << publish.getPacketIdentifier();
        LOG(DEBUG) << connectionName << " MQTT:   DUP: " << publish.getDup();
        LOG(DEBUG) << connectionName << " MQTT:   Retain: " << publish.getRetain();
 
        if (publish.getQoS() > 2) {
            LOG(ERROR) << connectionName << " MQTT:   Received invalid QoS: " << publish.getQoS();
            mqttContext->end(true);
            deliver = false;
        } else if (publish.getPacketIdentifier() == 0 && publish.getQoS() > 0) {
            LOG(ERROR) << connectionName << " MQTT:   Received QoS > 0 but no PackageIdentifier present";
            mqttContext->end(true);
            deliver = false;
        } else {
            switch (publish.getQoS()) {
                case 1:
                    sendPuback(publish.getPacketIdentifier());
 
                    break;
                case 2:
                    sendPubrec(publish.getPacketIdentifier());
 
                    if (session->publishPacketIdentifierSet.contains(publish.getPacketIdentifier())) {
                        deliver = false;
                    } else {
                        session->publishPacketIdentifierSet.insert(publish.getPacketIdentifier());
                    }
 
                    break;
            }
        }
 
        return deliver;
    }

_onPubrec()

void iot::mqtt::Mqtt::_onPubrec ( const iot::mqtt::packets::Pubrec & pubrec )

protected

Definition at line 320 of file Mqtt.cpp.

                                                           {
        if (pubrec.getPacketIdentifier() == 0) {
            LOG(ERROR) << connectionName << " MQTT:   PackageIdentifier missing";
            mqttContext->end(true);
        } else {
            LOG(DEBUG) << connectionName << " MQTT:   PacketIdentifier: 0x" << std::hex << std::setfill('0') << std::setw(4)
                       << pubrec.getPacketIdentifier();
 
            session->publishMap.erase(pubrec.getPacketIdentifier());
            session->pubrelPacketIdentifierSet.insert(pubrec.getPacketIdentifier());
 
            sendPubrel(pubrec.getPacketIdentifier());
        }
 
        onPubrec(pubrec);
    }

_onPubrel()

void iot::mqtt::Mqtt::_onPubrel ( const iot::mqtt::packets::Pubrel & pubrel )

protected

Definition at line 337 of file Mqtt.cpp.

                                                           {
        if (pubrel.getPacketIdentifier() == 0) {
            LOG(ERROR) << connectionName << " MQTT:   PackageIdentifier missing";
            mqttContext->end(true);
        } else {
            LOG(DEBUG) << connectionName << " MQTT:   PacketIdentifier: 0x" << std::hex << std::setfill('0') << std::setw(4)
                       << pubrel.getPacketIdentifier();
 
            session->publishPacketIdentifierSet.erase(pubrel.getPacketIdentifier());
 
            sendPubcomp(pubrel.getPacketIdentifier());
        }
 
        onPubrel(pubrel);
    }

createControlPacketDeserializer()

virtual iot::mqtt::ControlPacketDeserializer * iot::mqtt::Mqtt::createControlPacketDeserializer ( iot::mqtt::FixedHeader & staticHeader )

privatepure virtual

Implemented in iot::mqtt::client::Mqtt, and iot::mqtt::server::Mqtt.

deliverPacket()

virtual void iot::mqtt::Mqtt::deliverPacket ( iot::mqtt::ControlPacketDeserializer * controlPacketDeserializer )

privatepure virtual

Implemented in iot::mqtt::client::Mqtt, and iot::mqtt::server::Mqtt.

getConnectionName()

const std::string & iot::mqtt::Mqtt::getConnectionName

(

)

const

Definition at line 170 of file Mqtt.cpp.

                                                   {
        return connectionName;
    }

getMqttContext()

const MqttContext * iot::mqtt::Mqtt::getMqttContext

(

)

const

Definition at line 91 of file Mqtt.cpp.

                                                  {
        return mqttContext;
    }

getPacketIdentifier()

uint16_t iot::mqtt::Mqtt::getPacketIdentifier ( )

protected

Definition at line 398 of file Mqtt.cpp.

                                       {
        ++_packetIdentifier;
 
        if (_packetIdentifier == 0) {
            ++_packetIdentifier;
        }
 
        return _packetIdentifier;
    }

initSession()

void iot::mqtt::Mqtt::initSession ( Session * session, utils::Timeval keepAlive )

protected

Definition at line 174 of file Mqtt.cpp.

                                                                   {
        this->session = session;
 
        for (const auto& [packetIdentifier, publish] : session->publishMap) {
            LOG(INFO) << connectionName << " MQTT: PUBLISH Resend";
 
            send(publish);
        }
 
        for (const uint16_t packetIdentifier : session->pubrelPacketIdentifierSet) {
            LOG(INFO) << connectionName << " MQTT: PUBREL Resend";
 
            send(iot::mqtt::packets::Pubrel(packetIdentifier));
        }
 
        if (keepAlive > 0) {
            keepAlive *= 1.5;
 
            LOG(INFO) << connectionName << " MQTT: Keep alive initialized with: " << keepAlive;
 
            keepAliveTimer = core::timer::Timer::singleshotTimer(
                [this, keepAlive]() {
                    LOG(ERROR) << connectionName << " MQTT: Keep-alive timer expired. Interval was: " << keepAlive;
                    mqttContext->close();
                },
                keepAlive);
        }
 
        mqttContext->getSocketConnection()->setTimeout(0);
    }

onConnected()

void iot::mqtt::Mqtt::onConnected ( )

virtual

Definition at line 95 of file Mqtt.cpp.

                           {
        LOG(INFO) << "MQTT: Connected";
    }

Referenced by iot::mqtt::MqttContext::onConnected().

Here is the caller graph for this function:

onDisconnected()

void iot::mqtt::Mqtt::onDisconnected ( )

virtual

Definition at line 166 of file Mqtt.cpp.

                              {
        LOG(INFO) << connectionName << " MQTT: Disconnected";
    }

Referenced by iot::mqtt::MqttContext::onDisconnected().

Here is the caller graph for this function:

onPuback()

void iot::mqtt::Mqtt::onPuback ( const iot::mqtt::packets::Puback & puback )

protectedvirtual

Definition at line 256 of file Mqtt.cpp.

                                                                           {
    }

onPubcomp()

void iot::mqtt::Mqtt::onPubcomp ( const iot::mqtt::packets::Pubcomp & pubcomp )

protectedvirtual

Definition at line 265 of file Mqtt.cpp.

                                                                              {
    }

onPublish()

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

protectedvirtual

Definition at line 253 of file Mqtt.cpp.

                                                                       {
    }

onPubrec()

void iot::mqtt::Mqtt::onPubrec ( const iot::mqtt::packets::Pubrec & pubrec )

protectedvirtual

Definition at line 259 of file Mqtt.cpp.

                                                                           {
    }

onPubrel()

void iot::mqtt::Mqtt::onPubrel ( const iot::mqtt::packets::Pubrel & pubrel )

protectedvirtual

Definition at line 262 of file Mqtt.cpp.

                                                                           {
    }

onReceivedFromPeer()

std::size_t iot::mqtt::Mqtt::onReceivedFromPeer ( )

private

Definition at line 99 of file Mqtt.cpp.

                                       {
        std::size_t consumed = 0;
 
        switch (state) {
            case 0:
                consumed = fixedHeader.deserialize(mqttContext);
 
                if (!fixedHeader.isComplete()) {
                    break;
                }
                if (fixedHeader.isError()) {
                    mqttContext->close();
                    break;
                }
                printFixedHeader(fixedHeader);
 
                controlPacketDeserializer = createControlPacketDeserializer(fixedHeader);
 
                fixedHeader.reset();
 
                if (controlPacketDeserializer == nullptr) {
                    LOG(DEBUG) << connectionName << " MQTT: Received packet-type is unavailable ... closing connection";
 
                    mqttContext->end(true);
                    break;
                }
                if (controlPacketDeserializer->isError()) {
                    LOG(DEBUG) << connectionName << " MQTT: Fixed header has error ... closing connection";
 
                    delete controlPacketDeserializer;
                    controlPacketDeserializer = nullptr;
 
                    mqttContext->end(true);
                    break;
                }
 
                state++;
 
                [[fallthrough]];
            case 1:
                consumed += controlPacketDeserializer->deserialize(mqttContext);
 
                if (controlPacketDeserializer->isError()) {
                    LOG(DEBUG) << connectionName << " MQTT: Control packet has error ... closing connection";
                    mqttContext->end(true);
 
                    delete controlPacketDeserializer;
                    controlPacketDeserializer = nullptr;
 
                    state = 0;
                } else if (controlPacketDeserializer->isComplete()) {
                    deliverPacket(controlPacketDeserializer);
 
                    delete controlPacketDeserializer;
                    controlPacketDeserializer = nullptr;
 
                    state = 0;
 
                    keepAliveTimer.restart();
                }
 
                break;
        }
 
        return consumed;
    }

Referenced by iot::mqtt::MqttContext::onReceivedFromPeer().

Here is the caller graph for this function:

onSignal()

virtual bool iot::mqtt::Mqtt::onSignal ( int sig )

pure virtual

Implemented in iot::mqtt::client::Mqtt, and iot::mqtt::server::Mqtt.

Referenced by iot::mqtt::MqttContext::onSignal().

Here is the caller graph for this function:

operator=() [1/2]

Mqtt & iot::mqtt::Mqtt::operator= ( const Mqtt & )

delete

operator=() [2/2]

Mqtt & iot::mqtt::Mqtt::operator= ( Mqtt && )

delete

printFixedHeader()

void iot::mqtt::Mqtt::printFixedHeader ( const iot::mqtt::FixedHeader & fixedHeader ) const

protected

Definition at line 377 of file Mqtt.cpp.

                                                                    {
        LOG(INFO) << connectionName << " MQTT: ======================================================";
 
        LOG(TRACE) << connectionName << " MQTT: Received data (fixed header):\n" << toHexString(fixedHeader.serialize());
 
        LOG(DEBUG) << connectionName << " MQTT: Fixed Header: PacketType: 0x" << std::hex << std::setfill('0') << std::setw(2)
                   << static_cast<uint16_t>(fixedHeader.getType()) << " (" << iot::mqtt::mqttPackageName[fixedHeader.getType()] << ")";
        LOG(DEBUG) << connectionName << " MQTT:   PacketFlags: 0x" << std::hex << std::setfill('0') << std::setw(2)
                   << static_cast<uint16_t>(fixedHeader.getFlags()) << std::dec;
        LOG(DEBUG) << connectionName << " MQTT:   RemainingLength: " << fixedHeader.getRemainingLength();
    }

printVP()

void iot::mqtt::Mqtt::printVP ( const iot::mqtt::ControlPacket & packet ) const

protected

Definition at line 368 of file Mqtt.cpp.

                                                               {
        LOG(INFO) << connectionName << " MQTT: " << packet.getName() << " received: " << clientId;
 
        const std::string hexString = toHexString(packet.serializeVP());
        if (!hexString.empty()) {
            LOG(TRACE) << connectionName << " MQTT: Received data (variable header and payload):\n" << hexString;
        }
    }

send() [1/2]

void iot::mqtt::Mqtt::send ( const iot::mqtt::ControlPacket & controlPacket ) const

protected

Definition at line 205 of file Mqtt.cpp.

                                                            {
        LOG(INFO) << connectionName << " MQTT: " << controlPacket.getName() << " send: " << clientId;
 
        send(controlPacket.serialize());
    }

send() [2/2]

void iot::mqtt::Mqtt::send ( const std::vector< char > & data ) const

private

Definition at line 211 of file Mqtt.cpp.

                                                     {
        LOG(TRACE) << connectionName << " MQTT: Send data (full message):\n" << toHexString(data);
 
        mqttContext->send(data.data(), data.size());
    }

sendPuback()

void iot::mqtt::Mqtt::sendPuback

(

uint16_t

packetIdentifier

)

const

Definition at line 237 of file Mqtt.cpp.

                                                         { // Server & Client
        send(iot::mqtt::packets::Puback(packetIdentifier));
    }

sendPubcomp()

void iot::mqtt::Mqtt::sendPubcomp

(

uint16_t

packetIdentifier

)

const

Definition at line 249 of file Mqtt.cpp.

                                                          { // Server & Client
        send(iot::mqtt::packets::Pubcomp(packetIdentifier));
    }

sendPublish()

void iot::mqtt::Mqtt::sendPublish	(	const std::string &	topic,
		const std::string &	message,
		uint8_t	qoS,
		bool	retain )

Definition at line 217 of file Mqtt.cpp.

                                        { // Server & Client
 
        uint16_t packageIdentifier = qoS != 0 ? getPacketIdentifier() : 0;
 
        send(iot::mqtt::packets::Publish(packageIdentifier, topic, message, qoS, false, retain));
 
        LOG(INFO) << connectionName << " MQTT:   Topic: " << topic;
        LOG(INFO) << connectionName << " MQTT:   Message:\n" << toHexString(message);
        LOG(DEBUG) << connectionName << " MQTT:   QoS: " << static_cast<uint16_t>(qoS);
        LOG(DEBUG) << connectionName << " MQTT:   PacketIdentifier: " << _packetIdentifier;
        LOG(DEBUG) << connectionName << " MQTT:   DUP: " << false;
        LOG(DEBUG) << connectionName << " MQTT:   Retain: " << retain;
 
        if (qoS == 2) {
            session->publishMap.emplace(packageIdentifier,
                                        iot::mqtt::packets::Publish(packageIdentifier, topic, message, qoS, true, retain));
        }
    }

sendPubrec()

void iot::mqtt::Mqtt::sendPubrec

(

uint16_t

packetIdentifier

)

const

Definition at line 241 of file Mqtt.cpp.

                                                         { // Server & Client
        send(iot::mqtt::packets::Pubrec(packetIdentifier));
    }

sendPubrel()

void iot::mqtt::Mqtt::sendPubrel

(

uint16_t

packetIdentifier

)

const

Definition at line 245 of file Mqtt.cpp.

                                                         { // Server & Client
        send(iot::mqtt::packets::Pubrel(packetIdentifier));
    }

setMqttContext()

void iot::mqtt::Mqtt::setMqttContext ( MqttContext * mqttContext )

private

Definition at line 87 of file Mqtt.cpp.

                                                      {
        this->mqttContext = mqttContext;
    }

Referenced by iot::mqtt::MqttContext::MqttContext().

Here is the caller graph for this function:

toHexString() [1/2]

std::string iot::mqtt::Mqtt::toHexString ( const std::string & data )

static

Definition at line 394 of file Mqtt.cpp.

                                                     {
        return toHexString(std::vector<char>(data.begin(), data.end()));
    }

toHexString() [2/2]

std::string iot::mqtt::Mqtt::toHexString ( const std::vector< char > & data )

static

Definition at line 389 of file Mqtt.cpp.

                                                           {
        const std::string hexDump = utils::hexDump(data, 32);
        return !hexDump.empty() ? std::string(32, ' ').append(hexDump) : "";
    }

Friends And Related Symbol Documentation

MqttContext

friend class MqttContext

friend

Definition at line 160 of file Mqtt.h.

Member Data Documentation

_packetIdentifier

uint16_t iot::mqtt::Mqtt::_packetIdentifier = 0

private

Definition at line 149 of file Mqtt.h.

clientId

std::string iot::mqtt::Mqtt::clientId

protected

Definition at line 143 of file Mqtt.h.

connectionName

std::string iot::mqtt::Mqtt::connectionName

protected

Definition at line 142 of file Mqtt.h.

controlPacketDeserializer

iot::mqtt::ControlPacketDeserializer* iot::mqtt::Mqtt::controlPacketDeserializer = nullptr

private

Definition at line 147 of file Mqtt.h.

fixedHeader

iot::mqtt::FixedHeader iot::mqtt::Mqtt::fixedHeader

private

Definition at line 146 of file Mqtt.h.

keepAliveTimer

core::timer::Timer iot::mqtt::Mqtt::keepAliveTimer

private

Definition at line 151 of file Mqtt.h.

mqttContext

MqttContext* iot::mqtt::Mqtt::mqttContext = nullptr

protected

Definition at line 158 of file Mqtt.h.

session

Session* iot::mqtt::Mqtt::session = nullptr

private

Definition at line 155 of file Mqtt.h.

state

int iot::mqtt::Mqtt::state = 0

private

Definition at line 153 of file Mqtt.h.

---

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

• iot/mqtt/Mqtt.h
• iot/mqtt/Mqtt.cpp