From 12ae902cfc7b982ca442e7667d88ff7d3fd5856c Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Sun, 6 Nov 2022 18:13:02 +0200 Subject: [PATCH] Notification rules initial implementation --- .../main/data/upgrade/3.4.2/schema_update.sql | 11 +- .../server/actors/ActorSystemContext.java | 7 +- .../actors/ruleChain/DefaultTbContext.java | 6 +- .../controller/NotificationController.java | 17 ++- ...aultNotificationRuleProcessingService.java | 138 ++++++++++++++++++ ...faultNotificationSubscriptionService.java} | 82 +++++++---- .../NotificationRuleProcessingService.java | 28 ++++ .../NotificationSubscriptionService.java | 6 +- .../queue/DefaultTbCoreConsumerService.java | 44 +++--- .../DefaultSubscriptionManagerService.java | 23 ++- .../SubscriptionManagerService.java | 8 +- .../subscription/TbSubscriptionUtils.java | 31 ++-- .../AbstractSubscriptionService.java | 6 +- .../DefaultAlarmSubscriptionService.java | 8 +- .../DefaultTelemetrySubscriptionService.java | 8 +- .../DefaultNotificationCommandsHandler.java | 100 +++++++++---- .../sub/NotificationRequestUpdate.java | 33 +++++ .../notification/sub/NotificationUpdate.java | 31 ++++ .../sub/NotificationsSubscriptionUpdate.java | 19 ++- common/cluster-api/src/main/proto/queue.proto | 14 +- .../notification/NotificationRuleService.java | 26 ++++ .../dao/notification/NotificationService.java | 22 ++- .../server/common/data/alarm/Alarm.java | 3 + .../device/profile/DeviceProfileAlarm.java | 3 + .../common/data/id/NotificationRuleId.java | 36 +++++ .../data/notification/NotificationInfo.java | 21 ++- .../notification/NotificationRequest.java | 11 ++ .../NotificationRequestConfig.java | 2 +- .../NotificationRequestStatus.java | 21 +++ .../NonConfirmedNotificationEscalation.java | 15 +- .../notification/rule/NotificationRule.java | 31 ++-- .../server/dao/model/ModelConstants.java | 7 + .../dao/model/sql/AbstractAlarmEntity.java | 8 + .../model/sql/NotificationRequestEntity.java | 19 +++ .../dao/model/sql/NotificationRuleEntity.java | 65 +++++++++ .../DefaultNotificationRuleService.java | 35 +++++ .../DefaultNotificationService.java | 40 ++--- .../dao/notification/NotificationDao.java | 6 + .../notification/NotificationRequestDao.java | 6 + .../dao/notification/NotificationRuleDao.java | 22 +++ .../sql/notification/JpaNotificationDao.java | 13 ++ .../JpaNotificationRequestDao.java | 8 + .../notification/JpaNotificationRuleDao.java | 46 ++++++ .../notification/NotificationRepository.java | 8 +- .../NotificationRequestRepository.java | 3 + .../NotificationRuleRepository.java | 26 ++++ .../resources/sql/schema-entities-idx.sql | 4 +- .../main/resources/sql/schema-entities.sql | 5 +- .../api/RuleEngineNotificationService.java | 25 ++++ .../rule/engine/api/TbContext.java | 3 +- .../notification/TbNotificationNode.java | 2 +- .../rule/engine/profile/AlarmState.java | 1 + .../rule/engine/profile/DeviceState.java | 1 + 53 files changed, 964 insertions(+), 200 deletions(-) create mode 100644 application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationRuleProcessingService.java rename application/src/main/java/org/thingsboard/server/service/notification/{DefaultNotificationProcessingService.java => DefaultNotificationSubscriptionService.java} (61%) create mode 100644 application/src/main/java/org/thingsboard/server/service/notification/NotificationRuleProcessingService.java rename common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationProcessingService.java => application/src/main/java/org/thingsboard/server/service/notification/NotificationSubscriptionService.java (86%) create mode 100644 application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationRequestUpdate.java create mode 100644 application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationUpdate.java create mode 100644 common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationRuleService.java create mode 100644 common/data/src/main/java/org/thingsboard/server/common/data/id/NotificationRuleId.java create mode 100644 common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequestStatus.java create mode 100644 dao/src/main/java/org/thingsboard/server/dao/model/sql/NotificationRuleEntity.java create mode 100644 dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationRuleService.java create mode 100644 dao/src/main/java/org/thingsboard/server/dao/notification/NotificationRuleDao.java create mode 100644 dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRuleDao.java create mode 100644 dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRuleRepository.java create mode 100644 rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineNotificationService.java diff --git a/application/src/main/data/upgrade/3.4.2/schema_update.sql b/application/src/main/data/upgrade/3.4.2/schema_update.sql index 77f078e67f..7294916445 100644 --- a/application/src/main/data/upgrade/3.4.2/schema_update.sql +++ b/application/src/main/data/upgrade/3.4.2/schema_update.sql @@ -24,6 +24,10 @@ CREATE TABLE IF NOT EXISTS notification_target ( ); CREATE INDEX IF NOT EXISTS idx_notification_target_tenant_id_and_created_time ON notification_target(tenant_id, created_time DESC); +CREATE TABLE IF NOT EXISTS notification_rule ( + id UUID NOT NULL CONSTRAINT notification_rule_pkey PRIMARY KEY +); + CREATE TABLE IF NOT EXISTS notification_request ( id UUID NOT NULL CONSTRAINT notification_request_pkey PRIMARY KEY, created_time BIGINT NOT NULL, @@ -33,7 +37,10 @@ CREATE TABLE IF NOT EXISTS notification_request ( text_template VARCHAR NOT NULL, notification_info VARCHAR(1000), notification_severity VARCHAR(32), - additional_config VARCHAR(1000) + additional_config VARCHAR(1000), + status VARCHAR(32), + rule_id UUID NULL CONSTRAINT fk_notification_request_rule_id REFERENCES notification_rule(id), + alarm_id UUID ); CREATE INDEX IF NOT EXISTS idx_notification_request_tenant_id_and_created_time ON notification_request(tenant_id, created_time DESC); @@ -50,5 +57,5 @@ CREATE TABLE IF NOT EXISTS notification ( status VARCHAR(32) ) PARTITION BY RANGE (created_time); CREATE INDEX IF NOT EXISTS idx_notification_id ON notification(id); +CREATE INDEX IF NOT EXISTS idx_notification_notification_request_id ON notification(request_id); CREATE INDEX IF NOT EXISTS idx_notification_recipient_id_and_created_time ON notification(recipient_id, created_time DESC); -CREATE INDEX IF NOT EXISTS idx_notification_recipient_id_and_status_and_created_time ON notification(recipient_id, status, created_time DESC); diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index 2a574f5788..27d3d401ff 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -15,9 +15,7 @@ */ package org.thingsboard.server.actors; -import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.databind.node.ObjectNode; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; @@ -32,6 +30,7 @@ import org.springframework.data.redis.core.RedisTemplate; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import org.thingsboard.rule.engine.api.MailService; +import org.thingsboard.rule.engine.api.RuleEngineNotificationService; import org.thingsboard.rule.engine.api.SmsService; import org.thingsboard.rule.engine.api.sms.SmsSenderFactory; import org.thingsboard.server.actors.service.ActorService; @@ -64,7 +63,6 @@ import org.thingsboard.server.dao.entityview.EntityViewService; import org.thingsboard.server.dao.event.EventService; import org.thingsboard.server.dao.nosql.CassandraBufferedRateReadExecutor; import org.thingsboard.server.dao.nosql.CassandraBufferedRateWriteExecutor; -import org.thingsboard.server.dao.notification.NotificationProcessingService; import org.thingsboard.server.dao.ota.OtaPackageService; import org.thingsboard.server.dao.queue.QueueService; import org.thingsboard.server.dao.relation.RelationService; @@ -106,7 +104,6 @@ import javax.annotation.PostConstruct; import java.io.IOException; import java.io.PrintWriter; import java.io.StringWriter; -import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ScheduledExecutorService; @@ -307,7 +304,7 @@ public class ActorSystemContext { @Autowired @Getter - private NotificationProcessingService notificationProcessingService; + private RuleEngineNotificationService notificationService; @Lazy @Autowired(required = false) diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index c0197a0f89..82010dc984 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -27,6 +27,7 @@ import org.thingsboard.rule.engine.api.MailService; import org.thingsboard.rule.engine.api.RuleEngineAlarmService; import org.thingsboard.rule.engine.api.RuleEngineAssetProfileCache; import org.thingsboard.rule.engine.api.RuleEngineDeviceProfileCache; +import org.thingsboard.rule.engine.api.RuleEngineNotificationService; import org.thingsboard.rule.engine.api.RuleEngineRpcService; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; import org.thingsboard.rule.engine.api.ScriptEngine; @@ -79,7 +80,6 @@ import org.thingsboard.server.dao.edge.EdgeService; import org.thingsboard.server.dao.entityview.EntityViewService; import org.thingsboard.server.dao.nosql.CassandraStatementTask; import org.thingsboard.server.dao.nosql.TbResultSetFuture; -import org.thingsboard.server.dao.notification.NotificationProcessingService; import org.thingsboard.server.dao.ota.OtaPackageService; import org.thingsboard.server.dao.queue.QueueService; import org.thingsboard.server.dao.relation.RelationService; @@ -632,8 +632,8 @@ class DefaultTbContext implements TbContext { } @Override - public NotificationProcessingService getNotificationProcessingService() { - return mainCtx.getNotificationProcessingService(); + public RuleEngineNotificationService getNotificationService() { + return mainCtx.getNotificationService(); } @Override diff --git a/application/src/main/java/org/thingsboard/server/controller/NotificationController.java b/application/src/main/java/org/thingsboard/server/controller/NotificationController.java index 1bf771218a..a4f44b589d 100644 --- a/application/src/main/java/org/thingsboard/server/controller/NotificationController.java +++ b/application/src/main/java/org/thingsboard/server/controller/NotificationController.java @@ -39,7 +39,7 @@ import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.dao.notification.NotificationService; import org.thingsboard.server.queue.util.TbCoreComponent; -import org.thingsboard.server.dao.notification.NotificationProcessingService; +import org.thingsboard.server.service.notification.NotificationSubscriptionService; import org.thingsboard.server.service.security.model.SecurityUser; import org.thingsboard.server.service.security.permission.Operation; import org.thingsboard.server.service.security.permission.Resource; @@ -54,7 +54,7 @@ import java.util.UUID; public class NotificationController extends BaseController { private final NotificationService notificationService; - private final NotificationProcessingService notificationProcessingService; + private final NotificationSubscriptionService notificationSubscriptionService; @GetMapping("/notifications") @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") @@ -66,7 +66,7 @@ public class NotificationController extends BaseController { @RequestParam(defaultValue = "false") boolean unreadOnly, @AuthenticationPrincipal SecurityUser user) throws ThingsboardException { PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder); - return notificationService.findNotificationsByUserIdAndReadStatusAndPageLink(user.getTenantId(), user.getId(), unreadOnly, pageLink); + return notificationService.findNotificationsByUserIdAndReadStatus(user.getTenantId(), user.getId(), unreadOnly, pageLink); } @PutMapping("/notification/{id}/read") // or maybe to NotificationUpdateRequest for the future @@ -74,7 +74,7 @@ public class NotificationController extends BaseController { public void markNotificationAsRead(@PathVariable UUID id, @AuthenticationPrincipal SecurityUser user) { NotificationId notificationId = new NotificationId(id); - notificationProcessingService.markNotificationAsRead(user.getTenantId(), user.getId(), notificationId); + notificationSubscriptionService.markNotificationAsRead(user.getTenantId(), user.getId(), notificationId); } // delete notification? @@ -84,8 +84,11 @@ public class NotificationController extends BaseController { public NotificationRequest createNotificationRequest(@RequestBody NotificationRequest notificationRequest, @AuthenticationPrincipal SecurityUser user) throws ThingsboardException { accessControlService.checkPermission(user, Resource.NOTIFICATION_REQUEST, Operation.CREATE, null, notificationRequest); + if (notificationRequest.getId() != null) { + throw new IllegalArgumentException("Notification request cannot be changed. You can delete it and create a new one"); + } try { - NotificationRequest savedNotificationRequest = notificationProcessingService.processNotificationRequest(user.getTenantId(), notificationRequest); + NotificationRequest savedNotificationRequest = notificationSubscriptionService.processNotificationRequest(user.getTenantId(), notificationRequest); logEntityAction(user, EntityType.NOTIFICATION_REQUEST, savedNotificationRequest, ActionType.ADDED); return savedNotificationRequest; } catch (Exception e) { @@ -111,7 +114,7 @@ public class NotificationController extends BaseController { @RequestParam(required = false) String sortOrder, @AuthenticationPrincipal SecurityUser user) throws ThingsboardException { PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder); - return notificationService.findNotificationRequestsByTenantIdAndPageLink(user.getTenantId(), pageLink); + return notificationService.findNotificationRequestsByTenantId(user.getTenantId(), pageLink); } @DeleteMapping("/notification/request/{id}") @@ -121,7 +124,7 @@ public class NotificationController extends BaseController { NotificationRequest notificationRequest = notificationService.findNotificationRequestById(user.getTenantId(), notificationRequestId); accessControlService.checkPermission(user, Resource.NOTIFICATION_REQUEST, Operation.DELETE, notificationRequestId, notificationRequest); try { - notificationProcessingService.deleteNotificationRequest(user.getTenantId(), notificationRequestId); + notificationSubscriptionService.deleteNotificationRequest(user.getTenantId(), notificationRequestId); logEntityAction(user, EntityType.NOTIFICATION_REQUEST, notificationRequest, ActionType.DELETED); } catch (Exception e) { logEntityAction(user, EntityType.NOTIFICATION_REQUEST, notificationRequest, notificationRequest, ActionType.DELETED, e); diff --git a/application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationRuleProcessingService.java b/application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationRuleProcessingService.java new file mode 100644 index 0000000000..7c459b2e80 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationRuleProcessingService.java @@ -0,0 +1,138 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.notification; + +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import org.thingsboard.server.common.data.alarm.Alarm; +import org.thingsboard.server.common.data.id.NotificationRuleId; +import org.thingsboard.server.common.data.id.NotificationTargetId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.notification.NotificationInfo; +import org.thingsboard.server.common.data.notification.NotificationRequest; +import org.thingsboard.server.common.data.notification.NotificationRequestConfig; +import org.thingsboard.server.common.data.notification.NotificationRequestStatus; +import org.thingsboard.server.common.data.notification.NotificationSeverity; +import org.thingsboard.server.common.data.notification.rule.NonConfirmedNotificationEscalation; +import org.thingsboard.server.common.data.notification.rule.NotificationRule; +import org.thingsboard.server.dao.notification.NotificationRuleService; +import org.thingsboard.server.dao.notification.NotificationService; +import org.thingsboard.server.queue.util.TbCoreComponent; +import org.thingsboard.server.service.executors.DbCallbackExecutorService; + +import java.util.List; + +@Service +@TbCoreComponent +@RequiredArgsConstructor +public class DefaultNotificationRuleProcessingService implements NotificationRuleProcessingService { + + private final NotificationRuleService notificationRuleService; + private final NotificationService notificationService; + private final NotificationSubscriptionService notificationSubscriptionService; + private final DbCallbackExecutorService dbCallbackExecutorService; + + @Override + public ListenableFuture onAlarmCreatedOrUpdated(TenantId tenantId, Alarm alarm) { + return processAlarmUpdate(tenantId, alarm); + } + + @Override + public ListenableFuture onAlarmAcknowledged(TenantId tenantId, Alarm alarm) { + return processAlarmUpdate(tenantId, alarm); + } + + private ListenableFuture processAlarmUpdate(TenantId tenantId, Alarm alarm) { + if (alarm.getNotificationRuleId() == null) return Futures.immediateFuture(null); + return dbCallbackExecutorService.submit(() -> { + onAlarmUpdate(tenantId, alarm.getNotificationRuleId(), alarm); + }); + } + + private void onAlarmUpdate(TenantId tenantId, NotificationRuleId notificationRuleId, Alarm alarm) { + List notificationRequests = notificationService.findNotificationRequestsByRuleIdAndAlarmId(tenantId, notificationRuleId, alarm.getId()); + NotificationRule notificationRule = notificationRuleService.findNotificationRuleById(tenantId, notificationRuleId); + + if (notificationRequests.isEmpty()) { // in case it is first notification for alarm, or it was previously acked and now we need to send notifications again + NotificationTargetId initialNotificationTargetId = notificationRule.getInitialNotificationTargetId(); + submitNotificationRequest(tenantId, initialNotificationTargetId, notificationRule, alarm, 0); + + for (NonConfirmedNotificationEscalation escalation : notificationRule.getEscalations()) { + submitNotificationRequest(tenantId, escalation.getNotificationTargetId(), notificationRule, alarm, escalation.getDelayInMinutes()); + } + } else { + if (alarmAcknowledged(alarm)) { + for (NotificationRequest notificationRequest : notificationRequests) { + if (notificationRequest.getStatus() == NotificationRequestStatus.SCHEDULED) { + // using regular service due to no need to send an update to subscription manager + notificationService.deleteNotificationRequestById(tenantId, notificationRequest.getId()); + } else { + notificationSubscriptionService.deleteNotificationRequest(tenantId, notificationRequest.getId()); + // todo: or should we mark already sent notifications as read? + } + } + } else { + NotificationInfo newNotificationInfo = constructNotificationInfo(alarm, notificationRule); + for (NotificationRequest notificationRequest : notificationRequests) { + NotificationInfo previousNotificationInfo = notificationRequest.getNotificationInfo(); + if (!previousNotificationInfo.equals(newNotificationInfo)) { + notificationRequest.setNotificationInfo(newNotificationInfo); + notificationSubscriptionService.updateNotificationRequest(tenantId, notificationRequest); + } + // fixme: no need to send an update event for scheduled requests, only for sent + } + } + } + } + + private boolean alarmAcknowledged(Alarm alarm) { // todo: decide when to consider the alarm processed by notification target (not to escalate then) + return alarm.getStatus().isAck() && alarm.getStatus().isCleared(); + } + + private void submitNotificationRequest(TenantId tenantId, NotificationTargetId targetId, NotificationRule notificationRule, Alarm alarm, int delayInMinutes) { + NotificationRequestConfig config = new NotificationRequestConfig(); + if (delayInMinutes > 0) { + config.setSendingDelayInMinutes(delayInMinutes); + } + NotificationInfo notificationInfo = constructNotificationInfo(alarm, notificationRule); + + NotificationRequest notificationRequest = NotificationRequest.builder() + .tenantId(tenantId) + .targetId(targetId) + .notificationReason("Alarm") + .textTemplate(notificationRule.getNotificationTextTemplate()) // todo: format with alarm vars + .notificationInfo(notificationInfo) + .notificationSeverity(NotificationSeverity.NORMAL) // todo: from alarm severity + .additionalConfig(config) + .ruleId(notificationRule.getId()) + .alarmId(alarm.getId()) + .build(); + notificationSubscriptionService.processNotificationRequest(tenantId, notificationRequest); + } + + private NotificationInfo constructNotificationInfo(Alarm alarm, NotificationRule notificationRule) { + return NotificationInfo.builder() + .alarmId(alarm.getId()) + .alarmType(alarm.getType()) + .alarmOriginator(alarm.getOriginator()) + .alarmSeverity(alarm.getSeverity()) + .alarmStatus(alarm.getStatus()) + .build(); + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationProcessingService.java b/application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationSubscriptionService.java similarity index 61% rename from application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationProcessingService.java rename to application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationSubscriptionService.java index 99ede1fba9..8709f76179 100644 --- a/application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationProcessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationSubscriptionService.java @@ -18,6 +18,7 @@ package org.thingsboard.server.service.notification; import com.google.common.base.Strings; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; +import org.thingsboard.rule.engine.api.RuleEngineNotificationService; import org.thingsboard.rule.engine.api.util.TbNodeUtils; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.User; @@ -28,12 +29,12 @@ import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.notification.Notification; import org.thingsboard.server.common.data.notification.NotificationRequest; import org.thingsboard.server.common.data.notification.NotificationRequestConfig; +import org.thingsboard.server.common.data.notification.NotificationRequestStatus; import org.thingsboard.server.common.data.notification.NotificationStatus; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.dao.DaoUtil; -import org.thingsboard.server.dao.notification.NotificationProcessingService; import org.thingsboard.server.dao.notification.NotificationService; import org.thingsboard.server.dao.notification.NotificationTargetService; import org.thingsboard.server.gen.transport.TransportProtos; @@ -42,6 +43,8 @@ import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.service.executors.DbCallbackExecutorService; import org.thingsboard.server.service.subscription.TbSubscriptionUtils; import org.thingsboard.server.service.telemetry.AbstractSubscriptionService; +import org.thingsboard.server.service.ws.notification.sub.NotificationRequestUpdate; +import org.thingsboard.server.service.ws.notification.sub.NotificationUpdate; import java.util.HashSet; import java.util.Map; @@ -50,18 +53,18 @@ import java.util.UUID; @Service @Slf4j -public class DefaultNotificationProcessingService extends AbstractSubscriptionService implements NotificationProcessingService { +public class DefaultNotificationSubscriptionService extends AbstractSubscriptionService implements NotificationSubscriptionService, RuleEngineNotificationService { private final NotificationTargetService notificationTargetService; private final NotificationService notificationService; private final DbCallbackExecutorService dbCallbackExecutorService; private final NotificationsTopicService notificationsTopicService; - public DefaultNotificationProcessingService(TbClusterService clusterService, PartitionService partitionService, - NotificationTargetService notificationTargetService, - NotificationService notificationService, - DbCallbackExecutorService dbCallbackExecutorService, - NotificationsTopicService notificationsTopicService) { + public DefaultNotificationSubscriptionService(TbClusterService clusterService, PartitionService partitionService, + NotificationTargetService notificationTargetService, + NotificationService notificationService, + DbCallbackExecutorService dbCallbackExecutorService, + NotificationsTopicService notificationsTopicService) { super(clusterService, partitionService); this.notificationTargetService = notificationTargetService; this.notificationService = notificationService; @@ -72,13 +75,20 @@ public class DefaultNotificationProcessingService extends AbstractSubscriptionSe @Override public NotificationRequest processNotificationRequest(TenantId tenantId, NotificationRequest notificationRequest) { notificationRequest.setTenantId(tenantId); - NotificationRequest savedNotificationRequest = notificationService.createNotificationRequest(tenantId, notificationRequest); - if (notificationRequest.getAdditionalConfig() != null) { NotificationRequestConfig config = notificationRequest.getAdditionalConfig(); - // todo: delayed sending; check all delayed notification requests on start up, schedule send + if (config.getSendingDelayInMinutes() > 0) { + notificationRequest.setStatus(NotificationRequestStatus.SCHEDULED); + + // todo: delayed sending; check all delayed notification requests on start up, schedule send + } + } + if (notificationRequest.getStatus() == null) { + notificationRequest.setStatus(NotificationRequestStatus.PROCESSED); } + NotificationRequest savedNotificationRequest = notificationService.saveNotificationRequest(tenantId, notificationRequest); + DaoUtil.processBatches(pageLink -> { return notificationTargetService.findRecipientsForNotificationTarget(tenantId, notificationRequest.getTargetId(), pageLink); }, 100, recipients -> { @@ -99,7 +109,7 @@ public class DefaultNotificationProcessingService extends AbstractSubscriptionSe @Override public void markNotificationAsRead(TenantId tenantId, UserId recipientId, NotificationId notificationId) { - boolean updated = notificationService.updateNotificationStatus(tenantId, recipientId, notificationId, NotificationStatus.READ); + boolean updated = notificationService.markNotificationAsRead(tenantId, recipientId, notificationId); if (updated) { Notification notification = notificationService.findNotificationById(tenantId, notificationId); onNotificationUpdate(tenantId, recipientId, notification, false); @@ -108,8 +118,22 @@ public class DefaultNotificationProcessingService extends AbstractSubscriptionSe @Override public void deleteNotificationRequest(TenantId tenantId, NotificationRequestId notificationRequestId) { - notificationService.deleteNotificationRequest(tenantId, notificationRequestId); - onNotificationRequestDeleted(tenantId, notificationRequestId); + notificationService.deleteNotificationRequestById(tenantId, notificationRequestId); + onNotificationRequestUpdate(tenantId, NotificationRequestUpdate.builder() + .notificationRequestId(notificationRequestId) + .deleted(true) + .build()); + } + + @Override + public void updateNotificationRequest(TenantId tenantId, NotificationRequest notificationRequest) { + notificationService.saveNotificationRequest(tenantId, notificationRequest); + notificationService.updateNotificationsInfosByRequestId(tenantId, notificationRequest.getId(), notificationRequest.getNotificationInfo()); + onNotificationRequestUpdate(tenantId, NotificationRequestUpdate.builder() + .notificationRequestId(notificationRequest.getId()) + .notificationInfo(notificationRequest.getNotificationInfo()) + .deleted(false) + .build()); } private Notification createNotification(User recipient, NotificationRequest notificationRequest) { @@ -122,7 +146,7 @@ public class DefaultNotificationProcessingService extends AbstractSubscriptionSe .severity(notificationRequest.getNotificationSeverity()) .status(NotificationStatus.SENT) .build(); - return notificationService.createNotification(recipient.getTenantId(), notification); + return notificationService.saveNotification(recipient.getTenantId(), notification); } private String formatNotificationText(String template, User recipient) { @@ -135,20 +159,28 @@ public class DefaultNotificationProcessingService extends AbstractSubscriptionSe } private void onNotificationUpdate(TenantId tenantId, UserId recipientId, Notification notification, boolean isNew) { - forwardToSubscriptionManagerServiceOrSendToCore(tenantId, recipientId, subscriptionManagerService -> { - subscriptionManagerService.onNotificationUpdate(tenantId, recipientId, notification, isNew, TbCallback.EMPTY); - }, () -> { - return TbSubscriptionUtils.notificationUpdateToProto(tenantId, recipientId, notification, isNew); + NotificationUpdate notificationUpdate = NotificationUpdate.builder() + .notification(notification) + .isNew(isNew) + .build(); + wsCallBackExecutor.submit(() -> { + forwardToSubscriptionManagerService(tenantId, recipientId, subscriptionManagerService -> { + subscriptionManagerService.onNotificationUpdate(tenantId, recipientId, notificationUpdate, TbCallback.EMPTY); + }, () -> { + return TbSubscriptionUtils.notificationUpdateToProto(tenantId, recipientId, notificationUpdate); + }); }); } - public void onNotificationRequestDeleted(TenantId tenantId, NotificationRequestId notificationRequestId) { - TransportProtos.ToCoreMsg notificationRequestDeletedProto = TbSubscriptionUtils.notificationRequestDeletedToProto(tenantId, notificationRequestId); - Set coreServices = new HashSet<>(partitionService.getAllServiceIds(ServiceType.TB_CORE)); - for (String serviceId : coreServices) { - TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceId); - clusterService.pushMsgToCore(tpi, UUID.randomUUID(), notificationRequestDeletedProto, null); - } + private void onNotificationRequestUpdate(TenantId tenantId, NotificationRequestUpdate update) { + wsCallBackExecutor.submit(() -> { + TransportProtos.ToCoreMsg notificationRequestDeletedProto = TbSubscriptionUtils.notificationRequestUpdateToProto(tenantId, update); + Set coreServices = new HashSet<>(partitionService.getAllServiceIds(ServiceType.TB_CORE)); + for (String serviceId : coreServices) { + TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, serviceId); + clusterService.pushMsgToCore(tpi, UUID.randomUUID(), notificationRequestDeletedProto, null); + } + }); } @Override diff --git a/application/src/main/java/org/thingsboard/server/service/notification/NotificationRuleProcessingService.java b/application/src/main/java/org/thingsboard/server/service/notification/NotificationRuleProcessingService.java new file mode 100644 index 0000000000..4f3d0f7d9e --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/notification/NotificationRuleProcessingService.java @@ -0,0 +1,28 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.notification; + +import com.google.common.util.concurrent.ListenableFuture; +import org.thingsboard.server.common.data.alarm.Alarm; +import org.thingsboard.server.common.data.id.TenantId; + +public interface NotificationRuleProcessingService { + + ListenableFuture onAlarmCreatedOrUpdated(TenantId tenantId, Alarm alarm); + + ListenableFuture onAlarmAcknowledged(TenantId tenantId, Alarm alarm); + +} diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationProcessingService.java b/application/src/main/java/org/thingsboard/server/service/notification/NotificationSubscriptionService.java similarity index 86% rename from common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationProcessingService.java rename to application/src/main/java/org/thingsboard/server/service/notification/NotificationSubscriptionService.java index b25d4943b7..902ad39470 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationProcessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/notification/NotificationSubscriptionService.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.dao.notification; +package org.thingsboard.server.service.notification; import org.thingsboard.server.common.data.id.NotificationId; import org.thingsboard.server.common.data.id.NotificationRequestId; @@ -21,7 +21,7 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.notification.NotificationRequest; -public interface NotificationProcessingService { +public interface NotificationSubscriptionService { NotificationRequest processNotificationRequest(TenantId tenantId, NotificationRequest notificationRequest); @@ -29,4 +29,6 @@ public interface NotificationProcessingService { void deleteNotificationRequest(TenantId tenantId, NotificationRequestId notificationRequestId); + void updateNotificationRequest(TenantId tenantId, NotificationRequest notificationRequest); + } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index 1ee7c3cd16..cf93ea7f1d 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -18,6 +18,7 @@ package org.thingsboard.server.service.queue; import lombok.Getter; import lombok.Setter; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.context.event.ApplicationReadyEvent; import org.springframework.scheduling.annotation.Scheduled; @@ -31,6 +32,7 @@ import org.thingsboard.server.common.data.id.NotificationRequestId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.notification.Notification; +import org.thingsboard.server.common.data.notification.NotificationInfo; import org.thingsboard.server.common.data.rpc.RpcError; import org.thingsboard.server.common.msg.MsgType; import org.thingsboard.server.common.msg.TbActorMsg; @@ -80,6 +82,8 @@ import org.thingsboard.server.service.subscription.TbLocalSubscriptionService; import org.thingsboard.server.service.subscription.TbSubscriptionUtils; import org.thingsboard.server.service.sync.vc.GitVersionControlQueueService; import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper; +import org.thingsboard.server.service.ws.notification.sub.NotificationRequestUpdate; +import org.thingsboard.server.service.ws.notification.sub.NotificationUpdate; import org.thingsboard.server.service.ws.notification.sub.NotificationsSubscriptionUpdate; import javax.annotation.PostConstruct; @@ -461,13 +465,17 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService> subscriptionsByEntityId = new ConcurrentHashMap<>(); private final Map> subscriptionsByWsSessionId = new ConcurrentHashMap<>(); private final ConcurrentMap> partitionedSubscriptions = new ConcurrentHashMap<>(); @@ -309,6 +313,7 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene s -> alarm, false ); + notificationRuleProcessingService.onAlarmCreatedOrUpdated(tenantId, alarm); callback.onSuccess(); } @@ -330,13 +335,10 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene } @Override - public void onNotificationUpdate(TenantId tenantId, UserId recipientId, Notification notification, boolean isNew, TbCallback callback) { + public void onNotificationUpdate(TenantId tenantId, UserId recipientId, NotificationUpdate notificationUpdate, TbCallback callback) { Set subscriptions = subscriptionsByEntityId.get(recipientId); if (subscriptions != null) { - NotificationsSubscriptionUpdate subscriptionUpdate = NotificationsSubscriptionUpdate.builder() - .notification(notification) - .isNewNotification(isNew) - .build(); + NotificationsSubscriptionUpdate subscriptionUpdate = new NotificationsSubscriptionUpdate(notificationUpdate); subscriptions.stream() .filter(subscription -> subscription.getType() == TbSubscriptionType.NOTIFICATIONS || subscription.getType() == TbSubscriptionType.NOTIFICATIONS_COUNT) @@ -356,11 +358,8 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene } @Override - public void onNotificationRequestDeleted(TenantId tenantId, NotificationRequestId notificationRequestId, TbCallback callback) { - NotificationsSubscriptionUpdate subscriptionUpdate = NotificationsSubscriptionUpdate.builder() - .notificationRequestDeleted(true) - .notificationRequestId(notificationRequestId) - .build(); + public void onNotificationRequestUpdate(TenantId tenantId, NotificationRequestUpdate notificationRequestUpdate, TbCallback callback) { + NotificationsSubscriptionUpdate subscriptionUpdate = new NotificationsSubscriptionUpdate(notificationRequestUpdate); subscriptionsByEntityId.entrySet().stream() .filter(subEntry -> subEntry.getKey().getEntityType() == EntityType.USER) .flatMap(subEntry -> subEntry.getValue().stream() diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java b/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java index 239f9c94f9..8079bdbcce 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java @@ -18,14 +18,14 @@ package org.thingsboard.server.service.subscription; import org.springframework.context.ApplicationListener; import org.thingsboard.server.common.data.alarm.Alarm; import org.thingsboard.server.common.data.id.EntityId; -import org.thingsboard.server.common.data.id.NotificationRequestId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; -import org.thingsboard.server.common.data.notification.Notification; import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent; +import org.thingsboard.server.service.ws.notification.sub.NotificationRequestUpdate; +import org.thingsboard.server.service.ws.notification.sub.NotificationUpdate; import java.util.List; @@ -49,8 +49,8 @@ public interface SubscriptionManagerService extends ApplicationListener toSubscriptionManagerService, - Supplier toCore) { + protected void forwardToSubscriptionManagerService(TenantId tenantId, EntityId entityId, + Consumer toSubscriptionManagerService, + Supplier toCore) { TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId); if (currentPartitions.contains(tpi)) { if (subscriptionManagerService.isPresent()) { diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultAlarmSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultAlarmSubscriptionService.java index 386904918f..0e455bf9d5 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultAlarmSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultAlarmSubscriptionService.java @@ -37,12 +37,9 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.query.AlarmData; import org.thingsboard.server.common.data.query.AlarmDataQuery; -import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TbCallback; -import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.dao.alarm.AlarmOperationResult; import org.thingsboard.server.dao.alarm.AlarmService; -import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.usagestats.TbApiUsageClient; import org.thingsboard.server.service.apiusage.TbApiUsageStateService; @@ -164,13 +161,14 @@ public class DefaultAlarmSubscriptionService extends AbstractSubscriptionService Alarm alarm = result.getAlarm(); TenantId tenantId = result.getAlarm().getTenantId(); for (EntityId entityId : result.getPropagatedEntitiesList()) { - forwardToSubscriptionManagerServiceOrSendToCore(tenantId, entityId, subscriptionManagerService -> { + forwardToSubscriptionManagerService(tenantId, entityId, subscriptionManagerService -> { subscriptionManagerService.onAlarmUpdate(tenantId, entityId, alarm, TbCallback.EMPTY); }, () -> { return TbSubscriptionUtils.toAlarmUpdateProto(tenantId, entityId, alarm); }); } }); + // todo: handle notification rule } private void onAlarmDeleted(AlarmOperationResult result) { @@ -178,7 +176,7 @@ public class DefaultAlarmSubscriptionService extends AbstractSubscriptionService Alarm alarm = result.getAlarm(); TenantId tenantId = result.getAlarm().getTenantId(); for (EntityId entityId : result.getPropagatedEntitiesList()) { - forwardToSubscriptionManagerServiceOrSendToCore(tenantId, entityId, subscriptionManagerService -> { + forwardToSubscriptionManagerService(tenantId, entityId, subscriptionManagerService -> { subscriptionManagerService.onAlarmDeleted(tenantId, entityId, alarm, TbCallback.EMPTY); }, () -> { return TbSubscriptionUtils.toAlarmDeletedProto(tenantId, entityId, alarm); diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java index 567f5cb9ff..9521805337 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java @@ -366,7 +366,7 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer } private void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes, boolean notifyDevice) { - forwardToSubscriptionManagerServiceOrSendToCore(tenantId, entityId, subscriptionManagerService -> { + forwardToSubscriptionManagerService(tenantId, entityId, subscriptionManagerService -> { subscriptionManagerService.onAttributesUpdate(tenantId, entityId, scope, attributes, notifyDevice, TbCallback.EMPTY); }, () -> { return TbSubscriptionUtils.toAttributesUpdateProto(tenantId, entityId, scope, attributes); @@ -374,7 +374,7 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer } private void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List keys) { - forwardToSubscriptionManagerServiceOrSendToCore(tenantId, entityId, subscriptionManagerService -> { + forwardToSubscriptionManagerService(tenantId, entityId, subscriptionManagerService -> { subscriptionManagerService.onAttributesDelete(tenantId, entityId, scope, keys, TbCallback.EMPTY); }, () -> { return TbSubscriptionUtils.toAttributesDeleteProto(tenantId, entityId, scope, keys); @@ -382,7 +382,7 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer } private void onTimeSeriesUpdate(TenantId tenantId, EntityId entityId, List ts) { - forwardToSubscriptionManagerServiceOrSendToCore(tenantId, entityId, subscriptionManagerService -> { + forwardToSubscriptionManagerService(tenantId, entityId, subscriptionManagerService -> { subscriptionManagerService.onTimeSeriesUpdate(tenantId, entityId, ts, TbCallback.EMPTY); }, () -> { return TbSubscriptionUtils.toTimeseriesUpdateProto(tenantId, entityId, ts); @@ -390,7 +390,7 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer } private void onTimeSeriesDelete(TenantId tenantId, EntityId entityId, List keys, List ts) { - forwardToSubscriptionManagerServiceOrSendToCore(tenantId, entityId, subscriptionManagerService -> { + forwardToSubscriptionManagerService(tenantId, entityId, subscriptionManagerService -> { List updated = new ArrayList<>(); List deleted = new ArrayList<>(); 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 ee0afd3c79..6e4ff69a88 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 @@ -21,16 +21,20 @@ import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.thingsboard.server.common.data.id.IdBased; import org.thingsboard.server.common.data.id.NotificationId; +import org.thingsboard.server.common.data.id.NotificationRequestId; import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.notification.Notification; +import org.thingsboard.server.common.data.notification.NotificationInfo; import org.thingsboard.server.common.data.notification.NotificationStatus; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.dao.notification.NotificationService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.util.TbCoreComponent; -import org.thingsboard.server.dao.notification.NotificationProcessingService; +import org.thingsboard.server.service.notification.NotificationSubscriptionService; import org.thingsboard.server.service.security.model.SecurityUser; import org.thingsboard.server.service.ws.notification.cmd.NotificationsCountSubCmd; +import org.thingsboard.server.service.ws.notification.sub.NotificationRequestUpdate; +import org.thingsboard.server.service.ws.notification.sub.NotificationUpdate; import org.thingsboard.server.service.ws.notification.sub.NotificationsSubscription; import org.thingsboard.server.service.subscription.TbLocalSubscriptionService; import org.thingsboard.server.service.ws.WebSocketSessionRef; @@ -53,7 +57,7 @@ public class DefaultNotificationCommandsHandler implements NotificationCommandsH private final NotificationService notificationService; private final TbLocalSubscriptionService localSubscriptionService; - private final NotificationProcessingService notificationProcessingService; + private final NotificationSubscriptionService notificationSubscriptionService; private final TbServiceInfoProvider serviceInfoProvider; @Autowired @Lazy private WebSocketService wsService; @@ -106,47 +110,83 @@ public class DefaultNotificationCommandsHandler implements NotificationCommandsH subscription.getUnreadCounter().set(unreadCount); } + + /* Notifications subscription update handling */ private void handleNotificationsSubscriptionUpdate(NotificationsSubscription subscription, NotificationsSubscriptionUpdate subscriptionUpdate) { - if (subscriptionUpdate.getNotification() != null) { - Notification notification = subscriptionUpdate.getNotification(); - if (notification.getStatus() == NotificationStatus.READ) { - fetchUnreadNotifications(subscription); - sendUpdate(subscription.getSessionId(), subscription.createFullUpdate()); - } else { - subscription.getUnreadNotifications().put(notification.getUuidId(), notification); - if (subscriptionUpdate.isNewNotification()) { - subscription.getTotalUnreadCounter().incrementAndGet(); - Set beyondLimit = subscription.getUnreadNotifications().keySet().stream() - .skip(subscription.getLimit()) - .collect(Collectors.toSet()); - beyondLimit.forEach(notificationId -> subscription.getUnreadNotifications().remove(notificationId)); - } - sendUpdate(subscription.getSessionId(), subscription.createPartialUpdate(notification)); + if (subscriptionUpdate.getNotificationUpdate() != null) { + handleNotificationUpdate(subscription, subscriptionUpdate.getNotificationUpdate()); + } else if (subscriptionUpdate.getNotificationRequestUpdate() != null) { + handleNotificationRequestUpdate(subscription, subscriptionUpdate.getNotificationRequestUpdate()); + } + } + + private void handleNotificationUpdate(NotificationsSubscription subscription, NotificationUpdate update) { + Notification notification = update.getNotification(); + if (notification.getStatus() == NotificationStatus.READ) { + fetchUnreadNotifications(subscription); + sendUpdate(subscription.getSessionId(), subscription.createFullUpdate()); + } else { + subscription.getUnreadNotifications().put(notification.getUuidId(), notification); + if (update.isNew()) { + subscription.getTotalUnreadCounter().incrementAndGet(); + Set beyondLimit = subscription.getUnreadNotifications().keySet().stream() + .skip(subscription.getLimit()) + .collect(Collectors.toSet()); + beyondLimit.forEach(notificationId -> subscription.getUnreadNotifications().remove(notificationId)); } - } else if (subscriptionUpdate.isNotificationRequestDeleted()) { + sendUpdate(subscription.getSessionId(), subscription.createPartialUpdate(notification)); + } + } + + private void handleNotificationRequestUpdate(NotificationsSubscription subscription, NotificationRequestUpdate update) { + NotificationRequestId notificationRequestId = update.getNotificationRequestId(); + if (update.isDeleted()) { if (subscription.getUnreadNotifications().values().stream() - .anyMatch(notification -> notification.getRequestId().equals(subscriptionUpdate.getNotificationRequestId()))) { + .anyMatch(notification -> notification.getRequestId().equals(notificationRequestId))) { fetchUnreadNotifications(subscription); sendUpdate(subscription.getSessionId(), subscription.createFullUpdate()); } + } else { + NotificationInfo notificationInfo = update.getNotificationInfo(); + subscription.getUnreadNotifications().values().stream() + .filter(notification -> notification.getRequestId().equals(notificationRequestId)) + .forEach(notification -> { + notification.setInfo(notificationInfo); + sendUpdate(subscription.getSessionId(), subscription.createPartialUpdate(notification)); + }); } } + + /* Notifications count subscription update handling */ private void handleNotificationsCountSubscriptionUpdate(NotificationsCountSubscription subscription, NotificationsSubscriptionUpdate subscriptionUpdate) { - if (subscriptionUpdate.getNotification() != null) { - Notification notification = subscriptionUpdate.getNotification(); - if (subscriptionUpdate.isNewNotification()) { - subscription.getUnreadCounter().incrementAndGet(); - } else if (notification.getStatus() == NotificationStatus.READ) { - // for now this can only happen when user marks notification as read - subscription.getUnreadCounter().decrementAndGet(); - } - } else if (subscriptionUpdate.isNotificationRequestDeleted()) { - fetchUnreadNotificationsCount(subscription); + if (subscriptionUpdate.getNotificationUpdate() != null) { + handleNotificationUpdate(subscription, subscriptionUpdate.getNotificationUpdate()); + } else if (subscriptionUpdate.getNotificationRequestUpdate() != null) { + handleNotificationRequestUpdate(subscription, subscriptionUpdate.getNotificationRequestUpdate()); + } + sendUpdate(subscription.getSessionId(), subscription.createUpdate()); + } + + private void handleNotificationUpdate(NotificationsCountSubscription subscription, NotificationUpdate update) { + Notification notification = update.getNotification(); + if (update.isNew()) { + subscription.getUnreadCounter().incrementAndGet(); + } else if (notification.getStatus() == NotificationStatus.READ) { + // for now this can only happen when user marks notification as read + subscription.getUnreadCounter().decrementAndGet(); } sendUpdate(subscription.getSessionId(), subscription.createUpdate()); } + private void handleNotificationRequestUpdate(NotificationsCountSubscription subscription, NotificationRequestUpdate update) { + if (update.isDeleted()) { + fetchUnreadNotificationsCount(subscription); + sendUpdate(subscription.getSessionId(), subscription.createUpdate()); + } + } + + private void sendUpdate(String sessionId, CmdUpdate update) { wsService.sendWsMsg(sessionId, update); } @@ -155,7 +195,7 @@ public class DefaultNotificationCommandsHandler implements NotificationCommandsH @Override public void handleMarkAsReadCmd(WebSocketSessionRef sessionRef, MarkNotificationAsReadCmd cmd) { NotificationId notificationId = new NotificationId(cmd.getNotificationId()); - notificationProcessingService.markNotificationAsRead(sessionRef.getSecurityCtx().getTenantId(), sessionRef.getSecurityCtx().getId(), notificationId); + notificationSubscriptionService.markNotificationAsRead(sessionRef.getSecurityCtx().getTenantId(), sessionRef.getSecurityCtx().getId(), notificationId); } @Override diff --git a/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationRequestUpdate.java b/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationRequestUpdate.java new file mode 100644 index 0000000000..36c3b2a441 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationRequestUpdate.java @@ -0,0 +1,33 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.ws.notification.sub; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; +import org.thingsboard.server.common.data.id.NotificationRequestId; +import org.thingsboard.server.common.data.notification.NotificationInfo; + +@Data +@NoArgsConstructor +@AllArgsConstructor +@Builder +public class NotificationRequestUpdate { + private NotificationRequestId notificationRequestId; + private NotificationInfo notificationInfo; + private boolean deleted; +} diff --git a/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationUpdate.java b/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationUpdate.java new file mode 100644 index 0000000000..0e7db3e8cf --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationUpdate.java @@ -0,0 +1,31 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.ws.notification.sub; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; +import org.thingsboard.server.common.data.notification.Notification; + +@Data +@NoArgsConstructor +@AllArgsConstructor +@Builder +public class NotificationUpdate { + private Notification notification; + private boolean isNew; +} diff --git a/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsSubscriptionUpdate.java b/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsSubscriptionUpdate.java index 4ed7ca7341..cedd6041f7 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsSubscriptionUpdate.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsSubscriptionUpdate.java @@ -15,19 +15,22 @@ */ package org.thingsboard.server.service.ws.notification.sub; -import lombok.Builder; import lombok.Data; -import org.thingsboard.server.common.data.id.NotificationRequestId; -import org.thingsboard.server.common.data.notification.Notification; @Data -@Builder public class NotificationsSubscriptionUpdate { - private final Notification notification; - private final boolean isNewNotification; + private final NotificationUpdate notificationUpdate; + private final NotificationRequestUpdate notificationRequestUpdate; - private final boolean notificationRequestDeleted; - private final NotificationRequestId notificationRequestId; + public NotificationsSubscriptionUpdate(NotificationUpdate notificationUpdate) { + this.notificationUpdate = notificationUpdate; + this.notificationRequestUpdate = null; + } + + public NotificationsSubscriptionUpdate(NotificationRequestUpdate notificationRequestUpdate) { + this.notificationUpdate = null; + this.notificationRequestUpdate = notificationRequestUpdate; + } } diff --git a/common/cluster-api/src/main/proto/queue.proto b/common/cluster-api/src/main/proto/queue.proto index c1a6dd342e..74223de8c7 100644 --- a/common/cluster-api/src/main/proto/queue.proto +++ b/common/cluster-api/src/main/proto/queue.proto @@ -576,8 +576,8 @@ message TbAlarmSubscriptionUpdateProto { message NotificationsSubscriptionUpdateProto { string sessionId = 1; int32 subscriptionId = 2; - string notification = 3; - bool isNewNotification = 4; + string notificationUpdate = 3; + string notificationRequestUpdate = 4; } message NotificationUpdateProto { @@ -585,15 +585,13 @@ message NotificationUpdateProto { int64 tenantIdLSB = 2; int64 recipientIdMSB = 3; int64 recipientIdLSB = 4; - string notification = 5; - bool isNew = 6; + string update = 5; } -message NotificationRequestDeleteProto { +message NotificationRequestUpdateProto { int64 tenantIdMSB = 1; int64 tenantIdLSB = 2; - int64 notificationRequestIdMSB = 3; - int64 notificationRequestIdLSB = 4; + string update = 6; } message TbAttributeUpdateProto { @@ -700,7 +698,7 @@ message SubscriptionMgrMsgProto { NotificationsSubscriptionProto notificationsSub = 11; NotificationsCountSubscriptionProto notificationsCountSub = 12; NotificationUpdateProto notificationUpdate = 13; - NotificationRequestDeleteProto notificationRequestDelete = 14; + NotificationRequestUpdateProto notificationRequestUpdate = 14; } message LocalSubscriptionServiceMsgProto { diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationRuleService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationRuleService.java new file mode 100644 index 0000000000..291688a422 --- /dev/null +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationRuleService.java @@ -0,0 +1,26 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.dao.notification; + +import org.thingsboard.server.common.data.id.NotificationRuleId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.notification.rule.NotificationRule; + +public interface NotificationRuleService { + + NotificationRule findNotificationRuleById(TenantId tenantId, NotificationRuleId notificationRuleId); + +} diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationService.java index a4bebfd135..6b7f6267db 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationService.java @@ -15,37 +15,45 @@ */ package org.thingsboard.server.dao.notification; +import org.thingsboard.server.common.data.id.AlarmId; import org.thingsboard.server.common.data.id.NotificationId; import org.thingsboard.server.common.data.id.NotificationRequestId; +import org.thingsboard.server.common.data.id.NotificationRuleId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.notification.Notification; +import org.thingsboard.server.common.data.notification.NotificationInfo; import org.thingsboard.server.common.data.notification.NotificationRequest; -import org.thingsboard.server.common.data.notification.NotificationStatus; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; +import java.util.List; + public interface NotificationService { - NotificationRequest createNotificationRequest(TenantId tenantId, NotificationRequest notificationRequest); + NotificationRequest saveNotificationRequest(TenantId tenantId, NotificationRequest notificationRequest); NotificationRequest findNotificationRequestById(TenantId tenantId, NotificationRequestId id); - PageData findNotificationRequestsByTenantIdAndPageLink(TenantId tenantId, PageLink pageLink); + PageData findNotificationRequestsByTenantId(TenantId tenantId, PageLink pageLink); + + List findNotificationRequestsByRuleIdAndAlarmId(TenantId tenantId, NotificationRuleId ruleId, AlarmId alarmId); - void deleteNotificationRequest(TenantId tenantId, NotificationRequestId id); + void deleteNotificationRequestById(TenantId tenantId, NotificationRequestId id); - Notification createNotification(TenantId tenantId, Notification notification); + Notification saveNotification(TenantId tenantId, Notification notification); Notification findNotificationById(TenantId tenantId, NotificationId notificationId); - boolean updateNotificationStatus(TenantId tenantId, UserId userId, NotificationId notificationId, NotificationStatus status); + boolean markNotificationAsRead(TenantId tenantId, UserId userId, NotificationId notificationId); - PageData findNotificationsByUserIdAndReadStatusAndPageLink(TenantId tenantId, UserId userId, boolean unreadOnly, PageLink pageLink); + PageData findNotificationsByUserIdAndReadStatus(TenantId tenantId, UserId userId, boolean unreadOnly, PageLink pageLink); PageData findLatestUnreadNotificationsByUserId(TenantId tenantId, UserId userId, int limit); int countUnreadNotificationsByUserId(TenantId tenantId, UserId userId); + int updateNotificationsInfosByRequestId(TenantId tenantId, NotificationRequestId notificationRequestId, NotificationInfo notificationInfo); + } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/Alarm.java b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/Alarm.java index f13c84bc82..ba7d30fb85 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/Alarm.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/Alarm.java @@ -29,6 +29,7 @@ import org.thingsboard.server.common.data.HasTenantId; import org.thingsboard.server.common.data.id.AlarmId; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.NotificationRuleId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.validation.Length; @@ -78,6 +79,7 @@ public class Alarm extends BaseData implements HasName, HasTenantId, Ha "By default, 'propagateRelationTypes' array is empty which means that the alarm will be propagated based on any relation type to parent entities. " + "This parameter should be used only in case when 'propagate' parameter is set to true, otherwise, 'propagateRelationTypes' array will be ignored.") private List propagateRelationTypes; + private NotificationRuleId notificationRuleId; public Alarm() { super(); @@ -105,6 +107,7 @@ public class Alarm extends BaseData implements HasName, HasTenantId, Ha this.propagateToOwner = alarm.isPropagateToOwner(); this.propagateToTenant = alarm.isPropagateToTenant(); this.propagateRelationTypes = alarm.getPropagateRelationTypes(); + this.notificationRuleId = alarm.getNotificationRuleId(); } @Override diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DeviceProfileAlarm.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DeviceProfileAlarm.java index 71e7b5f415..33e5cf1391 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DeviceProfileAlarm.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DeviceProfileAlarm.java @@ -19,6 +19,7 @@ import io.swagger.annotations.ApiModel; import io.swagger.annotations.ApiModelProperty; import lombok.Data; import org.thingsboard.server.common.data.alarm.AlarmSeverity; +import org.thingsboard.server.common.data.id.NotificationRuleId; import org.thingsboard.server.common.data.validation.Length; import org.thingsboard.server.common.data.validation.NoXss; @@ -59,4 +60,6 @@ public class DeviceProfileAlarm implements Serializable { "This parameter should be used only in case when 'propagate' parameter is set to true, otherwise, 'propagateRelationTypes' array will be ignored.") private List propagateRelationTypes; + private NotificationRuleId notificationRuleId; + } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/id/NotificationRuleId.java b/common/data/src/main/java/org/thingsboard/server/common/data/id/NotificationRuleId.java new file mode 100644 index 0000000000..2b963c5b88 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/id/NotificationRuleId.java @@ -0,0 +1,36 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.id; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import org.thingsboard.server.common.data.EntityType; + +import java.util.UUID; + +public class NotificationRuleId extends UUIDBased { + + @JsonCreator + public NotificationRuleId(@JsonProperty("id") UUID id) { + super(id); + } + +// @Override +// public EntityType getEntityType() { +// return EntityType.NOTIFICATION_TARGET; +// } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationInfo.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationInfo.java index 4831e056cd..298998f8d7 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationInfo.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationInfo.java @@ -15,14 +15,22 @@ */ package org.thingsboard.server.common.data.notification; -import com.fasterxml.jackson.annotation.JsonIgnoreProperties; -import com.fasterxml.jackson.databind.node.ObjectNode; +import lombok.AllArgsConstructor; +import lombok.Builder; import lombok.Data; +import lombok.NoArgsConstructor; +import org.thingsboard.server.common.data.alarm.AlarmSeverity; +import org.thingsboard.server.common.data.alarm.AlarmStatus; +import org.thingsboard.server.common.data.id.AlarmId; import org.thingsboard.server.common.data.id.DashboardId; +import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.validation.NoXss; @Data +@AllArgsConstructor +@NoArgsConstructor +@Builder //@JsonIgnoreProperties(ignoreUnknown = true) //@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "notificationType", visible = true, defaultImpl = NotificationInfo.class) //@JsonSubTypes({ @@ -31,7 +39,12 @@ import org.thingsboard.server.common.data.validation.NoXss; public class NotificationInfo { @NoXss private String description; - - private ObjectNode alarmDetails; // move to child class private DashboardId dashboardId; + + private AlarmId alarmId; + private String alarmType; + private EntityId alarmOriginator; + private AlarmSeverity alarmSeverity; + private AlarmStatus alarmStatus; + } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequest.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequest.java index dd800ef3fc..af90eab7ee 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequest.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequest.java @@ -15,6 +15,8 @@ */ package org.thingsboard.server.common.data.notification; +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonProperty; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; @@ -23,7 +25,9 @@ import lombok.NoArgsConstructor; import org.thingsboard.server.common.data.BaseData; import org.thingsboard.server.common.data.HasName; import org.thingsboard.server.common.data.HasTenantId; +import org.thingsboard.server.common.data.id.AlarmId; import org.thingsboard.server.common.data.id.NotificationRequestId; +import org.thingsboard.server.common.data.id.NotificationRuleId; import org.thingsboard.server.common.data.id.NotificationTargetId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.validation.NoXss; @@ -51,6 +55,13 @@ public class NotificationRequest extends BaseData impleme private NotificationInfo notificationInfo; private NotificationSeverity notificationSeverity; private NotificationRequestConfig additionalConfig; + @JsonProperty(access = JsonProperty.Access.READ_ONLY) + private NotificationRequestStatus status; + + @JsonIgnore + private NotificationRuleId ruleId; // maybe move to child class + @JsonIgnore + private AlarmId alarmId; public static final String GENERAL_NOTIFICATION_REASON = "General"; public static final String ALARM_NOTIFICATION_REASON = "Alarm"; diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequestConfig.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequestConfig.java index 8f65427f21..fd01c91255 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequestConfig.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequestConfig.java @@ -19,5 +19,5 @@ import lombok.Data; @Data public class NotificationRequestConfig { - private Long sendingDelayMs; + private int sendingDelayInMinutes; } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequestStatus.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequestStatus.java new file mode 100644 index 0000000000..1c3eede15e --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequestStatus.java @@ -0,0 +1,21 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.notification; + +public enum NotificationRequestStatus { + PROCESSED, + SCHEDULED +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/NonConfirmedNotificationEscalation.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/NonConfirmedNotificationEscalation.java index befc666281..1174cedb06 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/NonConfirmedNotificationEscalation.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/NonConfirmedNotificationEscalation.java @@ -15,9 +15,18 @@ */ package org.thingsboard.server.common.data.notification.rule; -import java.util.UUID; +import lombok.Data; +import org.thingsboard.server.common.data.id.NotificationTargetId; +import javax.validation.constraints.Min; +import javax.validation.constraints.NotNull; + +@Data public class NonConfirmedNotificationEscalation { - private long delayMs; // delay since initial notification request // if no one from previous escalation item has read the notification, send notifications after this time to other recipients - private UUID notificationTargetId; + + @Min(1) + private int delayInMinutes; // delay since initial notification request // if no one from previous escalation item has read the notification, send notifications after this time to other recipients + @NotNull + private NotificationTargetId notificationTargetId; + } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/NotificationRule.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/NotificationRule.java index 340a49c54d..4d2a8ed184 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/NotificationRule.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/NotificationRule.java @@ -16,23 +16,36 @@ package org.thingsboard.server.common.data.notification.rule; import lombok.Data; +import lombok.EqualsAndHashCode; +import org.thingsboard.server.common.data.BaseData; +import org.thingsboard.server.common.data.HasName; +import org.thingsboard.server.common.data.HasTenantId; +import org.thingsboard.server.common.data.id.NotificationRuleId; +import org.thingsboard.server.common.data.id.NotificationTargetId; +import org.thingsboard.server.common.data.id.TenantId; +import javax.validation.Valid; +import javax.validation.constraints.NotBlank; +import javax.validation.constraints.NotNull; import java.util.List; import java.util.Map; +import java.util.TreeMap; import java.util.UUID; @Data -public class NotificationRule { +@EqualsAndHashCode(callSuper = true) +public class NotificationRule extends BaseData implements HasTenantId, HasName { - // we may choose it in the alarm rule config, or maybe it's better to configure evrth in the triggers? - - private UUID id; // NotificationRuleId id; + @NotNull + private TenantId tenantId; + @NotBlank private String name; - private Map triggers; // or maybe bad idea - // Map - concrete alarmRule or alarm rule search (e.g. alarm rule of device profiles of particular transport type with certain severity) - // triggerConfiguration (??) - alarm filter: severity, specific device profile, alarm rule - - private UUID initialNotificationTargetId; + @NotBlank + private String notificationTextTemplate; + @NotNull + private NotificationTargetId initialNotificationTargetId; + @NotNull + @Valid private List escalations; } 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 2ae524f1e2..27b17164ea 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,6 +295,7 @@ 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"; @@ -668,6 +669,12 @@ public class ModelConstants { public static final String NOTIFICATION_REQUEST_NOTIFICATION_INFO_PROPERTY = "notification_info"; public static final String NOTIFICATION_REQUEST_NOTIFICATION_SEVERITY_PROPERTY = "notification_severity"; public static final String NOTIFICATION_REQUEST_ADDITIONAL_CONFIG_PROPERTY = "additional_config"; + public static final String NOTIFICATION_REQUEST_STATUS_PROPERTY = "status"; + public static final String NOTIFICATION_REQUEST_RULE_ID_PROPERTY = "rule_id"; + public static final String NOTIFICATION_REQUEST_ALARM_ID_PROPERTY = "alarm_id"; + + public static final String NOTIFICATION_RULE_TABLE_NAME = "notification_rule"; + // ... protected static final String[] NONE_AGGREGATION_COLUMNS = new String[]{LONG_VALUE_COLUMN, DOUBLE_VALUE_COLUMN, BOOLEAN_VALUE_COLUMN, STRING_VALUE_COLUMN, JSON_VALUE_COLUMN, KEY_COLUMN, TS_COLUMN}; 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 56adfae6de..8bdb90d1b4 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 @@ -29,6 +29,7 @@ import org.thingsboard.server.common.data.alarm.AlarmStatus; import org.thingsboard.server.common.data.id.AlarmId; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.EntityIdFactory; +import org.thingsboard.server.common.data.id.NotificationRuleId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.dao.model.BaseEntity; import org.thingsboard.server.dao.model.BaseSqlEntity; @@ -47,6 +48,7 @@ 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; @@ -116,6 +118,9 @@ 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(); } @@ -150,6 +155,7 @@ public abstract class AbstractAlarmEntity extends BaseSqlEntity } else { this.propagateRelationTypes = null; } + this.notificationRuleId = getUuid(alarm.getNotificationRuleId()); } public AbstractAlarmEntity(AlarmEntity alarmEntity) { @@ -172,6 +178,7 @@ public abstract class AbstractAlarmEntity extends BaseSqlEntity this.clearTs = alarmEntity.getClearTs(); this.details = alarmEntity.getDetails(); this.propagateRelationTypes = alarmEntity.getPropagateRelationTypes(); + this.notificationRuleId = alarmEntity.getNotificationRuleId(); } protected Alarm toAlarm() { @@ -200,6 +207,7 @@ public abstract class AbstractAlarmEntity extends BaseSqlEntity } else { alarm.setPropagateRelationTypes(Collections.emptyList()); } + alarm.setNotificationRuleId(createId(notificationRuleId, NotificationRuleId::new)); return alarm; } } diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sql/NotificationRequestEntity.java b/dao/src/main/java/org/thingsboard/server/dao/model/sql/NotificationRequestEntity.java index 65e1fe8214..cd0bbd435a 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/model/sql/NotificationRequestEntity.java +++ b/dao/src/main/java/org/thingsboard/server/dao/model/sql/NotificationRequestEntity.java @@ -21,13 +21,16 @@ import lombok.EqualsAndHashCode; import org.hibernate.annotations.Type; import org.hibernate.annotations.TypeDef; import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.server.common.data.id.AlarmId; import org.thingsboard.server.common.data.id.NotificationRequestId; +import org.thingsboard.server.common.data.id.NotificationRuleId; import org.thingsboard.server.common.data.id.NotificationTargetId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.notification.NotificationInfo; import org.thingsboard.server.common.data.notification.NotificationRequest; import org.thingsboard.server.common.data.notification.NotificationRequestConfig; +import org.thingsboard.server.common.data.notification.NotificationRequestStatus; import org.thingsboard.server.common.data.notification.NotificationSeverity; import org.thingsboard.server.dao.model.BaseSqlEntity; import org.thingsboard.server.dao.model.ModelConstants; @@ -71,6 +74,16 @@ public class NotificationRequestEntity extends BaseSqlEntity { + + @Column(name = ModelConstants.TENANT_ID_PROPERTY, nullable = false) + private UUID tenantId; + + @Column(name = ModelConstants.NAME_PROPERTY, nullable = false) + private String name; + + + public NotificationRuleEntity() {} + + public NotificationRuleEntity(NotificationRule notificationRule) { + setId(notificationRule.getUuidId()); + setCreatedTime(notificationRule.getCreatedTime()); + setTenantId(getUuid(notificationRule.getTenantId())); + setName(notificationRule.getName()); + } + + @Override + public NotificationRule toData() { + NotificationRule notificationRule = new NotificationRule(); + notificationRule.setId(new NotificationRuleId(id)); + notificationRule.setCreatedTime(createdTime); + notificationRule.setTenantId(createId(tenantId, TenantId::fromUUID)); + notificationRule.setName(name); + return notificationRule; + } + +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationRuleService.java b/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationRuleService.java new file mode 100644 index 0000000000..7c9e404b3b --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationRuleService.java @@ -0,0 +1,35 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.dao.notification; + +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import org.thingsboard.server.common.data.id.NotificationRuleId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.notification.rule.NotificationRule; + +@Service +@RequiredArgsConstructor +public class DefaultNotificationRuleService implements NotificationRuleService { + + private final NotificationRuleDao notificationRuleDao; + + @Override + public NotificationRule findNotificationRuleById(TenantId tenantId, NotificationRuleId notificationRuleId) { + return notificationRuleDao.findById(tenantId, notificationRuleId.getId()); + } + +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationService.java b/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationService.java index c15480c747..a3ef857233 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationService.java @@ -19,22 +19,25 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.stereotype.Service; -import org.springframework.transaction.annotation.Transactional; +import org.thingsboard.server.common.data.id.AlarmId; import org.thingsboard.server.common.data.id.NotificationId; import org.thingsboard.server.common.data.id.NotificationRequestId; +import org.thingsboard.server.common.data.id.NotificationRuleId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.notification.Notification; +import org.thingsboard.server.common.data.notification.NotificationInfo; import org.thingsboard.server.common.data.notification.NotificationRequest; import org.thingsboard.server.common.data.notification.NotificationSeverity; import org.thingsboard.server.common.data.notification.NotificationStatus; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.common.data.page.SortOrder; -import org.thingsboard.server.dao.exception.DataValidationException; import org.thingsboard.server.dao.service.DataValidator; import org.thingsboard.server.dao.sql.query.EntityKeyMapping; +import java.util.List; + @Service @Slf4j @RequiredArgsConstructor @@ -46,7 +49,7 @@ public class DefaultNotificationService implements NotificationService { private final NotificationRequestValidator notificationRequestValidator = new NotificationRequestValidator(); @Override - public NotificationRequest createNotificationRequest(TenantId tenantId, NotificationRequest notificationRequest) { + public NotificationRequest saveNotificationRequest(TenantId tenantId, NotificationRequest notificationRequest) { if (StringUtils.isBlank(notificationRequest.getNotificationReason())) { notificationRequest.setNotificationReason(NotificationRequest.GENERAL_NOTIFICATION_REASON); } @@ -63,21 +66,23 @@ public class DefaultNotificationService implements NotificationService { } @Override - public PageData findNotificationRequestsByTenantIdAndPageLink(TenantId tenantId, PageLink pageLink) { + public PageData findNotificationRequestsByTenantId(TenantId tenantId, PageLink pageLink) { return notificationRequestDao.findByTenantIdAndPageLink(tenantId, pageLink); } + @Override + public List findNotificationRequestsByRuleIdAndAlarmId(TenantId tenantId, NotificationRuleId ruleId, AlarmId alarmId) { + return notificationRequestDao.findByRuleIdAndAlarmId(tenantId, ruleId, alarmId); + } + // ON DELETE CASCADE is used: notifications for request are deleted as well @Override - public void deleteNotificationRequest(TenantId tenantId, NotificationRequestId id) { + public void deleteNotificationRequestById(TenantId tenantId, NotificationRequestId id) { notificationRequestDao.removeById(tenantId, id.getId()); } @Override - public Notification createNotification(TenantId tenantId, Notification notification) { - if (notification.getId() != null) { - throw new DataValidationException("Notification cannot be updated"); // tmp ? - } + public Notification saveNotification(TenantId tenantId, Notification notification) { return notificationDao.save(tenantId, notification); } @@ -86,14 +91,13 @@ public class DefaultNotificationService implements NotificationService { return notificationDao.findById(tenantId, notificationId.getId()); } - @Transactional @Override - public boolean updateNotificationStatus(TenantId tenantId, UserId userId, NotificationId notificationId, NotificationStatus status) { - return notificationDao.updateStatusByIdAndUserId(tenantId, userId, notificationId, status); + public boolean markNotificationAsRead(TenantId tenantId, UserId userId, NotificationId notificationId) { + return notificationDao.updateStatusByIdAndUserId(tenantId, userId, notificationId, NotificationStatus.READ); } @Override - public PageData findNotificationsByUserIdAndReadStatusAndPageLink(TenantId tenantId, UserId userId, boolean unreadOnly, PageLink pageLink) { + public PageData findNotificationsByUserIdAndReadStatus(TenantId tenantId, UserId userId, boolean unreadOnly, PageLink pageLink) { if (unreadOnly) { return notificationDao.findUnreadByUserIdAndPageLink(tenantId, userId, pageLink); } else { @@ -105,7 +109,7 @@ public class DefaultNotificationService implements NotificationService { public PageData findLatestUnreadNotificationsByUserId(TenantId tenantId, UserId userId, int limit) { SortOrder sortOrder = new SortOrder(EntityKeyMapping.CREATED_TIME, SortOrder.Direction.DESC); PageLink pageLink = new PageLink(limit, 0, null, sortOrder); - return findNotificationsByUserIdAndReadStatusAndPageLink(tenantId, userId, true, pageLink); + return findNotificationsByUserIdAndReadStatus(tenantId, userId, true, pageLink); } @Override @@ -113,13 +117,15 @@ public class DefaultNotificationService implements NotificationService { return notificationDao.countUnreadByUserId(tenantId, userId); } + @Override + public int updateNotificationsInfosByRequestId(TenantId tenantId, NotificationRequestId notificationRequestId, NotificationInfo notificationInfo) { + return notificationDao.updateInfosByRequestId(tenantId, notificationRequestId, notificationInfo); + } + private static class NotificationRequestValidator extends DataValidator { @Override protected void validateDataImpl(TenantId tenantId, NotificationRequest notificationRequest) { - if (notificationRequest.getId() != null) { - throw new DataValidationException("Notification request cannot be changed once created"); - } } } diff --git a/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationDao.java b/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationDao.java index c16df8819a..dcfba4d704 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationDao.java @@ -16,9 +16,11 @@ package org.thingsboard.server.dao.notification; import org.thingsboard.server.common.data.id.NotificationId; +import org.thingsboard.server.common.data.id.NotificationRequestId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.notification.Notification; +import org.thingsboard.server.common.data.notification.NotificationInfo; import org.thingsboard.server.common.data.notification.NotificationStatus; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; @@ -34,4 +36,8 @@ public interface NotificationDao extends Dao { int countUnreadByUserId(TenantId tenantId, UserId userId); + PageData findByRequestId(TenantId tenantId, NotificationRequestId notificationRequestId, PageLink pageLink); + + int updateInfosByRequestId(TenantId tenantId, NotificationRequestId notificationRequestId, NotificationInfo notificationInfo); + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationRequestDao.java b/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationRequestDao.java index d518a4e245..1848e826e0 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationRequestDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationRequestDao.java @@ -15,14 +15,20 @@ */ package org.thingsboard.server.dao.notification; +import org.thingsboard.server.common.data.id.AlarmId; +import org.thingsboard.server.common.data.id.NotificationRuleId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.notification.NotificationRequest; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.dao.Dao; +import java.util.List; + public interface NotificationRequestDao extends Dao { PageData findByTenantIdAndPageLink(TenantId tenantId, PageLink pageLink); + List findByRuleIdAndAlarmId(TenantId tenantId, NotificationRuleId ruleId, AlarmId alarmId); + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationRuleDao.java b/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationRuleDao.java new file mode 100644 index 0000000000..3a7cbb777b --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationRuleDao.java @@ -0,0 +1,22 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.dao.notification; + +import org.thingsboard.server.common.data.notification.rule.NotificationRule; +import org.thingsboard.server.dao.Dao; + +public interface NotificationRuleDao extends Dao { +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationDao.java index 2611c32c90..cd98941011 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationDao.java @@ -20,10 +20,13 @@ import lombok.RequiredArgsConstructor; import org.springframework.beans.factory.annotation.Value; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.stereotype.Component; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.id.NotificationId; +import org.thingsboard.server.common.data.id.NotificationRequestId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.UserId; import org.thingsboard.server.common.data.notification.Notification; +import org.thingsboard.server.common.data.notification.NotificationInfo; import org.thingsboard.server.common.data.notification.NotificationStatus; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; @@ -82,6 +85,16 @@ public class JpaNotificationDao extends JpaAbstractDao findByRequestId(TenantId tenantId, NotificationRequestId notificationRequestId, PageLink pageLink) { + return DaoUtil.toPageData(notificationRepository.findByRequestId(notificationRequestId.getId(), DaoUtil.toPageable(pageLink))); + } + + @Override + public int updateInfosByRequestId(TenantId tenantId, NotificationRequestId notificationRequestId, NotificationInfo notificationInfo) { + return notificationRepository.updateInfosByRequestId(notificationRequestId.getId(), JacksonUtil.valueToTree(notificationInfo)); + } + @Override protected Class getEntityClass() { return NotificationEntity.class; 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 099218abcc..5e1ca5d3f7 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 @@ -19,6 +19,8 @@ import com.google.common.base.Strings; import lombok.RequiredArgsConstructor; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.id.AlarmId; +import org.thingsboard.server.common.data.id.NotificationRuleId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.notification.NotificationRequest; import org.thingsboard.server.common.data.page.PageData; @@ -29,6 +31,7 @@ import org.thingsboard.server.dao.notification.NotificationRequestDao; import org.thingsboard.server.dao.sql.JpaAbstractDao; import org.thingsboard.server.dao.util.SqlDao; +import java.util.List; import java.util.UUID; @Component @@ -44,6 +47,11 @@ public class JpaNotificationRequestDao extends JpaAbstractDao findByRuleIdAndAlarmId(TenantId tenantId, NotificationRuleId ruleId, AlarmId alarmId) { + return DaoUtil.convertDataList(notificationRequestRepository.findAllByRuleIdAndAlarmId(ruleId.getId(), alarmId.getId())); + } + @Override protected Class getEntityClass() { return NotificationRequestEntity.class; 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 new file mode 100644 index 0000000000..365a439bd5 --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRuleDao.java @@ -0,0 +1,46 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.dao.sql.notification; + +import lombok.RequiredArgsConstructor; +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.data.notification.rule.NotificationRule; +import org.thingsboard.server.dao.model.sql.NotificationRuleEntity; +import org.thingsboard.server.dao.notification.NotificationRuleDao; +import org.thingsboard.server.dao.sql.JpaAbstractDao; +import org.thingsboard.server.dao.util.SqlDao; + +import java.util.UUID; + +@Component +@SqlDao +@RequiredArgsConstructor +public class JpaNotificationRuleDao extends JpaAbstractDao implements NotificationRuleDao { + + private final NotificationRuleRepository notificationRuleRepository; + + @Override + protected Class getEntityClass() { + return NotificationRuleEntity.class; + } + + @Override + protected JpaRepository getRepository() { + return notificationRuleRepository; + } + +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRepository.java index e264689ce7..3f46a3dfa9 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRepository.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.dao.sql.notification; +import com.fasterxml.jackson.databind.JsonNode; import org.springframework.data.domain.Page; import org.springframework.data.domain.Pageable; import org.springframework.data.jpa.repository.JpaRepository; @@ -45,6 +46,11 @@ public interface NotificationRepository extends JpaRepository findByRequestId(UUID requestId, Pageable pageable); + + @Modifying + @Transactional + @Query("UPDATE NotificationEntity n SET n.info = :info WHERE n.requestId = :requestId") + int updateInfosByRequestId(@Param("requestId") UUID requestId, @Param("info") JsonNode info); } 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 1303c99034..4ff69f8797 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 @@ -23,6 +23,7 @@ import org.springframework.data.repository.query.Param; import org.springframework.stereotype.Repository; import org.thingsboard.server.dao.model.sql.NotificationRequestEntity; +import java.util.List; import java.util.UUID; @Repository @@ -34,4 +35,6 @@ public interface NotificationRequestRepository extends JpaRepository findByTenantIdAndSearchText(@Param("tenantId") UUID tenantId, @Param("searchText") String searchText, Pageable pageable); + List findAllByRuleIdAndAlarmId(UUID ruleId, UUID alarmId); + } 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 new file mode 100644 index 0000000000..a991407ab2 --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRuleRepository.java @@ -0,0 +1,26 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.dao.sql.notification; + +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.stereotype.Repository; +import org.thingsboard.server.dao.model.sql.NotificationRuleEntity; + +import java.util.UUID; + +@Repository +public interface NotificationRuleRepository extends JpaRepository { +} diff --git a/dao/src/main/resources/sql/schema-entities-idx.sql b/dao/src/main/resources/sql/schema-entities-idx.sql index 8f43b1ac4e..019a8b1771 100644 --- a/dao/src/main/resources/sql/schema-entities-idx.sql +++ b/dao/src/main/resources/sql/schema-entities-idx.sql @@ -80,6 +80,6 @@ CREATE INDEX IF NOT EXISTS idx_notification_request_tenant_id_and_created_time O CREATE INDEX IF NOT EXISTS idx_notification_id ON notification(id); -CREATE INDEX IF NOT EXISTS idx_notification_recipient_id_and_created_time ON notification(recipient_id, created_time DESC); +CREATE INDEX IF NOT EXISTS idx_notification_notification_request_id ON notification(request_id); -CREATE INDEX IF NOT EXISTS idx_notification_recipient_id_and_status_and_created_time ON notification(recipient_id, status, created_time DESC); +CREATE INDEX IF NOT EXISTS idx_notification_recipient_id_and_created_time ON notification(recipient_id, created_time DESC); diff --git a/dao/src/main/resources/sql/schema-entities.sql b/dao/src/main/resources/sql/schema-entities.sql index cae28a049a..1018abe42a 100644 --- a/dao/src/main/resources/sql/schema-entities.sql +++ b/dao/src/main/resources/sql/schema-entities.sql @@ -795,7 +795,10 @@ CREATE TABLE IF NOT EXISTS notification_request ( text_template VARCHAR NOT NULL, notification_info VARCHAR(1000), notification_severity VARCHAR(32), - additional_config VARCHAR(1000) + additional_config VARCHAR(1000), + status VARCHAR(32), + rule_id UUID NULL CONSTRAINT fk_notification_request_rule_id REFERENCES notification_rule(id), + alarm_id UUID ); CREATE TABLE IF NOT EXISTS notification ( diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineNotificationService.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineNotificationService.java new file mode 100644 index 0000000000..e7b87cf25d --- /dev/null +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineNotificationService.java @@ -0,0 +1,25 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.api; + +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.notification.NotificationRequest; + +public interface RuleEngineNotificationService { + + NotificationRequest processNotificationRequest(TenantId tenantId, NotificationRequest notificationRequest); + +} diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java index 6f6ebcd233..0978296ba5 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java @@ -53,7 +53,6 @@ import org.thingsboard.server.dao.edge.EdgeService; import org.thingsboard.server.dao.entityview.EntityViewService; import org.thingsboard.server.dao.nosql.CassandraStatementTask; import org.thingsboard.server.dao.nosql.TbResultSetFuture; -import org.thingsboard.server.dao.notification.NotificationProcessingService; import org.thingsboard.server.dao.ota.OtaPackageService; import org.thingsboard.server.dao.queue.QueueService; import org.thingsboard.server.dao.relation.RelationService; @@ -267,7 +266,7 @@ public interface TbContext { SmsSenderFactory getSmsSenderFactory(); - NotificationProcessingService getNotificationProcessingService(); + RuleEngineNotificationService getNotificationService(); ScriptEngine createJsScriptEngine(String script, String... argNames); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/notification/TbNotificationNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/notification/TbNotificationNode.java index 73ec3383fe..06fc0b110d 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/notification/TbNotificationNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/notification/TbNotificationNode.java @@ -60,7 +60,7 @@ public class TbNotificationNode implements TbNode { .notificationSeverity(config.getNotificationSeverity()) .build(); withCallback(ctx.getDbCallbackExecutor().executeAsync(() -> { - return ctx.getNotificationProcessingService().processNotificationRequest(ctx.getTenantId(), notificationRequest); + return ctx.getNotificationService().processNotificationRequest(ctx.getTenantId(), notificationRequest); }), r -> { TbMsgMetaData msgMetaData = msg.getMetaData().copy(); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmState.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmState.java index 1f90b4e6c1..29bd7d6177 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmState.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmState.java @@ -272,6 +272,7 @@ class AlarmState { if (alarmDefinition.getPropagateRelationTypes() != null) { currentAlarm.setPropagateRelationTypes(alarmDefinition.getPropagateRelationTypes()); } + currentAlarm.setNotificationRuleId(alarmDefinition.getNotificationRuleId()); currentAlarm = ctx.getAlarmService().createOrUpdateAlarm(currentAlarm); boolean updated = currentAlarm.getStartTs() != currentAlarm.getEndTs(); return new TbAlarmResult(!updated, updated, false, false, currentAlarm); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java index 5ab5adc7fc..1ef06547ba 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java @@ -191,6 +191,7 @@ class DeviceState { AlarmState alarmState = alarmStates.computeIfAbsent(alarm.getId(), a -> new AlarmState(this.deviceProfile, deviceId, alarm, getOrInitPersistedAlarmState(alarm), dynamicPredicateValueCtx)); alarmState.processAckAlarm(alarmNf); + // todo: process notification rule } ctx.tellSuccess(msg); }