diff --git a/application/src/main/java/org/thingsboard/server/controller/BaseController.java b/application/src/main/java/org/thingsboard/server/controller/BaseController.java index 6b43b83549..b0f0e0867f 100644 --- a/application/src/main/java/org/thingsboard/server/controller/BaseController.java +++ b/application/src/main/java/org/thingsboard/server/controller/BaseController.java @@ -747,65 +747,65 @@ public abstract class BaseController { return null; } - protected void sendNotificationMsgToEdgeService(TenantId tenantId, EdgeId edgeId, CustomerId customerId, ActionType edgeEventAction) { + protected void sendNotificationMsgToEdgeService(TenantId tenantId, EdgeId edgeId, CustomerId customerId, ActionType action) { if (!edgesSupportEnabled) { return; } try { - sendNotificationMsgToEdgeService(tenantId, edgeId, null, json.writeValueAsString(customerId), EdgeEventType.EDGE, edgeEventAction); + sendNotificationMsgToEdgeService(tenantId, edgeId, null, json.writeValueAsString(customerId), EdgeEventType.EDGE, action); } catch (Exception e) { log.warn("Failed to push assign/unassign to/from customer to core: {}", customerId, e); } } - protected void sendNotificationMsgToEdgeService(TenantId tenantId, EntityId entityId, CustomerId customerId, ActionType edgeEventAction) { + protected void sendNotificationMsgToEdgeService(TenantId tenantId, EntityId entityId, CustomerId customerId, ActionType action) { if (!edgesSupportEnabled) { return; } - EdgeEventType edgeEventType = EdgeUtils.getEdgeEventTypeByEntityType(entityId.getEntityType()); + EdgeEventType type = EdgeUtils.getEdgeEventTypeByEntityType(entityId.getEntityType()); try { - if (edgeEventType != null) { - sendNotificationMsgToEdgeService(tenantId, null, entityId, json.writeValueAsString(customerId), edgeEventType, edgeEventAction); + if (type != null) { + sendNotificationMsgToEdgeService(tenantId, null, entityId, json.writeValueAsString(customerId), type, action); } } catch (Exception e) { log.warn("Failed to push assign/unassign to/from customer to core: {}", customerId, e); } } - protected void sendNotificationMsgToEdgeService(TenantId tenantId, EntityRelation relation, ActionType edgeEventAction) { + protected void sendNotificationMsgToEdgeService(TenantId tenantId, EntityRelation relation, ActionType action) { if (!edgesSupportEnabled) { return; } try { if (!relation.getFrom().getEntityType().equals(EntityType.EDGE) && !relation.getTo().getEntityType().equals(EntityType.EDGE)) { - sendNotificationMsgToEdgeService(tenantId, null, null, json.writeValueAsString(relation), EdgeEventType.RELATION, edgeEventAction); + sendNotificationMsgToEdgeService(tenantId, null, null, json.writeValueAsString(relation), EdgeEventType.RELATION, action); } } catch (Exception e) { log.warn("Failed to push relation to core: {}", relation, e); } } - protected void sendNotificationMsgToEdgeService(TenantId tenantId, EntityId entityId, ActionType edgeEventAction) { - sendNotificationMsgToEdgeService(tenantId, null, entityId, edgeEventAction); + protected void sendNotificationMsgToEdgeService(TenantId tenantId, EntityId entityId, ActionType action) { + sendNotificationMsgToEdgeService(tenantId, null, entityId, action); } - protected void sendNotificationMsgToEdgeService(TenantId tenantId, EdgeId edgeId, EntityId entityId, ActionType edgeEventAction) { + protected void sendNotificationMsgToEdgeService(TenantId tenantId, EdgeId edgeId, EntityId entityId, ActionType action) { if (!edgesSupportEnabled) { return; } - EdgeEventType edgeEventType = EdgeUtils.getEdgeEventTypeByEntityType(entityId.getEntityType()); - if (edgeEventType != null) { - sendNotificationMsgToEdgeService(tenantId, edgeId, entityId, null, edgeEventType, edgeEventAction); + EdgeEventType type = EdgeUtils.getEdgeEventTypeByEntityType(entityId.getEntityType()); + if (type != null) { + sendNotificationMsgToEdgeService(tenantId, edgeId, entityId, null, type, action); } } - private void sendNotificationMsgToEdgeService(TenantId tenantId, EdgeId edgeId, EntityId entityId, String entityBody, EdgeEventType edgeEventType, ActionType edgeEventAction) { + private void sendNotificationMsgToEdgeService(TenantId tenantId, EdgeId edgeId, EntityId entityId, String body, EdgeEventType type, ActionType action) { TransportProtos.EdgeNotificationMsgProto.Builder builder = TransportProtos.EdgeNotificationMsgProto.newBuilder(); builder.setTenantIdMSB(tenantId.getId().getMostSignificantBits()); builder.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()); - builder.setEdgeEventType(edgeEventType.name()); - builder.setEdgeEventAction(edgeEventAction.name()); + builder.setType(type.name()); + builder.setAction(action.name()); if (entityId != null) { builder.setEntityIdMSB(entityId.getId().getMostSignificantBits()); builder.setEntityIdLSB(entityId.getId().getLeastSignificantBits()); @@ -815,8 +815,8 @@ public abstract class BaseController { builder.setEdgeIdMSB(edgeId.getId().getMostSignificantBits()); builder.setEdgeIdLSB(edgeId.getId().getLeastSignificantBits()); } - if (entityBody != null) { - builder.setEntityBody(entityBody); + if (body != null) { + builder.setBody(body); } TransportProtos.EdgeNotificationMsgProto msg = builder.build(); tbClusterService.pushMsgToCore(tenantId, entityId != null ? entityId : tenantId, diff --git a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java index a014ec3d30..e9ac598cfa 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java @@ -122,22 +122,22 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { private void saveEdgeEvent(TenantId tenantId, EdgeId edgeId, - EdgeEventType edgeEventType, - ActionType edgeEventAction, + EdgeEventType type, + ActionType action, EntityId entityId, - JsonNode entityBody) { - log.debug("Pushing edge event to edge queue. tenantId [{}], edgeId [{}], edgeEventType [{}], edgeEventAction[{}], entityId [{}], entityBody [{}]", - tenantId, edgeId, edgeEventType, edgeEventAction, entityId, entityBody); + JsonNode body) { + log.debug("Pushing edge event to edge queue. tenantId [{}], edgeId [{}], type [{}], action[{}], entityId [{}], body [{}]", + tenantId, edgeId, type, action, entityId, body); EdgeEvent edgeEvent = new EdgeEvent(); edgeEvent.setEdgeId(edgeId); edgeEvent.setTenantId(tenantId); - edgeEvent.setType(edgeEventType); - edgeEvent.setAction(edgeEventAction.name()); + edgeEvent.setType(type); + edgeEvent.setAction(action.name()); if (entityId != null) { edgeEvent.setEntityId(entityId.getId()); } - edgeEvent.setBody(entityBody); + edgeEvent.setBody(body); edgeEventService.saveAsync(edgeEvent); } @@ -145,8 +145,8 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { public void pushNotificationToEdge(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback) { try { TenantId tenantId = new TenantId(new UUID(edgeNotificationMsg.getTenantIdMSB(), edgeNotificationMsg.getTenantIdLSB())); - EdgeEventType edgeEventType = EdgeEventType.valueOf(edgeNotificationMsg.getEdgeEventType()); - switch (edgeEventType) { + EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType()); + switch (type) { case EDGE: processEdge(tenantId, edgeNotificationMsg); break; @@ -172,7 +172,7 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { processRelation(tenantId, edgeNotificationMsg); break; default: - log.debug("Edge event type [{}] is not designed to be pushed to edge", edgeEventType); + log.debug("Edge event type [{}] is not designed to be pushed to edge", type); } } catch (Exception e) { callback.onFailure(e); @@ -184,12 +184,12 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { private void processEdge(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { try { - ActionType edgeEventActionType = ActionType.valueOf(edgeNotificationMsg.getEdgeEventAction()); + ActionType actionType = ActionType.valueOf(edgeNotificationMsg.getAction()); EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB())); ListenableFuture edgeFuture; - switch (edgeEventActionType) { + switch (actionType) { case ASSIGNED_TO_CUSTOMER: - CustomerId customerId = mapper.readValue(edgeNotificationMsg.getEntityBody(), CustomerId.class); + CustomerId customerId = mapper.readValue(edgeNotificationMsg.getBody(), CustomerId.class); edgeFuture = edgeService.findEdgeByIdAsync(tenantId, edgeId); Futures.addCallback(edgeFuture, new FutureCallback() { @Override @@ -213,7 +213,7 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { }, dbCallbackExecutorService); break; case UNASSIGNED_FROM_CUSTOMER: - CustomerId customerIdToDelete = mapper.readValue(edgeNotificationMsg.getEntityBody(), CustomerId.class); + CustomerId customerIdToDelete = mapper.readValue(edgeNotificationMsg.getBody(), CustomerId.class); edgeFuture = edgeService.findEdgeByIdAsync(tenantId, edgeId); Futures.addCallback(edgeFuture, new FutureCallback() { @Override @@ -236,17 +236,17 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { } private void processWidgetBundleOrWidgetType(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { - ActionType edgeEventActionType = ActionType.valueOf(edgeNotificationMsg.getEdgeEventAction()); - EdgeEventType edgeEventType = EdgeEventType.valueOf(edgeNotificationMsg.getEdgeEventType()); - EntityId entityId = EntityIdFactory.getByEdgeEventTypeAndUuid(edgeEventType, new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); - switch (edgeEventActionType) { + ActionType actionType = ActionType.valueOf(edgeNotificationMsg.getAction()); + EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType()); + EntityId entityId = EntityIdFactory.getByEdgeEventTypeAndUuid(type, new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); + switch (actionType) { case ADDED: case UPDATED: case DELETED: TextPageData edgesByTenantId = edgeService.findEdgesByTenantId(tenantId, new TextPageLink(Integer.MAX_VALUE)); if (edgesByTenantId != null && edgesByTenantId.getData() != null && !edgesByTenantId.getData().isEmpty()) { for (Edge edge : edgesByTenantId.getData()) { - saveEdgeEvent(tenantId, edge.getId(), edgeEventType, edgeEventActionType, entityId, null); + saveEdgeEvent(tenantId, edge.getId(), type, actionType, entityId, null); } } break; @@ -254,20 +254,20 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { } private void processCustomer(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { - ActionType edgeEventActionType = ActionType.valueOf(edgeNotificationMsg.getEdgeEventAction()); - EdgeEventType edgeEventType = EdgeEventType.valueOf(edgeNotificationMsg.getEdgeEventType()); - EntityId entityId = EntityIdFactory.getByEdgeEventTypeAndUuid(edgeEventType, new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); + ActionType actionType = ActionType.valueOf(edgeNotificationMsg.getAction()); + EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType()); + EntityId entityId = EntityIdFactory.getByEdgeEventTypeAndUuid(type, new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); TextPageData edgesByTenantId = edgeService.findEdgesByTenantId(tenantId, new TextPageLink(Integer.MAX_VALUE)); if (edgesByTenantId != null && edgesByTenantId.getData() != null && !edgesByTenantId.getData().isEmpty()) { for (Edge edge : edgesByTenantId.getData()) { - switch (edgeEventActionType) { + switch (actionType) { case UPDATED: if (!edge.getCustomerId().isNullUid() && edge.getCustomerId().equals(entityId)) { - saveEdgeEvent(tenantId, edge.getId(), edgeEventType, edgeEventActionType, entityId, null); + saveEdgeEvent(tenantId, edge.getId(), type, actionType, entityId, null); } break; case DELETED: - saveEdgeEvent(tenantId, edge.getId(), edgeEventType, edgeEventActionType, entityId, null); + saveEdgeEvent(tenantId, edge.getId(), type, actionType, entityId, null); break; } } @@ -275,12 +275,12 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { } private void processEntity(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { - ActionType edgeEventActionType = ActionType.valueOf(edgeNotificationMsg.getEdgeEventAction()); - EdgeEventType edgeEventType = EdgeEventType.valueOf(edgeNotificationMsg.getEdgeEventType()); - EntityId entityId = EntityIdFactory.getByEdgeEventTypeAndUuid(edgeEventType, + ActionType actionType = ActionType.valueOf(edgeNotificationMsg.getAction()); + EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType()); + EntityId entityId = EntityIdFactory.getByEdgeEventTypeAndUuid(type, new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); ListenableFuture> edgeIdsFuture; - switch (edgeEventActionType) { + switch (actionType) { case ADDED: // used only for USER entity case UPDATED: case CREDENTIALS_UPDATED: @@ -290,7 +290,7 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { public void onSuccess(@Nullable List edgeIds) { if (edgeIds != null && !edgeIds.isEmpty()) { for (EdgeId edgeId : edgeIds) { - saveEdgeEvent(tenantId, edgeId, edgeEventType, edgeEventActionType, entityId, null); + saveEdgeEvent(tenantId, edgeId, type, actionType, entityId, null); } } } @@ -309,14 +309,14 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { if (edgeIds != null && !edgeIds.isEmpty()) { for (EdgeId edgeId : edgeIds) { try { - CustomerId customerId = mapper.readValue(edgeNotificationMsg.getEntityBody(), CustomerId.class); + CustomerId customerId = mapper.readValue(edgeNotificationMsg.getBody(), CustomerId.class); ListenableFuture future = edgeService.findEdgeByIdAsync(tenantId, edgeId); Futures.addCallback(future, new FutureCallback() { @Override public void onSuccess(@Nullable Edge edge) { if (edge != null && edge.getCustomerId() != null && !edge.getCustomerId().isNullUid() && edge.getCustomerId().equals(customerId)) { - saveEdgeEvent(tenantId, edgeId, edgeEventType, edgeEventActionType, entityId, null); + saveEdgeEvent(tenantId, edgeId, type, actionType, entityId, null); } } @Override @@ -341,15 +341,15 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { TextPageData edgesByTenantId = edgeService.findEdgesByTenantId(tenantId, new TextPageLink(Integer.MAX_VALUE)); if (edgesByTenantId != null && edgesByTenantId.getData() != null && !edgesByTenantId.getData().isEmpty()) { for (Edge edge : edgesByTenantId.getData()) { - saveEdgeEvent(tenantId, edge.getId(), edgeEventType, edgeEventActionType, entityId, null); + saveEdgeEvent(tenantId, edge.getId(), type, actionType, entityId, null); } } break; case ASSIGNED_TO_EDGE: case UNASSIGNED_FROM_EDGE: EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB())); - saveEdgeEvent(tenantId, edgeId, edgeEventType, edgeEventActionType, entityId, null); - if (edgeEventType.equals(EdgeEventType.RULE_CHAIN)) { + saveEdgeEvent(tenantId, edgeId, type, actionType, entityId, null); + if (type.equals(EdgeEventType.RULE_CHAIN)) { updateDependentRuleChains(tenantId, new RuleChainId(entityId.getId()), edgeId); } break; @@ -395,8 +395,8 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { ListenableFuture alarmFuture = alarmService.findAlarmByIdAsync(tenantId, alarmId); Futures.transform(alarmFuture, alarm -> { if (alarm != null) { - EdgeEventType edgeEventType = getEdgeQueueTypeByEntityType(alarm.getOriginator().getEntityType()); - if (edgeEventType != null) { + EdgeEventType type = getEdgeQueueTypeByEntityType(alarm.getOriginator().getEntityType()); + if (type != null) { ListenableFuture> relatedEdgeIdsByEntityIdFuture = edgeService.findRelatedEdgeIdsByEntityId(tenantId, alarm.getOriginator()); Futures.transform(relatedEdgeIdsByEntityIdFuture, relatedEdgeIdsByEntityId -> { if (relatedEdgeIdsByEntityId != null) { @@ -404,7 +404,7 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { saveEdgeEvent(tenantId, edgeId, EdgeEventType.ALARM, - ActionType.valueOf(edgeNotificationMsg.getEdgeEventAction()), + ActionType.valueOf(edgeNotificationMsg.getAction()), alarmId, null); } @@ -418,7 +418,7 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { } private void processRelation(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) throws JsonProcessingException { - EntityRelation relation = mapper.readValue(edgeNotificationMsg.getEntityBody(), EntityRelation.class); + EntityRelation relation = mapper.readValue(edgeNotificationMsg.getBody(), EntityRelation.class); if (!relation.getFrom().getEntityType().equals(EntityType.EDGE) && !relation.getTo().getEntityType().equals(EntityType.EDGE)) { List>> futures = new ArrayList<>(); @@ -439,7 +439,7 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { saveEdgeEvent(tenantId, edgeId, EdgeEventType.RELATION, - ActionType.valueOf(edgeNotificationMsg.getEdgeEventAction()), + ActionType.valueOf(edgeNotificationMsg.getAction()), null, mapper.valueToTree(relation)); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java index 4b76fb9731..d56f2b871b 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java @@ -309,8 +309,8 @@ public final class EdgeGrpcSession implements Closeable { log.trace("Processing edge event [{}]", edgeEvent); try { DownlinkMsg downlinkMsg = null; - ActionType edgeEventAction = ActionType.valueOf(edgeEvent.getAction()); - switch (edgeEventAction) { + ActionType action = ActionType.valueOf(edgeEvent.getAction()); + switch (action) { case UPDATED: case ADDED: case DELETED: @@ -323,7 +323,7 @@ public final class EdgeGrpcSession implements Closeable { case RELATION_DELETED: case ASSIGNED_TO_CUSTOMER: case UNASSIGNED_FROM_CUSTOMER: - downlinkMsg = processEntityMessage(edgeEvent, edgeEventAction); + downlinkMsg = processEntityMessage(edgeEvent, action); break; case ATTRIBUTES_UPDATED: case ATTRIBUTES_DELETED: @@ -444,37 +444,37 @@ public final class EdgeGrpcSession implements Closeable { return downlinkMsg; } - private DownlinkMsg processEntityMessage(EdgeEvent edgeEvent, ActionType edgeEventAction) { + private DownlinkMsg processEntityMessage(EdgeEvent edgeEvent, ActionType action) { UpdateMsgType msgType = getResponseMsgType(ActionType.valueOf(edgeEvent.getAction())); - log.trace("Executing processEntityMessage, edgeEvent [{}], edgeEventAction [{}], msgType [{}]", edgeEvent, edgeEventAction, msgType); + log.trace("Executing processEntityMessage, edgeEvent [{}], action [{}], msgType [{}]", edgeEvent, action, msgType); switch (edgeEvent.getType()) { case EDGE: // TODO: voba - add edge update logic return null; case DEVICE: - return processDevice(edgeEvent, msgType, edgeEventAction); + return processDevice(edgeEvent, msgType, action); case ASSET: - return processAsset(edgeEvent, msgType, edgeEventAction); + return processAsset(edgeEvent, msgType, action); case ENTITY_VIEW: - return processEntityView(edgeEvent, msgType, edgeEventAction); + return processEntityView(edgeEvent, msgType, action); case DASHBOARD: - return processDashboard(edgeEvent, msgType, edgeEventAction); + return processDashboard(edgeEvent, msgType, action); case CUSTOMER: - return processCustomer(edgeEvent, msgType, edgeEventAction); + return processCustomer(edgeEvent, msgType, action); case RULE_CHAIN: - return processRuleChain(edgeEvent, msgType, edgeEventAction); + return processRuleChain(edgeEvent, msgType, action); case RULE_CHAIN_METADATA: return processRuleChainMetadata(edgeEvent, msgType); case ALARM: return processAlarm(edgeEvent, msgType); case USER: - return processUser(edgeEvent, msgType, edgeEventAction); + return processUser(edgeEvent, msgType, action); case RELATION: return processRelation(edgeEvent, msgType); case WIDGETS_BUNDLE: - return processWidgetsBundle(edgeEvent, msgType, edgeEventAction); + return processWidgetsBundle(edgeEvent, msgType, action); case WIDGET_TYPE: - return processWidgetType(edgeEvent, msgType, edgeEventAction); + return processWidgetType(edgeEvent, msgType, action); case ADMIN_SETTINGS: return processAdminSettings(edgeEvent); default: @@ -524,10 +524,10 @@ public final class EdgeGrpcSession implements Closeable { return downlinkMsg; } - private DownlinkMsg processAsset(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeEventAction) { + private DownlinkMsg processAsset(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType action) { AssetId assetId = new AssetId(edgeEvent.getEntityId()); DownlinkMsg downlinkMsg = null; - switch (edgeEventAction) { + switch (action) { case ADDED: case UPDATED: case ASSIGNED_TO_EDGE: @@ -555,10 +555,10 @@ public final class EdgeGrpcSession implements Closeable { return downlinkMsg; } - private DownlinkMsg processEntityView(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeEventAction) { + private DownlinkMsg processEntityView(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType action) { EntityViewId entityViewId = new EntityViewId(edgeEvent.getEntityId()); DownlinkMsg downlinkMsg = null; - switch (edgeEventAction) { + switch (action) { case ADDED: case UPDATED: case ASSIGNED_TO_EDGE: @@ -586,10 +586,10 @@ public final class EdgeGrpcSession implements Closeable { return downlinkMsg; } - private DownlinkMsg processDashboard(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeEventAction) { + private DownlinkMsg processDashboard(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType action) { DashboardId dashboardId = new DashboardId(edgeEvent.getEntityId()); DownlinkMsg downlinkMsg = null; - switch (edgeEventAction) { + switch (action) { case ADDED: case UPDATED: case ASSIGNED_TO_EDGE: @@ -620,10 +620,10 @@ public final class EdgeGrpcSession implements Closeable { return downlinkMsg; } - private DownlinkMsg processCustomer(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeEventAction) { + private DownlinkMsg processCustomer(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType action) { CustomerId customerId = new CustomerId(edgeEvent.getEntityId()); DownlinkMsg downlinkMsg = null; - switch (edgeEventAction) { + switch (action) { case ADDED: case UPDATED: Customer customer = ctx.getCustomerService().findCustomerById(edgeEvent.getTenantId(), customerId); @@ -646,10 +646,10 @@ public final class EdgeGrpcSession implements Closeable { return downlinkMsg; } - private DownlinkMsg processRuleChain(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType edgeEventAction) { + private DownlinkMsg processRuleChain(EdgeEvent edgeEvent, UpdateMsgType msgType, ActionType action) { RuleChainId ruleChainId = new RuleChainId(edgeEvent.getEntityId()); DownlinkMsg downlinkMsg = null; - switch (edgeEventAction) { + switch (action) { case ADDED: case UPDATED: case ASSIGNED_TO_EDGE: diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultSyncEdgeService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultSyncEdgeService.java index 2195d15236..0c9b29a386 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultSyncEdgeService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultSyncEdgeService.java @@ -412,8 +412,8 @@ public class DefaultSyncEdgeService implements SyncEdgeService { EntityId entityId = EntityIdFactory.getByTypeAndUuid( EntityType.valueOf(attributesRequestMsg.getEntityType()), new UUID(attributesRequestMsg.getEntityIdMSB(), attributesRequestMsg.getEntityIdLSB())); - final EdgeEventType edgeEventType = getEdgeQueueTypeByEntityType(entityId.getEntityType()); - if (edgeEventType != null) { + final EdgeEventType type = getEdgeQueueTypeByEntityType(entityId.getEntityType()); + if (type != null) { SettableFuture futureToSet = SettableFuture.create(); ListenableFuture> ssAttrFuture = attributesService.findAll(edge.getTenantId(), entityId, DataConstants.SERVER_SCOPE); Futures.addCallback(ssAttrFuture, new FutureCallback>() { @@ -436,14 +436,14 @@ public class DefaultSyncEdgeService implements SyncEdgeService { } entityData.put("kv", attributes); entityData.put("scope", DataConstants.SERVER_SCOPE); - JsonNode entityBody = mapper.valueToTree(entityData); - log.debug("Sending attributes data msg, entityId [{}], attributes [{}]", entityId, entityBody); + JsonNode body = mapper.valueToTree(entityData); + log.debug("Sending attributes data msg, entityId [{}], attributes [{}]", entityId, body); saveEdgeEvent(edge.getTenantId(), edge.getId(), - edgeEventType, + type, ActionType.ATTRIBUTES_UPDATED, entityId, - entityBody); + body); } catch (Exception e) { log.error("[{}] Failed to send attribute updates to the edge", edge.getName(), e); throw new RuntimeException("[" + edge.getName() + "] Failed to send attribute updates to the edge", e); @@ -572,22 +572,22 @@ public class DefaultSyncEdgeService implements SyncEdgeService { private ListenableFuture saveEdgeEvent(TenantId tenantId, EdgeId edgeId, - EdgeEventType edgeEventType, - ActionType edgeEventAction, + EdgeEventType type, + ActionType action, EntityId entityId, - JsonNode entityBody) { - log.debug("Pushing edge event to edge queue. tenantId [{}], edgeId [{}], edgeEventType [{}], edgeEventAction[{}], entityId [{}], entityBody [{}]", - tenantId, edgeId, edgeEventType, edgeEventAction, entityId, entityBody); + JsonNode body) { + log.debug("Pushing edge event to edge queue. tenantId [{}], edgeId [{}], type [{}], action[{}], entityId [{}], body [{}]", + tenantId, edgeId, type, action, entityId, body); EdgeEvent edgeEvent = new EdgeEvent(); edgeEvent.setTenantId(tenantId); edgeEvent.setEdgeId(edgeId); - edgeEvent.setType(edgeEventType); - edgeEvent.setAction(edgeEventAction.name()); + edgeEvent.setType(type); + edgeEvent.setAction(action.name()); if (entityId != null) { edgeEvent.setEntityId(entityId.getId()); } - edgeEvent.setBody(entityBody); + edgeEvent.setBody(body); return edgeEventService.saveAsync(edgeEvent); } } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseProcessor.java index 6809501a0e..576c8ba667 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseProcessor.java @@ -91,23 +91,23 @@ public abstract class BaseProcessor { protected ListenableFuture saveEdgeEvent(TenantId tenantId, EdgeId edgeId, - EdgeEventType edgeEventType, - ActionType edgeEventAction, + EdgeEventType type, + ActionType action, EntityId entityId, - JsonNode entityBody) { - log.debug("Pushing event to edge queue. tenantId [{}], edgeId [{}], edgeEventType[{}], " + - "edgeEventAction [{}], entityId [{}], entityBody [{}]", - tenantId, edgeId, edgeEventType, edgeEventAction, entityId, entityBody); + JsonNode body) { + log.debug("Pushing event to edge queue. tenantId [{}], edgeId [{}], type[{}], " + + "action [{}], entityId [{}], body [{}]", + tenantId, edgeId, type, action, entityId, body); EdgeEvent edgeEvent = new EdgeEvent(); edgeEvent.setTenantId(tenantId); edgeEvent.setEdgeId(edgeId); - edgeEvent.setType(edgeEventType); - edgeEvent.setAction(edgeEventAction.name()); + edgeEvent.setType(type); + edgeEvent.setAction(action.name()); if (entityId != null) { edgeEvent.setEntityId(entityId.getId()); } - edgeEvent.setBody(entityBody); + edgeEvent.setBody(body); return edgeEventService.saveAsync(edgeEvent); } } diff --git a/common/queue/src/main/proto/queue.proto b/common/queue/src/main/proto/queue.proto index c32a72630d..b18907dd04 100644 --- a/common/queue/src/main/proto/queue.proto +++ b/common/queue/src/main/proto/queue.proto @@ -357,12 +357,12 @@ message EdgeNotificationMsgProto { int64 tenantIdLSB = 2; int64 edgeIdMSB = 3; int64 edgeIdLSB = 4; - string edgeEventType = 5; - string edgeEventAction = 6; + string type = 5; + string action = 6; int64 entityIdMSB = 7; int64 entityIdLSB = 8; string entityType = 9; - string entityBody = 10; + string body = 10; PostTelemetryMsg postTelemetryMsg = 11; PostAttributeMsg postAttributesMsg = 12; }