dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #5796 from desoliture1/gatewayFix

Browse Source 
[WIP][3.3.3] MQTT Gateway API - fix response data on attributes request
This commit is contained in:
Andrew Shvayka 2022-01-20 13:04:48 +02:00 committed by GitHub
commit a3f3578b65
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 85 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
View File

@ -452,6 +452,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
.setRequestId(requestId)
.setSharedStateMsg(true)
.addAllSharedAttributeList(toTsKvProtos(result))
.setIsMultipleAttributesRequest(request.getSharedAttributeNamesCount() > 1)
.build();
sendToTransport(responseMsg, sessionInfo);
}
@ -473,6 +474,8 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
.setRequestId(requestId)
.addAllClientAttributeList(toTsKvProtos(result.get(0)))
.addAllSharedAttributeList(toTsKvProtos(result.get(1)))
.setIsMultipleAttributesRequest(
request.getSharedAttributeNamesCount() + request.getClientAttributeNamesCount() > 1)
.build();
sendToTransport(responseMsg, sessionInfo);
}

1
common/cluster-api/src/main/proto/queue.proto
View File

@ -149,6 +149,7 @@ message GetAttributeResponseMsg {
int32 requestId = 1;
repeated TsKvProto clientAttributeList = 2;
repeated TsKvProto sharedAttributeList = 3;
bool isMultipleAttributesRequest = 4;
string error = 5;
bool sharedStateMsg = 6;
}

2
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptor.java
View File

