diff --git a/dao/src/main/java/org/thingsboard/server/dao/alarm/AlarmCommentDao.java b/dao/src/main/java/org/thingsboard/server/dao/alarm/AlarmCommentDao.java index 947db57532..442bff7516 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/alarm/AlarmCommentDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/alarm/AlarmCommentDao.java @@ -18,7 +18,6 @@ package org.thingsboard.server.dao.alarm; import com.google.common.util.concurrent.ListenableFuture; import org.thingsboard.server.common.data.alarm.AlarmComment; import org.thingsboard.server.common.data.alarm.AlarmCommentInfo; -import org.thingsboard.server.common.data.id.AlarmCommentId; import org.thingsboard.server.common.data.id.AlarmId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.page.PageData; @@ -29,13 +28,10 @@ import java.util.UUID; public interface AlarmCommentDao extends Dao { - AlarmComment createAlarmComment(TenantId tenantId, AlarmComment alarmComment); - - void deleteAlarmComment(TenantId tenantId, AlarmCommentId alarmCommentId); - AlarmComment findAlarmCommentById(TenantId tenantId, UUID key); PageData findAlarmComments(TenantId tenantId, AlarmId id, PageLink pageLink); ListenableFuture findAlarmCommentByIdAsync(TenantId tenantId, UUID key); + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmCommentService.java b/dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmCommentService.java index 6ded31fde3..56dd7a1d9b 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmCommentService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmCommentService.java @@ -86,7 +86,7 @@ public class BaseAlarmCommentService extends AbstractEntityService implements Al if (alarmComment.getType() == null) { alarmComment.setType(AlarmCommentType.OTHER); } - return alarmCommentDao.createAlarmComment(tenantId, alarmComment); + return alarmCommentDao.save(tenantId, alarmComment); } private AlarmComment updateAlarmComment(TenantId tenantId, AlarmComment newAlarmComment) { diff --git a/dao/src/main/java/org/thingsboard/server/dao/audit/AuditLogDao.java b/dao/src/main/java/org/thingsboard/server/dao/audit/AuditLogDao.java index fcbece7426..a3bdea1a37 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/audit/AuditLogDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/audit/AuditLogDao.java @@ -30,8 +30,6 @@ import java.util.UUID; public interface AuditLogDao extends Dao { - ListenableFuture saveByTenantId(AuditLog auditLog); - PageData findAuditLogsByTenantIdAndEntityId(UUID tenantId, EntityId entityId, List actionTypes, TimePageLink pageLink); PageData findAuditLogsByTenantIdAndCustomerId(UUID tenantId, CustomerId customerId, List actionTypes, TimePageLink pageLink); diff --git a/dao/src/main/java/org/thingsboard/server/dao/audit/AuditLogServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/audit/AuditLogServiceImpl.java index 94a9ac0442..98b40e52f8 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/audit/AuditLogServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/audit/AuditLogServiceImpl.java @@ -20,7 +20,6 @@ import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.ObjectNode; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; -import com.google.common.util.concurrent.MoreExecutors; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; @@ -48,6 +47,7 @@ import org.thingsboard.server.dao.audit.sink.AuditLogSink; import org.thingsboard.server.dao.device.provision.ProvisionRequest; import org.thingsboard.server.dao.entity.EntityService; import org.thingsboard.server.dao.service.DataValidator; +import org.thingsboard.server.dao.sql.JpaExecutorService; import java.io.PrintWriter; import java.io.StringWriter; @@ -76,6 +76,9 @@ public class AuditLogServiceImpl implements AuditLogService { @Autowired private AuditLogSink auditLogSink; + @Autowired + private JpaExecutorService executor; + @Autowired private DataValidator auditLogValidator; @@ -380,15 +383,15 @@ public class AuditLogServiceImpl implements AuditLogService { } private ListenableFuture logAction(TenantId tenantId, - EntityId entityId, - String entityName, - CustomerId customerId, - UserId userId, - String userName, - ActionType actionType, - JsonNode actionData, - ActionStatus actionStatus, - String actionFailureDetails) { + EntityId entityId, + String entityName, + CustomerId customerId, + UserId userId, + String userName, + ActionType actionType, + JsonNode actionData, + ActionStatus actionStatus, + String actionFailureDetails) { AuditLog auditLogEntry = createAuditLogEntry(tenantId, entityId, entityName, customerId, userId, userName, actionType, actionData, actionStatus, actionFailureDetails); log.trace("Executing logAction [{}]", auditLogEntry); @@ -402,11 +405,11 @@ public class AuditLogServiceImpl implements AuditLogService { } } - ListenableFuture future = auditLogDao.saveByTenantId(auditLogEntry); - return Futures.transform(future, auditLog -> { - auditLogSink.logAction(auditLogEntry); + return executor.submit(() -> { + AuditLog auditLog = auditLogDao.save(tenantId, auditLogEntry); + auditLogSink.logAction(auditLog); return null; - }, MoreExecutors.directExecutor()); + }); } } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/JpaAbstractDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/JpaAbstractDao.java index 7926eb0050..92bc4187c1 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/JpaAbstractDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/JpaAbstractDao.java @@ -19,12 +19,8 @@ import com.datastax.oss.driver.api.core.uuid.Uuids; import com.google.common.collect.Lists; import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.transaction.annotation.Transactional; -import org.springframework.transaction.support.TransactionSynchronizationManager; -import org.springframework.transaction.support.TransactionTemplate; -import org.thingsboard.server.common.data.id.HasId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.dao.Dao; import org.thingsboard.server.dao.DaoUtil; @@ -37,45 +33,26 @@ import java.util.Collection; import java.util.List; import java.util.Optional; import java.util.UUID; -import java.util.function.Consumer; /** * @author Valerii Sosliuk */ @Slf4j @SqlDao -public abstract class JpaAbstractDao, D extends HasId> +public abstract class JpaAbstractDao, D> extends JpaAbstractDaoListeningExecutorService implements Dao { @PersistenceContext private EntityManager entityManager; - @Autowired - private TransactionTemplate transactionTemplate; - protected abstract Class getEntityClass(); protected abstract JpaRepository getRepository(); - protected void setSearchText(E entity) { - } - @Override @Transactional public D save(TenantId tenantId, D domain) { - return save(tenantId, domain, null); - } - - @Override - @Transactional - public D saveAndFlush(TenantId tenantId, D domain) { - D d = save(tenantId, domain); - getRepository().flush(); - return d; - } - - protected D save(TenantId tenantId, D domain, Consumer preSaveAction) { E entity; try { entity = getEntityClass().getConstructor(domain.getClass()).newInstance(domain); @@ -83,7 +60,6 @@ public abstract class JpaAbstractDao, D extends HasId log.error("Can't create entity for domain object {}", domain, e); throw new IllegalArgumentException("Can't create entity for domain object {" + domain + "}", e); } - setSearchText(entity); log.debug("Saving entity {}", entity); boolean isNew = entity.getUuid() == null; if (isNew) { @@ -92,17 +68,9 @@ public abstract class JpaAbstractDao, D extends HasId entity.setCreatedTime(Uuids.unixTimestamp(uuid)); } - if (preSaveAction != null) { - preSaveAction.accept(entity); - } - if (TransactionSynchronizationManager.isActualTransactionActive()) { - return doSave(entity, isNew); - } else { - return transactionTemplate.execute(status -> doSave(entity, isNew)); + if (isPartitioned()) { + createPartition(entity); } - } - - private D doSave(E entity, boolean isNew) { if (isNew) { entityManager.persist(entity); } else { @@ -111,6 +79,14 @@ public abstract class JpaAbstractDao, D extends HasId return DaoUtil.getData(entity); } + @Override + @Transactional + public D saveAndFlush(TenantId tenantId, D domain) { + D d = save(tenantId, domain); + getRepository().flush(); + return d; + } + @Override public D findById(TenantId tenantId, UUID key) { log.debug("Get entity by key {}", key); @@ -155,4 +131,12 @@ public abstract class JpaAbstractDao, D extends HasId List entities = Lists.newArrayList(getRepository().findAll()); return DaoUtil.convertDataList(entities); } + + public boolean isPartitioned() { + return false; + } + + public void createPartition(E entity) { + } + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/alarm/JpaAlarmCommentDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/alarm/JpaAlarmCommentDao.java index 0e37e3ec4a..47df3bd0c6 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/alarm/JpaAlarmCommentDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/alarm/JpaAlarmCommentDao.java @@ -22,10 +22,8 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.stereotype.Component; -import org.springframework.transaction.annotation.Transactional; import org.thingsboard.server.common.data.alarm.AlarmComment; import org.thingsboard.server.common.data.alarm.AlarmCommentInfo; -import org.thingsboard.server.common.data.id.AlarmCommentId; import org.thingsboard.server.common.data.id.AlarmId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.page.PageData; @@ -54,21 +52,6 @@ public class JpaAlarmCommentDao extends JpaAbstractDao { - partitioningRepository.createPartitionIfNotExists(ALARM_COMMENT_TABLE_NAME, entity.getCreatedTime(), TimeUnit.HOURS.toMillis(partitionSizeInHours)); - }); - } - - @Override - public void deleteAlarmComment(TenantId tenantId, AlarmCommentId alarmCommentId){ - log.trace("Try to delete entity alarm comment by id using [{}]", alarmCommentId); - alarmCommentRepository.deleteById(alarmCommentId.getId()); - } - @Override public PageData findAlarmComments(TenantId tenantId, AlarmId id, PageLink pageLink){ log.trace("Try to find alarm comments by alarm id using [{}]", id); @@ -88,6 +71,16 @@ public class JpaAlarmCommentDao extends JpaAbstractDao getEntityClass() { return AlarmCommentEntity.class; diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/audit/JpaAuditLogDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/audit/JpaAuditLogDao.java index 9ea49fe339..ccada3a21b 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/audit/JpaAuditLogDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/audit/JpaAuditLogDao.java @@ -15,7 +15,6 @@ */ package org.thingsboard.server.dao.sql.audit; -import com.google.common.util.concurrent.ListenableFuture; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; @@ -26,7 +25,6 @@ import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.audit.AuditLog; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.EntityId; -import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.TimePageLink; @@ -69,18 +67,6 @@ public class JpaAuditLogDao extends JpaAbstractDao imp return auditLogRepository; } - @Override - public ListenableFuture saveByTenantId(AuditLog auditLog) { - return service.submit(() -> save(auditLog.getTenantId(), auditLog)); - } - - @Override - public AuditLog save(TenantId tenantId, AuditLog auditLog) { - return save(tenantId, auditLog, entity -> { - partitioningRepository.createPartitionIfNotExists(TABLE_NAME, entity.getCreatedTime(), TimeUnit.HOURS.toMillis(partitionSizeInHours)); - }); - } - @Override public PageData findAuditLogsByTenantIdAndEntityId(UUID tenantId, EntityId entityId, List actionTypes, TimePageLink pageLink) { return DaoUtil.toPageData( @@ -172,4 +158,14 @@ public class JpaAuditLogDao extends JpaAbstractDao imp jdbcTemplate.update("CALL migrate_audit_logs(?, ?, ?)", startTime, endTime, partitionSizeInMs); } + @Override + public boolean isPartitioned() { + return true; + } + + @Override + public void createPartition(AuditLogEntity entity) { + partitioningRepository.createPartitionIfNotExists(TABLE_NAME, entity.getCreatedTime(), TimeUnit.HOURS.toMillis(partitionSizeInHours)); + } + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/edge/JpaBaseEdgeEventDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/edge/JpaBaseEdgeEventDao.java index 91aa61baf4..1f8d0c8026 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/edge/JpaBaseEdgeEventDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/edge/JpaBaseEdgeEventDao.java @@ -47,7 +47,6 @@ import javax.annotation.PreDestroy; import java.util.ArrayList; import java.util.Comparator; import java.util.List; -import java.util.Objects; import java.util.UUID; import java.util.concurrent.TimeUnit; import java.util.function.Function; @@ -151,8 +150,9 @@ public class JpaBaseEdgeEventDao extends JpaAbstractDao save(EdgeEventEntity entity) { @@ -227,4 +227,15 @@ public class JpaBaseEdgeEventDao extends JpaAbstractDao { - partitioningRepository.createPartitionIfNotExists(ModelConstants.NOTIFICATION_TABLE_NAME, - entity.getCreatedTime(), TimeUnit.HOURS.toMillis(partitionSizeInHours)); - }); - } - @Override public PageData findUnreadByRecipientIdAndPageLink(TenantId tenantId, UserId recipientId, PageLink pageLink) { return DaoUtil.toPageData(notificationRepository.findByRecipientIdAndStatusNot(recipientId.getId(), NotificationStatus.READ, @@ -110,6 +100,17 @@ public class JpaNotificationDao extends JpaAbstractDao getEntityClass() { return NotificationEntity.class; diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/ota/JpaOtaPackageInfoDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/ota/JpaOtaPackageInfoDao.java index 8a65bee2e1..099aebb816 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/ota/JpaOtaPackageInfoDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/ota/JpaOtaPackageInfoDao.java @@ -19,6 +19,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.stereotype.Component; +import org.springframework.transaction.annotation.Transactional; import org.thingsboard.server.common.data.OtaPackageInfo; import org.thingsboard.server.common.data.id.DeviceProfileId; import org.thingsboard.server.common.data.id.OtaPackageId; @@ -58,6 +59,7 @@ public class JpaOtaPackageInfoDao extends JpaAbstractDao