Browse Source

Notification rules initial implementation

pull/7511/head
ViacheslavKlimov 4 years ago
parent
commit
12ae902cfc
  1. 11
      application/src/main/data/upgrade/3.4.2/schema_update.sql
  2. 7
      application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
  3. 6
      application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
  4. 17
      application/src/main/java/org/thingsboard/server/controller/NotificationController.java
  5. 138
      application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationRuleProcessingService.java
  6. 82
      application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationSubscriptionService.java
  7. 28
      application/src/main/java/org/thingsboard/server/service/notification/NotificationRuleProcessingService.java
  8. 6
      application/src/main/java/org/thingsboard/server/service/notification/NotificationSubscriptionService.java
  9. 44
      application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
  10. 23
      application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java
  11. 8
      application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java
  12. 31
      application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java
  13. 6
      application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java
  14. 8
      application/src/main/java/org/thingsboard/server/service/telemetry/DefaultAlarmSubscriptionService.java
  15. 8
      application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java
  16. 100
      application/src/main/java/org/thingsboard/server/service/ws/notification/DefaultNotificationCommandsHandler.java
  17. 33
      application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationRequestUpdate.java
  18. 31
      application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationUpdate.java
  19. 19
      application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsSubscriptionUpdate.java
  20. 14
      common/cluster-api/src/main/proto/queue.proto
  21. 26
      common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationRuleService.java
  22. 22
      common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationService.java
  23. 3
      common/data/src/main/java/org/thingsboard/server/common/data/alarm/Alarm.java
  24. 3
      common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DeviceProfileAlarm.java
  25. 36
      common/data/src/main/java/org/thingsboard/server/common/data/id/NotificationRuleId.java
  26. 21
      common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationInfo.java
  27. 11
      common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequest.java
  28. 2
      common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequestConfig.java
  29. 21
      common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequestStatus.java
  30. 15
      common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/NonConfirmedNotificationEscalation.java
  31. 31
      common/data/src/main/java/org/thingsboard/server/common/data/notification/rule/NotificationRule.java
  32. 7
      dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java
  33. 8
      dao/src/main/java/org/thingsboard/server/dao/model/sql/AbstractAlarmEntity.java
  34. 19
      dao/src/main/java/org/thingsboard/server/dao/model/sql/NotificationRequestEntity.java
  35. 65
      dao/src/main/java/org/thingsboard/server/dao/model/sql/NotificationRuleEntity.java
  36. 35
      dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationRuleService.java
  37. 40
      dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationService.java
  38. 6
      dao/src/main/java/org/thingsboard/server/dao/notification/NotificationDao.java
  39. 6
      dao/src/main/java/org/thingsboard/server/dao/notification/NotificationRequestDao.java
  40. 22
      dao/src/main/java/org/thingsboard/server/dao/notification/NotificationRuleDao.java
  41. 13
      dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationDao.java
  42. 8
      dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRequestDao.java
  43. 46
      dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRuleDao.java
  44. 8
      dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRepository.java
  45. 3
      dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRequestRepository.java
  46. 26
      dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRuleRepository.java
  47. 4
      dao/src/main/resources/sql/schema-entities-idx.sql
  48. 5
      dao/src/main/resources/sql/schema-entities.sql
  49. 25
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineNotificationService.java
  50. 3
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java
  51. 2
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/notification/TbNotificationNode.java
  52. 1
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmState.java
  53. 1
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/DeviceState.java

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

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

6
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

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

138
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<NotificationRequest> 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();
}
}

82
application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationProcessingService.java → 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<String> 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<String> 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

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

6
common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationProcessingService.java → 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);
}

