AWS IoTでは、MQTTクライアントのNode.js実装であるMQTT.jsをラップしたaws-iot-device-sdk-jsというライブラリを提供しています。
このライブラリではオフラインキューイングの仕組みが備わっているのですが、データ送信時に指定するQoSレベルと、さらには内部で使用しているMQTT.jsにもオフラインキューイングの仕組みが備わっているため、これらの関係がどうなっているかを調べてみました。
今回テストしたコード。
offlineQueueingをtrue/falseにすることで、オフラインキューイングを有効/無効にすることができます。
'use strict'
const awsIot = require('aws-iot-device-sdk')
require('dotenv').config()
const device = awsIot.device({
keyPath: process.env.KEY_FILE,
certPath: process.env.CERT_FILE,
caPath: process.env.CA_FILE,
host: process.env.IOT_HOST,
clientId: 'test-device',
offlineQueueing: true, // オフラインキューイング設定
debug: true
})
const topic = `test`
let timer = null
device.on('reconnect', () => {
console.info(new Date(), 'reconnect')
})
device.on('close', () => {
console.info(new Date(), 'closed')
})
device.on('offline', () => {
console.info(new Date(), 'offline')
})
device.on('error', (err) => {
console.error(new Date(), 'error', err)
})
device.on('connect', () => {
console.info(new Date(), 'connected')
if (timer !== null) {
return
}
let count = 0
timer = setInterval(() => {
count++
console.info(new Date(), `publishing {count: ${count}}`)
device.publish(topic, JSON.stringify({count: count}), (err) => {
if (err) {
return console.error(new Date(), err)
}
console.info(new Date(), 'published')
})
}, 5000)
})
実行ログ。
{ keyPath: './certs/private.pem.key',
certPath: './certs/certificate.pem.crt',
caPath: './certs/root-CA.crt',
clientId: 'test-device',
host: 'xxxxxxxxxxxxxx.iot.ap-northeast-1.amazonaws.com',
offlineQueueing: false,
debug: true,
reconnectPeriod: 1000,
fastDisconnectDetection: true,
protocol: 'mqtts',
port: 8883,
key: <Buffer 2d ... >,
cert: <Buffer 2d ...>,
ca: <Buffer 2d ... >,
requestCert: true,
rejectUnauthorized: true }
attempting new mqtt connection...
2017-08-23T08:28:05.391Z 'connected'
2017-08-23T08:28:10.398Z 'publishing {count: 1}'
2017-08-23T08:28:10.407Z 'published'
2017-08-23T08:28:15.415Z 'publishing {count: 2}'
2017-08-23T08:28:15.418Z 'published'
2017-08-23T08:28:20.424Z 'publishing {count: 3}'
2017-08-23T08:28:20.425Z 'published'
2017-08-23T08:28:25.426Z 'publishing {count: 4}'
2017-08-23T08:28:25.427Z 'published'
2017-08-23T08:28:30.433Z 'publishing {count: 5}'
2017-08-23T08:28:30.435Z 'published'
2017-08-23T08:28:35.440Z 'publishing {count: 6}'
2017-08-23T08:28:35.441Z 'published'
2017-08-23T08:28:40.447Z 'publishing {count: 7}'
2017-08-23T08:28:40.448Z 'published'
2017-08-23T08:28:45.454Z 'publishing {count: 8}'
2017-08-23T08:28:45.456Z 'published'
2017-08-23T08:28:50.462Z 'publishing {count: 9}'
2017-08-23T08:28:50.463Z 'published'
2017-08-23T08:28:55.469Z 'publishing {count: 10}'
2017-08-23T08:28:55.470Z 'published'
2017-08-23T08:29:00.476Z 'publishing {count: 11}'
2017-08-23T08:29:00.478Z 'published'
2017-08-23T08:29:05.484Z 'publishing {count: 12}'
2017-08-23T08:29:05.485Z 'published'
2017-08-23T08:29:10.491Z 'publishing {count: 13}'
2017-08-23T08:29:10.492Z 'published'
2017-08-23T08:29:15.498Z 'publishing {count: 14}'
2017-08-23T08:29:15.501Z 'published'
2017-08-23T08:29:20.508Z 'publishing {count: 15}'
2017-08-23T08:29:20.510Z 'published'
2017-08-23T08:29:25.515Z 'publishing {count: 16}'
2017-08-23T08:29:25.516Z 'published'
2017-08-23T08:29:30.522Z 'publishing {count: 17}'
2017-08-23T08:29:30.523Z 'published'
2017-08-23T08:29:35.529Z 'publishing {count: 18}'
2017-08-23T08:29:35.531Z 'published'
2017-08-23T08:29:40.536Z 'publishing {count: 19}'
2017-08-23T08:29:40.538Z 'published'
2017-08-23T08:29:45.544Z 'publishing {count: 20}'
2017-08-23T08:29:45.545Z 'published'
2017-08-23T08:29:50.551Z 'publishing {count: 21}'
2017-08-23T08:29:50.552Z 'published'
2017-08-23T08:29:55.559Z 'publishing {count: 22}'
2017-08-23T08:29:55.560Z 'published'
2017-08-23T08:30:00.520Z 'offline'
connection lost - will attempt reconnection in 1 seconds...
2017-08-23T08:30:00.523Z 'closed'
2017-08-23T08:30:00.561Z 'publishing {count: 23}'
2017-08-23T08:30:01.525Z 'reconnect'
2017-08-23T08:30:05.566Z 'publishing {count: 24}'
2017-08-23T08:30:10.573Z 'publishing {count: 25}'
2017-08-23T08:30:15.579Z 'publishing {count: 26}'
2017-08-23T08:30:20.585Z 'publishing {count: 27}'
2017-08-23T08:30:25.591Z 'publishing {count: 28}'
2017-08-23T08:30:30.597Z 'publishing {count: 29}'
connection lost - will attempt reconnection in 2 seconds...
2017-08-23T08:30:31.549Z 'closed'
2017-08-23T08:30:33.550Z 'reconnect'
2017-08-23T08:30:35.601Z 'publishing {count: 30}'
2017-08-23T08:30:39.428Z 'connected'
2017-08-23T08:30:40.603Z 'publishing {count: 31}'
2017-08-23T08:30:40.604Z 'published'
2017-08-23T08:30:45.609Z 'publishing {count: 32}'
2017-08-23T08:30:45.610Z 'published'
2017-08-23T08:30:50.616Z 'publishing {count: 33}'
2017-08-23T08:30:50.617Z 'published'
2017-08-23T08:30:55.628Z 'publishing {count: 34}'
2017-08-23T08:30:55.631Z 'published'
2017-08-23T08:31:00.640Z 'publishing {count: 35}'
2017-08-23T08:31:00.643Z 'published'
MQTTコネクション断の検知に時間がかかる
上記のテストコードを実行し、AWS IoTに接続していくつかのデータが送信されたことを確認したあと、意図的にネットワークを遮断という方法で実験を行いました。
注意すべきなのは、aws-iot-device-sdk-jsはネットワークが遮断されてすぐにオフライン(コード内部ではinactive)状態に入るわけではなく、1分ほど立ってからコネクションの断を検知してオフラインになるということです。
上記のログでいうと、{count: 10}を送信した直後にネットワークを遮断していますが、ライブラリがオフライン状態に入ったのはそこから約1分後です。
その後、{count: 30}を送信した直後にネットワークを復活させました。
コネクション断の検知は、MQTT.jsがネットワークソケットのcloseイベントに反応することによって行われていると思われますが、このcloseイベントが実際のコネクション断のあとなかなか発火されないのがなぜなのかはわかってません。
オフラインキューイング設定とQoSの関係
aws-iot-device-sdk-jsのオフラインキューイング設定と、publishメソッドに渡すQoSレベルの関係をまとめました。
QoSの設定は、上記コードにおいて、device.publishメソッドの第3引数に{qos: 1}を渡すことによって可能です。何も指定しない場合はQoS0になります。MQTTではQoS0から2までを定義していますが、AWS IoTではQoS0と1のみ利用可能です。
| # | オフラインキューイング | QoS | データの送信結果 |
|---|---|---|---|
| 1 | false | 0 | 実際のコネクション断から再接続までのデータがロストする |
| 2 | true | 0 | ライブラリがコネクション断を検知してから再接続するまでのデータがキューイングされ、再送される。実際のコネクション断からライブラリが検知するまでのデータはロストする。 |
| 3 | false | 1 | ライブラリがコネクション断を検知してから再接続するまでのデータのみロストするという不思議な動き。 |
| 4 | true | 1 | 全てのデータがキューイングされ、再送される。データロストはない。 |
#2の根拠
ライブラリがオフラインになってからキューイングが始まり、接続再開と同時にキューイングしたデータを送信するという動きになります。上述したように、実際のネットワーク断からライブラリがオフラインだと判断するまでに時間がかかるので注意が必要です。
#3の根拠
この不思議な動きはaws-iot-device-sdk-jsの実装を見るとわかります。
データロストが起きているのは、ライブラリがオフライン状態に入っているときです。
まず、オフラインキューイングはfalseなため最初のif文はパスします。次に、_filing()はライブラリがオフライン状態だとみなしているときにtrueになり、ここではオンラインであることを期待しているので、このif文もパスしてしまい、結果何も処理をされないままメソッドが終わってしまうことになります。
this.publish = function(topic, message, options, callback) {
//
// If filling or still draining, push this publish operation
// into the offline operations queue; otherwise, perform it
// immediately.
//
if (offlineQueueing === true && (_filling() || drainingTimer !== null)) {
if (_trimOfflinePublishQueueIfNecessary()) {
offlinePublishQueue.push({
topic: topic,
message: message,
options: options,
callback: callback
});
}
} else {
if (offlineQueueing === true || !_filling()) {
device.publish(topic, message, options, callback);
}
}
};
#1の根拠
単純にネットワークが遮断されている間のデータ送信が失敗しているということです。
MQTT.jsはデフォルトではQoS0のデータをオフラインキューイングするようになっていますが、上記のロジックによってMQTT.jsのpublishも呼ばれないため、このような動きになります。
#4の根拠
aws-iot-device-sdk-jsは実際のコネクション断に気づかずデータを送信するが、実際の送信処理を行っているMQTT.jsがQoSレベル1に従った動きをしているため、データのロストは起きません。つまり、データの送信は行うがAWS IoTからpubackが返ってこないために再送処理をしているということです。
まとめ
オフラインキューイングとQoSをどのように設定するかは送信するデータの重要度によって変わってくるので、どの組合せが良いかは一概には言えませんが、上記#2で指摘したようにオフラインキューイングが直感的な動きと少し異なる点には注意が必要です。