Move communication into QMqttConnection

client itself is now detached, and communication fully asynchronous.

src/mqtt/mqtt.pro

@@ -12,10 +12,12 @@ PUBLIC_HEADERS += \
     qmqttclient.h
 
 PRIVATE_HEADERS += \
-    qmqttclient_p.h
+    qmqttclient_p.h \
+    qmqttconnection_p.h
 
 SOURCES += \
-    qmqttclient.cpp
+    qmqttclient.cpp \
+    qmqttconnection.cpp
 
 HEADERS += $$PUBLIC_HEADERS $$PRIVATE_HEADERS
 

src/mqtt/qmqttclient.cpp

@@ -38,23 +38,25 @@ QT_BEGIN_NAMESPACE
 
 QMqttClient::QMqttClient(QObject *parent) : QObject(*(new QMqttClientPrivate), parent)
 {
+    Q_D(QMqttClient);
+    d->m_connection.setClient(this);
 }
 
 void QMqttClient::setTransport(QIODevice *device, QMqttClient::TransportType transport)
 {
     Q_D(QMqttClient);
-    if (d->m_transport && d->m_ownTransport)
-        delete d->m_transport;
 
-    d->m_transport = device;
-    d->m_transportType = transport;
-    d->m_ownTransport = false;
+    if (d->m_state != Disconnected) {
+        qWarning("Changing transport layer while connected is not possible");
+        return;
+    }
+    d->m_connection.setTransport(device, transport);
 }
 
 QIODevice *QMqttClient::transport() const
 {
     Q_D(const QMqttClient);
-    return d->m_transport;
+    return d->m_connection.transport();
 }
 
 bool QMqttClient::subscribe(const QString &topic)
@@ -82,124 +84,29 @@ quint16 QMqttClient::port() const
     return d->m_port;
 }
 