44
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<ToCore
} else if (msg.hasAlarmSubUpdate()) {
localSubscriptionService.onSubscriptionUpdate(msg.getAlarmSubUpdate().getSessionId(), TbSubscriptionUtils.fromProto(msg.getAlarmSubUpdate()), callback);
} else if (msg.hasNotificationsSubUpdate()) {
TransportProtos.NotificationsSubscriptionUpdateProto notificationsSubUpdateProto = msg.getNotificationsSubUpdate();
NotificationsSubscriptionUpdate notificationsSubscriptionUpdate = NotificationsSubscriptionUpdate.builder()
.notification(JacksonUtil.fromString(notificationsSubUpdateProto.getNotification(), Notification.class))
.isNewNotification(notificationsSubUpdateProto.getIsNewNotification())
.build();
localSubscriptionService.onSubscriptionUpdate(notificationsSubUpdateProto.getSessionId(),
notificationsSubUpdateProto.getSubscriptionId(), notificationsSubscriptionUpdate, callback);
TransportProtos.NotificationsSubscriptionUpdateProto subUpdateProto = msg.getNotificationsSubUpdate();
NotificationsSubscriptionUpdate notificationsSubscriptionUpdate;
if (StringUtils.isNotEmpty(subUpdateProto.getNotificationUpdate())) {
NotificationUpdate notificationUpdate = JacksonUtil.fromString(subUpdateProto.getNotificationUpdate(), NotificationUpdate.class);
notificationsSubscriptionUpdate = new NotificationsSubscriptionUpdate(notificationUpdate);
} else {
NotificationRequestUpdate notificationRequestUpdate = JacksonUtil.fromString(subUpdateProto.getNotificationRequestUpdate(), NotificationRequestUpdate.class);
notificationsSubscriptionUpdate = new NotificationsSubscriptionUpdate(notificationRequestUpdate);
}
localSubscriptionService.onSubscriptionUpdate(subUpdateProto.getSessionId(),
subUpdateProto.getSubscriptionId(), notificationsSubscriptionUpdate, callback);
} else {
throwNotHandled(msg, callback);
}
@ -524,18 +532,16 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
TbSubscriptionUtils.toEntityId(proto.getEntityType(), proto.getEntityIdMSB(), proto.getEntityIdLSB()),
JacksonUtil.fromString(proto.getAlarm(), Alarm.class), callback);
} else if (msg.hasNotificationUpdate()) {
TransportProtos.NotificationUpdateProto notificationUpdateProto = msg.getNotificationUpdate();
TenantId tenantId = TenantId.fromUUID(new UUID(notificationUpdateProto.getTenantIdMSB(), notificationUpdateProto.getTenantIdLSB()));
UserId recipientId = new UserId(new UUID(notificationUpdateProto.getRecipientIdMSB(), notificationUpdateProto.getRecipientIdLSB()));
Notification notification = JacksonUtil.fromString(notificationUpdateProto.getNotification(), Notification.class);
boolean isNew = notificationUpdateProto.getIsNew();
subscriptionManagerService.onNotificationUpdate(tenantId, recipientId, notification, isNew, callback);
} else if (msg.hasNotificationRequestDelete()) {
TransportProtos.NotificationRequestDeleteProto notificationRequestDeleteProto = msg.getNotificationRequestDelete();
TenantId tenantId = TenantId.fromUUID(new UUID(notificationRequestDeleteProto.getTenantIdMSB(), notificationRequestDeleteProto.getTenantIdLSB()));
NotificationRequestId notificationRequestId = new NotificationRequestId(new UUID(
notificationRequestDeleteProto.getNotificationRequestIdMSB(), notificationRequestDeleteProto.getNotificationRequestIdLSB()));
subscriptionManagerService.onNotificationRequestDeleted(tenantId, notificationRequestId, callback);
TransportProtos.NotificationUpdateProto updateProto = msg.getNotificationUpdate();
TenantId tenantId = TenantId.fromUUID(new UUID(updateProto.getTenantIdMSB(), updateProto.getTenantIdLSB()));
UserId recipientId = new UserId(new UUID(updateProto.getRecipientIdMSB(), updateProto.getRecipientIdLSB()));
NotificationUpdate update = JacksonUtil.fromString(updateProto.getUpdate(), NotificationUpdate.class);
subscriptionManagerService.onNotificationUpdate(tenantId, recipientId, update, callback);
} else if (msg.hasNotificationRequestUpdate()) {
TransportProtos.NotificationRequestUpdateProto updateProto = msg.getNotificationRequestUpdate();
TenantId tenantId = TenantId.fromUUID(new UUID(updateProto.getTenantIdMSB(), updateProto.getTenantIdLSB()));
NotificationRequestUpdate update = JacksonUtil.fromString(updateProto.getUpdate(), NotificationRequestUpdate.class);
subscriptionManagerService.onNotificationRequestUpdate(tenantId, update, callback);
} else {
throwNotHandled(msg, callback);
}

