mqtt_client.cpp

Go to the documentation of this file.

#include "mqtt_client.h"
#include <QDebug>
#include <QMessageBox>
#include <QString>

#ifdef WIN32
#include <windows.h>
#include <strsafe.h>
#endif

void connect_callback(struct mosquitto *mosq, void *context, int result,  int, const mosquitto_property *)
{
  MQTTClient* self = static_cast<MQTTClient*>(context);

  if( !result )
  {
    for(const auto& topic: self->config().topics)
    {
      mosquitto_subscribe(mosq, nullptr, topic.c_str(), self->config().qos);
    }
  }
  else
  {
    QMessageBox::warning(nullptr, "MQTT Client",
                         QString("Connection error: %1").arg(mosquitto_reason_string(result)),
                         QMessageBox::Ok);
  }
  self->_connected = true;
}

void disconnect_callback(struct mosquitto *mosq, void *context, int result)
{
  MQTTClient* self = static_cast<MQTTClient*>(context);

  if( self->isConnected() && result == MOSQ_ERR_CONN_LOST )
  {
    emit self->disconnected();
  }
}

void message_callback(struct mosquitto *mosq, void *context, const struct mosquitto_message *message, const mosquitto_property *)
{
  MQTTClient* self = static_cast<MQTTClient*>(context);
  self->onMessageReceived(message);
}

//----------------------------

MQTTClient::MQTTClient()
{
  mosquitto_lib_init();
  _mosq = mosquitto_new(nullptr, true, this);

  mosquitto_connect_v5_callback_set(_mosq, connect_callback);
  mosquitto_disconnect_callback_set(_mosq, disconnect_callback);
  mosquitto_message_v5_callback_set(_mosq, message_callback);
}

MQTTClient::~MQTTClient()
{
  if(_connected)
  {
    disconnect();
  }
  mosquitto_lib_cleanup();
}

bool MQTTClient::connect(const MosquittoConfig &config)
{
  if( _connected )
  {
    disconnect();
  }

  mosquitto_int_option(_mosq, MOSQ_OPT_PROTOCOL_VERSION, config.protocol_version);

  if(( !config.username.empty() || !config.password.empty()) )
  {
    if(mosquitto_username_pw_set(_mosq, config.username.c_str(), config.password.c_str()))
    {
      return false;
    }
  }

  if( config.cafile.empty() == false )
  {
    const char *cafile = config.cafile.c_str();
    const char *certfile = config.certfile.empty() ? nullptr: config.certfile.c_str();
    const char *keyfile = config.keyfile.empty() ? nullptr: config.keyfile.c_str();

    mosquitto_tls_set(_mosq, cafile, nullptr, certfile, keyfile, nullptr);
  }

  mosquitto_max_inflight_messages_set(_mosq, config.max_inflight);

  const mosquitto_property *properties = nullptr; // todo

  int rc = mosquitto_connect_bind_v5(_mosq,
                                     config.host.c_str(),
                                     config.port,
                                     config.keepalive,
                                     nullptr,
                                     properties);
  // TODO bind
  if(rc>0)
  {
    if(rc == MOSQ_ERR_ERRNO)
    {
      char err[1024];
#ifndef WIN32
      strerror_r(errno, err, 1024);
#else
      FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM, NULL, errno, 0, (LPTSTR)&err, 1024, NULL);
#endif
      QMessageBox::warning(nullptr, "MQTT Client",
                           QString("Error: %1").arg(err),  QMessageBox::Ok);
    }
    else
    {
      QMessageBox::warning(nullptr, "MQTT Client",
                           QString("Unable to connect (%1)").arg(mosquitto_strerror(rc)),
                           QMessageBox::Ok);
    }
    _connected = false;
    return false;
  }

  _connected = true;
  _config = config;
  mosquitto_loop_start(_mosq);
  return true;
}

void MQTTClient::disconnect()
{
  if( _connected )
  {
    mosquitto_disconnect(_mosq);
    mosquitto_loop_stop(_mosq, true);
  }
  _connected = false;
  _topics_set.clear();
  _message_callbacks.clear();
}

bool MQTTClient::isConnected() const
{
  return _connected;
}

void MQTTClient::addMessageCallback(const std::string &topic, MQTTClient::TopicCallback callback)
{
  std::unique_lock<std::mutex> lk(_mutex);
  _message_callbacks[topic] = callback;
}

void MQTTClient::onMessageReceived(const mosquitto_message * message)
{
  std::unique_lock<std::mutex> lk(_mutex);

  _topics_set.insert(message->topic);

  auto it = _message_callbacks.find(message->topic);
  if( it != _message_callbacks.end() )
  {
    it->second(message);
  }
}

const MosquittoConfig &MQTTClient::config() const
{
  return _config;
}

std::unordered_set<std::string> MQTTClient::getTopicList()
{
  std::unique_lock<std::mutex> lk(_mutex);
  return _topics_set;
}

void MQTTClient::subscribe(const std::string &topic, int qos)
{
  mosquitto_subscribe(_mosq, nullptr, topic.c_str(), qos);
}

void MQTTClient::unsubscribe(const std::string &topic)
{
  mosquitto_unsubscribe(_mosq, nullptr, topic.c_str());
}