-inline void addInt(QByteArray &data, quint16 n)
-{
-    const quint16 msb = qToBigEndian<quint16>(n);
-    const char * msb_c = reinterpret_cast<const char*>(&msb);
-    data.append(msb_c[0]);
-    data.append(msb_c[1]);
-    return;
-}
-
 void QMqttClient::connectToHost()
 {
     Q_D(QMqttClient);
-    if (d->m_transport == nullptr) {
-        // We are asked to create a transport layer
-        if (d->m_hostname.isEmpty() || d->m_port == 0) {
-            qWarning("No hostname specified");
-            return;
-        }
-        auto socket = new QTcpSocket();
-        d->m_transport = socket;
-        d->m_ownTransport = true;
-        d->m_transportType = TransportType::AbstractSocket;
 
-        connect(d->m_transport, &QIODevice::aboutToClose, this, &QMqttClient::disconnected);
+    if (!d->m_connection.ensureTransport()) {
+        qWarning("Could not ensure connection");
+        setState(Disconnected);
+        return;
     }
-
     setState(Connecting);
 
-    // Check for open / connected
-    if (!d->m_transport->isOpen()) {
-        if (d->m_transportType == TransportType::IODevice) {
-            if (d->m_transport->open(QIODevice::ReadWrite)) {
-                qWarning("Could not open Transport IO device");
-                setState(Disconnected);
-                return;
-            }
-        } else if (d->m_transportType == TransportType::AbstractSocket) {
-            auto socket = dynamic_cast<QTcpSocket*>(d->m_transport);
-            Q_ASSERT(socket);
-            socket->connectToHost(d->m_hostname, d->m_port);
-            if (!socket->waitForConnected()) {
-                qWarning("Could not establish socket connection for transport");
-                setState(Disconnected);
-                return;
-            }
-        }
-    }
-
-    QByteArray connectFrame;
-    // Fixed header
-    connectFrame += 0x10;
-
-    connectFrame += char(0); // Length to be filled later
-    // Variable header
-    // 3.1.2.1 Protocol Name
-    // 3.1.2.2 Protocol Level
-    if (d->m_protocolVersion == 3) {
-        const QByteArray other("MQIsdp");
-        addInt(connectFrame, other.size());
-        connectFrame += other.data();
-        connectFrame += char(3); // Version 3.1
-    } else if (d->m_protocolVersion == 4) {
-        connectFrame += char(0); // Length MSB
-        connectFrame += 4; // Length LSB
-        connectFrame += "MQTT";
-        connectFrame += char(4); // Version 3.1.1
-    } else {
-        qFatal("Illegal MQTT VERSION");
-    }
-
-    // 3.1.2.3 Connect Flags
-    quint8 flags = 0;
-    // Clean session
-    flags |= 1;
-    connectFrame += char(flags);
-
-    // 3.1.2.10 Keep Alive
-    addInt(connectFrame, d->m_keepAlive);
-
-    // 3.1.3 Payload
-    // 3.1.3.1 Client Identifier
-
-    if (d->m_protocolVersion == 3) { // no username
-        connectFrame += char(0);
-        connectFrame += char(0);
-    } else if (d->m_protocolVersion == 4) {
-        // Client id maximum left is 23
-        const QByteArray clientStringArray = d->m_clientId.left(23).toUtf8();
-        if (clientStringArray.size()) {
-            addInt(connectFrame, clientStringArray.size());
-            connectFrame += clientStringArray;
-        } else {
-            connectFrame += char(0);
-            connectFrame += char(0);
-        }
-    }
-
-    const quint8 frameLength = connectFrame.size();
-    connectFrame[1] = char(frameLength - 2); // The header bytes are not included
-
-    d->m_transport->write(connectFrame.constData(), frameLength);
-    if (!d->m_transport->waitForBytesWritten(30 * 1000)) {
-        qWarning("Could not send data");
-        qDebug() << "Error:" << d->m_transport->errorString();
+    if (!d->m_connection.ensureTransportOpen()) {
+        qWarning("Could not ensure that connection is open");
+        setState(Disconnected);
         return;
     }
 
-    if (!d->m_transport->waitForReadyRead(30 * 1000)) {
-        qWarning("Did not get an ACK within time");
-        qDebug() << "Error:" << d->m_transport->errorString();
+    if (!d->m_connection.sendControlConnect()) {
+        qWarning("Could not send CONNECT to broker");
+        // ### Who disconnects now? Connection or client?
+        setState(Disconnected);
         return;
     }
-
-    QByteArray result = d->m_transport->readAll();
-    qDebug() << "Result content:" << result;
-
-    // For now we assume everything is fine...
-    setState(Connected);
 }
 
 void QMqttClient::disconnectFromHost()
@@ -302,9 +209,6 @@ QMqttClientPrivate::QMqttClientPrivate()
 
 QMqttClientPrivate::~QMqttClientPrivate()
 {
-    // ### TODO: check for open connection and quit gracefully
-    if (m_ownTransport && m_transport)
-        delete m_transport;
 }
 
 QT_END_NAMESPACE

src/mqtt/qmqttclient_p.h

@@ -45,6 +45,7 @@
 
 #include <QtNetwork/QAbstractSocket>
 
+#include <private/qmqttconnection_p.h>
 #include <private/qobject_p.h>
 
 QT_BEGIN_NAMESPACE
@@ -57,9 +58,7 @@ public:
     ~QMqttClientPrivate();
     QString m_hostname;
     quint16 m_port{0};
-    QIODevice *m_transport{nullptr};
-    QMqttClient::TransportType m_transportType{QMqttClient::IODevice};
-    bool m_ownTransport{false};
+    QMqttConnection m_connection;
     QString m_clientId; // auto-generated
     quint16 m_keepAlive{60};
     // 3 == MQTT Standard 3.1

src/mqtt/qmqttconnection.cpp

@@ -0,0 +1,276 @@
+/****************************************************************************
+**
+** Copyright (C) 2017 The Qt Company Ltd.
+** Contact: https://www.qt.io/licensing/
+**
+** This file is part of the Qt Mqtt module of the Qt Toolkit.
+**
+** $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 <QtNetwork/QTcpSocket>
+
+QT_BEGIN_NAMESPACE
+
+QMqttConnection::QMqttConnection(QObject *parent) : QObject(parent)
+{
+
+}
+
+QMqttConnection::~QMqttConnection()
+{
+    // ### TODO: check for open connection and quit gracefully
+    if (m_ownTransport && m_transport)
+        delete m_transport;
+}
+
+void QMqttConnection::setTransport(QIODevice *device, QMqttClient::TransportType 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()
+{
+    if (m_transport)
+        return true;
+
+    // We are asked to create a transport layer
+    if (m_client->hostname().isEmpty() || m_client->port() == 0) {
+        qWarning("No hostname specified");
+        return false;
+    }
+    auto socket = new QTcpSocket();
+    m_transport = socket;
+    m_ownTransport = true;
+    m_transportType = QMqttClient::AbstractSocket;
+
+    connect(socket, &QAbstractSocket::disconnected, this, &QMqttConnection::transportConnectionClosed);
+    connect(m_transport, &QIODevice::aboutToClose, this, &QMqttConnection::transportConnectionClosed);
+    connect(m_transport, &QIODevice::readyRead, this, &QMqttConnection::transportReadReady);
+    return true;
+}
+
+bool QMqttConnection::ensureTransportOpen()
+{
+    if (m_transport->isOpen())
+        return true;
+
+    if (m_transportType == QMqttClient::IODevice) {
+        if (m_transport->open(QIODevice::ReadWrite)) {
+            qWarning("Could not open Transport IO device");
+            return false;
+        }
+    } else if (m_transportType == QMqttClient::AbstractSocket) {
+        auto socket = dynamic_cast<QTcpSocket*>(m_transport);
+        Q_ASSERT(socket);
+        socket->connectToHost(m_client->hostname(), m_client->port());
+        if (!socket->waitForConnected()) {
+            qWarning("Could not establish socket connection for transport");
+            return false;
+        }
+    }
+    return true;
+}
+
+inline void addInt(QByteArray &data, quint16 n)
+{
+    const quint16 msb = qToBigEndian<quint16>(n);
+    const char * msb_c = reinterpret_cast<const char*>(&msb);
+    data.append(msb_c[0]);
+    data.append(msb_c[1]);
+    return;
+}
+
+bool QMqttConnection::sendControlConnect()
+{
+    QByteArray connectFrame;
+    // Fixed header
+    connectFrame += PacketType::CONNECT;
+
+    connectFrame += char(0); // Length to be filled later
+    // Variable header
+    // 3.1.2.1 Protocol Name
+    // 3.1.2.2 Protocol Level
+    const quint8 protocolVersion = m_client->protocolVersion();
+    if (protocolVersion == 3) {
+        const QByteArray other("MQIsdp");
+        addInt(connectFrame, other.size());
+        connectFrame += other.data();
+        connectFrame += char(3); // Version 3.1
+    } else if (protocolVersion == 4) {
+        connectFrame += char(0); // Length MSB
+        connectFrame += 4; // Length LSB
+        connectFrame += "MQTT";
+        connectFrame += char(4); // Version 3.1.1
+    } else {
+        qFatal("Illegal MQTT VERSION");
+    }
+
+    // 3.1.2.3 Connect Flags
+    quint8 flags = 0;
+    // Clean session
+    flags |= 1;
+    connectFrame += char(flags);
+
+    // 3.1.2.10 Keep Alive
+    addInt(connectFrame, m_client->keepAlive());
+
+    // 3.1.3 Payload
+    // 3.1.3.1 Client Identifier
+    if (protocolVersion == 3) { // no username
+        connectFrame += char(0);
+        connectFrame += char(0);
+    } else if (protocolVersion == 4) {
+        // Client id maximum left is 23
+        const QByteArray clientStringArray = m_client->clientId().left(23).toUtf8();
+        if (clientStringArray.size()) {
+            addInt(connectFrame, clientStringArray.size());
+            connectFrame += clientStringArray;
+        } else {
+            connectFrame += char(0);
+            connectFrame += char(0);
+        }
+    }
+
+    const quint8 frameLength = connectFrame.size();
+    connectFrame[1] = char(frameLength - 2); // The header bytes are not included
+
+    const qint64 res = m_transport->write(connectFrame.constData(), frameLength);
+    if (Q_UNLIKELY(res == -1)) {
+        qWarning("Could not write CONNECT frame to transport");
+        return false;
+    }
+
+    m_internalState = BrokerWaitForConnectAck;
+    return true;
+}
+
+bool QMqttConnection::sendControlPublish()
+{
+    Q_UNIMPLEMENTED();
+    return false;
+}
+
+bool QMqttConnection::sendControlSubscribe()
+{
+    Q_UNIMPLEMENTED();
+    return false;
+}
+
+bool QMqttConnection::sendControlUnsubscribe()
+{
+    Q_UNIMPLEMENTED();
+    return false;
+}
+
+bool QMqttConnection::sendControlPingRequest()
+{
+    Q_UNIMPLEMENTED();
+    return false;
+}
+
+bool QMqttConnection::sendControlDisconnect()
+{
+    Q_UNIMPLEMENTED();
+    return false;
+}
+
+void QMqttConnection::setClient(QMqttClient *client)
+{
+    m_client = client;
+}
+
+void QMqttConnection::transportConnectionClosed()
+{
+    qWarning("Transport Connection closed!");
+    m_client->setState(QMqttClient::Disconnected);
+}
+
+void QMqttConnection::transportReadReady()
+{
+    qWarning("Connection can read stuff");
+    QByteArray data = m_transport->readAll();
+    Q_ASSERT(data.size());
+    const quint8 *ptr = reinterpret_cast<const quint8 *>(data.constData());
+    switch (data[0]) {
+    case PacketType::CONNACK: {
+        if (m_internalState != BrokerWaitForConnectAck) {
+            qWarning("Received CONNACK at unexpected time!");
+            break;
+        }
+        quint8 payloadSize = ptr[1];
+        if (payloadSize != 2 || (payloadSize != data.size() - 2)) {
+            qWarning("Unexpected FRAME size in CONNACK");
+            // ## SET SOME ERROR
+            break;
+        }
+        quint8 ackFlags = ptr[2];
+        if (ackFlags > 1) { // MQTT-3.2.2.1
+            qWarning("Unexpected CONACK Flags set");
+            // ## SET SOME ERROR
+            break;
+        }
+        bool sessionPresent = ackFlags == 1;
+
+        // ### TODO: MQTT-3.2.2-1
+        // ### TODO: MQTT-3.2.2-2
+        if (sessionPresent) {
+            qWarning("Connected with a clean session, ack contains session present");
+            // ## SET SOME ERROR
+            break;
+        }
+
+        quint8 connectResultValue = ptr[3];
+        if (connectResultValue != 0) {
+            qWarning("Connection has been rejected");
+            // MQTT-3.2.2-5
+            m_transport->close();
+        }
+        m_internalState = BrokerConnected;
+        m_client->setState(QMqttClient::Connected);
+        break;
+    }
+    default: {
+        qWarning("Transport received unknown data");
+    }
+    }
+}
+
+QT_END_NAMESPACE

src/mqtt/qmqttconnection_p.h

@@ -0,0 +1,110 @@
+/****************************************************************************
+**
+** Copyright (C) 2017 The Qt Company Ltd.
+** Contact: https://www.qt.io/licensing/
+**
+** This file is part of the Qt Mqtt module of the Qt Toolkit.
+**
+** $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$
+**
+****************************************************************************/
+
+#ifndef QMQTTCONNECTION_P_H
+#define QMQTTCONNECTION_P_H
+
+//
+//  W A R N I N G
+//  -------------
+//
+// This file is not part of the Qt API.  It exists purely as an
+// implementation detail.  This header file may change from version to
+// version without notice, or even be removed.
+//
+// We mean it.
+//
+
+#include "qmqttclient.h"
+#include <QObject>
+
+QT_BEGIN_NAMESPACE
+
+class QMqttConnection : public QObject
+{
+    Q_OBJECT
+    enum PacketType {
+        CONNECT     = 0x10,
+        CONNACK     = 0x20,
+        PUBLISH     = 0x30,
+        PUBACK      = 0x40,
+        PUBREC      = 0x50,
+        PUBREL      = 0x60,
+        PUBCOMP     = 0x70,
+        SUBSCRIBE   = 0x80,
+        SUBACK      = 0x90,
+        UNSUBSCRIBE = 0xA0,
+        UNSUBACK    = 0xB0,
+        PINGREQ     = 0xC0,
+        PINGRESP    = 0xD0,
+        DISCONNECT  = 0xE0
+    };
+
+    enum InternalConnectionState {
+        BrokerDisconnected = 0,
+        BrokerWaitForConnectAck,
+        BrokerConnected
+    };
+
+public:
+    explicit QMqttConnection(QObject *parent = 0);
+    ~QMqttConnection();
+
+    void setTransport(QIODevice *device, QMqttClient::TransportType transport);
+    QIODevice *transport() const;
+
+    bool ensureTransport();
+    bool ensureTransportOpen();
+
+    bool sendControlConnect();
+    bool sendControlPublish();
+    bool sendControlSubscribe();
+    bool sendControlUnsubscribe();
+    bool sendControlPingRequest();
+    bool sendControlDisconnect();
+
+    void setClient(QMqttClient *client);
+signals:
+
+public slots:
+    void transportConnectionClosed();
+    void transportReadReady();
+
+public:
+    QIODevice *m_transport{nullptr};
+    QMqttClient::TransportType m_transportType{QMqttClient::IODevice};
+    bool m_ownTransport{false};
+    QMqttClient *m_client{nullptr};
+private:
+    InternalConnectionState m_internalState{BrokerDisconnected};
+};
+
+QT_END_NAMESPACE
+
+#endif // QMQTTCONNECTION_P_H