23
application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java

@ -28,7 +28,6 @@ import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.alarm.Alarm;
import org.thingsboard.server.common.data.id.DeviceId;
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.Aggregation;
@ -39,11 +38,11 @@ import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
import org.thingsboard.server.common.data.kv.StringDataEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.common.data.notification.Notification;
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.attributes.AttributesService;
import org.thingsboard.server.service.notification.NotificationRuleProcessingService;
import org.thingsboard.server.dao.timeseries.TimeseriesService;
import org.thingsboard.server.gen.transport.TransportProtos.LocalSubscriptionServiceMsgProto;
import org.thingsboard.server.gen.transport.TransportProtos.TbAlarmSubscriptionUpdateProto;
@ -62,6 +61,8 @@ import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
import org.thingsboard.server.queue.util.TbCoreComponent;
import org.thingsboard.server.service.state.DefaultDeviceStateService;
import org.thingsboard.server.service.state.DeviceStateService;
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 org.thingsboard.server.service.ws.telemetry.sub.AlarmSubscriptionUpdate;
import org.thingsboard.server.service.ws.telemetry.sub.TelemetrySubscriptionUpdate;
@ -114,6 +115,9 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene
@Autowired
private TbClusterService clusterService;
@Autowired
private NotificationRuleProcessingService notificationRuleProcessingService;
private final Map<EntityId, Set<TbSubscription>> subscriptionsByEntityId = new ConcurrentHashMap<>();
private final Map<String, Map<Integer, TbSubscription>> subscriptionsByWsSessionId = new ConcurrentHashMap<>();
private final ConcurrentMap<TopicPartitionInfo, Set<TbSubscription>> 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<TbSubscription> 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()

8
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<Partitio
void onAlarmDeleted(TenantId tenantId, EntityId entityId, Alarm alarm, TbCallback callback);
void onNotificationUpdate(TenantId tenantId, UserId recipientId, Notification notification, boolean isNew, TbCallback callback);
void onNotificationUpdate(TenantId tenantId, UserId recipientId, NotificationUpdate notificationUpdate, TbCallback callback);
void onNotificationRequestDeleted(TenantId tenantId, NotificationRequestId notificationRequestId, TbCallback callback);
void onNotificationRequestUpdate(TenantId tenantId, NotificationRequestUpdate notificationRequestUpdate, TbCallback callback);
}

31
application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java

