WebSockets MQTT サブスクリプション

MQTTクライアントとWebSocket接続を組み合わせる。

「WebSockets MQTT サブスクリプション」では、WebSocket接続とQMqttClient を組み合わせたカスタムQIODevice の設計方法について解説します。

注： Qt 6.10以降、Qt MQTT では WebSockets がネイティブにサポートされています。この例は、カスタムトランスポートを実装する方法を示すために残されています。

カスタム QIODevice の作成

新しいカスタムデバイスであるWebSocketIODevice は、QIODevice のサブクラスである必要があります：

class WebSocketIODevice : public QIODevice
{
    Q_OBJECT
public:
    WebSocketIODevice(QObject *parent = nullptr);

    bool isSequential() const override;
    qint64 bytesAvailable() const override;

    bool open(OpenMode mode) override;
    void close() override;

    qint64 readData(char *data, qint64 maxlen) override;
    qint64 writeData(const char *data, qint64 len) override;

    void setUrl(const QUrl &url);
    void setProtocol(const QByteArray &data);
Q_SIGNALS:
    void socketConnected();

public slots:
    void handleBinaryMessage(const QByteArray &msg);
    void onSocketConnected();

private:
    QByteArray m_protocol;
    QByteArray m_buffer;
    QWebSocket m_socket;
    QUrl m_url;
};

接続とサブスクリプションを管理するクラスの設計

WebSocketIODevice は、QMqttClient およびQMqttSubscription と並んで、ClientSubscription クラスのプライベートメンバーとなります：

private:
    QMqttClient m_client;
    QMqttSubscription *m_subscription;
    QUrl m_url;
    QString m_topic;
    WebSocketIODevice m_device;
    int m_version;

メッセージの購読と受信

主なロジックは、ClientSubscription クラスのconnectAndSubscribe() メソッドに実装されています。 WebSocket経由でMQTT接続を初期化する前に、WebSocketが正常に接続されていることを確認する必要があります。MQTT接続が確立された後、QMqttClient はトピックを購読できます。購読に成功した場合、QMqttSubscription を使用して購読したトピックからのメッセージを受信し、そのメッセージはClientSubscription クラスのhandleMessage() メソッドによって処理されます。

voidClientSubscription::connectAndSubscribe()
{
    qCDebug(lcWebSocketMqtt)<< "ブローカー " <<m_url<< " に接続中";

    m_device.setUrl(m_url);
    m_device.setUrl(m_url);

    m_device.setProtocol(m_version== 3 ? "mqttv3.1":"mqtt");

    connect(&m_device, &WebSocketIODevice::socketConnected, this, [this]() {
        qCDebug(lcWebSocketMqtt)<< "WebSocketに接続しました。MQTT接続を初期化しています。";

        m_client.setProtocolVersion(m_version== 3 ?QMqttClient::MQTT_3_1 : QMqttClient::MQTT_3_1_1);
        m_client.setTransport(&m_device,QMqttClient::IODevice);

        connect(&m_client, &QMqttClient::connected, this, [this]() {
            qCDebug(lcWebSocketMqtt)<< "MQTT接続が確立されました";

            m_subscription=m_client.subscribe(m_topic);
            if(!m_subscription) {
                qDebug() << "Failed to subscribe to " << m_topic;
               emiterrorOccured();
            }

            connect(m_subscription, &QMqttSubscription::stateChanged,
                    [](QMqttSubscription::SubscriptionState s) {
                qCDebug(lcWebSocketMqtt)<< "サブスクリプションの状態が変更されました:" <<s;
            });

            connect(m_subscription, &QMqttSubscription::messageReceived,
                    [this](QMqttMessage msg) {
                handleMessage(msg.payload());
            });
        });

        m_client.connectToHost();
    });
    if(!m_device.open(QIODevice::ReadWrite))
        qDebug() << "Could not open socket device";
}

ファイル：

© 2026 The Qt Company Ltd. Documentation contributions included herein are the copyrights of their respective owners. The documentation provided herein is licensed under the terms of the GNU Free Documentation License version 1.3 as published by the Free Software Foundation. Qt and respective logos are trademarks of The Qt Company Ltd. in Finland and/or other countries worldwide. All other trademarks are property of their respective owners.