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

Merge pull request #8317 from volodymyr-babak/edge-attr-fix 

[3.5] Edge processor - delete attribute msg is not properly handled for non device type
pull/8328/head
Andrew Shvayka 4 years ago
committed by GitHub
parent
672c05d9b7 9e189c30e6
commit
87e48d9803
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 82 additions and 22 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. 10
      application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
  2. 40
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java
  3. 54
      application/src/test/java/org/thingsboard/server/edge/BaseTelemetryEdgeTest.java

10
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
View File

@ -423,6 +423,7 @@ public final class EdgeGrpcSession implements Closeable {
stopCurrentSendDownlinkMsgsTask(null);
}
} catch (Exception e) {
log.warn("[{}] Failed to send downlink msgs. Error msg {}", this.sessionId, e.getMessage(), e);
stopCurrentSendDownlinkMsgsTask(e);
}
};
@ -669,19 +670,18 @@ public final class EdgeGrpcSession implements Closeable {
}
private void interruptPreviousSendDownlinkMsgsTask() {
String msg = String.format("[%s] Previous send downlink future was not properly completed, stopping it now!", this.sessionId);
stopCurrentSendDownlinkMsgsTask(new RuntimeException(msg));
log.info("[{}]Previous send downlink future was not properly completed, stopping it now!", this.sessionId);
stopCurrentSendDownlinkMsgsTask(new RuntimeException());
}
private void interruptGeneralProcessingOnSync(TenantId tenantId, EdgeId edgeId) {
String msg = String.format("[%s][%s] Sync process started. General processing interrupted!", tenantId, edgeId);
stopCurrentSendDownlinkMsgsTask(new RuntimeException(msg));
log.info("[{}][{}][{}] Sync process started. General processing interrupted!", this.sessionId, tenantId, edgeId);
stopCurrentSendDownlinkMsgsTask(new RuntimeException());
}
public void stopCurrentSendDownlinkMsgsTask(Exception e) {
if (sessionState.getSendDownlinkMsgsFuture() != null && !sessionState.getSendDownlinkMsgsFuture().isDone()) {
if (e != null) {
log.warn(e.getMessage(), e);
sessionState.getSendDownlinkMsgsFuture().setException(e);
} else {
sessionState.getSendDownlinkMsgsFuture().set(null);

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,

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

@ -15,11 +15,17 @@
*/
package org.thingsboard.server.edge;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
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 +33,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 +241,50 @@ 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);
attributeDeleteMsg.addAllAttributeNames(List.of(attributeKey));
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