mqtt_client.zip

/****************************************************************************** ** ** Copyright (C) 2017 The Qt Company Ltd. ** Contact: https://www.qt.io/licensing/ ** ** This file is part of the QtMqtt module. ** ** $QT_BEGIN_LICENSE:GPL$ ** Commercial License Usage ** Licensees holding valid commercial Qt licenses may use this file in ** accordance with the commercial license agreement provided with the ** Software or, alternatively, in accordance with the terms contained in ** a written agreement between you and The Qt Company. For licensing terms ** and conditions see https://www.qt.io/terms-conditions. For further ** information use the contact form at https://www.qt.io/contact-us. ** ** GNU General Public License Usage ** Alternatively, this file may be used under the terms of the GNU ** General Public License version 3 or (at your option) any later version ** approved by the KDE Free Qt Foundation. The licenses are as published by ** the Free Software Foundation and appearing in the file LICENSE.GPL3 ** included in the packaging of this file. Please review the following ** information to ensure the GNU General Public License requirements will ** be met: https://www.gnu.org/licenses/gpl-3.0.html. ** ** $QT_END_LICENSE$ ** ******************************************************************************/ #include "qmqttconnection_p.h" #include "qmqttconnectionproperties_p.h" #include "qmqttcontrolpacket_p.h" #include "qmqttmessage_p.h" #include "qmqttpublishproperties_p.h" #include "qmqttsubscription_p.h" #include "qmqttclient_p.h" #include <QtCore/QLoggingCategory> #include <QtNetwork/QSslSocket> #include <QtNetwork/QTcpSocket> #include <limits> #include <cstdint> QT_BEGIN_NAMESPACE Q_LOGGING_CATEGORY(lcMqttConnection, "qt.mqtt.connection") Q_LOGGING_CATEGORY(lcMqttConnectionVerbose, "qt.mqtt.connection.verbose"); template <typename T> T QMqttConnection::readBufferTyped(qint64 *dataSize) { Q_STATIC_ASSERT(std::is_integral<T>::value); T result; readBuffer(reinterpret_cast<char *>(&result), sizeof(result)); if (dataSize != nullptr) *dataSize -= sizeof(result); return qFromBigEndian(result); } template<> QByteArray QMqttConnection::readBufferTyped(qint64 *dataSize) { const quint16 size = readBufferTyped<quint16>(dataSize); QByteArray ba(int(size), Qt::Uninitialized); readBuffer(ba.data(), size); if (dataSize) *dataSize -= size; return ba; } template<> QString QMqttConnection::readBufferTyped(qint64 *dataSize) { return QString::fromUtf8(readBufferTyped<QByteArray>(dataSize)); } QMqttConnection::QMqttConnection(QObject *parent) : QObject(parent) { } QMqttConnection::~QMqttConnection() { if (m_internalState == BrokerConnected) sendControlDisconnect(); if (m_ownTransport && m_transport) delete m_transport; } void QMqttConnection::timerEvent(QTimerEvent *event) { if (Q_LIKELY(event->timerId() == m_pingTimer.timerId())) { sendControlPingRequest(); return; } QObject::timerEvent(event); } void QMqttConnection::setTransport(QIODevice *device, QMqttClient::TransportType transport) { qCDebug(lcMqttConnection) << Q_FUNC_INFO << device << " Type:" << transport; if (m_transport) { disconnect(m_transport, &QIODevice::aboutToClose, this, &QMqttConnection::transportConnectionClosed); disconnect(m_transport, &QIODevice::readyRead, this, &QMqttConnection::transportReadyRead); if (m_ownTransport) delete m_transport; } m_transport = device; m_transportType = transport; m_ownTransport = false; connect(m_transport, &QIODevice::aboutToClose, this, &QMqttConnection::transportConnectionClosed); connect(m_transport, &QIODevice::readyRead, this, &QMqttConnection::transportReadyRead); } QIODevice *QMqttConnection::transport() const { return m_transport; } bool QMqttConnection::ensureTransport(bool createSecureIfNeeded) { Q_UNUSED(createSecureIfNeeded); // QT_NO_SSL qCDebug(lcMqttConnection) << Q_FUNC_INFO << m_transport; if (m_transport) { if (m_ownTransport) delete m_transport; else return true; } // We are asked to create a transport layer if (m_clientPrivate->m_hostname.isEmpty() || m_clientPrivate->m_port == 0) { qCDebug(lcMqttConnection) << "No hostname specified, not able to create a transport layer."; return false; } auto socket = #ifndef QT_NO_SSL createSecureIfNeeded ? new QSslSocket() : #endif new QTcpSocket(); m_transport = socket; m_ownTransport = true; m_transportType = #ifndef QT_NO_SSL createSecureIfNeeded ? QMqttClient::SecureSocket : #endif QMqttClient::AbstractSocket; #ifndef QT_NO_SSL if (QSslSocket *sslSocket = qobject_cast<QSslSocket *>(socket)) QObject::connect(sslSocket, &QSslSocket::encrypted, this, &QMqttConnection::transportConnectionEstablished); else #endif connect(socket, &QAbstractSocket::connected, this, &QMqttConnection::transportConnectionEstablished); connect(socket, &QAbstractSocket::disconnected, this, &QMqttConnection::transportConnectionClosed); connect(socket, static_cast<void (QAbstractSocket::*)(QAbstractSocket::SocketError)>(&QAbstractSocket::error), this, &QMqttConnection::transportError); connect(m_transport, &QIODevice::aboutToClose, this, &QMqttConnection::transportConnectionClosed); connect(m_transport, &QIODevice::readyRead, this, &QMqttConnection::transportReadyRead); return true; } bool QMqttConnection::ensureTransportOpen(const QString &sslPeerName) { qCDebug(lcMqttConnection) << Q_FUNC_INFO << m_transportType; if (m_transportType == QMqttClient::IODevice) { if (m_transport->isOpen()) return sendControlConnect(); if (!m_transport->open(QIODevice::ReadWrite)) { qCDebug(lcMqttConnection) << "Could not open Transport IO device."; m_internalState = BrokerDisconnected; return false; } return sendControlConnect(); } if (m_transportType == QMqttClient::AbstractSocket) { auto socket = qobject_cast<QTcpSocket *>(m_transport); Q_ASSERT(socket); if (socket->state() == QAbstractSocket::ConnectedState) return sendControlConnect(); m_internalState = BrokerConnecting; socket->connectToHost(m_clientPrivate->m_hostname, m_clientPrivate->m_port); } #ifndef QT_NO_SSL else if (m_transportType == QMqttClient::SecureSocket) { auto socket = qobject_cast<QSslSocket *>(m_transport); Q_ASSERT(socket); if (socket->state() == QAbstractSocket::ConnectedState) return sendControlConnect(); m_internalState = BrokerConnecting; if (!m_sslConfiguration.isNull()) socket->setSslConfiguration(m_sslConfiguration); socket->connectToHostEncrypted(m_clientPrivate->m_hostname, m_clientPrivate->m_port, sslPeerName); } #else Q_UNUSED(sslPeerName); #endif return true; } bool QMqttConnection::sendControlConnect() { qCDebug(lcMqttConnection) << Q_FUNC_INFO; QMqttControlPacket packet(QMqttControlPacket::CONNECT); // Variable header // 3.1.2.1 Protocol Name // 3.1.2.2 Protocol Level switch (m_clientPrivate->m_protocolVersion) { case QMqttClient::MQTT_3_1: packet.append("MQIsdp"); packet.append(char(3)); // Version 3.1 break; case QMqttClient::MQTT_3_1_1: packet.append("MQTT"); packet.append(char(4)); // Version 3.1.1 break; case QMqttClient::MQTT_5_0: packet.append("MQTT"); packet.append(char(5)); // Version 5.0 break; } // 3.1.2.3 Connect Flags quint8 flags = 0; // Clean session if (m_clientPrivate->m_cleanSession) flags |= 1 << 1; if (!m_clientPrivate-