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 299992783d..f41954c0b5 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 @@ -518,20 +518,18 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso systemContext.getDeviceStateService().onDeviceDisconnect(tenantId, deviceId); } - @SuppressWarnings("deprecation") // isMultipleAttributesRequest retained for the legacy gateway value/values response private void handleGetAttributesRequest(SessionInfoProto sessionInfo, GetAttributeRequestMsg request) { int requestId = request.getRequestId(); if (request.getOnlyShared()) { Futures.addCallback(findAllAttributesByScope(AttributeScope.SHARED_SCOPE), new FutureCallback<>() { @Override public void onSuccess(@Nullable List result) { - GetAttributeResponseMsg responseMsg = GetAttributeResponseMsg.newBuilder() + GetAttributeResponseMsg.Builder builder = GetAttributeResponseMsg.newBuilder() .setRequestId(requestId) .setSharedStateMsg(true) - .addAllSharedAttributeList(KvProtoUtil.attrToTsKvProtos(result)) - .setIsMultipleAttributesRequest(request.getSharedAttributeNamesCount() > 1) - .build(); - sendToTransport(responseMsg, sessionInfo); + .addAllSharedAttributeList(KvProtoUtil.attrToTsKvProtos(result)); + setMultipleAttributesRequest(builder, request.getSharedAttributeNamesCount() > 1); + sendToTransport(builder.build(), sessionInfo); } @Override @@ -547,15 +545,14 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso Futures.addCallback(getAttributesKvEntries(request), new FutureCallback<>() { @Override public void onSuccess(@Nullable List> result) { - GetAttributeResponseMsg responseMsg = GetAttributeResponseMsg.newBuilder() + GetAttributeResponseMsg.Builder builder = GetAttributeResponseMsg.newBuilder() .setRequestId(requestId) .setSeparateScopesResponse(request.getSeparateScopesResponse()) .addAllClientAttributeList(KvProtoUtil.attrToTsKvProtos(result.get(0))) - .addAllSharedAttributeList(KvProtoUtil.attrToTsKvProtos(result.get(1))) - .setIsMultipleAttributesRequest( - request.getSharedAttributeNamesCount() + request.getClientAttributeNamesCount() > 1) - .build(); - sendToTransport(responseMsg, sessionInfo); + .addAllSharedAttributeList(KvProtoUtil.attrToTsKvProtos(result.get(1))); + setMultipleAttributesRequest(builder, + request.getSharedAttributeNamesCount() + request.getClientAttributeNamesCount() > 1); + sendToTransport(builder.build(), sessionInfo); } @Override @@ -569,6 +566,11 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso } } + @SuppressWarnings("deprecation") // isMultipleAttributesRequest retained for the legacy gateway value/values response + private static void setMultipleAttributesRequest(GetAttributeResponseMsg.Builder builder, boolean multipleAttributesRequest) { + builder.setIsMultipleAttributesRequest(multipleAttributesRequest); + } + private ListenableFuture>> getAttributesKvEntries(GetAttributeRequestMsg request) { boolean clientAll = request.getAllClientAttributes(); boolean sharedAll = request.getAllSharedAttributes(); @@ -586,21 +588,23 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso clientAttributesFuture = findAllAttributesByScope(AttributeScope.CLIENT_SCOPE); sharedAttributesFuture = findAllAttributesByScope(AttributeScope.SHARED_SCOPE); } else { - // "all " wins over a specific key list for the same scope - clientAttributesFuture = clientAll - ? findAllAttributesByScope(AttributeScope.CLIENT_SCOPE) - : (clientSpecific - ? findAttributesByScope(toSet(request.getClientAttributeNamesList()), AttributeScope.CLIENT_SCOPE) - : Futures.immediateFuture(Collections.emptyList())); - sharedAttributesFuture = sharedAll - ? findAllAttributesByScope(AttributeScope.SHARED_SCOPE) - : (sharedSpecific - ? findAttributesByScope(toSet(request.getSharedAttributeNamesList()), AttributeScope.SHARED_SCOPE) - : Futures.immediateFuture(Collections.emptyList())); + clientAttributesFuture = resolveScopeFuture(clientAll, clientSpecific, request.getClientAttributeNamesList(), AttributeScope.CLIENT_SCOPE); + sharedAttributesFuture = resolveScopeFuture(sharedAll, sharedSpecific, request.getSharedAttributeNamesList(), AttributeScope.SHARED_SCOPE); } return Futures.allAsList(Arrays.asList(clientAttributesFuture, sharedAttributesFuture)); } + // "all " wins over a specific key list for the same scope; no signal => empty result. + private ListenableFuture> resolveScopeFuture(boolean all, boolean specific, List names, AttributeScope scope) { + if (all) { + return findAllAttributesByScope(scope); + } else if (specific) { + return findAttributesByScope(toSet(names), scope); + } else { + return Futures.immediateFuture(Collections.emptyList()); + } + } + private ListenableFuture> findAllAttributesByScope(AttributeScope scope) { return systemContext.getAttributesService().findAll(tenantId, deviceId, scope); } diff --git a/application/src/test/java/org/thingsboard/server/system/BaseHttpDeviceApiTest.java b/application/src/test/java/org/thingsboard/server/system/BaseHttpDeviceApiTest.java index b412567fb8..89825a1e58 100644 --- a/application/src/test/java/org/thingsboard/server/system/BaseHttpDeviceApiTest.java +++ b/application/src/test/java/org/thingsboard/server/system/BaseHttpDeviceApiTest.java @@ -16,6 +16,7 @@ package org.thingsboard.server.system; import com.fasterxml.jackson.databind.JsonNode; +import org.awaitility.Awaitility; import org.junit.Before; import org.junit.Test; import org.springframework.test.context.TestPropertySource; @@ -26,9 +27,11 @@ import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.controller.AbstractControllerTest; +import java.time.Duration; import java.util.HashMap; import java.util.Map; import java.util.Random; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; @@ -84,13 +87,21 @@ public abstract class BaseHttpDeviceApiTest extends AbstractControllerTest { mockMvc.perform( asyncDispatch(doPost("/api/v1/" + token + "/attributes", clientAttrs, new String[]{}).andReturn())) .andExpect(status().isOk()); - Thread.sleep(2000); - String body = doGetAsync("/api/v1/" + token + "/attributes?allClientKeys=true") + String allClientKeysUrl = "/api/v1/" + token + "/attributes?allClientKeys=true"; + Awaitility.await("client attributes are persisted and returned via allClientKeys=true") + .atMost(30, TimeUnit.SECONDS) + .pollInterval(Duration.ofMillis(100)) + .ignoreExceptions() + .until(() -> { + JsonNode r = JacksonUtil.toJsonNode(doGetAsync(allClientKeysUrl).andReturn().getResponse().getContentAsString()); + return r.has("client") && clientAttrs.keySet().stream().allMatch(r.get("client")::has); + }); + String body = doGetAsync(allClientKeysUrl) .andExpect(status().isOk()).andReturn().getResponse().getContentAsString(); JsonNode resp = JacksonUtil.toJsonNode(body); assertThat(resp.has("client")).isTrue(); - assertThat(resp.get("client").get("clientA").asText()).isEqualTo("valueA"); - assertThat(resp.get("client").get("clientB").asText()).isEqualTo("valueB"); + JsonNode client = resp.get("client"); + clientAttrs.forEach((key, value) -> assertThat(client.get(key).asText()).isEqualTo(value)); assertThat(resp.has("shared")).isFalse(); } diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/AbstractMqttAttributesIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/AbstractMqttAttributesIntegrationTest.java index 7e31704436..23cef0b49d 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/AbstractMqttAttributesIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/AbstractMqttAttributesIntegrationTest.java @@ -98,6 +98,9 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt protected static final String SHARED_ATTRIBUTES_PAYLOAD = "{\"sharedStr\":\"value1\",\"sharedBool\":true,\"sharedDbl\":42.0,\"sharedLong\":73," + "\"sharedJson\":{\"someNumber\":42,\"someArray\":[1,2,3],\"someNestedObject\":{\"key\":\"value\"}}}"; + protected static final String CLIENT_ATTRIBUTE_KEYS = "clientStr,clientBool,clientDbl,clientLong,clientJson"; + protected static final String SHARED_ATTRIBUTE_KEYS = "sharedStr,sharedBool,sharedDbl,sharedLong,sharedJson"; + private static final String SHARED_ATTRIBUTES_DELETED_RESPONSE = "{\"deleted\":[\"sharedJson\"]}"; private List getTsKvProtoList(String attributePrefix) { @@ -375,11 +378,7 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt client.connectAndWait(accessToken); SingleEntityFilter dtf = new SingleEntityFilter(); dtf.setSingleEntity(AliasEntityId.fromEntityId(savedDevice.getId())); - String clientKeysStr = "clientStr,clientBool,clientDbl,clientLong,clientJson"; - String sharedKeysStr = "sharedStr,sharedBool,sharedDbl,sharedLong,sharedJson"; - List keys = new ArrayList<>(); - keys.addAll(getEntityKeys(List.of(clientKeysStr.split(",")), CLIENT_ATTRIBUTE)); - keys.addAll(getEntityKeys(List.of(sharedKeysStr.split(",")), SHARED_ATTRIBUTE)); + List keys = buildClientAndSharedEntityKeys(); getWsClient().subscribeLatestUpdate(keys, dtf); getWsClient().registerWaitForUpdate(2); doPostAsync("/api/plugins/telemetry/DEVICE/" + savedDevice.getId().getId() + "/attributes/SHARED_SCOPE", @@ -505,8 +504,7 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt 20, 100); assertNotNull(device); - String clientKeysStr = "clientStr,clientBool,clientDbl,clientLong,clientJson"; - String attributeValuesUrl = "/api/plugins/telemetry/DEVICE/" + device.getId() + "/values/attributes/CLIENT_SCOPE?keys=" + clientKeysStr; + String attributeValuesUrl = "/api/plugins/telemetry/DEVICE/" + device.getId() + "/values/attributes/CLIENT_SCOPE?keys=" + CLIENT_ATTRIBUTE_KEYS; Awaitility.await() .atMost(10, TimeUnit.SECONDS) .ignoreExceptions() @@ -518,10 +516,7 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt SingleEntityFilter dtf = new SingleEntityFilter(); dtf.setSingleEntity(AliasEntityId.fromEntityId(device.getId())); - String sharedKeysStr = "sharedStr,sharedBool,sharedDbl,sharedLong,sharedJson"; - List keys = new ArrayList<>(); - keys.addAll(getEntityKeys(List.of(clientKeysStr.split(",")), CLIENT_ATTRIBUTE)); - keys.addAll(getEntityKeys(List.of(sharedKeysStr.split(",")), SHARED_ATTRIBUTE)); + List keys = buildClientAndSharedEntityKeys(); EntityDataUpdate initUpdate = getWsClient().subscribeLatestUpdate(keys, dtf); assertNotNull(initUpdate); assertFalse(initUpdate.getData().getData().isEmpty()); @@ -596,10 +591,80 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt client.disconnect(); } + protected void processProtoTestGatewayRequestAllSharedSeparated() throws Exception { + processProtoTestGatewayRequestAllSeparated(false); + } + + protected void processProtoTestGatewayRequestAllClientSeparated() throws Exception { + processProtoTestGatewayRequestAllSeparated(true); + } + + // allClient => request allClientKeys and expect only client in the separated response; otherwise the mirror for shared. + private void processProtoTestGatewayRequestAllSeparated(boolean allClient) throws Exception { + MqttTestClient client = new MqttTestClient(); + client.connectAndWait(gatewayAccessToken); + + String deviceName = "Gateway Device Request All " + (allClient ? "Client" : "Shared") + " Separated Proto"; + List clientKeysList = List.of(CLIENT_ATTRIBUTE_KEYS.split(",")); + client.publishAndWait(GATEWAY_ATTRIBUTES_TOPIC, getProtoGatewayDeviceClientAttributesPayload(deviceName, clientKeysList)); + + Device device = doExecuteWithRetriesAndInterval(() -> doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class), + 20, 100); + assertNotNull(device); + + SingleEntityFilter dtf = new SingleEntityFilter(); + dtf.setSingleEntity(AliasEntityId.fromEntityId(device.getId())); + List keys = buildClientAndSharedEntityKeys(); + EntityDataUpdate initUpdate = getWsClient().subscribeLatestUpdate(keys, dtf); + assertNotNull(initUpdate); + assertFalse(initUpdate.getData().getData().isEmpty()); + getWsClient().registerWaitForUpdate(); + + doPostAsync("/api/plugins/telemetry/DEVICE/" + device.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); + String update = getWsClient().waitForUpdate(); + assertThat(update).as("ws update received").isNotBlank(); + + client.subscribeAndWait(GATEWAY_ATTRIBUTES_RESPONSE_TOPIC, MqttQoS.AT_LEAST_ONCE); + awaitForDeviceActorToReceiveSubscription(device.getId(), FeatureType.ATTRIBUTES, 1); + + MqttTestCallback callback = new MqttTestSubscribeOnTopicCallback(GATEWAY_ATTRIBUTES_RESPONSE_TOPIC); + client.setCallback(callback); + // new proto format: allClientKeys/allSharedKeys => all of that scope, scope-separated response, other scope absent + TransportApiProtos.GatewayAttributesRequestMsg.Builder request = TransportApiProtos.GatewayAttributesRequestMsg.newBuilder() + .setDeviceName(deviceName) + .setId(1); + if (allClient) { + request.setAllClientKeys(true); + } else { + request.setAllSharedKeys(true); + } + client.publishAndWait(GATEWAY_ATTRIBUTES_REQUEST_TOPIC, request.build().toByteArray()); + + TransportApiProtos.GatewayAttributeResponseMsg actual; + if (allClient) { + validateProtoClientResponseGateway(callback, deviceName); + actual = TransportApiProtos.GatewayAttributeResponseMsg.parseFrom(callback.getPayloadBytes()); + assertTrue(actual.getResponseMsg().getSharedAttributeListList().isEmpty()); + } else { + validateProtoSharedResponseGateway(callback, deviceName); + actual = TransportApiProtos.GatewayAttributeResponseMsg.parseFrom(callback.getPayloadBytes()); + assertTrue(actual.getResponseMsg().getClientAttributeListList().isEmpty()); + } + + client.disconnect(); + } + private List getEntityKeys(List keys, EntityKeyType scope) { return keys.stream().map(key -> new EntityKey(scope, key)).collect(Collectors.toList()); } + private List buildClientAndSharedEntityKeys() { + List keys = new ArrayList<>(); + keys.addAll(getEntityKeys(List.of(CLIENT_ATTRIBUTE_KEYS.split(",")), CLIENT_ATTRIBUTE)); + keys.addAll(getEntityKeys(List.of(SHARED_ATTRIBUTE_KEYS.split(",")), SHARED_ATTRIBUTE)); + return keys; + } + private byte[] getAttributesProtoPayloadBytes() { DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration(); assertTrue(transportConfiguration instanceof MqttDeviceProfileTransportConfiguration); @@ -739,6 +804,7 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt return gatewayAttributeResponseMsg.build(); } + @SuppressWarnings("deprecation") // exercises the legacy single-scope gateway request (keys + client) private TransportApiProtos.GatewayAttributesRequestMsg getGatewayAttributesRequestMsg(String deviceName, List keysList, boolean client) { return TransportApiProtos.GatewayAttributesRequestMsg.newBuilder() .setDeviceName(deviceName) diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/request/MqttAttributesRequestProtoIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/request/MqttAttributesRequestProtoIntegrationTest.java index e368183459..b4be48d2e7 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/request/MqttAttributesRequestProtoIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/request/MqttAttributesRequestProtoIntegrationTest.java @@ -74,6 +74,26 @@ public class MqttAttributesRequestProtoIntegrationTest extends AbstractMqttAttri processProtoTestGatewayRequestAttributesValuesFromTheServer(); } + @Test + public void testRequestAllSharedAttributesFromTheServerGatewaySeparated() throws Exception { + MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() + .gatewayName("Gateway Test Request all shared attributes from the server proto") + .transportPayloadType(TransportPayloadType.PROTOBUF) + .build(); + processBeforeTest(configProperties); + processProtoTestGatewayRequestAllSharedSeparated(); + } + + @Test + public void testRequestAllClientAttributesFromTheServerGatewaySeparated() throws Exception { + MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() + .gatewayName("Gateway Test Request all client attributes from the server proto") + .transportPayloadType(TransportPayloadType.PROTOBUF) + .build(); + processBeforeTest(configProperties); + processProtoTestGatewayRequestAllClientSeparated(); + } + @Test public void testRequestAttributesValuesFromTheServerOnShortJsonTopic() throws Exception { MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() diff --git a/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java b/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java index a7ccead805..7284020aee 100644 --- a/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java +++ b/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java @@ -354,7 +354,6 @@ public class JsonConverter { return result; } - @SuppressWarnings("deprecation") // isMultipleAttributesRequest retained for the legacy gateway value/values response public static JsonObject getJsonObjectForGateway( String deviceName, TransportProtos.GetAttributeResponseMsg responseMsg @@ -363,27 +362,24 @@ public class JsonConverter { result.addProperty("id", responseMsg.getRequestId()); result.addProperty(DEVICE_PROPERTY, deviceName); if (responseMsg.getSeparateScopesResponse()) { - if (responseMsg.getClientAttributeListCount() > 0) { - JsonObject client = new JsonObject(); - responseMsg.getClientAttributeListList().forEach(addToObjectFromProto(client)); - result.add("client", client); - } - if (responseMsg.getSharedAttributeListCount() > 0) { - JsonObject shared = new JsonObject(); - responseMsg.getSharedAttributeListList().forEach(addToObjectFromProto(shared)); - result.add("shared", shared); - } + // Reuse the device-side scope-separated assembly so both responses stay identical in shape. + toJson(responseMsg).entrySet().forEach(entry -> result.add(entry.getKey(), entry.getValue())); } else { - if (responseMsg.getClientAttributeListCount() > 0) { - addValues(result, responseMsg.getClientAttributeListList(), responseMsg.getIsMultipleAttributesRequest()); - } - if (responseMsg.getSharedAttributeListCount() > 0) { - addValues(result, responseMsg.getSharedAttributeListList(), responseMsg.getIsMultipleAttributesRequest()); - } + addLegacyGatewayValues(result, responseMsg); } return result; } + @SuppressWarnings("deprecation") // isMultipleAttributesRequest retained for the legacy gateway value/values response + private static void addLegacyGatewayValues(JsonObject result, TransportProtos.GetAttributeResponseMsg responseMsg) { + if (responseMsg.getClientAttributeListCount() > 0) { + addValues(result, responseMsg.getClientAttributeListList(), responseMsg.getIsMultipleAttributesRequest()); + } + if (responseMsg.getSharedAttributeListCount() > 0) { + addValues(result, responseMsg.getSharedAttributeListList(), responseMsg.getIsMultipleAttributesRequest()); + } + } + public static JsonObject getJsonObjectForGateway(String deviceName, AttributeUpdateNotificationMsg notificationMsg) { JsonObject result = new JsonObject(); diff --git a/common/proto/src/main/proto/transport.proto b/common/proto/src/main/proto/transport.proto index e951079e2e..123913eca6 100644 --- a/common/proto/src/main/proto/transport.proto +++ b/common/proto/src/main/proto/transport.proto @@ -96,8 +96,8 @@ message GatewayDeviceRpcRequestMsg { message GatewayAttributesRequestMsg { int32 id = 1; string deviceName = 2; - bool client = 3; - repeated string keys = 4; + bool client = 3 [deprecated = true]; + repeated string keys = 4 [deprecated = true]; repeated string clientKeys = 5; repeated string sharedKeys = 6; bool allClientKeys = 7; diff --git a/common/proto/src/test/java/org/thingsboard/server/common/adaptor/JsonConverterGatewayResponseTest.java b/common/proto/src/test/java/org/thingsboard/server/common/adaptor/JsonConverterGatewayResponseTest.java index 5c9af2a679..a62d26f98b 100644 --- a/common/proto/src/test/java/org/thingsboard/server/common/adaptor/JsonConverterGatewayResponseTest.java +++ b/common/proto/src/test/java/org/thingsboard/server/common/adaptor/JsonConverterGatewayResponseTest.java @@ -16,7 +16,7 @@ package org.thingsboard.server.common.adaptor; import com.google.gson.JsonObject; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeResponseMsg; import org.thingsboard.server.gen.transport.TransportProtos.KeyValueProto; import org.thingsboard.server.gen.transport.TransportProtos.KeyValueType; diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/CoapAdaptorUtils.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/CoapAdaptorUtils.java index 7f6b886f36..2ec44e4976 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/CoapAdaptorUtils.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/CoapAdaptorUtils.java @@ -51,24 +51,20 @@ public class CoapAdaptorUtils { } private static String getQueryValue(List queryElements, String name) { + // Keep the last matching occurrence to preserve the original toKeys() behavior. + String value = null; for (String queryElement : queryElements) { String[] queryItem = queryElement.split("="); if (queryItem.length == 2 && queryItem[0].equals(name)) { - return queryItem[1]; + value = queryItem[1]; } } - return null; + return value; } - private static Set toKeys(List queryElements, String attributeName) throws AdaptorException { - String keys = null; - for (String queryElement : queryElements) { - String[] queryItem = queryElement.split("="); - if (queryItem.length == 2 && queryItem[0].equals(attributeName)) { - keys = queryItem[1]; - } - } - if (keys != null && !StringUtils.isEmpty(keys)) { + private static Set toKeys(List queryElements, String attributeName) { + String keys = getQueryValue(queryElements, attributeName); + if (!StringUtils.isEmpty(keys)) { return new HashSet<>(Arrays.asList(keys.split(","))); } else { return null; 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 0910dbec1e..72fd38aef4 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 @@ -37,9 +37,12 @@ import org.thingsboard.server.transport.mqtt.session.MqttDeviceAwareSessionConte import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; import java.util.Arrays; +import java.util.List; import java.util.Optional; import java.util.UUID; +import java.util.function.Consumer; import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_SOFTWARE_FIRMWARE_RESPONSES_TOPIC_FORMAT; @@ -175,8 +178,8 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor { result.setRequestId(getRequestId(topicName, topicBase)); String payload = inbound.payload().toString(UTF8); JsonObject json = JsonParser.parseString(payload).getAsJsonObject(); - parseAttrScope(json, "clientKeys", result::addAllClientAttributeNames, () -> result.setAllClientAttributes(true)); - parseAttrScope(json, "sharedKeys", result::addAllSharedAttributeNames, () -> result.setAllSharedAttributes(true)); + parseAttributeScope(json, "clientKeys", result::addAllClientAttributeNames, () -> result.setAllClientAttributes(true)); + parseAttributeScope(json, "sharedKeys", result::addAllSharedAttributeNames, () -> result.setAllSharedAttributes(true)); return result.build(); } catch (RuntimeException e) { log.debug("Failed to decode get attributes request", e); @@ -240,18 +243,37 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor { return new MqttPublishMessage(mqttFixedHeader, header, payload); } - // Three-state per scope: field absent => exclude; present + empty value => all in scope; present + list => those keys. - private static void parseAttrScope(JsonObject json, String field, - java.util.function.Consumer> setNames, - Runnable setAll) { + /** + * Three-state per-scope attribute selection shared by the device and gateway JSON request parsers: + *
    + *
  • field absent / null => the scope is excluded (neither names nor "all" is set);
    • + *
  • field present + empty value (empty string or empty array) => every key in that scope ({@code setAll});
    • + *
  • field present + a comma-separated string or a JSON array of names => only those keys ({@code setNames}).
    • + *