@ -19,7 +19,6 @@ import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.alarm.Alarm;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.EntityIdFactory;
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;
@ -33,7 +32,6 @@ import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.LongDataEntry;
import org.thingsboard.server.common.data.kv.StringDataEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.common.data.notification.Notification;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.KeyValueProto;
import org.thingsboard.server.gen.transport.TransportProtos.KeyValueType;
@ -53,6 +51,8 @@ import org.thingsboard.server.gen.transport.TransportProtos.TbTimeSeriesSubscrip
import org.thingsboard.server.gen.transport.TransportProtos.TbTimeSeriesUpdateProto;
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg;
import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto;
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.NotificationsCountSubscription;
import org.thingsboard.server.service.ws.notification.sub.NotificationsSubscription;
import org.thingsboard.server.service.ws.notification.sub.NotificationsSubscriptionUpdate;
@ -381,12 +381,15 @@ public class TbSubscriptionUtils {
}
public static TransportProtos.ToCoreNotificationMsg notificationsSubUpdateToProto(TbSubscription subscription, NotificationsSubscriptionUpdate update) {
TransportProtos.NotificationsSubscriptionUpdateProto updateProto = TransportProtos.NotificationsSubscriptionUpdateProto.newBuilder()
TransportProtos.NotificationsSubscriptionUpdateProto.Builder updateProto = TransportProtos.NotificationsSubscriptionUpdateProto.newBuilder()
.setSessionId(subscription.getSessionId())
.setSubscriptionId(subscription.getSubscriptionId())
.setNotification(JacksonUtil.toString(update.getNotification()))
.setIsNewNotification(update.isNewNotification())
.build();
.setSubscriptionId(subscription.getSubscriptionId());
if (update.getNotificationUpdate() != null) {
updateProto.setNotificationUpdate(JacksonUtil.toString(update.getNotificationUpdate()));
}
if (update.getNotificationRequestUpdate() != null) {
updateProto.setNotificationRequestUpdate(JacksonUtil.toString(update.getNotificationRequestUpdate()));
}
return TransportProtos.ToCoreNotificationMsg.newBuilder()
.setToLocalSubscriptionServiceMsg(TransportProtos.LocalSubscriptionServiceMsgProto.newBuilder()
.setNotificationsSubUpdate(updateProto)
@ -394,14 +397,13 @@ public class TbSubscriptionUtils {
.build();
}
public static ToCoreMsg notificationUpdateToProto(TenantId tenantId, UserId recipientId, Notification notification, boolean isNew) {
public static ToCoreMsg notificationUpdateToProto(TenantId tenantId, UserId recipientId, NotificationUpdate notificationUpdate) {
TransportProtos.NotificationUpdateProto updateProto = TransportProtos.NotificationUpdateProto.newBuilder()
.setTenantIdMSB(tenantId.getId().getMostSignificantBits())
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits())
.setRecipientIdMSB(recipientId.getId().getMostSignificantBits())
.setRecipientIdLSB(recipientId.getId().getLeastSignificantBits())
.setNotification(JacksonUtil.toString(notification))
.setIsNew(isNew)
.setUpdate(JacksonUtil.toString(notificationUpdate))
.build();
return ToCoreMsg.newBuilder()
.setToSubscriptionMgrMsg(SubscriptionMgrMsgProto.newBuilder()
@ -410,16 +412,15 @@ public class TbSubscriptionUtils {
.build();
}
public static ToCoreMsg notificationRequestDeletedToProto(TenantId tenantId, NotificationRequestId notificationRequestId) {
TransportProtos.NotificationRequestDeleteProto deleteProto = TransportProtos.NotificationRequestDeleteProto.newBuilder()
public static ToCoreMsg notificationRequestUpdateToProto(TenantId tenantId, NotificationRequestUpdate notificationRequestUpdate) {
TransportProtos.NotificationRequestUpdateProto updateProto = TransportProtos.NotificationRequestUpdateProto.newBuilder()
.setTenantIdMSB(tenantId.getId().getMostSignificantBits())
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits())
.setNotificationRequestIdMSB(notificationRequestId.getId().getMostSignificantBits())
.setNotificationRequestIdLSB(notificationRequestId.getId().getLeastSignificantBits())
.setUpdate(JacksonUtil.toString(notificationRequestUpdate))
.build();
return ToCoreMsg.newBuilder()
.setToSubscriptionMgrMsg(SubscriptionMgrMsgProto.newBuilder()
.setNotificationRequestDelete(deleteProto)
.setNotificationRequestUpdate(updateProto)
.build())
.build();
}

6
application/src/main/java/org/thingsboard/server/service/telemetry/AbstractSubscriptionService.java

@ -90,9 +90,9 @@ public abstract class AbstractSubscriptionService extends TbApplicationEventList
}
}
protected void forwardToSubscriptionManagerServiceOrSendToCore(TenantId tenantId, EntityId entityId,
Consumer<SubscriptionManagerService> toSubscriptionManagerService,
Supplier<TransportProtos.ToCoreMsg> toCore) {
protected void forwardToSubscriptionManagerService(TenantId tenantId, EntityId entityId,
Consumer<SubscriptionManagerService> toSubscriptionManagerService,
Supplier<TransportProtos.ToCoreMsg> toCore) {
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId);
if (currentPartitions.contains(tpi)) {
if (subscriptionManagerService.isPresent()) {

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

8
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<AttributeKvEntry> 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<String> 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<TsKvEntry> 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<String> keys, List<TsKvLatestRemovingResult> ts) {
forwardToSubscriptionManagerServiceOrSendToCore(tenantId, entityId, subscriptionManagerService -> {
forwardToSubscriptionManagerService(tenantId, entityId, subscriptionManagerService -> {
List<TsKvEntry> updated = new ArrayList<>();
List<String> deleted = new ArrayList<>();

100
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<UUID> 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<UUID> 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

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

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

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

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

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

22
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<NotificationRequest> findNotificationRequestsByTenantIdAndPageLink(TenantId tenantId, PageLink pageLink);
PageData<NotificationRequest> findNotificationRequestsByTenantId(TenantId tenantId, PageLink pageLink);
List<NotificationRequest> 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<Notification> findNotificationsByUserIdAndReadStatusAndPageLink(TenantId tenantId, UserId userId, boolean unreadOnly, PageLink pageLink);
PageData<Notification> findNotificationsByUserIdAndReadStatus(TenantId tenantId, UserId userId, boolean unreadOnly, PageLink pageLink);
PageData<Notification> findLatestUnreadNotificationsByUserId(TenantId tenantId, UserId userId, int limit);
int countUnreadNotificationsByUserId(TenantId tenantId, UserId userId);
int updateNotificationsInfosByRequestId(TenantId tenantId, NotificationRequestId notificationRequestId, NotificationInfo notificationInfo);
}

3
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<AlarmId> 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<String> propagateRelationTypes;
private NotificationRuleId notificationRuleId;
public Alarm() {
super();
@ -105,6 +107,7 @@ public class Alarm extends BaseData<AlarmId> implements HasName, HasTenantId, Ha
this.propagateToOwner = alarm.isPropagateToOwner();
this.propagateToTenant = alarm.isPropagateToTenant();
this.propagateRelationTypes = alarm.getPropagateRelationTypes();
this.notificationRuleId = alarm.getNotificationRuleId();
}
@Override

3
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<String> propagateRelationTypes;
private NotificationRuleId notificationRuleId;
}

36
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;
// }
}

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

11
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<NotificationRequestId> 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";

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

21
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
}

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

31
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<NotificationRuleId> 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<String, Object> triggers; // or maybe bad idea
// Map<NotificationTriggerType, NotificationTriggerConfig> - 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<NonConfirmedNotificationEscalation> escalations;
}

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

8
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<T extends Alarm> 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<T extends Alarm> extends BaseSqlEntity
} else {
this.propagateRelationTypes = null;
}
this.notificationRuleId = getUuid(alarm.getNotificationRuleId());
}
public AbstractAlarmEntity(AlarmEntity alarmEntity) {
@ -172,6 +178,7 @@ public abstract class AbstractAlarmEntity<T extends Alarm> 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<T extends Alarm> extends BaseSqlEntity
} else {
alarm.setPropagateRelationTypes(Collections.emptyList());
}
alarm.setNotificationRuleId(createId(notificationRuleId, NotificationRuleId::new));
return alarm;
}
}

