WebSockets MQTT サブスクリプション

MQTTクライアントとWebソケット接続の組み合わせ。

WebSockets MQTT Subscriptionでは、Web ソケット接続とQMqttClient を組み合わせたカスタムQIODevice を設計する方法を紹介します。

カスタム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() メソッドに実装されている。MQTT 接続を初期化する前に、Web ソケットが正常に接続されたことを確認する必要があります。MQTT 接続が確立されると、QMqttClient はトピックを購読できる。サブスクリプションに成功すると、QMqttSubscription を使用して、サブスクライブしたトピックからClientSubscription クラスのhandleMessage() メソッドで処理されるメッセージを受信できるようになります。

voidClientSubscription::connectAndSubscribe() { qCDebug(lcWebSocketMqtt)<< "Connecting to broker at "<<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) { if (!m_subscription)                qDebug() << "Failed to subscribe to " << m_topic;
                errorOccured() を発行する。QMqttSubscription::stateChanged, [](QMqttSubscription::SubscriptionState s) { qCDebug(lcWebSocketMqtt)<< "サブスクリプションの状態が変更されました:"<<s; }); connect(m_subscription, &QMqttSubscription::messageReceived, [this](QMqttMessagemsg) { handleMessage(msg.payload()); }); }); m_client.connectToHost(); });if(!m_device.open(QIODevice::ReadWrite))        qDebug() << "Could not open socket device";
}

ファイル

© 2025 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.