+ */ + public static void parseAttributeScope(JsonObject json, String field, + Consumer> setNames, Runnable setAll) { if (!json.has(field) || json.get(field).isJsonNull()) { return; } - String value = json.get(field).getAsString(); - if (value.trim().isEmpty()) { + JsonElement element = json.get(field); + List names = new ArrayList<>(); + if (element.isJsonArray()) { + for (JsonElement e : element.getAsJsonArray()) { + names.add(e.getAsString()); + } + } else { + String value = element.getAsString(); + if (value.trim().isEmpty()) { + setAll.run(); + return; + } + names.addAll(Arrays.asList(value.split(","))); + } + if (names.isEmpty()) { setAll.run(); } else { - setNames.accept(Arrays.asList(value.split(","))); + setNames.accept(names); } } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java index 4fced72e85..660eeaf954 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java @@ -658,8 +658,8 @@ public abstract class AbstractGatewaySessionHandler b.setAllClientAttributes(true)); - parseGatewayScope(jsonObj, "sharedKeys", b::addAllSharedAttributeNames, () -> b.setAllSharedAttributes(true)); + JsonMqttAdaptor.parseAttributeScope(jsonObj, "clientKeys", b::addAllClientAttributeNames, () -> b.setAllClientAttributes(true)); + JsonMqttAdaptor.parseAttributeScope(jsonObj, "sharedKeys", b::addAllSharedAttributeNames, () -> b.setAllSharedAttributes(true)); requestMsg = b.build(); } else if (jsonObj.has("client")) { // legacy format: client boolean + key/keys; keep the legacy value/values response @@ -683,33 +683,6 @@ public abstract class AbstractGatewaySessionHandler all in scope. - private static void parseGatewayScope(JsonObject json, String field, - Consumer> setNames, Runnable setAll) { - if (!json.has(field) || json.get(field).isJsonNull()) { - return; - } - JsonElement el = json.get(field); - List names = new ArrayList<>(); - if (el.isJsonArray()) { - for (JsonElement e : el.getAsJsonArray()) { - names.add(e.getAsString()); - } - } else { - String v = el.getAsString(); - if (v.trim().isEmpty()) { - setAll.run(); - return; - } - names.addAll(java.util.Arrays.asList(v.split(","))); - } - if (names.isEmpty()) { - setAll.run(); - } else { - setNames.accept(names); - } - } - private void onDeviceAttributesRequestProto(MqttPublishMessage mqttMsg) throws AdaptorException { try { TransportApiProtos.GatewayAttributesRequestMsg gw = TransportApiProtos.GatewayAttributesRequestMsg.parseFrom(getBytes(mqttMsg.payload())); @@ -733,9 +706,7 @@ public abstract class AbstractGatewaySessionHandler keys = new HashSet<>(gw.getKeysList()); - requestMsg = toGetAttributeRequestMsg(requestId, clientScope, keys); + requestMsg = toLegacyGatewayRequestMsg(requestId, gw); } processGetAttributeRequestMessage(mqttMsg, deviceName, requestMsg); } catch (RuntimeException | InvalidProtocolBufferException e) { @@ -743,6 +714,13 @@ public abstract class AbstractGatewaySessionHandler keys = new HashSet<>(gw.getKeysList()); + return toGetAttributeRequestMsg(requestId, clientScope, keys); + } + private void onDeviceRpcResponseJson(int msgId, ByteBuf payload) throws AdaptorException { JsonElement json = JsonMqttAdaptor.validateJsonPayload(sessionId, payload); validateJsonObject(json); diff --git a/common/transport/mqtt/src/test/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptorTest.java b/common/transport/mqtt/src/test/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptorTest.java new file mode 100644 index 0000000000..38c1f089f6 --- /dev/null +++ b/common/transport/mqtt/src/test/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptorTest.java @@ -0,0 +1,103 @@ +/** + * Copyright © 2016-2026 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.transport.mqtt.adaptors; + +import com.google.gson.JsonArray; +import com.google.gson.JsonNull; +import com.google.gson.JsonObject; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class JsonMqttAdaptorTest { + + private final List names = new ArrayList<>(); + private final AtomicBoolean allInvoked = new AtomicBoolean(false); + + private void parseClientKeys(JsonObject json) { + JsonMqttAdaptor.parseAttributeScope(json, "clientKeys", names::addAll, () -> allInvoked.set(true)); + } + + @Test + public void fieldAbsent_doesNothing() { + parseClientKeys(new JsonObject()); + assertTrue(names.isEmpty()); + assertFalse(allInvoked.get()); + } + + @Test + public void fieldNull_doesNothing() { + JsonObject json = new JsonObject(); + json.add("clientKeys", JsonNull.INSTANCE); + parseClientKeys(json); + assertTrue(names.isEmpty()); + assertFalse(allInvoked.get()); + } + + @Test + public void emptyString_setsAll() { + JsonObject json = new JsonObject(); + json.addProperty("clientKeys", ""); + parseClientKeys(json); + assertTrue(allInvoked.get()); + assertTrue(names.isEmpty()); + } + + @Test + public void blankString_setsAll() { + JsonObject json = new JsonObject(); + json.addProperty("clientKeys", " "); + parseClientKeys(json); + assertTrue(allInvoked.get()); + assertTrue(names.isEmpty()); + } + + @Test + public void emptyArray_setsAll() { + JsonObject json = new JsonObject(); + json.add("clientKeys", new JsonArray()); + parseClientKeys(json); + assertTrue(allInvoked.get()); + assertTrue(names.isEmpty()); + } + + @Test + public void commaSeparatedString_setsNames() { + JsonObject json = new JsonObject(); + json.addProperty("clientKeys", "a,b,c"); + parseClientKeys(json); + assertFalse(allInvoked.get()); + assertEquals(List.of("a", "b", "c"), names); + } + + @Test + public void jsonArray_setsNames() { + JsonObject json = new JsonObject(); + JsonArray arr = new JsonArray(); + arr.add("a"); + arr.add("b"); + json.add("clientKeys", arr); + parseClientKeys(json); + assertFalse(allInvoked.get()); + assertEquals(List.of("a", "b"), names); + } +}