19
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<NotificationRequest
@Column(name = ModelConstants.NOTIFICATION_REQUEST_ADDITIONAL_CONFIG_PROPERTY)
private JsonNode additionalConfig;
@Enumerated(EnumType.STRING)
@Column(name = ModelConstants.NOTIFICATION_REQUEST_STATUS_PROPERTY)
private NotificationRequestStatus status;
@Column(name = ModelConstants.NOTIFICATION_REQUEST_RULE_ID_PROPERTY)
private UUID ruleId;
@Column(name = ModelConstants.NOTIFICATION_REQUEST_ALARM_ID_PROPERTY)
private UUID alarmId;
public NotificationRequestEntity() {}
public NotificationRequestEntity(NotificationRequest notificationRequest) {
@ -83,6 +96,9 @@ public class NotificationRequestEntity extends BaseSqlEntity<NotificationRequest
setNotificationInfo(toJson(notificationRequest.getNotificationInfo()));
setNotificationSeverity(notificationRequest.getNotificationSeverity());
setAdditionalConfig(toJson(notificationRequest.getAdditionalConfig()));
setStatus(notificationRequest.getStatus());
setRuleId(getUuid(notificationRequest.getRuleId()));
setAlarmId(getUuid(notificationRequest.getAlarmId()));
}
@Override
@ -97,6 +113,9 @@ public class NotificationRequestEntity extends BaseSqlEntity<NotificationRequest
notificationRequest.setNotificationInfo(fromJson(notificationInfo));
notificationRequest.setNotificationSeverity(notificationSeverity);
notificationRequest.setAdditionalConfig(fromJson(additionalConfig));
notificationRequest.setStatus(status);
notificationRequest.setRuleId(createId(ruleId, NotificationRuleId::new));
notificationRequest.setAlarmId(createId(alarmId, AlarmId::new));
return notificationRequest;
}

