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

Rename edgeEventAction and edgeEventType to action and type

pull/2436/head
Volodymyr Babak 6 years ago
parent
6aa5655e6f
commit
4a71e8e6a4
6 changed files with 110 additions and 110 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. 38
      application/src/main/java/org/thingsboard/server/controller/BaseController.java
  2. 82
      application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java
  3. 48
      application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
  4. 28
      application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultSyncEdgeService.java
  5. 18
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseProcessor.java
  6. 6
      common/queue/src/main/proto/queue.proto

38
application/src/main/java/org/thingsboard/server/controller/BaseController.java
View File

@ -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,

82
application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java
View File

@ -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<Edge> 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<Edge>() {
@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<Edge>() {
@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<Edge> 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<Edge> 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<List<EdgeId>> 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<EdgeId> 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<Edge> future = edgeService.findEdgeByIdAsync(tenantId, edgeId);
Futures.addCallback(future, new FutureCallback<Edge>() {
@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<Edge> 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<Alarm> 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<List<EdgeId>> 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<ListenableFuture<List<EdgeId>>> 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));
}

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

@ -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:

28
application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultSyncEdgeService.java
View File

@ -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<Void> futureToSet = SettableFuture.create();
ListenableFuture<List<AttributeKvEntry>> ssAttrFuture = attributesService.findAll(edge.getTenantId(), entityId, DataConstants.SERVER_SCOPE);
Futures.addCallback(ssAttrFuture, new FutureCallback<List<AttributeKvEntry>>() {
@ -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<EdgeEvent> 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);
}
}

18
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseProcessor.java
View File

@ -91,23 +91,23 @@ public abstract class BaseProcessor {
protected ListenableFuture<EdgeEvent> 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);
}
}

6
common/queue/src/main/proto/queue.proto
View File

@ -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;
}

Write Preview
Loading…
Cancel
Save