qtmqtt客户端（支持ssl）亲测可用

/****************************************************************************** ** ** 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<> quint32 QMqttConnection::readBufferTyped(qint64 *dataSize) { if (dataSize) *dataSize -= sizeof(quint32); return qFromBigEndian<quint32>(reinterpret_cast<const quint32 *>(readBuffer(4).constData())); } template<> quint16 QMqttConnection::readBufferTyped(qint64 *dataSize) { if (dataSize) *dataSize -= sizeof(quint16); return qFromBigEndian<quint16>(reinterpret_cast<const quint16 *>(readBuffer(2).constData())); } template<> quint8 QMqttConnection::readBufferTyped(qint64 *dataSize) { quint8 result; readBuffer(reinterpret_cast<char *>(&result), 1); if (dataSize) *dataSize -= sizeof(quint8); return result; } template<> QString QMqttConnection::readBufferTyped(qint64 *dataSize) { const quint16 size = readBufferTyped<quint16>(dataSize); if (dataSize) *dataSize -= size; const QByteArray ba = readBuffer(size); return QString::fromUtf8(reinterpret_cast<const char *>(ba.constData()), ba.size()); } template<> QByteArray QMqttConnection::readBufferTyped(qint64 *dataSize) { const quint16 size = readBufferTyped<quint16>(dataSize); if (dataSize) *dataSize -= size; return readBuffer(size); } QMqttConnection::QMqttConnection(QObject *parent) : QObject(parent) { m_pingTimer.setSingleShot(false); m_pingTimer.connect(&m_pingTimer, &QTimer::timeout, this, &QMqttConnection::sendControlPingRequest); } QMqttConnection::~QMqttConnection() { if (m_internalState == BrokerConnected) sendControlDisconnect(); if (m_ownTransport && m_transport) delete m_transport; } 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::transportReadReady); 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::transportReadReady); } 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; 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::transportReadReady); 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 = dynamic_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 = dynamic_cast<QSslSocket*>(m_transport); Q_ASSERT(socket); if (socket->state() == QAbstractSocket::ConnectedState) return sendControlConnect(); m_internalState = BrokerConnecting; socket->connectToHostEncrypted(m_clientPrivate->m_hostname, m_clientPrivate->m_port, sslPeerName); if (!socket->waitForConnected()) { qCDebug(lcMqttConnection) << "Could not establish socket connection for transport."; return false; } if (!socket->waitForEncrypted()) { qCDebug(lcMqttConnection) << "Could not initiate encryption."; return false; } } #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 QMqttC