コンテンツにスキップ
machbase-neo
JSH
モジュール
mqtt

mqtt

Since v8.0.75

mqttは、JSHのMQTTクライアントモジュールです。

JSHアプリケーションでは、通常は以下のように使用します。

const mqtt = require('mqtt');

APIはイベント駆動で動作し、Clientを作成すると自動的にブローカーへの接続を試みます。

Client

MQTTクライアントオブジェクトです。

作成
new Client(options)
オプション
オプション型既定値説明
serversString[]MQTTブローカーURLの一覧（例：tcp://127.0.0.1:1883）
usernameStringブローカー認証のユーザー名
passwordStringブローカー認証のパスワード
keepAliveNumber30Keep Alive（秒）
connectRetryDelayNumber0再接続の遅延（ミリ秒）
cleanStartOnInitialConnectionBooleanfalse初回接続時のMQTT v5 clean startを有効にするかどうか
connectTimeoutNumber0接続のタイムアウト（ミリ秒）

プロパティ

config

client.configは、ネイティブMQTTクライアントが使用する解析済みの接続設定を提供します。

代表的なフィールドは以下のとおりです。

フィールド型説明
serverUrlsArray解析済みのブローカーURL一覧
connectUsernameString設定したユーザー名
connectPasswordbyte dataバイト形式のパスワード
keepAliveNumberKeep Aliveの間隔
reconnectBackoff(n)Function再接続の遅延を計算する関数
cleanStartOnInitialConnectionBooleanMQTT v5 clean startフラグ
connectTimeoutNumber接続のタイムアウト
console.println(client.config.serverUrls);
console.println(client.config.keepAlive);
使用例
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

const mqtt = require('mqtt');

const client = new mqtt.Client({
    servers: ['tcp://127.0.0.1:1883'],
    username: 'user',
    password: 'pass',
    keepAlive: 60,
    connectRetryDelay: 2000,
    connectTimeout: 10 * 1000,
    cleanStartOnInitialConnection: true,
});

client.on('open', () => {
    console.println('Connected');
    client.subscribe('test/topic', {
        qos: 0,
        properties: {
            subscriptionIdentifier: 7,
        },
    });
});

client.on('subscribed', (topic, reason) => {
    console.println('Subscribed:', topic, 'reason:', reason);
    client.publish('test/topic', 'Hello, MQTT!');
});

client.on('message', (msg) => {
    console.println('Message:', msg.topic, msg.payloadText);
    client.unsubscribe(msg.topic, {
        properties: {
            user: {
                source: 'example',
            },
        },
    });
});

client.on('unsubscribed', (topic, reason) => {
    console.println('Unsubscribed:', topic, 'reason:', reason);
    client.close();
});

client.on('error', (err) => {
    console.println('Error:', err.message);
});

client.on('close', () => {
    console.println('Disconnected');
});

メソッド

publish()

トピックにメッセージを発行します。

構文
publish(topic, message[, options])
パラメーター
  • topic String
  • message String | Uint8Array | Object | Array
  • options Object (省略可能)
オプション型既定値説明
qosNumber0QoSレベル
retainBooleanfalseRetainフラグ
propertiesObjectMQTT v5の発行プロパティ

options.propertiesのフィールド：

プロパティ型説明
payloadFormatNumberペイロード形式インジケーター
messageExpiryNumber有効期間
contentTypeStringコンテンツタイプ
responseTopicString応答トピック
correlationDataStringバイトに変換
topicAliasNumberトピックエイリアス
subscriptionIdentifierNumberサブスクリプション識別子
userObjectユーザー定義プロパティ（key: value）
戻り値

なし。結果はpublishedまたはerrorイベントで通知します。

subscribe()

トピックを購読します。

構文
subscribe(topic[, options])
パラメーター
  • topic String
  • options Object (省略可能)
オプション型既定値説明
qosNumber1QoSレベル
retainHandlingNumberMQTT v5のRetain処理
noLocalBooleanfalse同じクライアントが発行したメッセージの受信を抑制するかどうか
retainAsPublishedBooleanfalseブローカーのRetainフラグを保持するかどうか
propertiesObjectMQTT v5の購読プロパティ

options.propertiesのフィールド：

プロパティ型説明
subscriptionIdentifierNumberサブスクリプション識別子
userObjectユーザー定義プロパティ（key: value）
戻り値

なし。結果はsubscribedまたはerrorイベントで通知します。

