IoT platform realizes the best practice of RRPC synchronous call

Pub/Sub モードに基づく同期呼び出しの実践

1. 同期呼び出しのシーン

1.1 背景
MQTT プロトコルは PUB/SUB ベースの非同期通信モードであり、サーバーがデバイスにコマンドを送信し、デバイスから応答結果を返信させるシナリオを実現できません。
IoT プラットフォームは MQTT プロトコルをベースに、リクエストと応答の同期メカニズムを開発しました。これにより、MQTT プロトコルを変更せずに同期通信を実現できます。アプリケーションサーバーが POP API を通じて Rrpc 呼び出しを開始し、IoT デバイスはタイムアウト時間内に決められた形式で Pub メッセージに返信するだけで、サーバーは IoT デバイスの応答結果を同期的に取得できます。


具体的なフローは次のとおりです。



1.2 Topic 形式の規則
リクエスト： /sys/${productKey}/${deviceName}/rrpc/request/${messageId}**
** 応答： **/sys/${productKey}/${deviceName}/rrpc/ **response**/**${messageId}

$ は変数を表し、各デバイスごとに異なる値になります
messageId は IoT プラットフォームが生成するメッセージ ID です。
デバイスが返信する responseTopic 内の messageId は requestTopic のものと一致する必要があります

例：
デバイス側で以下の Topic をサブスクライブする必要があります。

/sys/${productKey}/${deviceName}/rrpc/request/+
実行中のデバイスが受信する Topic：

/sys/PK100101/DN213452/rrpc/request/443859344534
メッセージ受信後、タイムアウト時間内に以下の Topic に返信します。

/sys/PK100101/DN213452/rrpc/response/443859344534

2. Rrpc の同期呼び出し例

2.1 デバイス側のコード
const mqtt = require('aliyun-iot-mqtt');
//Device properties
const options = require("./iot-device-config.json");
//establish connection
const client = mqtt.getAliyunIotMqttClient(options);

client.subscribe(`/sys/${options.productKey}/${options.deviceName}/rrpc/request/+`)
client.on('message', function(topic, message) {

if(topic.indexOf(`/sys/${options.productKey}/${options.deviceName}/rrpc/request/`)>-1){
handleRrpc(topic, message)
}
})

function handleRrpc(topic, message){
topic = topic.replace('/request/','/response/');
console.log("topic=" + topic)
// Ordinary Rrpc, response payload customization
const payloadJson = {code:200,msg:"handle ok"};
client.publish(topic, JSON.stringify(payloadJson));
}

2.2 サーバー側の POP による Rrpc 呼び出し
const co = require('co');
const RPCClient = require('@alicloud/pop-core').RPCClient;

const options = require("./iot-ak-config.json");

//1. Initialize client
const client = new RPCClient({
accessKeyId: options.accessKey,
secretAccessKey: options. accessKeySecret,
endpoint: 'https://iot.cn-shanghai.aliyuncs.com',
apiVersion: '2018-01-20'
});

const payload = {
"msg": "hello Rrpc"
};

//2. Build request
const params = {
ProductKey: "a1gMu82K4m2",
DeviceName: "h5@nuwr5r9hf6l@1532088166923",
RequestBase64Byte: new Buffer(JSON.stringify(payload)).toString("base64"),
Timeout: 3000
};

co(function*() {
//3. Initiate an API call
const response = yield client.request('Rrpc', params);

console.log(JSON.stringify(response));
});
rrpc response:

{
"MessageId": "1037292594536681472",
"RequestId": "D2150496-2A61-4499-8B2A-4B3EC4B2A432",
"PayloadBase64Byte": "eyJjb2RlIjoyMDAsIm1zZyI6ImhhbmRsZSBvayJ9",
"Success": true,
"RrpcCode": "SUCCESS"
}

// PayloadBase64Byte decoding: {"code":200,"msg":"handle ok"}

3. Thing モデル - サービス同期呼び出し InvokeThingService の例
注：Thing モデルのサービス呼び出しインターフェースは InvokeThingService であり、Rrpc ではありません


デバイスがサブスクライブするサブ Topic
注：サービス同期呼び出し API は InvokeThingService です

/sys/${productKey}/${deviceName}/rrpc/request/+
IoT クラウドの下りリンクペイロード形式

{
"id": 3536123,
"version": "1.0",
"params": {
"Input parameter key1": "Input parameter value1",
"Input parameter key2": "Input parameter value2"
},
"method": "thing. service. {tsl. service. identifier}"
}

デバイスが応答する応答 Topic
Message Id of /sys/${productKey}/${deviceName}/rrpc/response/request
デバイスの応答ペイロード形式

{
"id": 3536123,
"code": 200,
"data": {
"Out parameter key1": "Out parameter value1",
"Out parameter key2": "Out parameter value2"
}
}

3.1 Thing モデル - 同期サービスの定義



3.2 デバイス側の実装
const mqtt = require('aliyun-iot-mqtt');
//Device properties
const options = require("./iot-device-config.json");
//establish connection
const client = mqtt.getAliyunIotMqttClient(options);

client.subscribe(`/sys/${options.productKey}/${options.deviceName}/rrpc/request/+`)
client.on('message', function(topic, message) {

if(topic.indexOf(`/sys/${options.productKey}/${options.deviceName}/rrpc/request/`)>-1){
handleRrpc(topic, message)
}
})
/*
* If there are multiple synchronous call services, they need to be distinguished by the method in the payload
*/
function handleRrpc(topic, message){

topic = topic.replace('/request/','/response/');
console.log("topic=" + topic)
//Object model Synchronous service call, response payload structure:
const payloadJson = {
id: Date.now(),
code:200,
data: {
currentMode: Math. floor((Math. random() * 20) + 10)
}
}

client.publish(topic, JSON.stringify(payloadJson));
}
注：デバイスが応答するペイロードは、オブジェクトモデルで定義された出力パラメーター構造に準拠している必要があります

3.3 サーバー側の POP インターフェース InvokeThingService
const co = require('co');
const RPCClient = require('@alicloud/pop-core').RPCClient;

const options = require("./iot-ak-config.json");

//1. Initialize client
const client = new RPCClient({
accessKeyId: options.accessKey,
secretAccessKey: options. accessKeySecret,
endpoint: 'https://iot.cn-shanghai.aliyuncs.com',
apiVersion: '2018-01-20'
});

const params = {
ProductKey: "a1gMu82K4m2",
DeviceName: "h5@nuwr5r9hf6l@1532088166923",
Args: JSON. stringify({ "mode": "1" }),
Identifier: "thing. service. setMode"
};

co(function*() {
try {
//3. Initiate an API call
const response = yield client.request('InvokeThingService', params);

console.log(JSON.stringify(response));
} catch (err) {
console. log(err);
}
});
呼び出し結果：

{
"Data": {
"MessageId": "1536145625658"
},
"RequestId": "29FD78CE-D1FF-48F7-B0A7-BD52C142DD7F",
"Success": true
}