Improve Relations API.

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

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) {

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

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<Boolean> saveRelation(EntityRelation relation) {
+    public ListenableFuture<Boolean> 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<Boolean> deleteRelation(EntityRelation relation) {
+    public boolean deleteRelation(EntityRelation relation) {
         return deleteRelation(relation.getFrom(), relation.getTo(), relation.getType(), relation.getTypeGroup());
     }
 
     @Override
-    public ListenableFuture<Boolean> deleteRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) {
+    public ListenableFuture<Boolean> 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<Boolean> 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<Boolean> 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<Boolean> deleteOutboundRelationsAsync(EntityId entity) {
         BoundStatement stmt = getDeleteAllByEntityStmt().bind()
                 .setUUID(0, entity.getId())
                 .setString(1, entity.getEntityType().name());

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<Boolean> saveRelation(EntityRelation relation) {
+    public boolean saveRelation(EntityRelation relation) {
         log.trace("Executing saveRelation [{}]", relation);
         validate(relation);
         return relationDao.saveRelation(relation);
     }
 
     @Override
-    public ListenableFuture<Boolean> deleteRelation(EntityRelation relation) {
+    public ListenableFuture<Boolean> 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<Boolean> deleteRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) {
+    public ListenableFuture<Boolean> 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<Boolean> deleteEntityRelations(EntityId entity) {
+    public ListenableFuture<Boolean> 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<ListenableFuture<List<EntityRelation>>> inboundRelationsList = new ArrayList<>();
@@ -90,12 +112,42 @@ public class BaseRelationService implements RelationService {
             inboundRelationsList.add(relationDao.findAllByTo(entity, typeGroup));
         }
         ListenableFuture<List<List<EntityRelation>>> inboundRelations = Futures.allAsList(inboundRelationsList);
+        ListenableFuture<List<Boolean>> inboundDeletions = Futures.transform(inboundRelations, new Function<List<List<EntityRelation>>, List<Boolean>>() {
+            @Override
+            public List<Boolean> apply(List<List<EntityRelation>> relations) {
+                List<Boolean> results = new ArrayList<>();
+                for (List<EntityRelation> relationList : relations) {
+                    relationList.stream().forEach(relation -> results.add(relationDao.deleteRelation(relation)));
+                }
+                return results;
+            }
+        });
+        ListenableFuture<Boolean> 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<Boolean> deleteEntityRelationsAsync(EntityId entity) {
+        log.trace("Executing deleteEntityRelationsAsync [{}]", entity);
+        validate(entity);
+        List<ListenableFuture<List<EntityRelation>>> inboundRelationsList = new ArrayList<>();
+        for (RelationTypeGroup typeGroup : RelationTypeGroup.values()) {
+            inboundRelationsList.add(relationDao.findAllByTo(entity, typeGroup));
+        }
+        ListenableFuture<List<List<EntityRelation>>> inboundRelations = Futures.allAsList(inboundRelationsList);
         ListenableFuture<List<Boolean>> inboundDeletions = Futures.transform(inboundRelations, new AsyncFunction<List<List<EntityRelation>>, List<Boolean>>() {
             @Override
             public ListenableFuture<List<Boolean>> apply(List<List<EntityRelation>> relations) throws Exception {
                 List<ListenableFuture<Boolean>> results = new ArrayList<>();
                 for (List<EntityRelation> 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<Boolean> inboundFuture = Futures.transform(inboundDeletions, getListToBooleanFunction());
 
-        ListenableFuture<Boolean> outboundFuture = relationDao.deleteOutboundRelations(entity);
+        ListenableFuture<Boolean> outboundFuture = relationDao.deleteOutboundRelationsAsync(entity);
 
         return Futures.transform(Futures.allAsList(Arrays.asList(inboundFuture, outboundFuture)), getListToBooleanFunction());
     }

dao/src/main/java/org/thingsboard/server/dao/relation/RelationDao.java

@@ -41,13 +41,21 @@ public interface RelationDao {
 
     ListenableFuture<EntityRelation> getRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup);
 
-    ListenableFuture<Boolean> saveRelation(EntityRelation relation);
+    boolean saveRelation(EntityRelation relation);
 
-    ListenableFuture<Boolean> deleteRelation(EntityRelation relation);
+    ListenableFuture<Boolean> saveRelationAsync(EntityRelation relation);
 
-    ListenableFuture<Boolean> deleteRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup);
+    boolean deleteRelation(EntityRelation relation);
 
-    ListenableFuture<Boolean> deleteOutboundRelations(EntityId entity);
+    ListenableFuture<Boolean> deleteRelationAsync(EntityRelation relation);
+
+    boolean deleteRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup);
+
+    ListenableFuture<Boolean> deleteRelationAsync(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup);
+
+    boolean deleteOutboundRelations(EntityId entity);
+
+    ListenableFuture<Boolean> deleteOutboundRelationsAsync(EntityId entity);
 
     ListenableFuture<List<EntityRelation>> findRelations(EntityId from, String relationType, RelationTypeGroup typeGroup, EntityType toType, TimePageLink pageLink);
 

dao/src/main/java/org/thingsboard/server/dao/relation/RelationService.java

@@ -33,13 +33,21 @@ public interface RelationService {
 
     ListenableFuture<EntityRelation> getRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup);
 
-    ListenableFuture<Boolean> saveRelation(EntityRelation relation);
+    boolean saveRelation(EntityRelation relation);
 
-    ListenableFuture<Boolean> deleteRelation(EntityRelation relation);
+    ListenableFuture<Boolean> saveRelationAsync(EntityRelation relation);
 
-    ListenableFuture<Boolean> deleteRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup);
+    boolean deleteRelation(EntityRelation relation);
 
-    ListenableFuture<Boolean> deleteEntityRelations(EntityId entity);
+    ListenableFuture<Boolean> deleteRelationAsync(EntityRelation relation);
+
+    boolean deleteRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup);
+
+    ListenableFuture<Boolean> deleteRelationAsync(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup);
+
+    boolean deleteEntityRelations(EntityId entity);
+
+    ListenableFuture<Boolean> deleteEntityRelationsAsync(EntityId entity);
 
     ListenableFuture<List<EntityRelation>> findByFrom(EntityId from, RelationTypeGroup typeGroup);
 

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<Boolean> saveRelation(EntityRelation relation) {
+    public boolean saveRelation(EntityRelation relation) {
+        return relationRepository.save(new RelationEntity(relation)) != null;
+    }
+
+    @Override
+    public ListenableFuture<Boolean> saveRelationAsync(EntityRelation relation) {
         return service.submit(() -> relationRepository.save(new RelationEntity(relation)) != null);
     }
 
     @Override
-    public ListenableFuture<Boolean> 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<Boolean> deleteRelationAsync(EntityRelation relation) {
         RelationCompositeKey key = new RelationCompositeKey(relation);
         return service.submit(
                 () -> {
@@ -136,7 +149,15 @@ public class JpaRelationDao extends JpaAbstractDaoListeningExecutorService imple
     }
 
     @Override
-    public ListenableFuture<Boolean> 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<Boolean> 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<Boolean> 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<Boolean> deleteOutboundRelationsAsync(EntityId entity) {
         return service.submit(
                 () -> {
                     boolean relationExistsBeforeDelete = relationRepository

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)

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

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);