From 1a5f9bc16d4bf1b4caf740658e0be738404d5a70 Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Mon, 23 Jan 2023 11:48:44 +0200 Subject: [PATCH] Minor refactoring --- .../main/data/upgrade/3.4.3/schema_update.sql | 4 +- .../DefaultNotificationCenter.java | 128 +++++++++--------- .../DefaultNotificationCommandsHandler.java | 6 +- .../server/dao/model/ModelConstants.java | 1 - .../dao/model/sql/AbstractAlarmEntity.java | 4 - .../sql/NotificationRequestInfoEntity.java | 25 +++- .../JpaNotificationRequestDao.java | 5 +- .../notification/JpaNotificationRuleDao.java | 2 +- .../NotificationRequestRepository.java | 12 +- .../NotificationRuleRepository.java | 8 +- .../query/DefaultAlarmQueryRepository.java | 1 - .../main/resources/sql/schema-entities.sql | 5 +- 12 files changed, 112 insertions(+), 89 deletions(-) diff --git a/application/src/main/data/upgrade/3.4.3/schema_update.sql b/application/src/main/data/upgrade/3.4.3/schema_update.sql index 55a972cc42..05a2e643fa 100644 --- a/application/src/main/data/upgrade/3.4.3/schema_update.sql +++ b/application/src/main/data/upgrade/3.4.3/schema_update.sql @@ -74,13 +74,11 @@ CREATE TABLE IF NOT EXISTS notification ( type VARCHAR(32) NOT NULL, subject VARCHAR(255), text VARCHAR(1000) NOT NULL, - additional_config VARCHAR(500), + additional_config VARCHAR(1000), info VARCHAR(1000), status VARCHAR(32) ) PARTITION BY RANGE (created_time); CREATE INDEX IF NOT EXISTS idx_notification_id_recipient_id ON notification(id, recipient_id); CREATE INDEX IF NOT EXISTS idx_notification_recipient_id_status_created_time ON notification(recipient_id, status, created_time DESC); -ALTER TABLE alarm ADD COLUMN IF NOT EXISTS notification_rule_id UUID; - ALTER TABLE tb_user ADD COLUMN IF NOT EXISTS phone VARCHAR(255); diff --git a/application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationCenter.java b/application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationCenter.java index 032add9d82..54726044b3 100644 --- a/application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationCenter.java +++ b/application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationCenter.java @@ -58,6 +58,7 @@ import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; import org.thingsboard.server.service.executors.DbCallbackExecutorService; +import org.thingsboard.server.service.executors.NotificationExecutorService; import org.thingsboard.server.service.notification.channels.NotificationChannel; import org.thingsboard.server.service.subscription.TbSubscriptionUtils; import org.thingsboard.server.service.telemetry.AbstractSubscriptionService; @@ -85,6 +86,7 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple private final NotificationService notificationService; private final NotificationTemplateService notificationTemplateService; private final NotificationSettingsService notificationSettingsService; + private final NotificationExecutorService notificationExecutor; private final DbCallbackExecutorService dbCallbackExecutorService; private final NotificationsTopicService notificationsTopicService; private final TbQueueProducerProvider producerProvider; @@ -124,74 +126,76 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple notificationRequest.setStatus(NotificationRequestStatus.PROCESSING); NotificationRequest savedNotificationRequest = notificationRequestService.saveNotificationRequest(tenantId, notificationRequest); - NotificationProcessingContext ctx = NotificationProcessingContext.builder() - .tenantId(tenantId) - .request(savedNotificationRequest) - .settings(settings) - .template(notificationTemplate) - .build(); - ctx.init(); - - Set deliveryMethods = ctx.getDeliveryMethods(); - List> results = new ArrayList<>(); - - for (UUID targetId : notificationRequest.getTargets()) { - DaoUtil.processBatches(pageLink -> { - return notificationTargetService.findRecipientsForNotificationTarget(tenantId, ctx.getCustomerId(), new NotificationTargetId(targetId), pageLink); - }, 200, recipientsBatch -> { - for (NotificationDeliveryMethod deliveryMethod : deliveryMethods) { - if (deliveryMethod.isStandalone()) continue; - - List recipients = recipientsBatch.getData(); - log.debug("Sending {} notifications for request {} to recipients batch ({})", deliveryMethod, savedNotificationRequest.getId(), recipients.size()); - NotificationChannel notificationChannel = channels.get(deliveryMethod); - for (User recipient : recipients) { - ListenableFuture resultFuture = process(notificationChannel, recipient, ctx); - DonAsynchron.withCallback(resultFuture, result -> { - ctx.getStats().reportSent(deliveryMethod, recipient); - }, error -> { - ctx.getStats().reportError(deliveryMethod, error, recipient); - }); - results.add(resultFuture); + notificationExecutor.submit(() -> { + NotificationProcessingContext ctx = NotificationProcessingContext.builder() + .tenantId(tenantId) + .request(savedNotificationRequest) + .settings(settings) + .template(notificationTemplate) + .build(); + ctx.init(); + + Set deliveryMethods = ctx.getDeliveryMethods(); + List> results = new ArrayList<>(); + + for (UUID targetId : notificationRequest.getTargets()) { + DaoUtil.processBatches(pageLink -> { + return notificationTargetService.findRecipientsForNotificationTarget(tenantId, ctx.getCustomerId(), new NotificationTargetId(targetId), pageLink); + }, 200, recipientsBatch -> { + for (NotificationDeliveryMethod deliveryMethod : deliveryMethods) { + if (deliveryMethod.isStandalone()) continue; + + List recipients = recipientsBatch.getData(); + log.debug("Sending {} notifications for request {} to recipients batch ({})", deliveryMethod, savedNotificationRequest.getId(), recipients.size()); + NotificationChannel notificationChannel = channels.get(deliveryMethod); + for (User recipient : recipients) { + ListenableFuture resultFuture = process(notificationChannel, recipient, ctx); + DonAsynchron.withCallback(resultFuture, result -> { + ctx.getStats().reportSent(deliveryMethod, recipient); + }, error -> { + ctx.getStats().reportError(deliveryMethod, error, recipient); + }); + results.add(resultFuture); + } } - } - }); - } - for (NotificationDeliveryMethod deliveryMethod : deliveryMethods) { - if (deliveryMethod.isStandalone()) { - NotificationChannel notificationChannel = channels.get(deliveryMethod); - ListenableFuture resultFuture = process(notificationChannel, null, ctx); - DonAsynchron.withCallback(resultFuture, result -> { - ctx.getStats().reportSent(deliveryMethod, null); - }, error -> { - ctx.getStats().reportError(deliveryMethod, error, null); }); - results.add(resultFuture); } - } - - Futures.whenAllComplete(results).run(() -> { - NotificationRequestStats stats = ctx.getStats(); - try { - notificationRequestService.updateNotificationRequest(tenantId, savedNotificationRequest.getId(), - NotificationRequestStatus.SENT, stats); - } catch (Exception e) { - log.error("Failed to update stats for notification request {}", savedNotificationRequest.getId(), e); + for (NotificationDeliveryMethod deliveryMethod : deliveryMethods) { + if (deliveryMethod.isStandalone()) { + NotificationChannel notificationChannel = channels.get(deliveryMethod); + ListenableFuture resultFuture = process(notificationChannel, null, ctx); + DonAsynchron.withCallback(resultFuture, result -> { + ctx.getStats().reportSent(deliveryMethod, null); + }, error -> { + ctx.getStats().reportError(deliveryMethod, error, null); + }); + results.add(resultFuture); + } } - UserId senderId = notificationRequest.getSenderId(); - if (senderId != null) { - if (stats.getErrors().isEmpty()) { - int sent = stats.getSent().values().stream().mapToInt(AtomicInteger::get).sum(); - sendBasicNotification(tenantId, senderId, "Notifications sent", - "All notifications were successfully sent (" + sent + ")"); - } else { - int failures = stats.getErrors().values().stream().mapToInt(Map::size).sum(); - sendBasicNotification(tenantId, senderId, "Notification failure", - "Some notifications were not sent (" + failures + ")"); // TODO: 'Go to request' button + Futures.whenAllComplete(results).run(() -> { + NotificationRequestStats stats = ctx.getStats(); + try { + notificationRequestService.updateNotificationRequest(tenantId, savedNotificationRequest.getId(), + NotificationRequestStatus.SENT, stats); + } catch (Exception e) { + log.error("Failed to update stats for notification request {}", savedNotificationRequest.getId(), e); } - } - }, dbCallbackExecutorService); + + UserId senderId = notificationRequest.getSenderId(); + if (senderId != null) { + if (stats.getErrors().isEmpty()) { + int sent = stats.getSent().values().stream().mapToInt(AtomicInteger::get).sum(); + sendBasicNotification(tenantId, senderId, "Notifications sent", + "All notifications were successfully sent (" + sent + ")"); + } else { + int failures = stats.getErrors().values().stream().mapToInt(Map::size).sum(); + sendBasicNotification(tenantId, senderId, "Notification failure", + "Some notifications were not sent (" + failures + ")"); // TODO: 'Go to request' button + } + } + }, dbCallbackExecutorService); + }); return savedNotificationRequest; } diff --git a/application/src/main/java/org/thingsboard/server/service/ws/notification/DefaultNotificationCommandsHandler.java b/application/src/main/java/org/thingsboard/server/service/ws/notification/DefaultNotificationCommandsHandler.java index 3dc3670b94..50146ef206 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/notification/DefaultNotificationCommandsHandler.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/notification/DefaultNotificationCommandsHandler.java @@ -200,7 +200,11 @@ public class DefaultNotificationCommandsHandler implements NotificationCommandsH } case UPDATED: { if (update.getUpdatedStatus() == NotificationStatus.READ) { - subscription.getUnreadCounter().decrementAndGet(); + if (update.isAllNotifications()) { + fetchUnreadNotificationsCount(subscription); + } else { + subscription.getUnreadCounter().decrementAndGet(); + } sendUpdate(subscription.getSessionId(), subscription.createUpdate()); } break; diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java b/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java index ad548b93f9..72b564bfec 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java +++ b/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java @@ -295,7 +295,6 @@ public class ModelConstants { public static final String ALARM_PROPAGATE_TO_OWNER_PROPERTY = "propagate_to_owner"; public static final String ALARM_PROPAGATE_TO_TENANT_PROPERTY = "propagate_to_tenant"; public static final String ALARM_PROPAGATE_RELATION_TYPES = "propagate_relation_types"; - public static final String ALARM_NOTIFICATION_RULE_ID = "notification_rule_id"; public static final String ALARM_BY_ID_VIEW_NAME = "alarm_by_id"; diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sql/AbstractAlarmEntity.java b/dao/src/main/java/org/thingsboard/server/dao/model/sql/AbstractAlarmEntity.java index 2982132dec..4633565563 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/model/sql/AbstractAlarmEntity.java +++ b/dao/src/main/java/org/thingsboard/server/dao/model/sql/AbstractAlarmEntity.java @@ -48,7 +48,6 @@ import static org.thingsboard.server.dao.model.ModelConstants.ALARM_ACK_TS_PROPE import static org.thingsboard.server.dao.model.ModelConstants.ALARM_CLEAR_TS_PROPERTY; import static org.thingsboard.server.dao.model.ModelConstants.ALARM_CUSTOMER_ID_PROPERTY; import static org.thingsboard.server.dao.model.ModelConstants.ALARM_END_TS_PROPERTY; -import static org.thingsboard.server.dao.model.ModelConstants.ALARM_NOTIFICATION_RULE_ID; import static org.thingsboard.server.dao.model.ModelConstants.ALARM_ORIGINATOR_ID_PROPERTY; import static org.thingsboard.server.dao.model.ModelConstants.ALARM_ORIGINATOR_TYPE_PROPERTY; import static org.thingsboard.server.dao.model.ModelConstants.ALARM_PROPAGATE_PROPERTY; @@ -118,9 +117,6 @@ public abstract class AbstractAlarmEntity extends BaseSqlEntity @Column(name = ALARM_PROPAGATE_RELATION_TYPES) private String propagateRelationTypes; - @Column(name = ALARM_NOTIFICATION_RULE_ID) - private UUID notificationRuleId; - public AbstractAlarmEntity() { super(); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sql/NotificationRequestInfoEntity.java b/dao/src/main/java/org/thingsboard/server/dao/model/sql/NotificationRequestInfoEntity.java index 64115fceda..4486b8821d 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/model/sql/NotificationRequestInfoEntity.java +++ b/dao/src/main/java/org/thingsboard/server/dao/model/sql/NotificationRequestInfoEntity.java @@ -19,11 +19,16 @@ import com.fasterxml.jackson.databind.JsonNode; import lombok.Data; import lombok.EqualsAndHashCode; import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod; +import org.thingsboard.server.common.data.notification.NotificationRequest; import org.thingsboard.server.common.data.notification.NotificationRequestInfo; import org.thingsboard.server.common.data.notification.template.NotificationTemplateConfig; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.stream.Collectors; @Data @@ -41,11 +46,21 @@ public class NotificationRequestInfoEntity extends NotificationRequestEntity { @Override public NotificationRequestInfo toData() { - List deliveryMethods = fromJson(templateConfig, NotificationTemplateConfig.class) - .getDeliveryMethodsTemplates().entrySet().stream() - .filter(entry -> entry.getValue().isEnabled()) - .map(Map.Entry::getKey).collect(Collectors.toList()); - return new NotificationRequestInfo(super.toData(), templateName, deliveryMethods); + NotificationRequest request = super.toData(); + List deliveryMethods; + if (templateConfig != null) { + deliveryMethods = fromJson(templateConfig, NotificationTemplateConfig.class) + .getDeliveryMethodsTemplates().entrySet().stream() + .filter(entry -> entry.getValue().isEnabled()) + .map(Map.Entry::getKey).collect(Collectors.toList()); + } else if (request.getStats() != null) { + Set methods = new HashSet<>(request.getStats().getSent().keySet()); + methods.addAll(request.getStats().getErrors().keySet()); + deliveryMethods = new ArrayList<>(methods); + } else { + deliveryMethods = Collections.emptyList(); + } + return new NotificationRequestInfo(request, templateName, deliveryMethods); } } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRequestDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRequestDao.java index e5403286ee..4f56eeeb1b 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRequestDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRequestDao.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.dao.sql.notification; +import com.google.common.base.Strings; import lombok.RequiredArgsConstructor; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.stereotype.Component; @@ -60,8 +61,8 @@ public class JpaNotificationRequestDao extends JpaAbstractDao findInfosByTenantIdAndOriginatorTypeAndPageLink(TenantId tenantId, EntityType originatorType, PageLink pageLink) { - return DaoUtil.pageToPageData(notificationRequestRepository.findInfosByTenantIdAndOriginatorEntityType(getId(tenantId, true), - originatorType, DaoUtil.toPageable(pageLink))).mapData(NotificationRequestInfoEntity::toData); + return DaoUtil.pageToPageData(notificationRequestRepository.findInfosByTenantIdAndOriginatorEntityTypeAndSearchText(getId(tenantId, true), + originatorType, Strings.nullToEmpty(pageLink.getTextSearch()), DaoUtil.toPageable(pageLink))).mapData(NotificationRequestInfoEntity::toData); } @Override diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRuleDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRuleDao.java index e967ca318e..8341002623 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRuleDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRuleDao.java @@ -46,7 +46,7 @@ public class JpaNotificationRuleDao extends JpaAbstractDao findByTenantIdAndPageLink(TenantId tenantId, PageLink pageLink) { - return DaoUtil.toPageData(notificationRuleRepository.findByTenantIdAndNameContainingIgnoreCase(getId(tenantId, true), + return DaoUtil.toPageData(notificationRuleRepository.findByTenantIdAndSearchText(getId(tenantId, true), Strings.nullToEmpty(pageLink.getTextSearch()), DaoUtil.toPageable(pageLink))); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRequestRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRequestRepository.java index 8efbdfb569..cd615bf6a0 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRequestRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRequestRepository.java @@ -36,14 +36,16 @@ import java.util.UUID; public interface NotificationRequestRepository extends JpaRepository { String REQUEST_INFO_QUERY = "SELECT new org.thingsboard.server.dao.model.sql.NotificationRequestInfoEntity(r, t.name, t.configuration) " + - "FROM NotificationRequestEntity r INNER JOIN NotificationTemplateEntity t ON r.templateId = t.id"; + "FROM NotificationRequestEntity r LEFT JOIN NotificationTemplateEntity t ON r.templateId = t.id"; Page findByTenantIdAndOriginatorEntityType(UUID tenantId, EntityType originatorType, Pageable pageable); - @Query(REQUEST_INFO_QUERY + " WHERE r.tenantId = :tenantId AND r.originatorEntityType = :originatorType") - Page findInfosByTenantIdAndOriginatorEntityType(@Param("tenantId") UUID tenantId, - @Param("originatorType") EntityType originatorType, - Pageable pageable); + @Query(REQUEST_INFO_QUERY + " WHERE r.tenantId = :tenantId AND r.originatorEntityType = :originatorType " + + "AND (:searchText = '' OR (t.name IS NOT NULL AND lower(t.name) LIKE lower(concat('%', :searchText, '%'))))") + Page findInfosByTenantIdAndOriginatorEntityTypeAndSearchText(@Param("tenantId") UUID tenantId, + @Param("originatorType") EntityType originatorType, + @Param("searchText") String searchText, + Pageable pageable); @Query(REQUEST_INFO_QUERY + " WHERE r.id = :id") NotificationRequestInfoEntity findInfoById(@Param("id") UUID id); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRuleRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRuleRepository.java index 4fcc22dd69..27c7753ede 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRuleRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRuleRepository.java @@ -18,6 +18,8 @@ package org.thingsboard.server.dao.sql.notification; import org.springframework.data.domain.Page; import org.springframework.data.domain.Pageable; import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.query.Param; import org.springframework.stereotype.Repository; import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType; import org.thingsboard.server.dao.model.sql.NotificationRuleEntity; @@ -28,7 +30,11 @@ import java.util.UUID; @Repository public interface NotificationRuleRepository extends JpaRepository { - Page findByTenantIdAndNameContainingIgnoreCase(UUID tenantId, String searchText, Pageable pageable); + @Query("SELECT r FROM NotificationRuleEntity r WHERE r.tenantId = :tenantId " + + "AND lower(r.name) LIKE lower(concat('%', :searchText, '%')) ") + Page findByTenantIdAndSearchText(@Param("tenantId") UUID tenantId, + @Param("searchText") String searchText, + Pageable pageable); boolean existsByRecipientsConfigContaining(String string); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultAlarmQueryRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultAlarmQueryRepository.java index ce48e9ec95..d91a8c5fc4 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultAlarmQueryRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultAlarmQueryRepository.java @@ -104,7 +104,6 @@ public class DefaultAlarmQueryRepository implements AlarmQueryRepository { " a.tenant_id as tenant_id, " + " a.customer_id as customer_id, " + " a.propagate_relation_types as propagate_relation_types, " + - " a.notification_rule_id as notification_rule_id, " + " a.type as type," + SELECT_ORIGINATOR_NAME + ", "; private static final String JOIN_ENTITY_ALARMS = "inner join entity_alarm ea on a.id = ea.alarm_id"; diff --git a/dao/src/main/resources/sql/schema-entities.sql b/dao/src/main/resources/sql/schema-entities.sql index 3c7e6f83f1..1a6ab00d2b 100644 --- a/dao/src/main/resources/sql/schema-entities.sql +++ b/dao/src/main/resources/sql/schema-entities.sql @@ -58,8 +58,7 @@ CREATE TABLE IF NOT EXISTS alarm ( propagate_relation_types varchar, type varchar(255), propagate_to_owner boolean, - propagate_to_tenant boolean, - notification_rule_id uuid + propagate_to_tenant boolean ); CREATE TABLE IF NOT EXISTS entity_alarm ( @@ -834,7 +833,7 @@ CREATE TABLE IF NOT EXISTS notification ( type VARCHAR(32) NOT NULL, subject VARCHAR(255), text VARCHAR(1000) NOT NULL, - additional_config VARCHAR(500), + additional_config VARCHAR(1000), info VARCHAR(1000), status VARCHAR(32) ) PARTITION BY RANGE (created_time);