|
|
|
@ -32,6 +32,7 @@ import lombok.Data; |
|
|
|
import lombok.extern.slf4j.Slf4j; |
|
|
|
import org.apache.commons.lang.RandomStringUtils; |
|
|
|
import org.checkerframework.checker.nullness.qual.Nullable; |
|
|
|
import org.thingsboard.rule.engine.api.msg.DeviceAttributesEventNotificationMsg; |
|
|
|
import org.thingsboard.server.common.data.AdminSettings; |
|
|
|
import org.thingsboard.server.common.data.Customer; |
|
|
|
import org.thingsboard.server.common.data.Dashboard; |
|
|
|
@ -64,6 +65,7 @@ import org.thingsboard.server.common.data.id.TenantId; |
|
|
|
import org.thingsboard.server.common.data.id.UserId; |
|
|
|
import org.thingsboard.server.common.data.id.WidgetTypeId; |
|
|
|
import org.thingsboard.server.common.data.id.WidgetsBundleId; |
|
|
|
import org.thingsboard.server.common.data.kv.AttributeKey; |
|
|
|
import org.thingsboard.server.common.data.kv.AttributeKvEntry; |
|
|
|
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; |
|
|
|
import org.thingsboard.server.common.data.kv.LongDataEntry; |
|
|
|
@ -87,6 +89,7 @@ import org.thingsboard.server.common.transport.util.JsonUtils; |
|
|
|
import org.thingsboard.server.gen.edge.AdminSettingsUpdateMsg; |
|
|
|
import org.thingsboard.server.gen.edge.AlarmUpdateMsg; |
|
|
|
import org.thingsboard.server.gen.edge.AssetUpdateMsg; |
|
|
|
import org.thingsboard.server.gen.edge.AttributeDeleteMsg; |
|
|
|
import org.thingsboard.server.gen.edge.AttributesRequestMsg; |
|
|
|
import org.thingsboard.server.gen.edge.ConnectRequestMsg; |
|
|
|
import org.thingsboard.server.gen.edge.ConnectResponseCode; |
|
|
|
@ -124,11 +127,12 @@ import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
|
|
|
import org.thingsboard.server.service.edge.EdgeContextComponent; |
|
|
|
|
|
|
|
import java.io.Closeable; |
|
|
|
import java.io.IOException; |
|
|
|
import java.util.ArrayList; |
|
|
|
import java.util.Collections; |
|
|
|
import java.util.HashSet; |
|
|
|
import java.util.List; |
|
|
|
import java.util.Optional; |
|
|
|
import java.util.Set; |
|
|
|
import java.util.UUID; |
|
|
|
import java.util.concurrent.CountDownLatch; |
|
|
|
import java.util.concurrent.ExecutionException; |
|
|
|
@ -387,7 +391,7 @@ public final class EdgeGrpcSession implements Closeable { |
|
|
|
ctx.getAttributesService().save(edge.getTenantId(), edge.getId(), DataConstants.SERVER_SCOPE, attributes); |
|
|
|
} |
|
|
|
|
|
|
|
private DownlinkMsg processTelemetryMessage(EdgeEvent edgeEvent) throws IOException { |
|
|
|
private DownlinkMsg processTelemetryMessage(EdgeEvent edgeEvent) { |
|
|
|
log.trace("Executing processTelemetryMessage, edgeEvent [{}]", edgeEvent); |
|
|
|
EntityId entityId = null; |
|
|
|
switch (edgeEvent.getEdgeEventType()) { |
|
|
|
@ -840,6 +844,9 @@ public final class EdgeGrpcSession implements Closeable { |
|
|
|
result.add(processPostTelemetry(entityId, entityData.getPostTelemetryMsg(), metaData)); |
|
|
|
} |
|
|
|
} |
|
|
|
if (entityData.hasAttributeDeleteMsg()) { |
|
|
|
result.add(processAttributeDeleteMsg(entityId, entityData.getAttributeDeleteMsg(), entityData.getEntityType())); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
@ -962,6 +969,7 @@ public final class EdgeGrpcSession implements Closeable { |
|
|
|
|
|
|
|
@Override |
|
|
|
public void onFailure(Throwable t) { |
|
|
|
log.error("Can't process post telemetry [{}]", msg, t); |
|
|
|
futureToSet.setException(t); |
|
|
|
} |
|
|
|
}); |
|
|
|
@ -970,11 +978,49 @@ public final class EdgeGrpcSession implements Closeable { |
|
|
|
} |
|
|
|
|
|
|
|
private ListenableFuture<Void> processPostAttributes(EntityId entityId, TransportProtos.PostAttributeMsg msg, TbMsgMetaData metaData) { |
|
|
|
SettableFuture<Void> futureToSet = SettableFuture.create(); |
|
|
|
JsonObject json = JsonUtils.getJsonObject(msg.getKvList()); |
|
|
|
TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), entityId, metaData, gson.toJson(json)); |
|
|
|
// TODO: voba - verify that null callback is OK
|
|
|
|
ctx.getTbClusterService().pushMsgToRuleEngine(edge.getTenantId(), tbMsg.getOriginator(), tbMsg, null); |
|
|
|
return Futures.immediateFuture(null); |
|
|
|
ctx.getTbClusterService().pushMsgToRuleEngine(edge.getTenantId(), tbMsg.getOriginator(), tbMsg, new TbQueueCallback() { |
|
|
|
@Override |
|
|
|
public void onSuccess(TbQueueMsgMetadata metadata) { |
|
|
|
futureToSet.set(null); |
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
|
public void onFailure(Throwable t) { |
|
|
|
log.error("Can't process post attributes [{}]", msg, t); |
|
|
|
futureToSet.setException(t); |
|
|
|
} |
|
|
|
}); |
|
|
|
return futureToSet; |
|
|
|
} |
|
|
|
|
|
|
|
private ListenableFuture<Void> processAttributeDeleteMsg(EntityId entityId, AttributeDeleteMsg attributeDeleteMsg, String entityType) { |
|
|
|
SettableFuture<Void> futureToSet = SettableFuture.create(); |
|
|
|
String scope = attributeDeleteMsg.getScope(); |
|
|
|
List<String> attributeNames = attributeDeleteMsg.getAttributeNamesList(); |
|
|
|
ctx.getAttributesService().removeAll(edge.getTenantId(), entityId, scope, attributeNames); |
|
|
|
if (EntityType.DEVICE.name().equals(entityType)) { |
|
|
|
Set<AttributeKey> attributeKeys = new HashSet<>(); |
|
|
|
for (String attributeName : attributeNames) { |
|
|
|
attributeKeys.add(new AttributeKey(scope, attributeName)); |
|
|
|
} |
|
|
|
ctx.getTbClusterService().pushMsgToCore(DeviceAttributesEventNotificationMsg.onDelete( |
|
|
|
edge.getTenantId(), (DeviceId) entityId, 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; |
|
|
|
} |
|
|
|
|
|
|
|
private ListenableFuture<Void> onDeviceUpdate(DeviceUpdateMsg deviceUpdateMsg) { |
|
|
|
|
Volodymyr Babak
6 years ago