tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Edge telemetry processor - fixed processing of delete attribute request for non device entities

pull/8317/head
Volodymyr Babak 4 years ago
parent
28a90e965a
commit
463ad698c4
2 changed files with 83 additions and 17 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 40
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java
  2. 60
      application/src/test/java/org/thingsboard/server/edge/BaseTelemetryEdgeTest.java

40
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java
View File

@ -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<Void> processAttributeDeleteMsg(TenantId tenantId, EntityId entityId, AttributeDeleteMsg attributeDeleteMsg,
String entityType) {
SettableFuture<Void> futureToSet = SettableFuture.create();
String scope = attributeDeleteMsg.getScope();
List<String> 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<List<String>> removeAllFuture = attributesService.removeAll(tenantId, entityId, scope, attributeKeys);
return Futures.transformAsync(removeAllFuture, removeAttributes -> {
if (EntityType.DEVICE.name().equals(entityType)) {
SettableFuture<Void> 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,

60
application/src/test/java/org/thingsboard/server/edge/BaseTelemetryEdgeTest.java
View File

@ -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<String> 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<String> 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<String> actualKeys = doGetAsyncTyped(urlTemplate, new TypeReference<>() {});
return actualKeys != null && actualKeys.isEmpty();
});
}
}

Write Preview
Loading…
Cancel
Save