Practice of Synchronous Call Based on Pub/Sub Model

Synchronous Call Scenarios

The MQTT protocol is an asynchronous communication model based on pub/sub, and is not applicable for scenarios where the server sends instructions to the device and the device responds with the results.

Alibaba Cloud IoT platform has developed a request and response synchronization mechanism based on the MQTT protocol, enabling simultaneous communication without modifying the MQTT protocol. The application server initiates a revert-RPC through the POP API. The IoT device only needs to reply to the pub message in a fixed format within the timeout, and the server can synchronously obtain the response from the IoT device.

The pub/sub procedure can be visualized as follows:

Topic Format Convention

Request:

Response:

$ is a variable that varies from devices

The messageId is the message ID generated by the IoT platform, and the messageId in the responseTopic of the device is consistent with the requestTopic.

Example

The device needs to subscribe:

/sys/${productKey}/${deviceName}/rrpc/request/+

The running device receives the topic:

/sys/PK100101/DN213452/rrpc/request/443859344534

After receiving the message, reply to the topic within the timeout:

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

Example of Synchronous Revert-RPC

Code on the Device

const mqtt = require('aliyun-iot-mqtt');
//Device property
const options = require("./iot-device-config.json");
//Establish the 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)
//Normal revert-RPC, responding to the payload customization
const payloadJson = {code:200,msg:"handle ok"};
client.publish(topic, JSON.stringify(payloadJson));
}

Revert-RPC through POP on the Server

const co = require('co');
const RPCClient = require('@alicloud/pop-core').RPCClient;
const options = require("./iot-ak-config.json");//1. Initialize the client
const client = new RPCClient({
accessKeyId: options.accessKey,
secretAccessKey: options.accessKeySecret,
endpoint: 'https://iot.cn-shanghai.aliyuncs.com',
apiVersion: '2017-04-20'
});
const payload = {
"msg": "hello Rrpc"
};
//2. Construct a 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));
});

Revert-RPC response:

{
"MessageId": "1037292594536681472",
"RequestId": "D2150496-2A61-4499-8B2A-4B3EC4B2A432",
"PayloadBase64Byte": "eyJjb2RlIjoyMDAsIm1zZyI6ImhhbmRsZSBvayJ9",
"Success": true,
"RrpcCode": "SUCCESS"
}
// PayloadBase64Byte decoding: {"code":200,"msg":"handle ok"}

“Thing Model” — An Example of Synchronous Service Invocation Using InvokeThingService

Note: Thing model InvokeThingService is not a revert-RPC

Thing Model — Synchronous Service Definition

Implementation on the Device

const mqtt = require('aliyun-iot-mqtt');
//Device property
const options = require("./iot-device-config.json");
//Establish the 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 invocation services, they should be distinguished by the method in the payload.
*/
function handleRrpc(topic, message){
topic = topic.replace('/request/','/response/');
console.log("topic=" + topic)
//Thing model, synchronous service invocation, response payload structure:
const payloadJson = {
id: Date.now(),
code:200,
data: {
currentMode: Math.floor((Math.random() * 20) + 10)
}
}
client.publish(topic, JSON.stringify(payloadJson));
}

Note: The payload of the device-response must satisfy the output parameter structure defined in the thing model.

Server Side POP API InvokeThingService

const co = require('co');
const RPCClient = require('@alicloud/pop-core').RPCClient;
const options = require("./iot-ak-config.json");//1. Initialize the 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);
}
});

Result of the call:

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

To learn more about Alibaba Cloud IoT Platform, visit www.alibabacloud.com/product/iot

Reference:https://www.alibabacloud.com/blog/practice-of-synchronous-call-based-on-pub%2Fsub-model_594095?spm=a2c41.12185799.0.0

Follow me to keep abreast with the latest technology news, industry insights, and developer trends.

Get the Medium app

A button that says 'Download on the App Store', and if clicked it will lead you to the iOS App store
A button that says 'Get it on, Google Play', and if clicked it will lead you to the Google Play store