IoT Platform:MQTTプロトкоルの仕様

サポート機能

使用方法

セッションの有効期限

• 接続時に Clean Start と Session Expiry Interval を設定します：

  MqttConnectionOptions options = new MqttConnectionOptions();
  options.setCleanStart(true);
  options.setSessionExpiryInterval(60L);// 単位：秒。
  
  MqttClient mqttClient = new MqttClient(host, clientId, new MemoryPersistence());
  mqttClient.connect(options);

• 切断時に Session Expiry Interval を設定します：

  MqttProperties mqttProperties = new MqttProperties();
  mqttProperties.setSessionExpiryInterval(60L);// 単位：秒。
  
  MqttAsyncClient mqttAsyncClient = new MqttAsyncClient(host, clientId, new MemoryPersistence());
  mqttAsyncClient.disconnect(30000, null, null, MqttReturnCode.RETURN_CODE_SUCCESS, mqttProperties);

メッセージの有効期限

発行時に Message Expiry Interval を設定します：

// メッセージを作成します。
String content = "Hello World";
byte[] payload = content.getBytes();

// メッセージを作成します。
MqttMessage message = new MqttMessage(payload);
// メッセージの QoS を設定します。
message.setQos(1);

MqttProperties mqttProperties = new MqttProperties();

// メッセージの有効期間 (TTL) を設定します。
mqttProperties.setMessageExpiryInterval(600L);

message.setProperties(mqttProperties);

// メッセージを発行します。
MqttClient mqttClient = new MqttClient(host, clientId, new MemoryPersistence());
mqttClient.publish(topic, message);

サブスクリプションオプション

利用可能なサブスクリプションオプション：

• QoS：MQTT メッセージの QoS レベル。0 (QoS 0 メッセージ) または 1 (QoS 1 メッセージ) に設定できます。

• No Local：クライアントが自身で発行したメッセージを受信するかどうかを指定します。

  MQTT 3.1.1 では、クライアントはサブスクライブしたトピックで自身が発行したメッセージを受信します。MQTT 5.0 では、このオプションを true に設定すると、自己配信を防ぐことができます。

  値：

  • true：受信しません。

  • false：受信します。

• Retain As Published：サーバーがメッセージをクライアントに転送する際に RETAIN フラグを保持するかどうかを指定します。

  値：

  • true：メッセージに RETAIN フラグがある場合、そのフラグは保持されます。メッセージに RETAIN フラグがない場合、このオプションは効果がありません。

  • false：元のメッセージに RETAIN フラグがあるかどうかに関わらず、RETAIN フラグは保持されません。

  重要

  Retain As Published の設定は、保持されたメッセージの RETAIN フラグには影響しません。

• Retain Handling：サブスクリプションが確立されたときに、サーバーが保持されたメッセージをクライアントに送信するかどうかを指定します。

  値：

  • 0：クライアントのサブスクリプションが成功した場合、サーバーは保持されたメッセージを送信します。

  • 1：クライアントのサブスクリプションが成功し、かつそのサブスクリプションが以前に存在しなかった場合にのみ、サーバーは保持されたメッセージを送信します。

  • 2：クライアントのサブスクリプションが成功しても、サーバーは保持されたメッセージを送信しません。

// QoS サブスクリプションオプションを設定します。
MqttSubscription mqttSubscription = new MqttSubscription("aaa/bbb");

// QoS サブスクリプションオプションを設定します。
mqttSubscription.setQos(1);

// No Local サブスクリプションオプションを設定します。
mqttSubscription.setNoLocal(true);

// Retain As Published サブスクリプションオプションを設定します。
mqttSubscription.setRetainAsPublished(true);

// Retain Handling サブスクリプションオプションを設定します。
mqttSubscription.setRetainHandling(1);

MqttClient mqttClient = new MqttClient(host, clientId, new MemoryPersistence());
mqttClient.subscribe(new MqttSubscription[]{mqttSubscription});

保持されたメッセージ

// 保持されたメッセージを作成します。
String content = "Hello World";
byte[] payload = content.getBytes();
MqttMessage message = new MqttMessage(payload);
// メッセージを保持されたメッセージとして設定します。
message.setRetained(true);

// メッセージを発行します。
MqttClient mqttClient = new MqttClient(host, clientId, new MemoryPersistence());
mqttClient.publish(topic, message);

遺言メッセージ

// 遺言メッセージを作成します。
String content = "Will Message";
byte[] payload = content.getBytes();
MqttMessage message = new MqttMessage(payload);

MqttConnectionOptions options = new MqttConnectionOptions();
options.setUserName(USERNAME);
options.setPassword(PASSWORD.getBytes());

// 遺言メッセージを設定します。
options.setWill(topic, message);

// 遺言の遅延を設定します。
MqttProperties willMessageProperties = new MqttProperties();
willMessageProperties.setWillDelayInterval(60L);
options.setWillMessageProperties(willMessageProperties);

// 接続を確立します。
MqttClient mqttClient = new MqttClient(host, clientId, new MemoryPersistence());
mqttClient.connect(options);

接続ネゴシエーション

MqttConnectionOptions connOpts = new MqttConnectionOptions();
connOpts.setMaximumPacketSize(1024L);

ユーザープロパティ

MqttProperties properties = new MqttProperties();
List<UserProperty> userPropertys = new ArrayList<>();
userPropertys.add(new UserProperty("key1","value1"));
properties.setUserProperties(userPropertys);

デバイスが MQTT 5.0 経由で接続した後、報告された UserProperty データは IoT Platform のログで確認できます。

リクエスト/レスポンスパターン

リクエスターがデバイスで、レシーバーがビジネスサーバーである場合、AMQP サブスクリプションまたはルール転送の後にメッセージプロパティから ResponseTopic と CorrelationData を解析します。その後、Pub API を呼び出してデバイスに応答します。

MqttProperties properties = new MqttProperties();
properties.setCorrelationData("requestId12345".getBytes());
properties.setResponseTopic("/" + productKey + "/" + deviceName + "/user/get");

拡張エラーコード

エラーのトラブルシューティングをご参照ください。

トピックエイリアス

該当なし。

共有サブスクリプション

共有サブスクリプションのトピック形式： $share/${ShareName}/${filter}。

• $share：静的フィールド。共有サブスクリプションのトピックは $share で始まる必要があります。

• ${ShareName}：英字、数字、アンダースコア (_) のみを含む文字列。

  同じ ${ShareName} を持つセッションは1つのサブスクリプションを共有します。一致する各メッセージは1つのセッションにのみ配信されます。

• ${filter}：非共有サブスクリプションのトピックフィルター。英字、数字、アンダースコア (_) が使用できます。

MqttConnectionOptions options = new MqttConnectionOptions();
options.setUserName(username);
options.setPassword(password);

MqttClient mqttClient = new MqttClient(host, clientId, new MemoryPersistence());
mqttClient.connect(options);

mqttClient.subscribe("$share/testGroup/user/post", 1);