diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java index e50ffd047a..fece86b92a 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java @@ -18,6 +18,7 @@ package org.thingsboard.server.service.edge.rpc.processor.telemetry; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.JsonNode; import com.google.common.util.concurrent.FutureCallback; +import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.SettableFuture; import com.google.gson.Gson; @@ -283,26 +284,31 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor { private ListenableFuture processAttributeDeleteMsg(TenantId tenantId, EntityId entityId, AttributeDeleteMsg attributeDeleteMsg, String entityType) { - SettableFuture futureToSet = SettableFuture.create(); + String scope = attributeDeleteMsg.getScope(); List attributeKeys = attributeDeleteMsg.getAttributeNamesList(); - attributesService.removeAll(tenantId, entityId, scope, attributeKeys); - if (EntityType.DEVICE.name().equals(entityType)) { - tbClusterService.pushMsgToCore(DeviceAttributesEventNotificationMsg.onDelete( - tenantId, (DeviceId) entityId, scope, attributeKeys), new TbQueueCallback() { - @Override - public void onSuccess(TbQueueMsgMetadata metadata) { - futureToSet.set(null); - } + ListenableFuture> removeAllFuture = attributesService.removeAll(tenantId, entityId, scope, attributeKeys); + return Futures.transformAsync(removeAllFuture, removeAttributes -> { + if (EntityType.DEVICE.name().equals(entityType)) { + SettableFuture futureToSet = SettableFuture.create(); + tbClusterService.pushMsgToCore(DeviceAttributesEventNotificationMsg.onDelete( + tenantId, (DeviceId) entityId, scope, attributeKeys), new TbQueueCallback() { + @Override + public void onSuccess(TbQueueMsgMetadata metadata) { + futureToSet.set(null); + } - @Override - public void onFailure(Throwable t) { - log.error("Can't process attribute delete msg [{}]", attributeDeleteMsg, t); - futureToSet.setException(t); - } - }); - } - return futureToSet; + @Override + public void onFailure(Throwable t) { + log.error("Can't process attribute delete msg [{}]", attributeDeleteMsg, t); + futureToSet.setException(t); + } + }); + return futureToSet; + } else { + return Futures.immediateFuture(null); + } + }, dbCallbackExecutorService); } public EntityDataProto convertTelemetryEventToEntityDataProto(EntityType entityType, diff --git a/application/src/test/java/org/thingsboard/server/edge/BaseTelemetryEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/BaseTelemetryEdgeTest.java index ff6728cecc..8793495b53 100644 --- a/application/src/test/java/org/thingsboard/server/edge/BaseTelemetryEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/BaseTelemetryEdgeTest.java @@ -15,11 +15,20 @@ */ package org.thingsboard.server.edge; +import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.google.gson.Gson; +import com.google.gson.reflect.TypeToken; import com.google.protobuf.AbstractMessage; +import org.awaitility.Awaitility; import org.junit.Assert; import org.junit.Test; +import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; import org.thingsboard.server.common.data.edge.EdgeEventType; @@ -27,9 +36,11 @@ import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.gen.edge.v1.AttributeDeleteMsg; import org.thingsboard.server.gen.edge.v1.DeviceUpdateMsg; import org.thingsboard.server.gen.edge.v1.EntityDataProto; +import org.thingsboard.server.gen.edge.v1.UplinkMsg; import org.thingsboard.server.gen.transport.TransportProtos; import java.util.List; +import java.util.concurrent.TimeUnit; abstract public class BaseTelemetryEdgeTest extends AbstractEdgeTest { @@ -233,4 +244,53 @@ abstract public class BaseTelemetryEdgeTest extends AbstractEdgeTest { Assert.assertEquals("value1", keyValueProto.getStringV()); } + @Test + public void testSendAttributesDeleteRequestToCloud_nonDeviceEntity() throws Exception { + edgeImitator.expectMessageAmount(2); + Asset savedAsset = saveAsset("Delete Attribute Test"); + doPost("/api/edge/" + edge.getUuidId() + "/asset/" + savedAsset.getUuidId(), Asset.class); + Assert.assertTrue(edgeImitator.waitForMessages()); + + final String attributeKey = "key1"; + ObjectNode attributesData = JacksonUtil.OBJECT_MAPPER.createObjectNode(); + attributesData.put(attributeKey, "value1"); + doPost("/api/plugins/telemetry/ASSET/" + savedAsset.getId() + "/attributes/" + DataConstants.SERVER_SCOPE, attributesData); + + // Wait before device attributes saved to database before deleting them + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .until(() -> { + String urlTemplate = "/api/plugins/telemetry/ASSET/" + savedAsset.getId() + "/keys/attributes/" + DataConstants.SERVER_SCOPE; + List actualKeys = doGetAsyncTyped(urlTemplate, new TypeReference<>() {}); + return actualKeys != null && !actualKeys.isEmpty() && actualKeys.contains(attributeKey); + }); + + EntityDataProto.Builder builder = EntityDataProto.newBuilder() + .setEntityIdMSB(savedAsset.getUuidId().getMostSignificantBits()) + .setEntityIdLSB(savedAsset.getUuidId().getLeastSignificantBits()) + .setEntityType(savedAsset.getId().getEntityType().name()); + AttributeDeleteMsg.Builder attributeDeleteMsg = AttributeDeleteMsg.newBuilder(); + attributeDeleteMsg.setScope(DataConstants.SERVER_SCOPE); + ArrayNode arrayNode = JacksonUtil.OBJECT_MAPPER.createArrayNode(); + arrayNode.add(attributeKey); + List keys = new Gson().fromJson(arrayNode.toString(), new TypeToken<>(){}.getType()); + attributeDeleteMsg.addAllAttributeNames(keys); + attributeDeleteMsg.build(); + builder.setAttributeDeleteMsg(attributeDeleteMsg); + UplinkMsg.Builder uplinkMsgBuilder = UplinkMsg.newBuilder(); + uplinkMsgBuilder.addEntityData(builder.build()); + + edgeImitator.expectResponsesAmount(1); + edgeImitator.sendUplinkMsg(uplinkMsgBuilder.build()); + Assert.assertTrue(edgeImitator.waitForResponses()); + + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .until(() -> { + String urlTemplate = "/api/plugins/telemetry/ASSET/" + savedAsset.getId() + "/keys/attributes/" + DataConstants.SERVER_SCOPE; + List actualKeys = doGetAsyncTyped(urlTemplate, new TypeReference<>() {}); + return actualKeys != null && actualKeys.isEmpty(); + }); + } + }