简柏林 / thingskit · Commits

Commit 125130193d9308b87c64aefe95d058347495882a

Authored by Volodymyr Babak
1 parent 486ea8e0

Ignore EDGE relation

Inline Side-by-side
Showing 3 changed files with 39 additions and 30 deletions
@@ -725,7 +725,10 @@ public abstract class BaseController { @@ -725,7 +725,10 @@ public abstract class BaseController {
725 725
726 protected void sendNotificationMsgToEdgeService(TenantId tenantId, EntityRelation relation, ActionType edgeEventAction) { 726 protected void sendNotificationMsgToEdgeService(TenantId tenantId, EntityRelation relation, ActionType edgeEventAction) {
727 try { 727 try {
728 - sendNotificationMsgToEdgeService(tenantId, null, null, json.writeValueAsString(relation), EdgeEventType.RELATION, edgeEventAction); 728 + if (!relation.getFrom().getEntityType().equals(EntityType.EDGE) &&
  729 + !relation.getTo().getEntityType().equals(EntityType.EDGE)) {
  730 + sendNotificationMsgToEdgeService(tenantId, null, null, json.writeValueAsString(relation), EdgeEventType.RELATION, edgeEventAction);
  731 + }
729 } catch (Exception e) { 732 } catch (Exception e) {
730 log.warn("Failed to push relation to core: {}", relation, e); 733 log.warn("Failed to push relation to core: {}", relation, e);
731 } 734 }
@@ -236,32 +236,35 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { @@ -236,32 +236,35 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
236 } 236 }
237 237
238 private void processRelation(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { 238 private void processRelation(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) {
239 - EntityRelation entityRelation = mapper.convertValue(edgeNotificationMsg.getEntityBody(), EntityRelation.class);  
240 - List<ListenableFuture<List<EdgeId>>> futures = new ArrayList<>();  
241 - futures.add(findRelatedEdgeIdsByEntityId(tenantId, entityRelation.getTo()));  
242 - futures.add(findRelatedEdgeIdsByEntityId(tenantId, entityRelation.getFrom()));  
243 - ListenableFuture<List<List<EdgeId>>> combinedFuture = Futures.allAsList(futures);  
244 - Futures.transform(combinedFuture, listOfListsEdgeIds -> {  
245 - Set<EdgeId> uniqueEdgeIds = new HashSet<>();  
246 - if (listOfListsEdgeIds != null && !listOfListsEdgeIds.isEmpty()) {  
247 - for (List<EdgeId> listOfListsEdgeId : listOfListsEdgeIds) {  
248 - if (listOfListsEdgeId != null) {  
249 - uniqueEdgeIds.addAll(listOfListsEdgeId); 239 + EntityRelation relation = mapper.convertValue(edgeNotificationMsg.getEntityBody(), EntityRelation.class);
  240 + if (!relation.getFrom().getEntityType().equals(EntityType.EDGE) &&
  241 + !relation.getTo().getEntityType().equals(EntityType.EDGE)) {
  242 + List<ListenableFuture<List<EdgeId>>> futures = new ArrayList<>();
  243 + futures.add(findRelatedEdgeIdsByEntityId(tenantId, relation.getTo()));
  244 + futures.add(findRelatedEdgeIdsByEntityId(tenantId, relation.getFrom()));
  245 + ListenableFuture<List<List<EdgeId>>> combinedFuture = Futures.allAsList(futures);
  246 + Futures.transform(combinedFuture, listOfListsEdgeIds -> {
  247 + Set<EdgeId> uniqueEdgeIds = new HashSet<>();
  248 + if (listOfListsEdgeIds != null && !listOfListsEdgeIds.isEmpty()) {
  249 + for (List<EdgeId> listOfListsEdgeId : listOfListsEdgeIds) {
  250 + if (listOfListsEdgeId != null) {
  251 + uniqueEdgeIds.addAll(listOfListsEdgeId);
  252 + }
250 } 253 }
251 } 254 }
252 - }  
253 - if (!uniqueEdgeIds.isEmpty()) {  
254 - for (EdgeId edgeId : uniqueEdgeIds) {  
255 - saveEdgeEvent(tenantId,  
256 - edgeId,  
257 - EdgeEventType.RELATION,  
258 - ActionType.valueOf(edgeNotificationMsg.getEdgeEventAction()),  
259 - null,  
260 - mapper.valueToTree(entityRelation)); 255 + if (!uniqueEdgeIds.isEmpty()) {
  256 + for (EdgeId edgeId : uniqueEdgeIds) {
  257 + saveEdgeEvent(tenantId,
  258 + edgeId,
  259 + EdgeEventType.RELATION,
  260 + ActionType.valueOf(edgeNotificationMsg.getEdgeEventAction()),
  261 + null,
  262 + mapper.valueToTree(relation));
  263 + }
261 } 264 }
262 - }  
263 - return null;  
264 - }, dbCallbackExecutorService); 265 + return null;
  266 + }, dbCallbackExecutorService);
  267 + }
265 } 268 }
266 269
267 private ListenableFuture<List<EdgeId>> findRelatedEdgeIdsByEntityId(TenantId tenantId, EntityId entityId) { 270 private ListenableFuture<List<EdgeId>> findRelatedEdgeIdsByEntityId(TenantId tenantId, EntityId entityId) {
@@ -358,12 +358,15 @@ public class DefaultSyncEdgeService implements SyncEdgeService { @@ -358,12 +358,15 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
358 log.trace("[{}] [{}] [{}] relation(s) are going to be pushed to edge.", edge.getId(), entityId, entityRelations.size()); 358 log.trace("[{}] [{}] [{}] relation(s) are going to be pushed to edge.", edge.getId(), entityId, entityRelations.size());
359 for (EntityRelation relation : entityRelations) { 359 for (EntityRelation relation : entityRelations) {
360 try { 360 try {
361 - saveEdgeEvent(edge.getTenantId(),  
362 - edge.getId(),  
363 - EdgeEventType.RELATION,  
364 - ActionType.ADDED,  
365 - null,  
366 - mapper.valueToTree(relation)); 361 + if (!relation.getFrom().getEntityType().equals(EntityType.EDGE) &&
  362 + !relation.getTo().getEntityType().equals(EntityType.EDGE)) {
  363 + saveEdgeEvent(edge.getTenantId(),
  364 + edge.getId(),
  365 + EdgeEventType.RELATION,
  366 + ActionType.ADDED,
  367 + null,
  368 + mapper.valueToTree(relation));
  369 + }
367 } catch (Exception e) { 370 } catch (Exception e) {
368 log.error("Exception during loading relation [{}] to edge on sync!", relation, e); 371 log.error("Exception during loading relation [{}] to edge on sync!", relation, e);
369 } 372 }