@ -122,7 +122,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
public Optional<MqttMessage> convertToGatewayPublish(MqttDeviceAwareSessionContext ctx, String deviceName, TransportProtos.GetAttributeResponseMsg responseMsg) throws AdaptorException {
return processConvertFromGatewayAttributeResponseMsg(ctx, deviceName, responseMsg);
}
@Override
public Optional<MqttMessage> convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.AttributeUpdateNotificationMsg notificationMsg, String topic) {
return Optional.of(createMqttPublishMsg(ctx, topic, JsonConverter.toJson(notificationMsg)));

14
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java
View File

@ -313,16 +313,18 @@ public class JsonConverter {
return result;
}
public static JsonObject getJsonObjectForGateway(String deviceName, TransportProtos.GetAttributeResponseMsg
responseMsg) {
public static JsonObject getJsonObjectForGateway(
String deviceName,
TransportProtos.GetAttributeResponseMsg responseMsg
) {
JsonObject result = new JsonObject();
result.addProperty("id", responseMsg.getRequestId());
result.addProperty(DEVICE_PROPERTY, deviceName);
if (responseMsg.getClientAttributeListCount() > 0) {
addValues(result, responseMsg.getClientAttributeListList());
addValues(result, responseMsg.getClientAttributeListList(), responseMsg.getIsMultipleAttributesRequest());
}
if (responseMsg.getSharedAttributeListCount() > 0) {
addValues(result, responseMsg.getSharedAttributeListList());
addValues(result, responseMsg.getSharedAttributeListList(), responseMsg.getIsMultipleAttributesRequest());
}
return result;
}
@ -335,8 +337,8 @@ public class JsonConverter {
return result;
}
private static void addValues(JsonObject result, List<TransportProtos.TsKvProto> kvList) {
if (kvList.size() == 1) {
private static void addValues(JsonObject result, List<TransportProtos.TsKvProto> kvList, boolean multipleAttrKeysRequested) {
if (kvList.size() == 1 && !multipleAttrKeysRequested) {
addValueToJson(result, "value", kvList.get(0).getKv());
} else {
JsonObject values;

2
msa/black-box-tests/src/test/java/org/thingsboard/server/msa/AbstractContainerTest.java
View File

@ -22,6 +22,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableMap;
import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import lombok.extern.slf4j.Slf4j;
import org.apache.cassandra.cql3.Json;
import org.apache.commons.lang3.RandomStringUtils;
@ -69,6 +70,7 @@ public abstract class AbstractContainerTest {
protected static String TB_TOKEN;
protected static RestClient restClient;
protected ObjectMapper mapper = new ObjectMapper();
protected JsonParser jsonParser = new JsonParser();
@BeforeClass
public static void before() throws Exception {

70
msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttGatewayClientTest.java
View File

@ -20,6 +20,7 @@ import com.google.common.collect.Sets;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import io.netty.buffer.ByteBuf;
@ -151,6 +152,75 @@ public class MqttGatewayClientTest extends AbstractContainerTest {
Assert.assertTrue(verify(actualLatestTelemetry, "attr4", Long.toString(73)));
}
@Test
public void responseDataOnAttributesRequestCheck() throws Exception {
Optional<DeviceCredentials> createdDeviceCredentials = restClient.getDeviceCredentialsByDeviceId(createdDevice.getId());
Assert.assertTrue(createdDeviceCredentials.isPresent());
JsonObject sharedAttributes = new JsonObject();
sharedAttributes.addProperty("attr1", "value1");
sharedAttributes.addProperty("attr2", true);
sharedAttributes.addProperty("attr3", 42.0);
sharedAttributes.addProperty("attr4", 73);
mqttClient.on("v1/gateway/attributes/response", listener, MqttQoS.AT_LEAST_ONCE).get();
ResponseEntity sharedAttributesResponse = restClient.getRestTemplate()
.postForEntity(HTTPS_URL + "/api/plugins/telemetry/DEVICE/{deviceId}/SHARED_SCOPE",
mapper.readTree(sharedAttributes.toString()), ResponseEntity.class,
createdDevice.getId());
Assert.assertTrue(sharedAttributesResponse.getStatusCode().is2xxSuccessful());
var event = listener.getEvents().poll(10, TimeUnit.SECONDS);
JsonObject requestData = new JsonObject();
requestData.addProperty("id", 1);
requestData.addProperty("device", createdDevice.getName());
requestData.addProperty("client", false);
requestData.addProperty("key", "attr1");
mqttClient.on("v1/gateway/attributes/response", listener, MqttQoS.AT_LEAST_ONCE).get();
mqttClient.publish("v1/gateway/attributes/request", Unpooled.wrappedBuffer(requestData.toString().getBytes())).get();
event = listener.getEvents().poll(10, TimeUnit.SECONDS);
JsonObject responseData = jsonParser.parse(Objects.requireNonNull(event).getMessage()).getAsJsonObject();
Assert.assertTrue(responseData.has("value"));
Assert.assertEquals(sharedAttributes.get("attr1").getAsString(), responseData.get("value").getAsString());
requestData = new JsonObject();
requestData.addProperty("id", 1);
requestData.addProperty("device", createdDevice.getName());
requestData.addProperty("client", false);
JsonArray keys = new JsonArray();
keys.add("attr1");
keys.add("attr2");
requestData.add("keys", keys);
mqttClient.on("v1/gateway/attributes/response", listener, MqttQoS.AT_LEAST_ONCE).get();
mqttClient.publish("v1/gateway/attributes/request", Unpooled.wrappedBuffer(requestData.toString().getBytes())).get();
event = listener.getEvents().poll(10, TimeUnit.SECONDS);
responseData = jsonParser.parse(Objects.requireNonNull(event).getMessage()).getAsJsonObject();
Assert.assertTrue(responseData.has("values"));
Assert.assertEquals(sharedAttributes.get("attr1").getAsString(), responseData.get("values").getAsJsonObject().get("attr1").getAsString());
Assert.assertEquals(sharedAttributes.get("attr2").getAsString(), responseData.get("values").getAsJsonObject().get("attr2").getAsString());
requestData = new JsonObject();
requestData.addProperty("id", 1);
requestData.addProperty("device", createdDevice.getName());
requestData.addProperty("client", false);
keys = new JsonArray();
keys.add("attr1");
keys.add("undefined");
requestData.add("keys", keys);
mqttClient.on("v1/gateway/attributes/response", listener, MqttQoS.AT_LEAST_ONCE).get();
mqttClient.publish("v1/gateway/attributes/request", Unpooled.wrappedBuffer(requestData.toString().getBytes())).get();
event = listener.getEvents().poll(10, TimeUnit.SECONDS);
responseData = jsonParser.parse(Objects.requireNonNull(event).getMessage()).getAsJsonObject();
Assert.assertTrue(responseData.has("values"));
Assert.assertEquals(sharedAttributes.get("attr1").getAsString(), responseData.get("values").getAsJsonObject().get("attr1").getAsString());
Assert.assertEquals(1, responseData.get("values").getAsJsonObject().entrySet().size());
}
@Test
public void requestAttributeValuesFromServer() throws Exception {
WsClient wsClient = subscribeToWebSocket(createdDevice.getId(), "CLIENT_SCOPE", CmdsType.ATTR_SUB_CMDS);