Overview
You can use the revert-RPC (RRPC) feature of IoT Platform to initiate an RRPC request to a device. A response from the device is synchronously returned. The device receives a synchronous request topic in the format of /ext/rrpc/{messageId}/{rrpc_topic} or /sys/{pk}/{dn}/rrpc/request/${msgId}. After the device receives the message, it processes the message and publishes the result to /ext/rrpc/{messageId}/{rrpc_topic} or /ext/rrpc/${msgId}/${topic}. The device can receive the synchronous RRPC request from IoT Platform only after it subscribes to the corresponding downstream topic from IoT Platform.
API operation for messaging - RRpc
You can call the RRPC operation of IoT Platform to trigger IoT Platform to send a synchronous RRPC request to a device.
API operation for device management - InvokeThingsService
You can call the InvokeThingsService operation of IoT Platform to initiate a synchronous RRPC request to a device. You can also define a service of the device as a synchronous service in the IoT Platform console and trigger a synchronous service invocation request. That is, a synchronous RRPC request is initiated to the device.
Call to the RRpc operation
In the entire RRPC process, a device must subscribe to an RRPC topic and immediately returns a response when IoT Platform sends a request. Before you start the following invocation process, ensure that the SDK initialization is successful. That is, a Message Queuing Telemetry Transport (MQTT) connection has been created.
RRPC subscription
Replace ${topic} with a custom topic, such as /a1pMraiYxxx/test/user/get, where a1pMraiYxxx indicates the ProductKey and test indicates the DeviceName. You do not need to subscribe to a system RRPC topic. When the device processes a request from IoT Platform, the system RRPC topic is /sys/${pk}/${dn}/rrpc/request/${requestId}.
MqttSubscribeRequest subscribeRequest = new MqttSubscribeRequest();
subscribeRequest.isSubscribe = true;
// Replace ${topic} with a custom topic.
subscribeRequest.topic = "/ext/rrpc/+/${topic}";
LinkKit.getInstance().subscribe(subscribeRequest, new IConnectSubscribeListener() {
@Override
public void onSuccess() {
// The subscription is successful.
}
@Override
public void onFailure(AError aError) {
// The subscription failed.
}
);
RRPC listening and response
The listener that listens to downstream data can receive RRPC requests from IoT Platform. You can specify the topic that can be received.
// This operation is used to set a global listener that listens to downstream data.
LinkKit.getInstance().registerOnPushListener(notifyListener);
/**
* The downstream data listener. All MQTT data from IoT Platform is called back at this listener.
*/
private static IConnectNotifyListener notifyListener = new IConnectNotifyListener() {
/**
* The onNotify listener is triggered if shouldHandle does not specify that the topic is not processed.
* @param connectId indicates the connection type. This parameter is used to check whether the connection is a persistent connection. connectId == ConnectSDK.getInstance().getPersistentConnectId()
* @param topic indicates the downstream topic, which hosts data sent from IoT Platform.
* @param aMessage indicates the content of downstream data.
*/
@Override
public void onNotify(String connectId, String topic, AMessage aMessage) {
String data = new String((byte[]) aMessage.data);
// An example of data returned from IoT Platform: data = {"method":"thing.service.test_service","id":"123374967","params":{"vv":60},"version":"1.0.0"}
}
/**
* @param connectId indicates the connection type. This parameter is used to check whether the connection is a persistent connection. connectId == ConnectSDK.getInstance().getPersistentConnectId()
* @param topic indicates the downstream topic, which hosts downstream data from IoT Platform.
* @ return specifies whether to process this topic. If this parameter is set to true, data of this topic is called back to onNotify. If this parameter is set to false, data of this topic is not called back to onNotify. We recommend that you use the default value true.
*/
@Override
public boolean shouldHandle(String connectId, String topic) {
return true;
}
/**
* @param connectId indicates the connection type. This parameter is used to check whether the connection is a persistent connection. connectId == ConnectSDK.getInstance().getPersistentConnectId()
* @param connectState {@link ConnectState}
* CONNECTED indicates that the connection is successful.
* DISCONNECTED indicates that the connection is broken.
* CONNECTING indicates that the connection is being established.
* CONNECTFAIL indicates that the connection failed.
*/
@Override
public void onConnectStateChange(String connectId, ConnectState connectState) {
Log.d(TAG, "onConnectStateChange() called with: connectId = [" + connectId + "], connectState = [" + connectState + "]");
}
};
Demo
Devices
Subscribe to a custom topic, that is, "/a1ExY4afKY1/testDevice/user/get", and listen to downstream data from IoT Platform. An RRPC response contains the returned data that you need. In the following demo, data is null.
MqttSubscribeRequest subscribeRequest = new MqttSubscribeRequest();
subscribeRequest.isSubscribe = true;
subscribeRequest.topic = "/a1ExY4afKY1/testDevice/user/get";
LinkKit.getInstance().subscribe(subscribeRequest, new IConnectSubscribeListener() {
@Override
public void onSuccess() {
// The subscription is successful.
}
@Override
public void onFailure(AError aError) {
// The subscription failed.
}
);
// You can use the global listener to listen to data from IoT Platform.
LinkKit.getInstance().registerOnPushListener(notifyListener);
private IConnectNotifyListener notifyListener = new IConnectNotifyListener() {
@Override
public void onNotify(String connectId, String topic, AMessage aMessage) {
if (CONNECT_ID.equals(connectId) && ! TextUtils.isEmpty(topic) &&
topic.startsWith("/ext/rrpc/")) {
// Example: topic=/ext/rrpc/1138654706478941696//a1ExY4afKY1/testDevice/user/get
//ALog.d(TAG, "receice Message=" + new String((byte[]) aMessage.data));
// An example of data returned from IoT Platform: {"method":"thing.service.test_service","id":"123374967","params":{"vv":60},"version":"1.0.0"}
MqttPublishRequest request = new MqttPublishRequest();
request.isRPC = false;
request.topic = topic;
String[] array = topic.split("/");
String resId = array[3];
request.msgId = resId;
// TODO: Enter information as needed. The information here is for reference only.
request.payloadObj = "{\"id\":\"" + resId + "\", \"code\":\"200\"" + ",\"data\":{} }";
LinkKit.getInstance().publish(request, new IConnectSendListener() {
@Override
public void onResponse(ARequest aRequest, AResponse aResponse) {
// The response is successful.
}
@Override
public void onFailure(ARequest aRequest, AError aError) {
// No successful response is returned.
}
});
}
}
@Override
public boolean shouldHandle(String connectId, String topic) {
return true;
}
@Override
public void onConnectStateChange(String connectId, ConnectState connectState) {
Log.d(TAG, "onConnectStateChange() called with: connectId = [" + connectId + "], connectState = [" + connectState + "]");
}
};