MosquittoClient.h

#pragma once

#include <list>
#include <queue>
#include <set>
#include <string>

#include <Poco/AtomicCounter.h>
#include <Poco/Event.h>
#include <Poco/Mutex.h>
#include <Poco/SharedPtr.h>
#include <Poco/Timespan.h>

#include <mosquittopp.h>

#include "loop/StoppableRunnable.h"
#include "net/MqttClient.h"
#include "util/Loggable.h"

namespace BeeeOn {

class MosquittoClient:
        Loggable,
        protected mosqpp::mosquittopp,
        public StoppableRunnable,
        public MqttClient {
public:
        typedef Poco::SharedPtr<MosquittoClient> Ptr;

        MosquittoClient();
        ~MosquittoClient();

        void run() override;
        void stop() override;

        void setHost(const std::string &host);
        void setPort(int port);

        void setSubTopics(const std::list<std::string> &subTopics);

        void setReconnectTimeout(const Poco::Timespan &timeout);

        void setClientID(const std::string &id);
        std::string clientID() const;

        void publish(const MqttMessage &msq) override;

        MqttMessage receive(const Poco::Timespan &timeout) override;

protected:
        virtual std::string buildClientID() const;

        void connect();

private:
        bool initConnection();

        void subscribeToAll();

        void throwMosquittoError(int returnCode) const;

        void on_message(const struct mosquitto_message *message) override;

        MqttMessage nextMessage();

private:
        std::string m_clientID;
        std::string m_host;
        Poco::Timespan m_reconnectTimeout;
        int m_port;
        std::set<std::string> m_subTopics;
        Poco::AtomicCounter m_stop;
        Poco::Event m_receiveEvent;
        Poco::Event m_reconnectEvent;
        Poco::FastMutex m_queueMutex;
        std::queue<MqttMessage> m_msgQueue;
};

}