From 7b14fdaf5e0e6901a7a509299df744bb380fd63d Mon Sep 17 00:00:00 2001 From: desoliture Date: Mon, 17 Jan 2022 18:59:39 +0200 Subject: [PATCH] add isMultipleAttributesRequest field in queue.proto in GetAttributeResponseMsg message --- .../device/DeviceActorMessageProcessor.java | 3 +++ common/cluster-api/src/main/proto/queue.proto | 1 + .../adaptors/BackwardCompatibilityAdaptor.java | 5 ----- .../transport/mqtt/adaptors/JsonMqttAdaptor.java | 15 ++++----------- .../mqtt/adaptors/MqttTransportAdaptor.java | 3 --- .../transport/mqtt/adaptors/ProtoMqttAdaptor.java | 5 ----- .../mqtt/session/GatewayDeviceSessionCtx.java | 9 +-------- .../mqtt/session/GatewaySessionHandler.java | 9 --------- .../common/transport/adaptor/JsonConverter.java | 7 +++---- .../msa/connectivity/MqttGatewayClientTest.java | 1 - 10 files changed, 12 insertions(+), 46 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java index 20109ad074..bfcb95a71c 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java @@ -450,6 +450,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { .setRequestId(requestId) .setSharedStateMsg(true) .addAllSharedAttributeList(toTsKvProtos(result)) + .setIsMultipleAttributesRequest(request.getSharedAttributeNamesCount() > 1) .build(); sendToTransport(responseMsg, sessionInfo); } @@ -471,6 +472,8 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { .setRequestId(requestId) .addAllClientAttributeList(toTsKvProtos(result.get(0))) .addAllSharedAttributeList(toTsKvProtos(result.get(1))) + .setIsMultipleAttributesRequest( + request.getSharedAttributeNamesCount() > 1 || request.getClientAttributeNamesCount() > 1) .build(); sendToTransport(responseMsg, sessionInfo); } diff --git a/common/cluster-api/src/main/proto/queue.proto b/common/cluster-api/src/main/proto/queue.proto index dca92c1c7a..6565a45740 100644 --- a/common/cluster-api/src/main/proto/queue.proto +++ b/common/cluster-api/src/main/proto/queue.proto @@ -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; } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/BackwardCompatibilityAdaptor.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/BackwardCompatibilityAdaptor.java index 0cc65da1b0..a608d6c52d 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/BackwardCompatibilityAdaptor.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/BackwardCompatibilityAdaptor.java @@ -128,11 +128,6 @@ public class BackwardCompatibilityAdaptor implements MqttTransportAdaptor { return protoAdaptor.convertToGatewayPublish(ctx, deviceName, rpcRequest); } - @Override - public Optional convertToGatewayPublish(MqttDeviceAwareSessionContext ctx, String deviceName, TransportProtos.GetAttributeResponseMsg responseMsg, boolean multipleAttributeKeysRequested) throws AdaptorException { - return protoAdaptor.convertToGatewayPublish(ctx, deviceName, responseMsg, multipleAttributeKeysRequested); - } - @Override public Optional convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.ToServerRpcResponseMsg rpcResponse, String topicBase) throws AdaptorException { log.warn("[{}] invoked not implemented adaptor method! ToServerRpcResponseMsg: {} TopicBase: {}", ctx.getSessionId(), rpcResponse, topicBase); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptor.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptor.java index d71b87efdc..8687b02f56 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptor.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptor.java @@ -34,7 +34,6 @@ import org.thingsboard.server.common.data.ota.OtaPackageType; import org.thingsboard.server.common.transport.adaptor.AdaptorException; import org.thingsboard.server.common.transport.adaptor.JsonConverter; import org.thingsboard.server.gen.transport.TransportProtos; -import org.thingsboard.server.transport.mqtt.session.GatewayDeviceSessionCtx; import org.thingsboard.server.transport.mqtt.session.MqttDeviceAwareSessionContext; import java.nio.charset.Charset; @@ -43,7 +42,6 @@ import java.util.Arrays; import java.util.HashSet; import java.util.Optional; import java.util.Set; -import java.util.Map; import java.util.UUID; import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_SOFTWARE_FIRMWARE_RESPONSES_TOPIC_FORMAT; @@ -125,14 +123,9 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor { @Override public Optional convertToGatewayPublish(MqttDeviceAwareSessionContext ctx, String deviceName, TransportProtos.GetAttributeResponseMsg responseMsg) throws AdaptorException { - return convertToGatewayPublish(ctx, deviceName, responseMsg, false); + return processConvertFromGatewayAttributeResponseMsg(ctx, deviceName, responseMsg); } - - @Override - public Optional convertToGatewayPublish(MqttDeviceAwareSessionContext ctx, String deviceName, TransportProtos.GetAttributeResponseMsg responseMsg, boolean multipleAttributeKeysRequested) throws AdaptorException { - return processConvertFromGatewayAttributeResponseMsg(ctx, deviceName, responseMsg, multipleAttributeKeysRequested); - } - + @Override public Optional convertToPublish(MqttDeviceAwareSessionContext ctx, TransportProtos.AttributeUpdateNotificationMsg notificationMsg, String topic) { return Optional.of(createMqttPublishMsg(ctx, topic, JsonConverter.toJson(notificationMsg))); @@ -239,11 +232,11 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor { } } - private Optional processConvertFromGatewayAttributeResponseMsg(MqttDeviceAwareSessionContext ctx, String deviceName, TransportProtos.GetAttributeResponseMsg responseMsg, boolean multipleAttrKeysRequested) throws AdaptorException { + private Optional processConvertFromGatewayAttributeResponseMsg(MqttDeviceAwareSessionContext ctx, String deviceName, TransportProtos.GetAttributeResponseMsg responseMsg) throws AdaptorException { if (!StringUtils.isEmpty(responseMsg.getError())) { throw new AdaptorException(responseMsg.getError()); } else { - JsonObject result = JsonConverter.getJsonObjectForGateway(deviceName, responseMsg, multipleAttrKeysRequested); + JsonObject result = JsonConverter.getJsonObjectForGateway(deviceName, responseMsg); return Optional.of(createMqttPublishMsg(ctx, MqttTopics.GATEWAY_ATTRIBUTES_RESPONSE_TOPIC, result)); } } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/MqttTransportAdaptor.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/MqttTransportAdaptor.java index a980e91b53..53ef439eda 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/MqttTransportAdaptor.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/MqttTransportAdaptor.java @@ -25,7 +25,6 @@ import io.netty.handler.codec.mqtt.MqttPublishMessage; import io.netty.handler.codec.mqtt.MqttPublishVariableHeader; import org.thingsboard.server.common.data.ota.OtaPackageType; import org.thingsboard.server.common.transport.adaptor.AdaptorException; -import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.AttributeUpdateNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ClaimDeviceMsg; import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg; @@ -73,8 +72,6 @@ public interface MqttTransportAdaptor { Optional convertToGatewayPublish(MqttDeviceAwareSessionContext ctx, String deviceName, ToDeviceRpcRequestMsg rpcRequest) throws AdaptorException; - Optional convertToGatewayPublish(MqttDeviceAwareSessionContext ctx, String deviceName, TransportProtos.GetAttributeResponseMsg responseMsg, boolean multipleAttributeKeysRequested) throws AdaptorException; - Optional convertToPublish(MqttDeviceAwareSessionContext ctx, ToServerRpcResponseMsg rpcResponse, String topicBase) throws AdaptorException; ProvisionDeviceRequestMsg convertToProvisionRequestMsg(MqttDeviceAwareSessionContext ctx, MqttPublishMessage inbound) throws AdaptorException; diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/ProtoMqttAdaptor.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/ProtoMqttAdaptor.java index 4dc9a95d6b..4fa47b367d 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/ProtoMqttAdaptor.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/ProtoMqttAdaptor.java @@ -209,11 +209,6 @@ public class ProtoMqttAdaptor implements MqttTransportAdaptor { return Optional.of(createMqttPublishMsg(ctx, MqttTopics.GATEWAY_RPC_TOPIC, payloadBytes)); } - @Override - public Optional convertToGatewayPublish(MqttDeviceAwareSessionContext ctx, String deviceName, TransportProtos.GetAttributeResponseMsg responseMsg, boolean multipleAttributeKeysRequested) throws AdaptorException { - return convertToGatewayPublish(ctx, deviceName, responseMsg); - } - public static byte[] toBytes(ByteBuf inbound) { byte[] bytes = new byte[inbound.readableBytes()]; int readerIndex = inbound.readerIndex(); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java index 02324fd3e7..f41c5668cd 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java @@ -80,10 +80,7 @@ public class GatewayDeviceSessionCtx extends MqttDeviceAwareSessionContext imple @Override public void onGetAttributesResponse(TransportProtos.GetAttributeResponseMsg response) { try { - boolean multipleAttrKeysReq = isMultipleAttributeKeysRequested(response.getRequestId()); - parent.getPayloadAdaptor() - .convertToGatewayPublish(this, getDeviceInfo().getDeviceName(), response, multipleAttrKeysReq) - .ifPresent(parent::writeAndFlush); + parent.getPayloadAdaptor().convertToGatewayPublish(this, getDeviceInfo().getDeviceName(), response).ifPresent(parent::writeAndFlush); } catch (Exception e) { log.trace("[{}] Failed to convert device attributes response to MQTT msg", sessionId, e); } @@ -139,10 +136,6 @@ public class GatewayDeviceSessionCtx extends MqttDeviceAwareSessionContext imple // This feature is not supported in the TB IoT Gateway yet. } - public boolean isMultipleAttributeKeysRequested(int requestId) { - return parent.isMultipleAttributeKeysRequested(requestId); - } - private boolean isAckExpected(MqttMessage message) { return message.fixedHeader().qosLevel().value() > 0; } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java index 12e9f786bf..7b45b5f9d9 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java @@ -89,7 +89,6 @@ public class GatewaySessionHandler { private final ConcurrentMap mqttQoSMap; private final ChannelHandlerContext channel; private final DeviceSessionCtx deviceSessionCtx; - private final Map pendingAttributesRequests = new ConcurrentHashMap<>(); public GatewaySessionHandler(DeviceSessionCtx deviceSessionCtx, UUID sessionId) { this.context = deviceSessionCtx.getContext(); @@ -108,13 +107,6 @@ public class GatewaySessionHandler { return new ConcurrentReferenceHashMap<>(16, ReferenceType.WEAK); } - public boolean isMultipleAttributeKeysRequested(int requestId) { - JsonObject request = pendingAttributesRequests.getOrDefault(requestId, new JsonObject()); - boolean multipleAttributeKeysRequested = request.has("keys") && request.get("keys").getAsJsonArray().size() > 1; - pendingAttributesRequests.remove(requestId); - return multipleAttributeKeysRequested; - } - public void onDeviceConnect(MqttPublishMessage mqttMsg) throws AdaptorException { if (isJsonPayloadType()) { onDeviceConnectJson(mqttMsg); @@ -566,7 +558,6 @@ public class GatewaySessionHandler { if (json.isJsonObject()) { JsonObject jsonObj = json.getAsJsonObject(); int requestId = jsonObj.get("id").getAsInt(); - pendingAttributesRequests.put(requestId, jsonObj); String deviceName = jsonObj.get(DEVICE_PROPERTY).getAsString(); boolean clientScope = jsonObj.get("client").getAsBoolean(); Set keys; diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java index 86f90d786f..1e6267ca33 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java @@ -315,17 +315,16 @@ public class JsonConverter { public static JsonObject getJsonObjectForGateway( String deviceName, - TransportProtos.GetAttributeResponseMsg responseMsg, - boolean multipleAttrKeysRequested + 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(), multipleAttrKeysRequested); + addValues(result, responseMsg.getClientAttributeListList(), responseMsg.getIsMultipleAttributesRequest()); } if (responseMsg.getSharedAttributeListCount() > 0) { - addValues(result, responseMsg.getSharedAttributeListList(), multipleAttrKeysRequested); + addValues(result, responseMsg.getSharedAttributeListList(), responseMsg.getIsMultipleAttributesRequest()); } return result; } diff --git a/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttGatewayClientTest.java b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttGatewayClientTest.java index dfdc68d9d5..74dfc2107c 100644 --- a/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttGatewayClientTest.java +++ b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttGatewayClientTest.java @@ -156,7 +156,6 @@ public class MqttGatewayClientTest extends AbstractContainerTest { public void responseDataOnAttributesRequestCheck() throws Exception { Optional createdDeviceCredentials = restClient.getDeviceCredentialsByDeviceId(createdDevice.getId()); Assert.assertTrue(createdDeviceCredentials.isPresent()); - WsClient wsClient = subscribeToWebSocket(createdDevice.getId(), "CLIENT_SCOPE", CmdsType.ATTR_SUB_CMDS); JsonObject sharedAttributes = new JsonObject(); sharedAttributes.addProperty("attr1", "value1"); sharedAttributes.addProperty("attr2", true);