diff --git a/application/src/main/java/org/thingsboard/server/controller/EntityRelationController.java b/application/src/main/java/org/thingsboard/server/controller/EntityRelationController.java index 551e73f3ed..ec7be1276a 100644 --- a/application/src/main/java/org/thingsboard/server/controller/EntityRelationController.java +++ b/application/src/main/java/org/thingsboard/server/controller/EntityRelationController.java @@ -51,7 +51,7 @@ public class EntityRelationController extends BaseController { if (relation.getTypeGroup() == null) { relation.setTypeGroup(RelationTypeGroup.COMMON); } - relationService.saveRelation(relation).get(); + relationService.saveRelation(relation); } catch (Exception e) { throw handleException(e); } @@ -76,7 +76,7 @@ public class EntityRelationController extends BaseController { checkEntityId(toId); RelationTypeGroup relationTypeGroup = parseRelationTypeGroup(strRelationTypeGroup, RelationTypeGroup.COMMON); try { - Boolean found = relationService.deleteRelation(fromId, toId, strRelationType, relationTypeGroup).get(); + Boolean found = relationService.deleteRelation(fromId, toId, strRelationType, relationTypeGroup); if (!found) { throw new ThingsboardException("Requested item wasn't found!", ThingsboardErrorCode.ITEM_NOT_FOUND); } @@ -95,7 +95,7 @@ public class EntityRelationController extends BaseController { EntityId entityId = EntityIdFactory.getByTypeAndId(strType, strId); checkEntityId(entityId); try { - relationService.deleteEntityRelations(entityId).get(); + relationService.deleteEntityRelations(entityId); } catch (Exception e) { throw handleException(e); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmService.java b/dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmService.java index cd64f32ca7..be7e44f008 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmService.java @@ -307,12 +307,12 @@ public class BaseAlarmService extends AbstractEntityService implements AlarmServ private void deleteRelation(EntityRelation alarmRelation) throws ExecutionException, InterruptedException { log.debug("Deleting Alarm relation: {}", alarmRelation); - relationService.deleteRelation(alarmRelation).get(); + relationService.deleteRelationAsync(alarmRelation).get(); } private void createRelation(EntityRelation alarmRelation) throws ExecutionException, InterruptedException { log.debug("Creating Alarm relation: {}", alarmRelation); - relationService.saveRelation(alarmRelation).get(); + relationService.saveRelationAsync(alarmRelation).get(); } private Alarm merge(Alarm existing, Alarm alarm) { diff --git a/dao/src/main/java/org/thingsboard/server/dao/entity/AbstractEntityService.java b/dao/src/main/java/org/thingsboard/server/dao/entity/AbstractEntityService.java index b75196daf2..a7f1530b62 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/entity/AbstractEntityService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/entity/AbstractEntityService.java @@ -31,12 +31,7 @@ public abstract class AbstractEntityService { protected void deleteEntityRelations(EntityId entityId) { log.trace("Executing deleteEntityRelations [{}]", entityId); - try { - relationService.deleteEntityRelations(entityId).get(); - } catch (InterruptedException | ExecutionException e) { - throw new RuntimeException(e); - } + relationService.deleteEntityRelations(entityId); } - } diff --git a/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationDao.java b/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationDao.java index 9627db3649..c313f0b966 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationDao.java @@ -140,9 +140,21 @@ public class BaseRelationDao extends CassandraAbstractAsyncDao implements Relati return getFuture(executeAsyncRead(stmt), rs -> rs != null ? getEntityRelation(rs.one()) : null); } + @Override + public boolean saveRelation(EntityRelation relation) { + BoundStatement stmt = getSaveRelationStatement(relation); + ResultSet rs = executeWrite(stmt); + return rs.wasApplied(); + } @Override - public ListenableFuture saveRelation(EntityRelation relation) { + public ListenableFuture saveRelationAsync(EntityRelation relation) { + BoundStatement stmt = getSaveRelationStatement(relation); + ResultSetFuture future = executeAsyncWrite(stmt); + return getBooleanListenableFuture(future); + } + + private BoundStatement getSaveRelationStatement(EntityRelation relation) { BoundStatement stmt = getSaveStmt().bind() .setUUID(0, relation.getFrom().getId()) .setString(1, relation.getFrom().getEntityType().name()) @@ -151,17 +163,34 @@ public class BaseRelationDao extends CassandraAbstractAsyncDao implements Relati .set(4, relation.getTypeGroup(), relationTypeGroupCodec) .setString(5, relation.getType()) .set(6, relation.getAdditionalInfo(), JsonNode.class); - ResultSetFuture future = executeAsyncWrite(stmt); - return getBooleanListenableFuture(future); + return stmt; } @Override - public ListenableFuture deleteRelation(EntityRelation relation) { + public boolean deleteRelation(EntityRelation relation) { return deleteRelation(relation.getFrom(), relation.getTo(), relation.getType(), relation.getTypeGroup()); } @Override - public ListenableFuture deleteRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) { + public ListenableFuture deleteRelationAsync(EntityRelation relation) { + return deleteRelationAsync(relation.getFrom(), relation.getTo(), relation.getType(), relation.getTypeGroup()); + } + + @Override + public boolean deleteRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) { + BoundStatement stmt = getDeleteRelationStatement(from, to, relationType, typeGroup); + ResultSet rs = executeWrite(stmt); + return rs.wasApplied(); + } + + @Override + public ListenableFuture deleteRelationAsync(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) { + BoundStatement stmt = getDeleteRelationStatement(from, to, relationType, typeGroup); + ResultSetFuture future = executeAsyncWrite(stmt); + return getBooleanListenableFuture(future); + } + + private BoundStatement getDeleteRelationStatement(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) { BoundStatement stmt = getDeleteStmt().bind() .setUUID(0, from.getId()) .setString(1, from.getEntityType().name()) @@ -169,12 +198,21 @@ public class BaseRelationDao extends CassandraAbstractAsyncDao implements Relati .setString(3, to.getEntityType().name()) .set(4, typeGroup, relationTypeGroupCodec) .setString(5, relationType); - ResultSetFuture future = executeAsyncWrite(stmt); - return getBooleanListenableFuture(future); + return stmt; } @Override - public ListenableFuture deleteOutboundRelations(EntityId entity) { + public boolean deleteOutboundRelations(EntityId entity) { + BoundStatement stmt = getDeleteAllByEntityStmt().bind() + .setUUID(0, entity.getId()) + .setString(1, entity.getEntityType().name()); + ResultSet rs = executeWrite(stmt); + return rs.wasApplied(); + } + + + @Override + public ListenableFuture deleteOutboundRelationsAsync(EntityId entity) { BoundStatement stmt = getDeleteAllByEntityStmt().bind() .setUUID(0, entity.getId()) .setString(1, entity.getEntityType().name()); diff --git a/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java b/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java index 5c29a37d1e..3574e83d7d 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java @@ -31,6 +31,7 @@ import org.thingsboard.server.dao.exception.DataValidationException; import javax.annotation.Nullable; import java.util.*; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; import java.util.function.BiConsumer; /** @@ -61,28 +62,49 @@ public class BaseRelationService implements RelationService { } @Override - public ListenableFuture saveRelation(EntityRelation relation) { + public boolean saveRelation(EntityRelation relation) { log.trace("Executing saveRelation [{}]", relation); validate(relation); return relationDao.saveRelation(relation); } @Override - public ListenableFuture deleteRelation(EntityRelation relation) { + public ListenableFuture saveRelationAsync(EntityRelation relation) { + log.trace("Executing saveRelationAsync [{}]", relation); + validate(relation); + return relationDao.saveRelationAsync(relation); + } + + @Override + public boolean deleteRelation(EntityRelation relation) { log.trace("Executing deleteRelation [{}]", relation); validate(relation); return relationDao.deleteRelation(relation); } @Override - public ListenableFuture deleteRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) { + public ListenableFuture deleteRelationAsync(EntityRelation relation) { + log.trace("Executing deleteRelationAsync [{}]", relation); + validate(relation); + return relationDao.deleteRelationAsync(relation); + } + + @Override + public boolean deleteRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) { log.trace("Executing deleteRelation [{}][{}][{}][{}]", from, to, relationType, typeGroup); validate(from, to, relationType, typeGroup); return relationDao.deleteRelation(from, to, relationType, typeGroup); } @Override - public ListenableFuture deleteEntityRelations(EntityId entity) { + public ListenableFuture deleteRelationAsync(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) { + log.trace("Executing deleteRelationAsync [{}][{}][{}][{}]", from, to, relationType, typeGroup); + validate(from, to, relationType, typeGroup); + return relationDao.deleteRelationAsync(from, to, relationType, typeGroup); + } + + @Override + public boolean deleteEntityRelations(EntityId entity) { log.trace("Executing deleteEntityRelations [{}]", entity); validate(entity); List>> inboundRelationsList = new ArrayList<>(); @@ -90,12 +112,42 @@ public class BaseRelationService implements RelationService { inboundRelationsList.add(relationDao.findAllByTo(entity, typeGroup)); } ListenableFuture>> inboundRelations = Futures.allAsList(inboundRelationsList); + ListenableFuture> inboundDeletions = Futures.transform(inboundRelations, new Function>, List>() { + @Override + public List apply(List> relations) { + List results = new ArrayList<>(); + for (List relationList : relations) { + relationList.stream().forEach(relation -> results.add(relationDao.deleteRelation(relation))); + } + return results; + } + }); + ListenableFuture inboundFuture = Futures.transform(inboundDeletions, getListToBooleanFunction()); + boolean inboundDeleteResult = false; + try { + inboundDeleteResult = inboundFuture.get(); + } catch (InterruptedException | ExecutionException e) { + log.error("Error deleting entity inbound relations", e); + } + boolean outboundDeleteResult = relationDao.deleteOutboundRelations(entity); + return inboundDeleteResult && outboundDeleteResult; + } + + @Override + public ListenableFuture deleteEntityRelationsAsync(EntityId entity) { + log.trace("Executing deleteEntityRelationsAsync [{}]", entity); + validate(entity); + List>> inboundRelationsList = new ArrayList<>(); + for (RelationTypeGroup typeGroup : RelationTypeGroup.values()) { + inboundRelationsList.add(relationDao.findAllByTo(entity, typeGroup)); + } + ListenableFuture>> inboundRelations = Futures.allAsList(inboundRelationsList); ListenableFuture> inboundDeletions = Futures.transform(inboundRelations, new AsyncFunction>, List>() { @Override public ListenableFuture> apply(List> relations) throws Exception { List> results = new ArrayList<>(); for (List relationList : relations) { - relationList.stream().forEach(relation -> results.add(relationDao.deleteRelation(relation))); + relationList.stream().forEach(relation -> results.add(relationDao.deleteRelationAsync(relation))); } return Futures.allAsList(results); } @@ -103,7 +155,7 @@ public class BaseRelationService implements RelationService { ListenableFuture inboundFuture = Futures.transform(inboundDeletions, getListToBooleanFunction()); - ListenableFuture outboundFuture = relationDao.deleteOutboundRelations(entity); + ListenableFuture outboundFuture = relationDao.deleteOutboundRelationsAsync(entity); return Futures.transform(Futures.allAsList(Arrays.asList(inboundFuture, outboundFuture)), getListToBooleanFunction()); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/relation/RelationDao.java b/dao/src/main/java/org/thingsboard/server/dao/relation/RelationDao.java index ec946d0ed1..edd6d879a5 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/relation/RelationDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/relation/RelationDao.java @@ -41,13 +41,21 @@ public interface RelationDao { ListenableFuture getRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup); - ListenableFuture saveRelation(EntityRelation relation); + boolean saveRelation(EntityRelation relation); - ListenableFuture deleteRelation(EntityRelation relation); + ListenableFuture saveRelationAsync(EntityRelation relation); - ListenableFuture deleteRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup); + boolean deleteRelation(EntityRelation relation); - ListenableFuture deleteOutboundRelations(EntityId entity); + ListenableFuture deleteRelationAsync(EntityRelation relation); + + boolean deleteRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup); + + ListenableFuture deleteRelationAsync(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup); + + boolean deleteOutboundRelations(EntityId entity); + + ListenableFuture deleteOutboundRelationsAsync(EntityId entity); ListenableFuture> findRelations(EntityId from, String relationType, RelationTypeGroup typeGroup, EntityType toType, TimePageLink pageLink); diff --git a/dao/src/main/java/org/thingsboard/server/dao/relation/RelationService.java b/dao/src/main/java/org/thingsboard/server/dao/relation/RelationService.java index 89aae8e5e5..cef93935e6 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/relation/RelationService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/relation/RelationService.java @@ -33,13 +33,21 @@ public interface RelationService { ListenableFuture getRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup); - ListenableFuture saveRelation(EntityRelation relation); + boolean saveRelation(EntityRelation relation); - ListenableFuture deleteRelation(EntityRelation relation); + ListenableFuture saveRelationAsync(EntityRelation relation); - ListenableFuture deleteRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup); + boolean deleteRelation(EntityRelation relation); - ListenableFuture deleteEntityRelations(EntityId entity); + ListenableFuture deleteRelationAsync(EntityRelation relation); + + boolean deleteRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup); + + ListenableFuture deleteRelationAsync(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup); + + boolean deleteEntityRelations(EntityId entity); + + ListenableFuture deleteEntityRelationsAsync(EntityId entity); ListenableFuture> findByFrom(EntityId from, RelationTypeGroup typeGroup); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/relation/JpaRelationDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/relation/JpaRelationDao.java index 16aabac0cc..b996aeac33 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/relation/JpaRelationDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/relation/JpaRelationDao.java @@ -120,12 +120,25 @@ public class JpaRelationDao extends JpaAbstractDaoListeningExecutorService imple } @Override - public ListenableFuture saveRelation(EntityRelation relation) { + public boolean saveRelation(EntityRelation relation) { + return relationRepository.save(new RelationEntity(relation)) != null; + } + + @Override + public ListenableFuture saveRelationAsync(EntityRelation relation) { return service.submit(() -> relationRepository.save(new RelationEntity(relation)) != null); } @Override - public ListenableFuture deleteRelation(EntityRelation relation) { + public boolean deleteRelation(EntityRelation relation) { + RelationCompositeKey key = new RelationCompositeKey(relation); + boolean relationExistsBeforeDelete = relationRepository.exists(key); + relationRepository.delete(key); + return relationExistsBeforeDelete; + } + + @Override + public ListenableFuture deleteRelationAsync(EntityRelation relation) { RelationCompositeKey key = new RelationCompositeKey(relation); return service.submit( () -> { @@ -136,7 +149,15 @@ public class JpaRelationDao extends JpaAbstractDaoListeningExecutorService imple } @Override - public ListenableFuture deleteRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) { + public boolean deleteRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) { + RelationCompositeKey key = getRelationCompositeKey(from, to, relationType, typeGroup); + boolean relationExistsBeforeDelete = relationRepository.exists(key); + relationRepository.delete(key); + return relationExistsBeforeDelete; + } + + @Override + public ListenableFuture deleteRelationAsync(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) { RelationCompositeKey key = getRelationCompositeKey(from, to, relationType, typeGroup); return service.submit( () -> { @@ -147,7 +168,16 @@ public class JpaRelationDao extends JpaAbstractDaoListeningExecutorService imple } @Override - public ListenableFuture deleteOutboundRelations(EntityId entity) { + public boolean deleteOutboundRelations(EntityId entity) { + boolean relationExistsBeforeDelete = relationRepository + .findAllByFromIdAndFromType(UUIDConverter.fromTimeUUID(entity.getId()), entity.getEntityType().name()) + .size() > 0; + relationRepository.deleteByFromIdAndFromType(UUIDConverter.fromTimeUUID(entity.getId()), entity.getEntityType().name()); + return relationExistsBeforeDelete; + } + + @Override + public ListenableFuture deleteOutboundRelationsAsync(EntityId entity) { return service.submit( () -> { boolean relationExistsBeforeDelete = relationRepository diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/BaseAlarmServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/BaseAlarmServiceTest.java index b208e75d29..23fc3cc1a3 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/BaseAlarmServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/BaseAlarmServiceTest.java @@ -57,7 +57,7 @@ public abstract class BaseAlarmServiceTest extends AbstractServiceTest { EntityRelation relation = new EntityRelation(parentId, childId, EntityRelation.CONTAINS_TYPE); - Assert.assertTrue(relationService.saveRelation(relation).get()); + Assert.assertTrue(relationService.saveRelationAsync(relation).get()); long ts = System.currentTimeMillis(); Alarm alarm = Alarm.builder().tenantId(tenantId).originator(childId) @@ -94,7 +94,7 @@ public abstract class BaseAlarmServiceTest extends AbstractServiceTest { EntityRelation relation = new EntityRelation(parentId, childId, EntityRelation.CONTAINS_TYPE); - Assert.assertTrue(relationService.saveRelation(relation).get()); + Assert.assertTrue(relationService.saveRelationAsync(relation).get()); long ts = System.currentTimeMillis(); Alarm alarm = Alarm.builder().tenantId(tenantId).originator(childId) diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/BaseRelationServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/BaseRelationServiceTest.java index d76265dd7d..6b58af8e19 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/BaseRelationServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/BaseRelationServiceTest.java @@ -75,13 +75,13 @@ public abstract class BaseRelationServiceTest extends AbstractServiceTest { saveRelation(relationA); saveRelation(relationB); - Assert.assertTrue(relationService.deleteRelation(relationA).get()); + Assert.assertTrue(relationService.deleteRelationAsync(relationA).get()); Assert.assertFalse(relationService.checkRelation(parentId, childId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.COMMON).get()); Assert.assertTrue(relationService.checkRelation(childId, subChildId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.COMMON).get()); - Assert.assertTrue(relationService.deleteRelation(childId, subChildId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.COMMON).get()); + Assert.assertTrue(relationService.deleteRelationAsync(childId, subChildId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.COMMON).get()); } @Test @@ -96,7 +96,7 @@ public abstract class BaseRelationServiceTest extends AbstractServiceTest { saveRelation(relationA); saveRelation(relationB); - Assert.assertTrue(relationService.deleteEntityRelations(childId).get()); + Assert.assertTrue(relationService.deleteEntityRelationsAsync(childId).get()); Assert.assertFalse(relationService.checkRelation(parentId, childId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.COMMON).get()); @@ -152,7 +152,7 @@ public abstract class BaseRelationServiceTest extends AbstractServiceTest { } private Boolean saveRelation(EntityRelation relationA1) throws ExecutionException, InterruptedException { - return relationService.saveRelation(relationA1).get(); + return relationService.saveRelationAsync(relationA1).get(); } @Test diff --git a/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionCtx.java b/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionCtx.java index 27c4f884fe..00d1f0c6d7 100644 --- a/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionCtx.java +++ b/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionCtx.java @@ -94,7 +94,7 @@ public class GatewaySessionCtx { newDevice.setName(deviceName); newDevice.setType(deviceType); newDevice = deviceService.saveDevice(newDevice); - relationService.saveRelation(new EntityRelation(gateway.getId(), newDevice.getId(), "Created")); + relationService.saveRelationAsync(new EntityRelation(gateway.getId(), newDevice.getId(), "Created")); return newDevice; }); GatewayDeviceSessionCtx ctx = new GatewayDeviceSessionCtx(this, device);