使用例
client.subscribe('test/topic', {
    qos: 0,
    properties: {
        subscriptionIdentifier: 7,
        user: {
            source: 'example',
        },
    },
});

unsubscribe()

トピックの購読を解除します。

構文
unsubscribe(topic[, options])
パラメーター
  • topic String
  • options Object (省略可能)
オプション型説明
propertiesObjectMQTT v5の購読解除プロパティ

options.propertiesのフィールド：

プロパティ型説明
userObjectユーザー定義プロパティ（key: value）
戻り値

なし。結果はunsubscribedまたはerrorイベントで通知します。

使用例
client.unsubscribe('test/topic', {
    properties: {
        user: {
            source: 'example',
        },
    },
});

close()

クライアント接続を閉じます。

構文
close()
戻り値

なし。closeイベントが発生します。

イベント

open

クライアントの接続が完了すると発生します。

client.on('open', () => { ... })

message

購読したメッセージを受信すると発生します。

client.on('message', (msg) => { ... })

msgのフィールド：

プロパティ型説明
topicStringトピック名
payloadBufferバイナリデータを保持するメッセージペイロード
payloadTextStringUTF-8にデコードしたテキスト用の便利なフィールド
propertiesObjectMQTT v5の発行プロパティ

msg.propertiesのフィールド：

プロパティ型説明
payloadFormatNumberペイロード形式インジケーター
messageExpiryNumber有効期間
contentTypeStringコンテンツタイプ
responseTopicString応答トピック
correlationDataBufferバイナリデータを保持する相関データ
topicAliasNumberトピックエイリアス
subscriptionIdentifierNumberサブスクリプション識別子
userObjectユーザー定義プロパティ

テキストメッセージは、msg.payloadTextまたはmsg.payload.toString()で読み取れます。

バイナリメッセージは、msg.payloadをそのまま使用します。

client.on('message', (msg) => {
    console.println('Payload is buffer:', Buffer.isBuffer(msg.payload));
    console.println('Payload bytes:', Array.from(msg.payload).join(','));
});

MQTT v5の発行プロパティには、msg.propertiesでアクセスできます。

client.on('message', (msg) => {
    console.println('Content type:', msg.properties.contentType);
    console.println('Response topic:', msg.properties.responseTopic);
    console.println('Correlation data:', msg.properties.correlationData.toString());
    console.println('User source:', msg.properties.user.source);
});

subscribed

購読のACKを受信すると発生します。

client.on('subscribed', (topic, reason) => { ... })
  • topic String
  • reason Number （MQTT理由コード）

published

発行のACKを受信すると発生します。

client.on('published', (topic, reason) => { ... })
  • topic String
  • reason Number （MQTT理由コード）

unsubscribed

購読解除のACKを受信すると発生します。

client.on('unsubscribed', (topic, reason) => { ... })
  • topic String
  • reason Number （MQTT理由コード）

error

接続、購読、購読解除、発行でエラーが発生すると通知します。

client.on('error', (err) => { ... })
  • err Error

接続前またはclose()の呼び出し後に、publish()、subscribe()、unsubscribe()を呼び出すと、errorイベントでエラーを通知します。

close

close()の呼び出し時に発生します。

client.on('close', () => { ... })

動作に関する注意

  • Clientは、コンストラクターで自動的にブローカーへの接続を試みます。
  • JavaScriptラッパーは、可能な場合、受信したバイナリペイロードをBufferに変換します。
  • テキスト表現が可能なバッファーペイロードでは、msg.payloadTextが自動的に設定されます。
  • MQTT v5のcorrelationDataは、バイナリデータを保持するBufferとして提供されます。

MQTT v5の書き込みプロパティの例

以下の例は、MQTT v5 write APIのユーザープロパティを使い、db/write/{table}トピックにデータを書き込みます。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

const mqtt = require('mqtt');

const client = new mqtt.Client({
    servers: ['tcp://127.0.0.1:5653'],
});

const rows = [
    ['my-car', Date.now(), 32.1],
    ['my-car', Date.now() + 1000, 65.4],
];

client.on('open', () => {
    client.publish('db/write/EXAMPLE', rows, {
        qos: 1,
        properties: {
            user: {
                method: 'append',
                timeformat: 'ms',
            },
        },
    });
});

client.on('published', () => client.close());
client.on('error', (err) => console.println(err.message));
最終更新日
machclinats