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 d8710eaa2c..6c19ac83e7 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 @@ -23,7 +23,6 @@ import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; import jakarta.annotation.Nullable; import lombok.extern.slf4j.Slf4j; -import org.apache.commons.collections4.CollectionUtils; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.LinkedHashMapRemoveEldest; import org.thingsboard.server.actors.ActorSystemContext; @@ -97,7 +96,6 @@ import org.thingsboard.server.service.state.DefaultDeviceStateService; import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; @@ -524,13 +522,11 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso Futures.addCallback(findAllAttributesByScope(AttributeScope.SHARED_SCOPE), new FutureCallback<>() { @Override public void onSuccess(@Nullable List result) { - GetAttributeResponseMsg responseMsg = GetAttributeResponseMsg.newBuilder() + sendToTransport(GetAttributeResponseMsg.newBuilder() .setRequestId(requestId) .setSharedStateMsg(true) .addAllSharedAttributeList(KvProtoUtil.attrToTsKvProtos(result)) - .setIsMultipleAttributesRequest(request.getSharedAttributeNamesCount() > 1) - .build(); - sendToTransport(responseMsg, sessionInfo); + .build(), sessionInfo); } @Override @@ -545,15 +541,19 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso } else { Futures.addCallback(getAttributesKvEntries(request), new FutureCallback<>() { @Override - public void onSuccess(@Nullable List> result) { - GetAttributeResponseMsg responseMsg = GetAttributeResponseMsg.newBuilder() + public void onSuccess(ClientSharedAttributes result) { + GetAttributeResponseMsg.Builder builder = GetAttributeResponseMsg.newBuilder() .setRequestId(requestId) - .addAllClientAttributeList(KvProtoUtil.attrToTsKvProtos(result.get(0))) - .addAllSharedAttributeList(KvProtoUtil.attrToTsKvProtos(result.get(1))) - .setIsMultipleAttributesRequest( - request.getSharedAttributeNamesCount() + request.getClientAttributeNamesCount() > 1) - .build(); - sendToTransport(responseMsg, sessionInfo); + .addAllClientAttributeList(KvProtoUtil.attrToTsKvProtos(result.client())) + .addAllSharedAttributeList(KvProtoUtil.attrToTsKvProtos(result.shared())); + // legacy gateway value/values response relies on the deprecated isMultipleAttributesRequest flag + if (!request.getSeparateScopesResponse()) { + builder.setIsMultipleAttributesRequest( + request.getSharedAttributeNamesCount() + request.getClientAttributeNamesCount() > 1); + } else { + builder.setSeparateScopesResponse(true); + } + sendToTransport(builder.build(), sessionInfo); } @Override @@ -567,23 +567,35 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso } } - private ListenableFuture>> getAttributesKvEntries(GetAttributeRequestMsg request) { - ListenableFuture> clientAttributesFuture; - ListenableFuture> sharedAttributesFuture; - if (CollectionUtils.isEmpty(request.getClientAttributeNamesList()) && CollectionUtils.isEmpty(request.getSharedAttributeNamesList())) { - clientAttributesFuture = findAllAttributesByScope(AttributeScope.CLIENT_SCOPE); - sharedAttributesFuture = findAllAttributesByScope(AttributeScope.SHARED_SCOPE); - } else if (!CollectionUtils.isEmpty(request.getClientAttributeNamesList()) && !CollectionUtils.isEmpty(request.getSharedAttributeNamesList())) { - clientAttributesFuture = findAttributesByScope(toSet(request.getClientAttributeNamesList()), AttributeScope.CLIENT_SCOPE); - sharedAttributesFuture = findAttributesByScope(toSet(request.getSharedAttributeNamesList()), AttributeScope.SHARED_SCOPE); - } else if (CollectionUtils.isEmpty(request.getClientAttributeNamesList()) && !CollectionUtils.isEmpty(request.getSharedAttributeNamesList())) { - clientAttributesFuture = Futures.immediateFuture(Collections.emptyList()); - sharedAttributesFuture = findAttributesByScope(toSet(request.getSharedAttributeNamesList()), AttributeScope.SHARED_SCOPE); + private record ClientSharedAttributes(List client, List shared) {} + + private ListenableFuture getAttributesKvEntries(GetAttributeRequestMsg request) { + boolean clientAll = request.getAllClientAttributes(); + boolean sharedAll = request.getAllSharedAttributes(); + List clientNames = request.getClientAttributeNamesList(); + List sharedNames = request.getSharedAttributeNamesList(); + + // backward-compat: an empty request (no scope signalled at all) fetches everything from both scopes + boolean fetchAll = !clientAll && !sharedAll && clientNames.isEmpty() && sharedNames.isEmpty(); + + ListenableFuture> clientFuture = resolveScopeFuture(clientAll || fetchAll, clientNames, AttributeScope.CLIENT_SCOPE); + ListenableFuture> sharedFuture = resolveScopeFuture(sharedAll || fetchAll, sharedNames, AttributeScope.SHARED_SCOPE); + + return Futures.whenAllSucceed(clientFuture, sharedFuture) + .call(() -> new ClientSharedAttributes(Futures.getDone(clientFuture), Futures.getDone(sharedFuture)), + MoreExecutors.directExecutor()); + } + + // "all " wins over a specific key list for the same scope; no signal => empty result. + // Same precedence as JsonConverter.applyScope (the request-build side); keep the two in sync. + private ListenableFuture> resolveScopeFuture(boolean all, List names, AttributeScope scope) { + if (all) { + return findAllAttributesByScope(scope); + } else if (!names.isEmpty()) { + return findAttributesByScope(new HashSet<>(names), scope); } else { - sharedAttributesFuture = Futures.immediateFuture(Collections.emptyList()); - clientAttributesFuture = findAttributesByScope(toSet(request.getClientAttributeNamesList()), AttributeScope.CLIENT_SCOPE); + return Futures.immediateFuture(Collections.emptyList()); } - return Futures.allAsList(Arrays.asList(clientAttributesFuture, sharedAttributesFuture)); } private ListenableFuture> findAllAttributesByScope(AttributeScope scope) { @@ -594,10 +606,6 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso return systemContext.getAttributesService().find(tenantId, deviceId, scope, attributesSet); } - private Set toSet(List strings) { - return new HashSet<>(strings); - } - private SessionType getSessionType(UUID sessionId) { return sessions.containsKey(sessionId) ? SessionType.ASYNC : SessionType.SYNC; } 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 dc28913851..9dff237eee 100644 --- a/application/src/test/java/org/thingsboard/server/system/BaseHttpDeviceApiTest.java +++ b/application/src/test/java/org/thingsboard/server/system/BaseHttpDeviceApiTest.java @@ -15,6 +15,8 @@ */ 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; @@ -25,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; @@ -74,6 +78,60 @@ public abstract class BaseHttpDeviceApiTest extends AbstractControllerTest { doGetAsync("/api/v1/" + deviceCredentials.getCredentialsId() + "/attributes?clientKeys=keyA,keyB,keyC").andExpect(status().isOk()); } + @Test + public void testGetAllClientAttributesViaAllClientKeysParam() throws Exception { + String token = deviceCredentials.getCredentialsId(); + Map clientAttrs = new HashMap<>(); + clientAttrs.put("clientA", "valueA"); + clientAttrs.put("clientB", "valueB"); + mockMvc.perform( + asyncDispatch(doPost("/api/v1/" + token + "/attributes", clientAttrs, new String[]{}).andReturn())) + .andExpect(status().isOk()); + 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(); + JsonNode client = resp.get("client"); + clientAttrs.forEach((key, value) -> assertThat(client.get(key).asText()).isEqualTo(value)); + assertThat(resp.has("shared")).isFalse(); + } + + @Test + public void testGetAllSharedAttributesViaAllSharedKeysParam() throws Exception { + String token = deviceCredentials.getCredentialsId(); + Map sharedAttrs = new HashMap<>(); + sharedAttrs.put("sharedA", "valueA"); + sharedAttrs.put("sharedB", "valueB"); + mockMvc.perform( + asyncDispatch(doPost("/api/plugins/telemetry/DEVICE/" + device.getId().getId() + "/attributes/SHARED_SCOPE", sharedAttrs).andReturn())) + .andExpect(status().isOk()); + String allSharedKeysUrl = "/api/v1/" + token + "/attributes?allSharedKeys=true"; + Awaitility.await("shared attributes are persisted and returned via allSharedKeys=true") + .atMost(30, TimeUnit.SECONDS) + .pollInterval(Duration.ofMillis(100)) + .ignoreExceptions() + .until(() -> { + JsonNode r = JacksonUtil.toJsonNode(doGetAsync(allSharedKeysUrl).andReturn().getResponse().getContentAsString()); + return r.has("shared") && sharedAttrs.keySet().stream().allMatch(r.get("shared")::has); + }); + String body = doGetAsync(allSharedKeysUrl) + .andExpect(status().isOk()).andReturn().getResponse().getContentAsString(); + JsonNode resp = JacksonUtil.toJsonNode(body); + assertThat(resp.has("shared")).isTrue(); + JsonNode shared = resp.get("shared"); + sharedAttrs.forEach((key, value) -> assertThat(shared.get(key).asText()).isEqualTo(value)); + assertThat(resp.has("client")).isFalse(); + } + @Test public void testReplyToCommandWithLargeResponse() throws Exception { String errorResponse = doPost("/api/v1/" + deviceCredentials.getCredentialsId() + "/rpc/5", diff --git a/application/src/test/java/org/thingsboard/server/transport/AbstractTransportIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/AbstractTransportIntegrationTest.java index c91fea282d..03d8a0e7f4 100644 --- a/application/src/test/java/org/thingsboard/server/transport/AbstractTransportIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/AbstractTransportIntegrationTest.java @@ -30,6 +30,9 @@ public abstract class AbstractTransportIntegrationTest extends AbstractControlle public static final int DEFAULT_WAIT_TIMEOUT_SECONDS = 30; + protected static final String CLIENT_ATTRIBUTE_KEYS = "clientStr,clientBool,clientDbl,clientLong,clientJson"; + protected static final String SHARED_ATTRIBUTE_KEYS = "sharedStr,sharedBool,sharedDbl,sharedLong,sharedJson"; + protected static final AtomicInteger atomicInteger = new AtomicInteger(2); public static final String DEVICE_TELEMETRY_PROTO_SCHEMA = "syntax =\"proto3\";\n" + diff --git a/application/src/test/java/org/thingsboard/server/transport/coap/attributes/AbstractCoapAttributesIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/coap/attributes/AbstractCoapAttributesIntegrationTest.java index de5423abf1..a47bf82372 100644 --- a/application/src/test/java/org/thingsboard/server/transport/coap/attributes/AbstractCoapAttributesIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/coap/attributes/AbstractCoapAttributesIntegrationTest.java @@ -89,12 +89,13 @@ public abstract class AbstractCoapAttributesIntegrationTest extends AbstractCoap " }\n" + "}"; - private static final String CLIENT_ATTRIBUTES_PAYLOAD = "{\"clientStr\":\"value1\",\"clientBool\":true,\"clientDbl\":42.0,\"clientLong\":73," + + protected static final String CLIENT_ATTRIBUTES_PAYLOAD = "{\"clientStr\":\"value1\",\"clientBool\":true,\"clientDbl\":42.0,\"clientLong\":73," + "\"clientJson\":{\"someNumber\":42,\"someArray\":[1,2,3],\"someNestedObject\":{\"key\":\"value\"}}}"; - private static final String SHARED_ATTRIBUTES_PAYLOAD = "{\"sharedStr\":\"value1\",\"sharedBool\":true,\"sharedDbl\":42.0,\"sharedLong\":73," + + 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 SHARED_ATTRIBUTES_PAYLOAD_ON_CURRENT_STATE_NOTIFICATION = "{\"sharedStr\":\"value\",\"sharedBool\":false,\"sharedDbl\":41.0,\"sharedLong\":72," + "\"sharedJson\":{\"someNumber\":41,\"someArray\":[],\"someNestedObject\":{\"key\":\"value\"}}}"; @@ -169,18 +170,28 @@ public abstract class AbstractCoapAttributesIntegrationTest extends AbstractCoap } protected void processJsonTestRequestAttributesValuesFromTheServer() throws Exception { + postAttributesAndAwaitWsUpdate(); + String featureTokenUrl = CoapTestClient.getFeatureTokenUrl(accessToken, FeatureType.ATTRIBUTES) + "?clientKeys=" + CLIENT_ATTRIBUTE_KEYS + "&sharedKeys=" + SHARED_ATTRIBUTE_KEYS; + client.setURI(featureTokenUrl); + validateJsonResponse(client.getMethod()); + } + + protected void processJsonTestRequestAttributesWithQuery(String querySuffix, String expectedResponse) throws Exception { + postAttributesAndAwaitWsUpdate(); + String featureTokenUrl = CoapTestClient.getFeatureTokenUrl(accessToken, FeatureType.ATTRIBUTES) + querySuffix; + client.setURI(featureTokenUrl); + CoapResponse response = client.getMethod(); + assertEquals(CoAP.ResponseCode.CONTENT, response.getCode()); + assertEquals(JacksonUtil.toJsonNode(expectedResponse), JacksonUtil.fromBytes(response.getPayload())); + } + + private void postAttributesAndAwaitWsUpdate() throws Exception { client = new CoapTestClient(accessToken, FeatureType.ATTRIBUTES); 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 clientKeysList = List.of(clientKeysStr.split(",")); - List sharedKeysList = List.of(sharedKeysStr.split(",")); - List csKeys = getEntityKeys(clientKeysList, CLIENT_ATTRIBUTE); - List shKeys = getEntityKeys(sharedKeysList, SHARED_ATTRIBUTE); List keys = new ArrayList<>(); - keys.addAll(csKeys); - keys.addAll(shKeys); + keys.addAll(getEntityKeys(List.of(CLIENT_ATTRIBUTE_KEYS.split(",")), CLIENT_ATTRIBUTE)); + keys.addAll(getEntityKeys(List.of(SHARED_ATTRIBUTE_KEYS.split(",")), SHARED_ATTRIBUTE)); getWsClient().subscribeLatestUpdate(keys, dtf); getWsClient().registerWaitForUpdate(2); @@ -192,20 +203,14 @@ public abstract class AbstractCoapAttributesIntegrationTest extends AbstractCoap String update = getWsClient().waitForUpdate(); assertThat(update).as("ws update received").isNotBlank(); - - String featureTokenUrl = CoapTestClient.getFeatureTokenUrl(accessToken, FeatureType.ATTRIBUTES) + "?clientKeys=" + clientKeysStr + "&sharedKeys=" + sharedKeysStr; - client.setURI(featureTokenUrl); - validateJsonResponse(client.getMethod()); } protected void processProtoTestRequestAttributesValuesFromTheServer() throws Exception { client = new CoapTestClient(accessToken, FeatureType.ATTRIBUTES); 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 clientKeysList = List.of(clientKeysStr.split(",")); - List sharedKeysList = List.of(sharedKeysStr.split(",")); + List clientKeysList = List.of(CLIENT_ATTRIBUTE_KEYS.split(",")); + List sharedKeysList = List.of(SHARED_ATTRIBUTE_KEYS.split(",")); List csKeys = getEntityKeys(clientKeysList, CLIENT_ATTRIBUTE); List shKeys = getEntityKeys(sharedKeysList, SHARED_ATTRIBUTE); List keys = new ArrayList<>(); @@ -223,7 +228,7 @@ public abstract class AbstractCoapAttributesIntegrationTest extends AbstractCoap String update = getWsClient().waitForUpdate(); assertThat(update).as("ws update received").isNotBlank(); - String featureTokenUrl = CoapTestClient.getFeatureTokenUrl(accessToken, FeatureType.ATTRIBUTES) + "?clientKeys=" + clientKeysStr + "&sharedKeys=" + sharedKeysStr; + String featureTokenUrl = CoapTestClient.getFeatureTokenUrl(accessToken, FeatureType.ATTRIBUTES) + "?clientKeys=" + CLIENT_ATTRIBUTE_KEYS + "&sharedKeys=" + SHARED_ATTRIBUTE_KEYS; client.setURI(featureTokenUrl); validateProtoResponse(client.getMethod()); } diff --git a/application/src/test/java/org/thingsboard/server/transport/coap/attributes/request/CoapAttributesRequestJsonIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/coap/attributes/request/CoapAttributesRequestJsonIntegrationTest.java index e9f75574a4..0449e590ac 100644 --- a/application/src/test/java/org/thingsboard/server/transport/coap/attributes/request/CoapAttributesRequestJsonIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/coap/attributes/request/CoapAttributesRequestJsonIntegrationTest.java @@ -47,4 +47,14 @@ public class CoapAttributesRequestJsonIntegrationTest extends CoapAttributesRequ public void testRequestAttributesValuesFromTheServer() throws Exception { processJsonTestRequestAttributesValuesFromTheServer(); } + + @Test + public void testRequestAllSharedAttributes() throws Exception { + processJsonTestRequestAttributesWithQuery("?allSharedKeys=true", "{\"shared\":" + SHARED_ATTRIBUTES_PAYLOAD + "}"); + } + + @Test + public void testRequestAllClientAttributes() throws Exception { + processJsonTestRequestAttributesWithQuery("?allClientKeys=true", "{\"client\":" + CLIENT_ATTRIBUTES_PAYLOAD + "}"); + } } 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 5f43aa7e00..f7c02c0144 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 @@ -92,12 +92,13 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt " }\n" + "}"; - private static final String CLIENT_ATTRIBUTES_PAYLOAD = "{\"clientStr\":\"value1\",\"clientBool\":true,\"clientDbl\":42.0,\"clientLong\":73," + + protected static final String CLIENT_ATTRIBUTES_PAYLOAD = "{\"clientStr\":\"value1\",\"clientBool\":true,\"clientDbl\":42.0,\"clientLong\":73," + "\"clientJson\":{\"someNumber\":42,\"someArray\":[1,2,3],\"someNestedObject\":{\"key\":\"value\"}}}"; - private static final String SHARED_ATTRIBUTES_PAYLOAD = "{\"sharedStr\":\"value1\",\"sharedBool\":true,\"sharedDbl\":42.0,\"sharedLong\":73," + + 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\"}}}"; + private static final String SHARED_ATTRIBUTES_DELETED_RESPONSE = "{\"deleted\":[\"sharedJson\"]}"; private List getTsKvProtoList(String attributePrefix) { @@ -341,10 +342,8 @@ 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 clientKeysList = List.of(clientKeysStr.split(",")); - List sharedKeysList = List.of(sharedKeysStr.split(",")); + List clientKeysList = List.of(CLIENT_ATTRIBUTE_KEYS.split(",")); + List sharedKeysList = List.of(SHARED_ATTRIBUTE_KEYS.split(",")); List csKeys = getEntityKeys(clientKeysList, CLIENT_ATTRIBUTE); List shKeys = getEntityKeys(sharedKeysList, SHARED_ATTRIBUTE); List keys = new ArrayList<>(); @@ -362,21 +361,41 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt assertThat(update).as("ws update received").isNotBlank(); MqttTestCallback callback = new MqttTestSubscribeOnTopicCallback(attrSubTopic.replace("+", "1")); client.setCallback(callback); - String payloadStr = "{\"clientKeys\":\"" + clientKeysStr + "\", \"sharedKeys\":\"" + sharedKeysStr + "\"}"; + String payloadStr = "{\"clientKeys\":\"" + CLIENT_ATTRIBUTE_KEYS + "\", \"sharedKeys\":\"" + SHARED_ATTRIBUTE_KEYS + "\"}"; client.publishAndWait(attrReqTopicPrefix + "1", payloadStr.getBytes()); String expectedResponse = "{\"client\":" + CLIENT_ATTRIBUTES_PAYLOAD + ",\"shared\":" + SHARED_ATTRIBUTES_PAYLOAD + "}"; validateJsonResponse(callback, expectedResponse); client.disconnect(); } + protected void processJsonTestRequestAttributesWithPayload(String attrPubTopic, String attrSubTopic, String attrReqTopicPrefix, + String requestPayload, String expectedResponse) throws Exception { + MqttTestClient client = new MqttTestClient(); + client.connectAndWait(accessToken); + SingleEntityFilter dtf = new SingleEntityFilter(); + dtf.setSingleEntity(AliasEntityId.fromEntityId(savedDevice.getId())); + List keys = buildClientAndSharedEntityKeys(); + getWsClient().subscribeLatestUpdate(keys, dtf); + getWsClient().registerWaitForUpdate(2); + doPostAsync("/api/plugins/telemetry/DEVICE/" + savedDevice.getId().getId() + "/attributes/SHARED_SCOPE", + SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); + client.publishAndWait(attrPubTopic, CLIENT_ATTRIBUTES_PAYLOAD.getBytes()); + client.subscribeAndWait(attrSubTopic, MqttQoS.AT_MOST_ONCE); + String update = getWsClient().waitForUpdate(); + assertThat(update).as("ws update received").isNotBlank(); + MqttTestCallback callback = new MqttTestSubscribeOnTopicCallback(attrSubTopic.replace("+", "1")); + client.setCallback(callback); + client.publishAndWait(attrReqTopicPrefix + "1", requestPayload.getBytes()); + validateJsonResponse(callback, expectedResponse); + client.disconnect(); + } + protected void processProtoTestRequestAttributesValuesFromTheServer(String attrPubTopic, String attrSubTopic, String attrReqTopicPrefix) throws Exception { MqttTestClient client = new MqttTestClient(); client.connectAndWait(accessToken); DeviceTypeFilter dtf = new DeviceTypeFilter(List.of(savedDevice.getType()), savedDevice.getName()); - String clientKeysStr = "clientStr,clientBool,clientDbl,clientLong,clientJson"; - String sharedKeysStr = "sharedStr,sharedBool,sharedDbl,sharedLong,sharedJson"; - List clientKeysList = List.of(clientKeysStr.split(",")); - List sharedKeysList = List.of(sharedKeysStr.split(",")); + List clientKeysList = List.of(CLIENT_ATTRIBUTE_KEYS.split(",")); + List sharedKeysList = List.of(SHARED_ATTRIBUTE_KEYS.split(",")); List csKeys = getEntityKeys(clientKeysList, CLIENT_ATTRIBUTE); List shKeys = getEntityKeys(sharedKeysList, SHARED_ATTRIBUTE); List keys = new ArrayList<>(); @@ -394,8 +413,8 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt MqttTestCallback callback = new MqttTestSubscribeOnTopicCallback(attrSubTopic.replace("+", "1")); client.setCallback(callback); TransportApiProtos.AttributesRequest.Builder attributesRequestBuilder = TransportApiProtos.AttributesRequest.newBuilder(); - attributesRequestBuilder.setClientKeys(clientKeysStr); - attributesRequestBuilder.setSharedKeys(sharedKeysStr); + attributesRequestBuilder.setClientKeys(CLIENT_ATTRIBUTE_KEYS); + attributesRequestBuilder.setSharedKeys(SHARED_ATTRIBUTE_KEYS); TransportApiProtos.AttributesRequest attributesRequest = attributesRequestBuilder.build(); client.publishAndWait(attrReqTopicPrefix + "1", attributesRequest.toByteArray()); validateProtoResponse(callback, getExpectedAttributeResponseMsg()); @@ -414,9 +433,8 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt 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) @@ -429,9 +447,8 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt SingleEntityFilter dtf = new SingleEntityFilter(); dtf.setSingleEntity(AliasEntityId.fromEntityId(device.getId())); - String sharedKeysStr = "sharedStr,sharedBool,sharedDbl,sharedLong,sharedJson"; - List clientKeysList = List.of(clientKeysStr.split(",")); - List sharedKeysList = List.of(sharedKeysStr.split(",")); + List clientKeysList = List.of(CLIENT_ATTRIBUTE_KEYS.split(",")); + List sharedKeysList = List.of(SHARED_ATTRIBUTE_KEYS.split(",")); List csKeys = getEntityKeys(clientKeysList, CLIENT_ATTRIBUTE); List shKeys = getEntityKeys(sharedKeysList, SHARED_ATTRIBUTE); List keys = new ArrayList<>(); @@ -468,13 +485,58 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt client.disconnect(); } + protected void processJsonTestGatewayRequestAttributesSeparated(String deviceName, String requestPayloadSuffix, String expectedBody) throws Exception { + MqttTestClient client = new MqttTestClient(); + client.connectAndWait(gatewayAccessToken); + String postClientAttributes = "{\"" + deviceName + "\":" + CLIENT_ATTRIBUTES_PAYLOAD + "}"; + client.publishAndWait(GATEWAY_ATTRIBUTES_TOPIC, postClientAttributes.getBytes()); + + Device device = doExecuteWithRetriesAndInterval(() -> doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class), + 20, 100); + assertNotNull(device); + + String attributeValuesUrl = "/api/plugins/telemetry/DEVICE/" + device.getId() + "/values/attributes/CLIENT_SCOPE?keys=" + CLIENT_ATTRIBUTE_KEYS; + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .ignoreExceptions() + .until(() -> { + List> attributes = doGetAsyncTyped(attributeValuesUrl, new TypeReference<>() { + }); + return attributes.size() == 5; + }); + + 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); + String requestPayloadStr = "{\"id\": 1, \"device\": \"" + deviceName + "\"" + requestPayloadSuffix + "}"; + client.publishAndWait(GATEWAY_ATTRIBUTES_REQUEST_TOPIC, requestPayloadStr.getBytes()); + + assertThat(callback.getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)).as("await callback").isTrue(); + assertEquals(MqttQoS.AT_LEAST_ONCE.value(), callback.getMessageArrivedQoS()); + String expected = "{\"id\":1,\"device\":\"" + deviceName + "\"" + expectedBody + "}"; + assertEquals(JacksonUtil.toJsonNode(expected), JacksonUtil.fromBytes(callback.getPayloadBytes())); + client.disconnect(); + } + protected void processProtoTestGatewayRequestAttributesValuesFromTheServer() throws Exception { MqttTestClient client = new MqttTestClient(); client.connectAndWait(gatewayAccessToken); String deviceName = "Gateway Device Request Attributes"; - String clientKeysStr = "clientStr,clientBool,clientDbl,clientLong,clientJson"; - List clientKeysList = List.of(clientKeysStr.split(",")); + 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), @@ -484,8 +546,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 sharedKeysList = List.of(sharedKeysStr.split(",")); + List sharedKeysList = List.of(SHARED_ATTRIBUTE_KEYS.split(",")); List csKeys = getEntityKeys(clientKeysList, CLIENT_ATTRIBUTE); List shKeys = getEntityKeys(sharedKeysList, SHARED_ATTRIBUTE); List keys = new ArrayList<>(); @@ -520,10 +581,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); @@ -663,6 +794,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/MqttAttributesRequestJsonIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/request/MqttAttributesRequestJsonIntegrationTest.java index 3786a133ef..df5ed3c1ea 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/request/MqttAttributesRequestJsonIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/attributes/request/MqttAttributesRequestJsonIntegrationTest.java @@ -57,4 +57,67 @@ public class MqttAttributesRequestJsonIntegrationTest extends AbstractMqttAttrib public void testRequestAttributesValuesFromTheServerGateway() throws Exception { processJsonTestGatewayRequestAttributesValuesFromTheServer(); } + + @Test + public void testRequestAllSharedAttributesViaEmptyValue() throws Exception { + // sharedKeys present + empty => all shared; clientKeys absent => no client + processJsonTestRequestAttributesWithPayload( + MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX, + "{\"sharedKeys\":\"\"}", + "{\"shared\":" + SHARED_ATTRIBUTES_PAYLOAD + "}"); + } + + @Test + public void testRequestAllClientAttributesViaEmptyValue() throws Exception { + // clientKeys present + empty => all client; sharedKeys absent => no shared + processJsonTestRequestAttributesWithPayload( + MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX, + "{\"clientKeys\":\"\"}", + "{\"client\":" + CLIENT_ATTRIBUTES_PAYLOAD + "}"); + } + + @Test + public void testRequestAllAttributesViaEmptyObject() throws Exception { + // both fields absent => fetch everything (both scopes) + processJsonTestRequestAttributesWithPayload( + MqttTopics.DEVICE_ATTRIBUTES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_RESPONSES_TOPIC, MqttTopics.DEVICE_ATTRIBUTES_REQUEST_TOPIC_PREFIX, + "{}", + "{\"client\":" + CLIENT_ATTRIBUTES_PAYLOAD + ",\"shared\":" + SHARED_ATTRIBUTES_PAYLOAD + "}"); + } + + @Test + public void testGatewayRequestAllAttributesSeparated() throws Exception { + // new format: empty clientKeys + empty sharedKeys => all both, scope-separated response + processJsonTestGatewayRequestAttributesSeparated( + "Gateway Device Request All Both Separated Json", + ", \"clientKeys\": \"\", \"sharedKeys\": \"\"", + ",\"client\":" + CLIENT_ATTRIBUTES_PAYLOAD + ",\"shared\":" + SHARED_ATTRIBUTES_PAYLOAD); + } + + @Test + public void testGatewayRequestAllSharedSeparated() throws Exception { + // new format: empty sharedKeys only => all shared, no client in the separated response + processJsonTestGatewayRequestAttributesSeparated( + "Gateway Device Request All Shared Separated Json", + ", \"sharedKeys\": \"\"", + ",\"shared\":" + SHARED_ATTRIBUTES_PAYLOAD); + } + + @Test + public void testGatewayRequestAllClientSeparated() throws Exception { + // new format: empty clientKeys only => all client, no shared in the separated response + processJsonTestGatewayRequestAttributesSeparated( + "Gateway Device Request All Client Separated Json", + ", \"clientKeys\": \"\"", + ",\"client\":" + CLIENT_ATTRIBUTES_PAYLOAD); + } + + @Test + public void testGatewayRequestSpecificSharedKeysSeparated() throws Exception { + // new format: specific shared key list => only those shared keys, scope-separated response + processJsonTestGatewayRequestAttributesSeparated( + "Gateway Device Request Specific Shared Keys Json", + ", \"sharedKeys\": \"sharedStr,sharedBool\"", + ",\"shared\":{\"sharedStr\":\"value1\",\"sharedBool\":true}"); + } } 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 8a40b1bc76..2a930b31db 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 @@ -55,6 +55,8 @@ import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509Ce import java.math.BigDecimal; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -361,13 +363,70 @@ public class JsonConverter { JsonObject result = new JsonObject(); result.addProperty("id", responseMsg.getRequestId()); result.addProperty(DEVICE_PROPERTY, deviceName); + if (responseMsg.getSeparateScopesResponse()) { + // 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 { + 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()); } - return result; + } + + /** + * Three-state per-scope attribute selection for the JSON {@code clientKeys}/{@code sharedKeys} format + * (device and gateway MQTT APIs). Depending on the value of {@code keysField}: + *
    + *
  • absent or null: the scope is excluded (neither callback is invoked);
    • + *
  • present but empty (or blank): every key in that scope is selected ({@code selectAllKeys});
    • + *
  • present with a comma-separated list of names: only those keys are selected ({@code selectKeys}).
    • + *
+ */ + public static void parseAttributeScope(JsonObject json, String keysField, + Runnable selectAllKeys, Consumer> selectKeys) { + if (!json.has(keysField) || json.get(keysField).isJsonNull()) { + return; + } + String rawKeys = json.get(keysField).getAsString(); + if (rawKeys.trim().isEmpty()) { + selectAllKeys.run(); + } else { + selectKeys.accept(Arrays.asList(rawKeys.split(","))); + } + } + + /** + * Populate the client scope of an attribute request: "all" wins over a specific key list; a missing/empty + * list leaves the scope unset. Shared by the HTTP and CoAP query-parameter parsers. + */ + public static void applyClientScope(TransportProtos.GetAttributeRequestMsg.Builder builder, boolean all, Collection names) { + applyScope(all, names, () -> builder.setAllClientAttributes(true), builder::addAllClientAttributeNames); + } + + /** + * Shared-scope counterpart of {@link #applyClientScope}. + */ + public static void applySharedScope(TransportProtos.GetAttributeRequestMsg.Builder builder, boolean all, Collection names) { + applyScope(all, names, () -> builder.setAllSharedAttributes(true), builder::addAllSharedAttributeNames); + } + + // "all" wins over a specific key list; a missing/empty list leaves the scope unset. + // Same precedence as DeviceActorMessageProcessor.resolveScopeFuture (the fetch side); keep the two in sync. + private static void applyScope(boolean all, Collection names, Runnable setAll, Consumer> addNames) { + if (all) { + setAll.run(); + } else if (names != null && !names.isEmpty()) { + addNames.accept(names); + } } public static JsonObject getJsonObjectForGateway(String deviceName, AttributeUpdateNotificationMsg diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 2e5a382185..0d8138ff8b 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -425,15 +425,19 @@ message GetAttributeRequestMsg { repeated string clientAttributeNames = 2; repeated string sharedAttributeNames = 3; bool onlyShared = 4; + bool allClientAttributes = 5; + bool allSharedAttributes = 6; + bool separateScopesResponse = 7; } message GetAttributeResponseMsg { int32 requestId = 1; repeated TsKvProto clientAttributeList = 2; repeated TsKvProto sharedAttributeList = 3; - bool isMultipleAttributesRequest = 4; + bool isMultipleAttributesRequest = 4 [deprecated = true]; string error = 5; bool sharedStateMsg = 6; + bool separateScopesResponse = 7; } message AttributeUpdateNotificationMsg { diff --git a/common/proto/src/main/proto/transport.proto b/common/proto/src/main/proto/transport.proto index 8adebf62b7..123913eca6 100644 --- a/common/proto/src/main/proto/transport.proto +++ b/common/proto/src/main/proto/transport.proto @@ -96,6 +96,10 @@ 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; + bool allSharedKeys = 8; } diff --git a/common/proto/src/test/java/org/thingsboard/server/common/adaptor/JsonConverterAttributeRequestTest.java b/common/proto/src/test/java/org/thingsboard/server/common/adaptor/JsonConverterAttributeRequestTest.java new file mode 100644 index 0000000000..4c3cd39b44 --- /dev/null +++ b/common/proto/src/test/java/org/thingsboard/server/common/adaptor/JsonConverterAttributeRequestTest.java @@ -0,0 +1,124 @@ +/** + * 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.common.adaptor; + +import com.google.gson.JsonNull; +import com.google.gson.JsonObject; +import org.junit.jupiter.api.Test; +import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.assertj.core.api.Assertions.assertThat; +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 JsonConverterAttributeRequestTest { + + private final List names = new ArrayList<>(); + private final AtomicBoolean allInvoked = new AtomicBoolean(false); + + private void parseClientKeys(JsonObject json) { + JsonConverter.parseAttributeScope(json, "clientKeys", () -> allInvoked.set(true), names::addAll); + } + + @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 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 applyClientScope_all_winsOverNames() { + GetAttributeRequestMsg.Builder b = GetAttributeRequestMsg.newBuilder(); + JsonConverter.applyClientScope(b, true, List.of("ignored")); + assertThat(b.getAllClientAttributes()).isTrue(); + assertThat(b.getClientAttributeNamesList()).isEmpty(); + } + + @Test + public void applyClientScope_names_addsNames() { + GetAttributeRequestMsg.Builder b = GetAttributeRequestMsg.newBuilder(); + JsonConverter.applyClientScope(b, false, List.of("a", "b")); + assertThat(b.getAllClientAttributes()).isFalse(); + assertThat(b.getClientAttributeNamesList()).containsExactly("a", "b"); + } + + @Test + public void applyClientScope_nullOrEmpty_setsNothing() { + GetAttributeRequestMsg.Builder b = GetAttributeRequestMsg.newBuilder(); + JsonConverter.applyClientScope(b, false, null); + JsonConverter.applyClientScope(b, false, List.of()); + assertThat(b.getAllClientAttributes()).isFalse(); + assertThat(b.getClientAttributeNamesList()).isEmpty(); + } + + @Test + public void applySharedScope_all_winsOverNames() { + GetAttributeRequestMsg.Builder b = GetAttributeRequestMsg.newBuilder(); + JsonConverter.applySharedScope(b, true, null); + assertThat(b.getAllSharedAttributes()).isTrue(); + assertThat(b.getSharedAttributeNamesList()).isEmpty(); + } + + @Test + public void applySharedScope_names_addsNames() { + GetAttributeRequestMsg.Builder b = GetAttributeRequestMsg.newBuilder(); + JsonConverter.applySharedScope(b, false, List.of("s1")); + assertThat(b.getAllSharedAttributes()).isFalse(); + assertThat(b.getSharedAttributeNamesList()).containsExactly("s1"); + } +} 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 new file mode 100644 index 0000000000..7915c10554 --- /dev/null +++ b/common/proto/src/test/java/org/thingsboard/server/common/adaptor/JsonConverterGatewayResponseTest.java @@ -0,0 +1,74 @@ +/** + * 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.common.adaptor; + +import com.google.gson.JsonObject; +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; +import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto; + +import static org.assertj.core.api.Assertions.assertThat; + +public class JsonConverterGatewayResponseTest { + + private TsKvProto kv(String key, long v) { + return TsKvProto.newBuilder().setTs(1L).setKv( + KeyValueProto.newBuilder().setKey(key).setType(KeyValueType.LONG_V).setLongV(v).build()).build(); + } + + @Test + public void separateScopes_keepsScopeLabels_keyNames_andOverlap() { + GetAttributeResponseMsg msg = GetAttributeResponseMsg.newBuilder() + .setRequestId(1) + .setSeparateScopesResponse(true) + .addClientAttributeList(kv("test", 27)) + .addSharedAttributeList(kv("test", 99)) + .build(); + + JsonObject out = JsonConverter.getJsonObjectForGateway("DeviceA", msg); + + assertThat(out.get("id").getAsInt()).isEqualTo(1); + assertThat(out.get("device").getAsString()).isEqualTo("DeviceA"); + assertThat(out.getAsJsonObject("client").get("test").getAsLong()).isEqualTo(27); + assertThat(out.getAsJsonObject("shared").get("test").getAsLong()).isEqualTo(99); + assertThat(out.has("value")).isFalse(); + assertThat(out.has("values")).isFalse(); + } + + @Test + public void separateScopes_omitsEmptyScope() { + GetAttributeResponseMsg msg = GetAttributeResponseMsg.newBuilder() + .setRequestId(2).setSeparateScopesResponse(true) + .addSharedAttributeList(kv("s", 1)).build(); + JsonObject out = JsonConverter.getJsonObjectForGateway("DeviceA", msg); + assertThat(out.has("client")).isFalse(); + assertThat(out.getAsJsonObject("shared").get("s").getAsLong()).isEqualTo(1); + } + + @Test + @SuppressWarnings("deprecation") + public void legacy_singleValue_unchanged() { + GetAttributeResponseMsg msg = GetAttributeResponseMsg.newBuilder() + .setRequestId(3).setSeparateScopesResponse(false) + .setIsMultipleAttributesRequest(false) + .addSharedAttributeList(kv("s", 5)).build(); + JsonObject out = JsonConverter.getJsonObjectForGateway("DeviceA", msg); + assertThat(out.get("value").getAsLong()).isEqualTo(5); + assertThat(out.has("shared")).isFalse(); + } +} 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 15838564e8..6878b4b29b 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 @@ -17,6 +17,7 @@ package org.thingsboard.server.transport.coap.adaptors; import org.eclipse.californium.core.coap.Request; import org.thingsboard.server.common.adaptor.AdaptorException; +import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.gen.transport.TransportProtos; @@ -31,28 +32,32 @@ public class CoapAdaptorUtils { List queryElements = inbound.getOptions().getUriQuery(); TransportProtos.GetAttributeRequestMsg.Builder result = TransportProtos.GetAttributeRequestMsg.newBuilder(); if (queryElements != null && queryElements.size() > 0) { - Set clientKeys = toKeys(queryElements, "clientKeys"); - Set sharedKeys = toKeys(queryElements, "sharedKeys"); - if (clientKeys != null) { - result.addAllClientAttributeNames(clientKeys); - } - if (sharedKeys != null) { - result.addAllSharedAttributeNames(sharedKeys); - } + boolean allClient = "true".equalsIgnoreCase(getQueryValue(queryElements, "allClientKeys")); + boolean allShared = "true".equalsIgnoreCase(getQueryValue(queryElements, "allSharedKeys")); + Set clientKeys = allClient ? null : toKeys(queryElements, "clientKeys"); + Set sharedKeys = allShared ? null : toKeys(queryElements, "sharedKeys"); + JsonConverter.applyClientScope(result, allClient, clientKeys); + JsonConverter.applySharedScope(result, allShared, sharedKeys); } result.setOnlyShared(false); return result.build(); } - private static Set toKeys(List queryElements, String attributeName) throws AdaptorException { - String keys = null; + 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(attributeName)) { - keys = queryItem[1]; + if (queryItem.length == 2 && queryItem[0].equals(name)) { + value = queryItem[1]; } } - if (keys != null && !StringUtils.isEmpty(keys)) { + return value; + } + + 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/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java b/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java index d0205314bc..9466249cdc 100644 --- a/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java +++ b/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java @@ -141,22 +141,22 @@ public class DeviceApiController implements TbTransportService { public DeferredResult getDeviceAttributes( @Parameter(description = ACCESS_TOKEN_PARAM_DESCRIPTION, required = true, schema = @Schema(defaultValue = "YOUR_DEVICE_ACCESS_TOKEN")) @PathVariable("deviceToken") String deviceToken, - @Parameter(description = "Comma separated key names for attribute with client scope", required = true , schema = @Schema(defaultValue = "state")) + @Parameter(description = "Comma separated key names for attribute with client scope. Leave empty and set allClientKeys=true to fetch all client-scope attributes", required = false) @RequestParam(value = "clientKeys", required = false, defaultValue = "") String clientKeys, - @Parameter(description = "Comma separated key names for attribute with shared scope", required = true , schema = @Schema(defaultValue = "configuration")) - @RequestParam(value = "sharedKeys", required = false, defaultValue = "") String sharedKeys) { + @Parameter(description = "Comma separated key names for attribute with shared scope. Leave empty and set allSharedKeys=true to fetch all shared-scope attributes", required = false) + @RequestParam(value = "sharedKeys", required = false, defaultValue = "") String sharedKeys, + @Parameter(description = "Set to true to return ALL client-scope attributes (ignores clientKeys)") + @RequestParam(value = "allClientKeys", required = false, defaultValue = "false") boolean allClientKeys, + @Parameter(description = "Set to true to return ALL shared-scope attributes (ignores sharedKeys)") + @RequestParam(value = "allSharedKeys", required = false, defaultValue = "false") boolean allSharedKeys) { DeferredResult responseWriter = new DeferredResult<>(); transportContext.getTransportService().process(DeviceTransportType.DEFAULT, ValidateDeviceTokenRequestMsg.newBuilder().setToken(deviceToken).build(), new DeviceAuthCallback(transportContext, responseWriter, sessionInfo -> { GetAttributeRequestMsg.Builder request = GetAttributeRequestMsg.newBuilder().setRequestId(0); List clientKeySet = !StringUtils.isEmpty(clientKeys) ? Arrays.asList(clientKeys.split(",")) : null; List sharedKeySet = !StringUtils.isEmpty(sharedKeys) ? Arrays.asList(sharedKeys.split(",")) : null; - if (clientKeySet != null) { - request.addAllClientAttributeNames(clientKeySet); - } - if (sharedKeySet != null) { - request.addAllSharedAttributeNames(sharedKeySet); - } + JsonConverter.applyClientScope(request, allClientKeys, clientKeySet); + JsonConverter.applySharedScope(request, allSharedKeys, sharedKeySet); TransportService transportService = transportContext.getTransportService(); transportService.registerSyncSession(sessionInfo, new HttpSessionListener(responseWriter, transportContext.getTransportService(), sessionInfo), 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 2516e155fd..7aa648786e 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,10 +37,7 @@ import org.thingsboard.server.transport.mqtt.session.MqttDeviceAwareSessionConte import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; -import java.util.Arrays; -import java.util.HashSet; import java.util.Optional; -import java.util.Set; import java.util.UUID; import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_SOFTWARE_FIRMWARE_RESPONSES_TOPIC_FORMAT; @@ -176,15 +173,9 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor { TransportProtos.GetAttributeRequestMsg.Builder result = TransportProtos.GetAttributeRequestMsg.newBuilder(); result.setRequestId(getRequestId(topicName, topicBase)); String payload = inbound.payload().toString(UTF8); - JsonElement requestBody = JsonParser.parseString(payload); - Set clientKeys = toStringSet(requestBody, "clientKeys"); - Set sharedKeys = toStringSet(requestBody, "sharedKeys"); - if (clientKeys != null) { - result.addAllClientAttributeNames(clientKeys); - } - if (sharedKeys != null) { - result.addAllSharedAttributeNames(sharedKeys); - } + JsonObject json = JsonParser.parseString(payload).getAsJsonObject(); + JsonConverter.parseAttributeScope(json, "clientKeys", () -> result.setAllClientAttributes(true), result::addAllClientAttributeNames); + JsonConverter.parseAttributeScope(json, "sharedKeys", () -> result.setAllSharedAttributes(true), result::addAllSharedAttributeNames); return result.build(); } catch (RuntimeException e) { log.debug("Failed to decode get attributes request", e); @@ -248,15 +239,6 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor { return new MqttPublishMessage(mqttFixedHeader, header, payload); } - private Set toStringSet(JsonElement requestBody, String name) { - JsonElement element = requestBody.getAsJsonObject().get(name); - if (element != null) { - return new HashSet<>(Arrays.asList(element.getAsString().split(","))); - } else { - return null; - } - } - private static String validatePayload(UUID sessionId, ByteBuf payloadData, boolean isEmptyPayloadAllowed) throws AdaptorException { String payload = payloadData.toString(UTF8); if (payload == null) { 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 8ea6480142..69afa30d80 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 @@ -27,7 +27,6 @@ import com.google.gson.JsonNull; import com.google.gson.JsonObject; import com.google.gson.JsonSyntaxException; import com.google.protobuf.InvalidProtocolBufferException; -import com.google.protobuf.ProtocolStringList; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandlerContext; @@ -654,36 +653,74 @@ public abstract class AbstractGatewaySessionHandler keys; - if (jsonObj.has("key")) { - keys = Collections.singleton(jsonObj.get("key").getAsString()); - } else { - JsonArray keysArray = jsonObj.get("keys").getAsJsonArray(); - keys = new HashSet<>(); - for (JsonElement keyObj : keysArray) { - keys.add(keyObj.getAsString()); + TransportProtos.GetAttributeRequestMsg requestMsg; + if (jsonObj.has("clientKeys") || jsonObj.has("sharedKeys")) { + // new unified format: clientKeys/sharedKeys + empty-value="all"; emit the separated response + TransportProtos.GetAttributeRequestMsg.Builder requestBuilder = TransportProtos.GetAttributeRequestMsg.newBuilder() + .setRequestId(requestId).setSeparateScopesResponse(true); + JsonConverter.parseAttributeScope(jsonObj, "clientKeys", () -> requestBuilder.setAllClientAttributes(true), requestBuilder::addAllClientAttributeNames); + JsonConverter.parseAttributeScope(jsonObj, "sharedKeys", () -> requestBuilder.setAllSharedAttributes(true), requestBuilder::addAllSharedAttributeNames); + requestMsg = requestBuilder.build(); + } else if (jsonObj.has("client")) { + // legacy format: client boolean + key/keys; keep the legacy value/values response + boolean clientScope = jsonObj.get("client").getAsBoolean(); + Set keys; + if (jsonObj.has("key")) { + keys = Collections.singleton(jsonObj.get("key").getAsString()); + } else { + JsonArray keysArray = jsonObj.get("keys").getAsJsonArray(); + keys = new HashSet<>(); + for (JsonElement keyObj : keysArray) { + keys.add(keyObj.getAsString()); + } } + requestMsg = toGetAttributeRequestMsg(requestId, clientScope, keys); + } else { + // neither marker present: fetch everything, separated response + requestMsg = TransportProtos.GetAttributeRequestMsg.newBuilder() + .setRequestId(requestId).setSeparateScopesResponse(true).build(); } - TransportProtos.GetAttributeRequestMsg requestMsg = toGetAttributeRequestMsg(requestId, clientScope, keys); processGetAttributeRequestMessage(msg, deviceName, requestMsg); } private void onDeviceAttributesRequestProto(MqttPublishMessage mqttMsg) throws AdaptorException { try { - TransportApiProtos.GatewayAttributesRequestMsg gatewayAttributesRequestMsg = TransportApiProtos.GatewayAttributesRequestMsg.parseFrom(getBytes(mqttMsg.payload())); - String deviceName = checkDeviceName(gatewayAttributesRequestMsg.getDeviceName()); - int requestId = gatewayAttributesRequestMsg.getId(); - boolean clientScope = gatewayAttributesRequestMsg.getClient(); - ProtocolStringList keysList = gatewayAttributesRequestMsg.getKeysList(); - Set keys = new HashSet<>(keysList); - TransportProtos.GetAttributeRequestMsg requestMsg = toGetAttributeRequestMsg(requestId, clientScope, keys); + TransportApiProtos.GatewayAttributesRequestMsg gw = TransportApiProtos.GatewayAttributesRequestMsg.parseFrom(getBytes(mqttMsg.payload())); + String deviceName = checkDeviceName(gw.getDeviceName()); + int requestId = gw.getId(); + boolean newFormat = gw.getAllClientKeys() || gw.getAllSharedKeys() + || gw.getClientKeysCount() > 0 || gw.getSharedKeysCount() > 0; + TransportProtos.GetAttributeRequestMsg requestMsg; + if (newFormat) { + TransportProtos.GetAttributeRequestMsg.Builder b = TransportProtos.GetAttributeRequestMsg.newBuilder() + .setRequestId(requestId).setSeparateScopesResponse(true); + if (gw.getAllClientKeys()) { + b.setAllClientAttributes(true); + } else { + b.addAllClientAttributeNames(gw.getClientKeysList()); + } + if (gw.getAllSharedKeys()) { + b.setAllSharedAttributes(true); + } else { + b.addAllSharedAttributeNames(gw.getSharedKeysList()); + } + requestMsg = b.build(); + } else { + requestMsg = toLegacyGatewayRequestMsg(requestId, gw); + } processGetAttributeRequestMessage(mqttMsg, deviceName, requestMsg); } catch (RuntimeException | InvalidProtocolBufferException e) { throw new AdaptorException(e); } } + @SuppressWarnings("deprecation") // gw.getClient()/getKeysList() retained for the legacy single-scope gateway request + private TransportProtos.GetAttributeRequestMsg toLegacyGatewayRequestMsg(int requestId, TransportApiProtos.GatewayAttributesRequestMsg gw) { + boolean clientScope = gw.getClient(); + Set 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/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java index 5c5914105b..c566763788 100644 --- a/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java +++ b/msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java @@ -234,6 +234,33 @@ public class MqttClientTest extends AbstractContainerTest { assertThat(attributes.getShared().get("sharedAttr")).isEqualTo(sharedAttributeValue); } + @Test + public void requestAllSharedAttributesFromServerViaEmptyValue() throws Exception { + DeviceCredentials deviceCredentials = testRestClient.getDeviceCredentialsByDeviceId(device.getId()); + MqttMessageListener listener = new MqttMessageListener(); + MqttClient mqttClient = getMqttClient(deviceCredentials, listener); + + // Add a new shared attribute + JsonObject sharedAttributes = new JsonObject(); + String sharedAttributeValue = StringUtils.randomAlphanumeric(8); + sharedAttributes.addProperty("sharedAttr", sharedAttributeValue); + testRestClient.postTelemetryAttribute(device.getId(), SHARED_SCOPE, mapper.readTree(sharedAttributes.toString())); + + mqttClient.on("v1/devices/me/attributes/response/+", listener, MqttQoS.AT_LEAST_ONCE).get(); + TimeUnit.SECONDS.sleep(3 * timeoutMultiplier); + + // empty sharedKeys => all shared; clientKeys absent => no client + JsonObject request = new JsonObject(); + request.addProperty("sharedKeys", ""); + mqttClient.publish("v1/devices/me/attributes/request/" + new Random().nextInt(100), Unpooled.wrappedBuffer(request.toString().getBytes())).get(); + MqttEvent event = listener.getEvents().poll(10 * timeoutMultiplier, TimeUnit.SECONDS); + AttributesResponse attributes = mapper.readValue(Objects.requireNonNull(event).getMessage(), AttributesResponse.class); + + assertThat(attributes.getShared()).hasSize(1); + assertThat(attributes.getShared().get("sharedAttr")).isEqualTo(sharedAttributeValue); + assertThat(attributes.getClient()).isNullOrEmpty(); + } + @Test public void subscribeToAttributeUpdatesFromServer() throws Exception { DeviceCredentials deviceCredentials = testRestClient.getDeviceCredentialsByDeviceId(device.getId()); 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 c9fa76cc9a..bd52349e63 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 @@ -235,6 +235,34 @@ public class MqttGatewayClientTest extends AbstractContainerTest { assertThat(responseData.get("values").getAsJsonObject().entrySet()).hasSize(1); } + @Test + public void gatewayRequestAllSharedReturnsSeparatedResponse() throws Exception { + JsonObject sharedAttributes = new JsonObject(); + sharedAttributes.addProperty("attr1", "value1"); + sharedAttributes.addProperty("attr2", true); + testRestClient.postTelemetryAttribute(createdDevice.getId(), SHARED_SCOPE, mapper.readTree(sharedAttributes.toString())); + + mqttClient.on("v1/gateway/attributes/response", listener, MqttQoS.AT_LEAST_ONCE).get(); + + // new format: empty sharedKeys => all shared; scope-separated response + JsonObject requestData = new JsonObject(); + requestData.addProperty("id", 1); + requestData.addProperty("device", createdDevice.getName()); + requestData.addProperty("sharedKeys", ""); + mqttClient.publish("v1/gateway/attributes/request", Unpooled.wrappedBuffer(requestData.toString().getBytes())).get(); + + // Saving the shared attribute above also triggers an attribute-update push on + // v1/gateway/attributes, which can land in the queue before the request response. + // Skip those and read the event from the response topic. + MqttEvent event = pollEventForTopic("v1/gateway/attributes/response"); + JsonObject responseData = JsonParser.parseString(Objects.requireNonNull(event).getMessage()).getAsJsonObject(); + assertThat(responseData.has("shared")).isTrue(); + assertThat(responseData.has("value")).isFalse(); + assertThat(responseData.has("values")).isFalse(); + assertThat(responseData.getAsJsonObject("shared").get("attr1").getAsString()).isEqualTo("value1"); + assertThat(responseData.getAsJsonObject("shared").get("attr2").getAsBoolean()).isTrue(); + } + @Test public void requestAttributeValuesFromServer() throws Exception { WsClient wsClient = subscribeToWebSocket(createdDevice.getId(), "CLIENT_SCOPE", CmdsType.ATTR_SUB_CMDS); @@ -371,6 +399,21 @@ public class MqttGatewayClientTest extends AbstractContainerTest { this.createdDevice = createDeviceThroughGateway(mqttClient, gatewayDevice); } + private MqttEvent pollEventForTopic(String topic) throws InterruptedException { + long deadline = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(10 * timeoutMultiplier); + long remaining; + while ((remaining = deadline - System.currentTimeMillis()) > 0) { + MqttEvent event = listener.getEvents().poll(remaining, TimeUnit.MILLISECONDS); + if (event == null) { + break; + } + if (topic.equals(event.getTopic())) { + return event; + } + } + return null; + } + private void checkAttribute(boolean client, String expectedValue) throws Exception { JsonObject gatewayAttributesRequest = new JsonObject(); int messageId = new Random().nextInt(100);