65
dao/src/main/java/org/thingsboard/server/dao/model/sql/NotificationRuleEntity.java

@ -0,0 +1,65 @@
/**
* 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.model.sql;
import lombok.Data;
import lombok.EqualsAndHashCode;
import org.hibernate.annotations.TypeDef;
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;
import org.thingsboard.server.dao.model.BaseSqlEntity;
import org.thingsboard.server.dao.model.ModelConstants;
import org.thingsboard.server.dao.util.mapping.JsonStringType;
import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.Table;
import java.util.UUID;
@Data @EqualsAndHashCode(callSuper = true)
@Entity
@TypeDef(name = "json", typeClass = JsonStringType.class)
@Table(name = ModelConstants.NOTIFICATION_RULE_TABLE_NAME)
public class NotificationRuleEntity extends BaseSqlEntity<NotificationRule> {
@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;
}
}

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

40
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<NotificationRequest> findNotificationRequestsByTenantIdAndPageLink(TenantId tenantId, PageLink pageLink) {
public PageData<NotificationRequest> findNotificationRequestsByTenantId(TenantId tenantId, PageLink pageLink) {
return notificationRequestDao.findByTenantIdAndPageLink(tenantId, pageLink);
}
@Override
public List<NotificationRequest> 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<Notification> findNotificationsByUserIdAndReadStatusAndPageLink(TenantId tenantId, UserId userId, boolean unreadOnly, PageLink pageLink) {
public PageData<Notification> 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<Notification> 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<NotificationRequest> {
@Override
protected void validateDataImpl(TenantId tenantId, NotificationRequest notificationRequest) {
if (notificationRequest.getId() != null) {
throw new DataValidationException("Notification request cannot be changed once created");
}
}
}

6
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<Notification> {
int countUnreadByUserId(TenantId tenantId, UserId userId);
PageData<Notification> findByRequestId(TenantId tenantId, NotificationRequestId notificationRequestId, PageLink pageLink);
int updateInfosByRequestId(TenantId tenantId, NotificationRequestId notificationRequestId, NotificationInfo notificationInfo);
}

6
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<NotificationRequest> {
PageData<NotificationRequest> findByTenantIdAndPageLink(TenantId tenantId, PageLink pageLink);
List<NotificationRequest> findByRuleIdAndAlarmId(TenantId tenantId, NotificationRuleId ruleId, AlarmId alarmId);
}

22
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<NotificationRule> {
}

13
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<NotificationEntity, Notif
return notificationRepository.countByRecipientIdAndStatusNot(userId.getId(), NotificationStatus.READ);
}
@Override
public PageData<Notification> 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<NotificationEntity> getEntityClass() {
return NotificationEntity.class;

8
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<NotificationReques
Strings.nullToEmpty(pageLink.getTextSearch()), DaoUtil.toPageable(pageLink)));
}
@Override
public List<NotificationRequest> findByRuleIdAndAlarmId(TenantId tenantId, NotificationRuleId ruleId, AlarmId alarmId) {
return DaoUtil.convertDataList(notificationRequestRepository.findAllByRuleIdAndAlarmId(ruleId.getId(), alarmId.getId()));
}
@Override
protected Class<NotificationRequestEntity> getEntityClass() {
return NotificationRequestEntity.class;

46
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<NotificationRuleEntity, NotificationRule> implements NotificationRuleDao {
private final NotificationRuleRepository notificationRuleRepository;
@Override
protected Class<NotificationRuleEntity> getEntityClass() {
return NotificationRuleEntity.class;
}
@Override
protected JpaRepository<NotificationRuleEntity, UUID> getRepository() {
return notificationRuleRepository;
}
}

8
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<NotificationEntity
int countByRecipientIdAndStatusNot(UUID recipientId, NotificationStatus status);
void deleteByRequestId(UUID requestId);
Page<NotificationEntity> 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);
}

3
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<Notificatio
Page<NotificationRequestEntity> findByTenantIdAndSearchText(@Param("tenantId") UUID tenantId,
@Param("searchText") String searchText, Pageable pageable);
List<NotificationRequestEntity> findAllByRuleIdAndAlarmId(UUID ruleId, UUID alarmId);
}

26
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<NotificationRuleEntity, UUID> {
}

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

5
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 (

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

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

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

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

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

Loading…
Cancel
Save