/* **************************************************************************** * Copyright 2019 Open Systems Development BV * * * * Permission is hereby granted, free of charge, to any person obtaining a * * copy of this software and associated documentation files (the "Software"), * * to deal in the Software without restriction, including without limitation * * the rights to use, copy, modify, merge, publish, distribute, sublicense, * * and/or sell copies of the Software, and to permit persons to whom the * * Software is furnished to do so, subject to the following conditions: * * * * The above copyright notice and this permission notice shall be included in * * all copies or substantial portions of the Software. * * * * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR * * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, * * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL * * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER * * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING * * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER * * DEALINGS IN THE SOFTWARE. * * ***************************************************************************/ #include "qmqtt.h" #include "log.h" using namespace osdev::components; QMQTTClient::QMQTTClient( const QMqttConfigSettings& connectionSettings, QObject *parent ) : QObject(parent) , m_qhConnections() , m_connConfig( connectionSettings ) { } QMQTTClient::~QMQTTClient() { // Clean up any connections we have. for( auto key : m_qhConnections.keys()) { QPointer pTemp = m_qhConnections.take( key ); if(pTemp) { delete pTemp.data(); } } } void QMQTTClient::subscribe( const QString& mqtt_topic ) { if(!topicExists( mqtt_topic ) ) { if( !createNewThread( mqtt_topic ) ) { LogError("[QMQTTClient::subscribe]", QString("There was an error creating a subscription to : %1").arg(mqtt_topic)); } } if( topicExists( mqtt_topic ) ) { m_qhConnections.value( mqtt_topic )->subscribe( mqtt_topic ); } } void QMQTTClient::unsubscribe(const QString& mqtt_topic) { if( topicExists( mqtt_topic ) ) { m_qhConnections.value( mqtt_topic )->unsubscribe( mqtt_topic ); } } void QMQTTClient::publish(const QString& mqtt_topic, const QString& mqtt_message) { if( !topicExists( mqtt_topic ) ) { if( !createNewThread( mqtt_topic ) ) { LogError("[QMQTTClient::publish]", QString("There was an error creating publishing to : %1").arg( mqtt_topic ) ); } } if(topicExists(mqtt_topic)) { m_qhConnections.value(mqtt_topic)->publish(mqtt_topic, mqtt_message); } } bool QMQTTClient::createNewThread(const QString& topic) { QPointer pThread = new QMqttPubSubClient(m_connConfig.getHostAddress(), m_connConfig.getPortNumber() ); if(pThread) { connect( pThread.data(), &QMqttPubSubClient::signalMessageReceived, this, &QMQTTClient::signalNewMessageReceived ); connect( pThread.data(), &QMqttPubSubClient::signalReInitialise, this, &QMQTTClient::signalReInitialize ); m_qhConnections.insert( topic, pThread ); } return m_qhConnections.contains( topic ); } bool QMQTTClient::topicExists( const QString& topic ) { if( m_qhConnections.contains( topic ) ) return true; else return false; }