datastream_mqtt.cpp

Go to the documentation of this file.

#include "datastream_mqtt.h"
#include "ui_datastream_mqtt.h"
#include <QMessageBox>
#include <QSettings>
#include <QDebug>
#include <QUuid>
#include <QIntValidator>
#include <QMessageBox>

DataStreamMQTT::DataStreamMQTT():
  _running(false)
{
   _notification_action = new QAction(this);

  connect(_notification_action, &QAction::triggered, this, [this]() {
    QMessageBox::warning(nullptr, "MQTT error",
                         QString("Failed to parse %1 messages").arg(_failed_parsing),
                         QMessageBox::Ok);

    if (_failed_parsing > 0)
    {
      _failed_parsing = 0;
      emit notificationsChanged(_failed_parsing);
    }
  });

  _mosq = std::make_shared<MQTTClient>();
  _dialog = new MQTT_Dialog(_mosq);
}

DataStreamMQTT::~DataStreamMQTT()
{
  shutdown();
  delete _dialog;
}

bool DataStreamMQTT::start(QStringList *)
{
  if (_running)
  {
    return _running;
  }

  //cleanup notifications
  _failed_parsing = 0;
  emit notificationsChanged(0);

  if( parserFactories() == nullptr || parserFactories()->empty() )
  {
    QMessageBox::warning(nullptr,tr("MQTT Client"), tr("No available MessageParsers"),  QMessageBox::Ok);
    _running = false;
    return false;
  }

  bool first_start = _dialog->ui->comboBoxProtocol->count() == 0;

  if( first_start )
  {
    // change the section of the dialog related to protocols
    for( const auto& it: *parserFactories())
    {
      _dialog->ui->comboBoxProtocol->addItem( it.first );

      if(auto widget = it.second->optionsWidget() )
      {
        widget->setVisible(false);
        _dialog->ui->layoutOptions->addWidget( widget );
      }
    }

    connect(_dialog->ui->comboBoxProtocol,
            qOverload<const QString &>(&QComboBox::currentIndexChanged),
            this, &DataStreamMQTT::onComboProtocolChanged);
  }

  _running = false;

  QSettings settings;
  _protocol = settings.value("MosquittoMQTT::serialization_protocol", "JSON").toString();
  _dialog->ui->comboBoxProtocol->setCurrentText(_protocol);

  if( _dialog->exec() == QDialog::Rejected )
  {
    return false;
  }
  _protocol = _dialog->ui->comboBoxProtocol->currentText();

  // remove all previous subscriptions and create new ones
  for( const auto& topic: _mosq->config().topics )
  {
    _mosq->unsubscribe(topic);
  }

  for (const auto& item: _dialog->ui->listWidget->selectedItems())
  {
    MQTTClient::TopicCallback callback = [this](const mosquitto_message* message)
    {
      onMessageReceived(message);
    };

    std::string topic_name = item->text().toStdString();
    _mosq->subscribe(topic_name, _mosq->config().qos);
    _mosq->addMessageCallback(topic_name, callback);
  }

  _running = true;
  return _running;
}

void DataStreamMQTT::shutdown()
{
  if( _running )
  {
    _running = false;
    _parsers.clear();
    _topic_to_parse.clear();
    dataMap().clear();
  }
}

bool DataStreamMQTT::isRunning() const
{
  return _running;
}


void DataStreamMQTT::onComboProtocolChanged(const QString& selected_protocol)
{
  if (_current_parser_creator)
  {
    if( auto prev_widget = _current_parser_creator->optionsWidget())
    {
      prev_widget->setVisible(false);
    }
  }
  _current_parser_creator = parserFactories()->at(selected_protocol);

  if (auto widget = _current_parser_creator->optionsWidget())
  {
    widget->setVisible(true);
  }
}

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

  auto it = _parsers.find(message->topic);
  if( it == _parsers.end() )
  {
    auto& parser_factory = parserFactories()->at( _protocol );
    auto parser = parser_factory->createParser({message->topic}, {}, {}, dataMap());
    it = _parsers.insert( {message->topic, parser} ).first;
  }
  auto& parser = it->second;

  bool result = false;
  try {
    MessageRef msg( static_cast<uint8_t*>(message->payload), message->payloadlen);

    using namespace std::chrono;
    auto ts = high_resolution_clock::now().time_since_epoch();
    double timestamp = 1e-6* double( duration_cast<microseconds>(ts).count() );

    result = parser->parseMessage(msg, timestamp);
  }
  catch (std::exception& ) {}

  emit dataReceived();

  if( !result )
  {
    _failed_parsing++;
    emit notificationsChanged(_failed_parsing);
  }
}