YevhenBondarenko
3 years ago
436 changed files with 21504 additions and 1298 deletions
@ -0,0 +1,43 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.actors.ruleChain; |
|||
|
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.service.ComponentActor; |
|||
import org.thingsboard.server.actors.shared.ComponentMsgProcessor; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; |
|||
|
|||
public abstract class RuleEngineComponentActor<T extends EntityId, P extends ComponentMsgProcessor<T>> extends ComponentActor<T, P> { |
|||
|
|||
public RuleEngineComponentActor(ActorSystemContext systemContext, TenantId tenantId, T id) { |
|||
super(systemContext, tenantId, id); |
|||
} |
|||
|
|||
@Override |
|||
protected void logLifecycleEvent(ComponentLifecycleEvent event, Exception e) { |
|||
super.logLifecycleEvent(event, e); |
|||
systemContext.getNotificationRuleProcessingService().process(tenantId, getRuleChainId(), getRuleChainName(), |
|||
id, processor.getComponentName(), event, e); |
|||
} |
|||
|
|||
protected abstract RuleChainId getRuleChainId(); |
|||
|
|||
protected abstract String getRuleChainName(); |
|||
|
|||
} |
|||
@ -0,0 +1,297 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.controller; |
|||
|
|||
import io.swagger.annotations.ApiOperation; |
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.security.access.prepost.PreAuthorize; |
|||
import org.springframework.security.core.annotation.AuthenticationPrincipal; |
|||
import org.springframework.web.bind.annotation.DeleteMapping; |
|||
import org.springframework.web.bind.annotation.GetMapping; |
|||
import org.springframework.web.bind.annotation.PathVariable; |
|||
import org.springframework.web.bind.annotation.PostMapping; |
|||
import org.springframework.web.bind.annotation.PutMapping; |
|||
import org.springframework.web.bind.annotation.RequestBody; |
|||
import org.springframework.web.bind.annotation.RequestMapping; |
|||
import org.springframework.web.bind.annotation.RequestParam; |
|||
import org.springframework.web.bind.annotation.RestController; |
|||
import org.thingsboard.rule.engine.api.NotificationCenter; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.exception.ThingsboardException; |
|||
import org.thingsboard.server.common.data.id.NotificationId; |
|||
import org.thingsboard.server.common.data.id.NotificationRequestId; |
|||
import org.thingsboard.server.common.data.id.NotificationTargetId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.notification.Notification; |
|||
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod; |
|||
import org.thingsboard.server.common.data.notification.NotificationProcessingContext; |
|||
import org.thingsboard.server.common.data.notification.NotificationRequest; |
|||
import org.thingsboard.server.common.data.notification.NotificationRequestInfo; |
|||
import org.thingsboard.server.common.data.notification.NotificationRequestPreview; |
|||
import org.thingsboard.server.common.data.notification.info.UserOriginatedNotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.settings.NotificationSettings; |
|||
import org.thingsboard.server.common.data.notification.targets.NotificationTarget; |
|||
import org.thingsboard.server.common.data.notification.targets.NotificationTargetType; |
|||
import org.thingsboard.server.common.data.notification.template.DeliveryMethodNotificationTemplate; |
|||
import org.thingsboard.server.common.data.notification.template.NotificationTemplate; |
|||
import org.thingsboard.server.common.data.page.PageData; |
|||
import org.thingsboard.server.common.data.page.PageLink; |
|||
import org.thingsboard.server.dao.notification.NotificationRequestService; |
|||
import org.thingsboard.server.dao.notification.NotificationService; |
|||
import org.thingsboard.server.dao.notification.NotificationSettingsService; |
|||
import org.thingsboard.server.dao.notification.NotificationTargetService; |
|||
import org.thingsboard.server.dao.notification.NotificationTemplateService; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.security.model.SecurityUser; |
|||
import org.thingsboard.server.service.security.permission.Operation; |
|||
|
|||
import javax.validation.Valid; |
|||
import java.util.Collections; |
|||
import java.util.HashMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.UUID; |
|||
import java.util.stream.Collectors; |
|||
|
|||
import static org.thingsboard.server.service.security.permission.Resource.NOTIFICATION; |
|||
|
|||
@RestController |
|||
@TbCoreComponent |
|||
@RequestMapping("/api") |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class NotificationController extends BaseController { |
|||
|
|||
private final NotificationService notificationService; |
|||
private final NotificationRequestService notificationRequestService; |
|||
private final NotificationTemplateService notificationTemplateService; |
|||
private final NotificationTargetService notificationTargetService; |
|||
private final NotificationCenter notificationCenter; |
|||
private final NotificationSettingsService notificationSettingsService; |
|||
|
|||
@ApiOperation(value = "Get notifications (getNotifications)", |
|||
notes = "**WebSocket API**:\n\n" + |
|||
"There are 2 types of subscriptions: one for unread notifications count, another for unread notifications themselves.\n\n" + |
|||
"The URI for opening WS session for notifications: `/api/ws/plugins/notifications`.\n\n" + |
|||
"Subscription command for unread notifications count:\n" + |
|||
"```\n{\n \"unreadCountSubCmd\": {\n \"cmdId\": 1234\n }\n}\n```\n" + |
|||
"To subscribe for latest unread notifications:\n" + |
|||
"```\n{\n \"unreadSubCmd\": {\n \"cmdId\": 1234,\n \"limit\": 10\n }\n}\n```\n" + |
|||
"To unsubscribe from any subscription:\n" + |
|||
"```\n{\n \"unsubCmd\": {\n \"cmdId\": 1234\n }\n}\n```\n" + |
|||
"To mark certain notifications as read, use following command:\n" + |
|||
"```\n{\n \"markAsReadCmd\": {\n \"cmdId\": 1234,\n \"notifications\": [\n \"6f860330-7fc2-11ed-b855-7dd3b7d2faa9\",\n \"5b6dfee0-8d0d-11ed-b61f-35a57b03dade\"\n ]\n }\n}\n\n```\n" + |
|||
"To mark all notifications as read:\n" + |
|||
"```\n{\n \"markAllAsReadCmd\": {\n \"cmdId\": 1234\n }\n}\n```\n" + |
|||
"\n\n" + |
|||
"Update structure for unread **notifications count subscription**:\n" + |
|||
"```\n{\n \"cmdId\": 1234,\n \"totalUnreadCount\": 55\n}\n```\n" + |
|||
"For **notifications subscription**:\n" + |
|||
"- full update of latest unread notifications:\n" + |
|||
"```\n{\n" + |
|||
" \"cmdId\": 1234,\n" + |
|||
" \"notifications\": [\n" + |
|||
" {\n" + |
|||
" \"id\": {\n" + |
|||
" \"entityType\": \"NOTIFICATION\",\n" + |
|||
" \"id\": \"6f860330-7fc2-11ed-b855-7dd3b7d2faa9\"\n" + |
|||
" },\n" + |
|||
" ...\n" + |
|||
" }\n" + |
|||
" ],\n" + |
|||
" \"totalUnreadCount\": 1\n" + |
|||
"}\n```\n" + |
|||
"- when new notification arrives or shown notification is updated:\n" + |
|||
"```\n{\n" + |
|||
" \"cmdId\": 1234,\n" + |
|||
" \"update\": {\n" + |
|||
" \"id\": {\n" + |
|||
" \"entityType\": \"NOTIFICATION\",\n" + |
|||
" \"id\": \"6f860330-7fc2-11ed-b855-7dd3b7d2faa9\"\n" + |
|||
" },\n" + |
|||
" # updated notification info, text, subject etc.\n" + |
|||
" ...\n" + |
|||
" },\n" + |
|||
" \"totalUnreadCount\": 2\n" + |
|||
"}\n```\n" + |
|||
"- when unread notifications count changes:\n" + |
|||
"```\n{\n" + |
|||
" \"cmdId\": 1234,\n" + |
|||
" \"totalUnreadCount\": 5\n" + |
|||
"}\n```") |
|||
@GetMapping("/notifications") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") |
|||
public PageData<Notification> getNotifications(@RequestParam int pageSize, |
|||
@RequestParam int page, |
|||
@RequestParam(required = false) String textSearch, |
|||
@RequestParam(required = false) String sortProperty, |
|||
@RequestParam(required = false) String sortOrder, |
|||
@RequestParam(defaultValue = "false") boolean unreadOnly, |
|||
@AuthenticationPrincipal SecurityUser user) throws ThingsboardException { |
|||
// no permissions
|
|||
PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder); |
|||
return notificationService.findNotificationsByRecipientIdAndReadStatus(user.getTenantId(), user.getId(), unreadOnly, pageLink); |
|||
} |
|||
|
|||
@PutMapping("/notification/{id}/read") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") |
|||
public void markNotificationAsRead(@PathVariable UUID id, |
|||
@AuthenticationPrincipal SecurityUser user) { |
|||
// no permissions
|
|||
NotificationId notificationId = new NotificationId(id); |
|||
notificationCenter.markNotificationAsRead(user.getTenantId(), user.getId(), notificationId); |
|||
} |
|||
|
|||
@PutMapping("/notifications/read") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") |
|||
public void markAllNotificationsAsRead(@AuthenticationPrincipal SecurityUser user) { |
|||
// no permissions
|
|||
notificationCenter.markAllNotificationsAsRead(user.getTenantId(), user.getId()); |
|||
} |
|||
|
|||
@DeleteMapping("/notification/{id}") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") |
|||
public void deleteNotification(@PathVariable UUID id, |
|||
@AuthenticationPrincipal SecurityUser user) { |
|||
// no permissions
|
|||
NotificationId notificationId = new NotificationId(id); |
|||
notificationCenter.deleteNotification(user.getTenantId(), user.getId(), notificationId); |
|||
} |
|||
|
|||
@PostMapping("/notification/request") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public NotificationRequest createNotificationRequest(@RequestBody @Valid NotificationRequest notificationRequest, |
|||
@AuthenticationPrincipal SecurityUser user) throws Exception { |
|||
if (notificationRequest.getId() != null) { |
|||
throw new IllegalArgumentException("Notification request cannot be updated. You may only cancel/delete it"); |
|||
} |
|||
notificationRequest.setTenantId(user.getTenantId()); |
|||
checkEntity(notificationRequest.getId(), notificationRequest, NOTIFICATION); |
|||
|
|||
notificationRequest.setOriginatorEntityId(user.getId()); |
|||
if (notificationRequest.getInfo() != null && !(notificationRequest.getInfo() instanceof UserOriginatedNotificationInfo)) { |
|||
throw new IllegalArgumentException("Unsupported notification info type"); |
|||
} |
|||
notificationRequest.setRuleId(null); |
|||
notificationRequest.setStatus(null); |
|||
notificationRequest.setStats(null); |
|||
|
|||
return doSaveAndLog(EntityType.NOTIFICATION_REQUEST, notificationRequest, notificationCenter::processNotificationRequest); |
|||
} |
|||
|
|||
@PostMapping("/notification/request/preview") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public NotificationRequestPreview getNotificationRequestPreview(@RequestBody @Valid NotificationRequest request, |
|||
@AuthenticationPrincipal SecurityUser user) throws ThingsboardException { |
|||
NotificationRequestPreview preview = new NotificationRequestPreview(); |
|||
|
|||
request.setOriginatorEntityId(user.getId()); |
|||
NotificationTemplate template; |
|||
if (request.getTemplateId() != null) { |
|||
template = checkEntityId(request.getTemplateId(), notificationTemplateService::findNotificationTemplateById, Operation.READ); |
|||
} else { |
|||
template = request.getTemplate(); |
|||
} |
|||
if (template == null) { |
|||
throw new IllegalArgumentException("Template is missing"); |
|||
} |
|||
NotificationProcessingContext tmpProcessingCtx = NotificationProcessingContext.builder() |
|||
.tenantId(user.getTenantId()) |
|||
.request(request) |
|||
.settings(null) |
|||
.template(template) |
|||
.build(); |
|||
|
|||
Map<NotificationDeliveryMethod, DeliveryMethodNotificationTemplate> processedTemplates = tmpProcessingCtx.getDeliveryMethods().stream() |
|||
.collect(Collectors.toMap(m -> m, deliveryMethod -> { |
|||
Map<String, String> templateContext; |
|||
if (NotificationTargetType.PLATFORM_USERS.getSupportedDeliveryMethods().contains(deliveryMethod)) { |
|||
templateContext = tmpProcessingCtx.createTemplateContext(user); |
|||
} else { |
|||
templateContext = Collections.emptyMap(); |
|||
} |
|||
return tmpProcessingCtx.getProcessedTemplate(deliveryMethod, templateContext); |
|||
})); |
|||
preview.setProcessedTemplates(processedTemplates); |
|||
|
|||
// generic permission
|
|||
Map<String, Integer> recipientsCountByTarget = new HashMap<>(); |
|||
List<NotificationTarget> targets = notificationTargetService.findNotificationTargetsByTenantIdAndIds(user.getTenantId(), |
|||
request.getTargets().stream().map(NotificationTargetId::new).collect(Collectors.toList())); |
|||
for (NotificationTarget target : targets) { |
|||
int recipientsCount; |
|||
if (target.getConfiguration().getType() == NotificationTargetType.PLATFORM_USERS) { |
|||
recipientsCount = notificationTargetService.countRecipientsForNotificationTargetConfig(user.getTenantId(), target.getConfiguration()); |
|||
} else { |
|||
recipientsCount = 1; |
|||
} |
|||
recipientsCountByTarget.put(target.getName(), recipientsCount); |
|||
} |
|||
preview.setRecipientsCountByTarget(recipientsCountByTarget); |
|||
preview.setTotalRecipientsCount(recipientsCountByTarget.values().stream().mapToInt(Integer::intValue).sum()); |
|||
|
|||
return preview; |
|||
} |
|||
|
|||
@GetMapping("/notification/request/{id}") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public NotificationRequestInfo getNotificationRequestById(@PathVariable UUID id) throws ThingsboardException { |
|||
NotificationRequestId notificationRequestId = new NotificationRequestId(id); |
|||
return checkEntityId(notificationRequestId, notificationRequestService::findNotificationRequestInfoById, Operation.READ); |
|||
} |
|||
|
|||
@GetMapping("/notification/requests") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public PageData<NotificationRequestInfo> getNotificationRequests(@RequestParam int pageSize, |
|||
@RequestParam int page, |
|||
@RequestParam(required = false) String textSearch, |
|||
@RequestParam(required = false) String sortProperty, |
|||
@RequestParam(required = false) String sortOrder, |
|||
@AuthenticationPrincipal SecurityUser user) throws ThingsboardException { |
|||
// generic permission
|
|||
PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder); |
|||
return notificationRequestService.findNotificationRequestsInfosByTenantIdAndOriginatorType(user.getTenantId(), EntityType.USER, pageLink); |
|||
} |
|||
|
|||
@DeleteMapping("/notification/request/{id}") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public void deleteNotificationRequest(@PathVariable UUID id) throws Exception { |
|||
NotificationRequestId notificationRequestId = new NotificationRequestId(id); |
|||
NotificationRequest notificationRequest = checkEntityId(notificationRequestId, notificationRequestService::findNotificationRequestById, Operation.DELETE); |
|||
doDeleteAndLog(EntityType.NOTIFICATION_REQUEST, notificationRequest, notificationCenter::deleteNotificationRequest); |
|||
} |
|||
|
|||
|
|||
@PostMapping("/notification/settings") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public NotificationSettings saveNotificationSettings(@RequestBody @Valid NotificationSettings notificationSettings, |
|||
@AuthenticationPrincipal SecurityUser user) { |
|||
// generic permission
|
|||
TenantId tenantId = user.isSystemAdmin() ? TenantId.SYS_TENANT_ID : user.getTenantId(); |
|||
notificationSettingsService.saveNotificationSettings(tenantId, notificationSettings); |
|||
return notificationSettings; |
|||
} |
|||
|
|||
@GetMapping("/notification/settings") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public NotificationSettings getNotificationSettings(@AuthenticationPrincipal SecurityUser user) { |
|||
// generic permission
|
|||
TenantId tenantId = user.isSystemAdmin() ? TenantId.SYS_TENANT_ID : user.getTenantId(); |
|||
return notificationSettingsService.findNotificationSettings(tenantId); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,95 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.controller; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.security.access.prepost.PreAuthorize; |
|||
import org.springframework.security.core.annotation.AuthenticationPrincipal; |
|||
import org.springframework.web.bind.annotation.DeleteMapping; |
|||
import org.springframework.web.bind.annotation.GetMapping; |
|||
import org.springframework.web.bind.annotation.PathVariable; |
|||
import org.springframework.web.bind.annotation.PostMapping; |
|||
import org.springframework.web.bind.annotation.RequestBody; |
|||
import org.springframework.web.bind.annotation.RequestMapping; |
|||
import org.springframework.web.bind.annotation.RequestParam; |
|||
import org.springframework.web.bind.annotation.RestController; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.exception.ThingsboardException; |
|||
import org.thingsboard.server.common.data.id.NotificationRuleId; |
|||
import org.thingsboard.server.common.data.notification.rule.NotificationRule; |
|||
import org.thingsboard.server.common.data.notification.rule.NotificationRuleInfo; |
|||
import org.thingsboard.server.common.data.page.PageData; |
|||
import org.thingsboard.server.common.data.page.PageLink; |
|||
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; |
|||
import org.thingsboard.server.dao.notification.NotificationRuleService; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.security.model.SecurityUser; |
|||
import org.thingsboard.server.service.security.permission.Operation; |
|||
|
|||
import javax.validation.Valid; |
|||
import java.util.UUID; |
|||
|
|||
import static org.thingsboard.server.service.security.permission.Resource.NOTIFICATION; |
|||
|
|||
@RestController |
|||
@TbCoreComponent |
|||
@RequestMapping("/api/notification") |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class NotificationRuleController extends BaseController { |
|||
|
|||
private final NotificationRuleService notificationRuleService; |
|||
|
|||
@PostMapping("/rule") |
|||
@PreAuthorize("hasAnyAuthority('TENANT_ADMIN')") |
|||
public NotificationRule saveNotificationRule(@RequestBody @Valid NotificationRule notificationRule) throws Exception { |
|||
notificationRule.setTenantId(getTenantId()); |
|||
checkEntity(notificationRule.getId(), notificationRule, NOTIFICATION); |
|||
return doSaveAndLog(EntityType.NOTIFICATION_RULE, notificationRule, notificationRuleService::saveNotificationRule); |
|||
} |
|||
|
|||
@GetMapping("/rule/{id}") |
|||
@PreAuthorize("hasAnyAuthority('TENANT_ADMIN')") |
|||
public NotificationRuleInfo getNotificationRuleById(@PathVariable UUID id) throws ThingsboardException { |
|||
NotificationRuleId notificationRuleId = new NotificationRuleId(id); |
|||
return checkEntityId(notificationRuleId, notificationRuleService::findNotificationRuleInfoById, Operation.READ); |
|||
} |
|||
|
|||
@GetMapping("/rules") |
|||
@PreAuthorize("hasAnyAuthority('TENANT_ADMIN')") |
|||
public PageData<NotificationRuleInfo> getNotificationRules(@RequestParam int pageSize, |
|||
@RequestParam int page, |
|||
@RequestParam(required = false) String textSearch, |
|||
@RequestParam(required = false) String sortProperty, |
|||
@RequestParam(required = false) String sortOrder, |
|||
@AuthenticationPrincipal SecurityUser user) throws ThingsboardException { |
|||
// generic permission
|
|||
PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder); |
|||
return notificationRuleService.findNotificationRulesInfosByTenantId(user.getTenantId(), pageLink); |
|||
} |
|||
|
|||
@DeleteMapping("/rule/{id}") |
|||
@PreAuthorize("hasAnyAuthority('TENANT_ADMIN')") |
|||
public void deleteNotificationRule(@PathVariable UUID id, |
|||
@AuthenticationPrincipal SecurityUser user) throws Exception { |
|||
NotificationRuleId notificationRuleId = new NotificationRuleId(id); |
|||
NotificationRule notificationRule = checkEntityId(notificationRuleId, notificationRuleService::findNotificationRuleById, Operation.DELETE); |
|||
doDeleteAndLog(EntityType.NOTIFICATION_RULE, notificationRule, notificationRuleService::deleteNotificationRuleById); |
|||
tbClusterService.broadcastEntityStateChangeEvent(user.getTenantId(), notificationRuleId, ComponentLifecycleEvent.DELETED); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,174 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.controller; |
|||
|
|||
import io.swagger.annotations.ApiOperation; |
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.security.access.prepost.PreAuthorize; |
|||
import org.springframework.security.core.annotation.AuthenticationPrincipal; |
|||
import org.springframework.web.bind.annotation.DeleteMapping; |
|||
import org.springframework.web.bind.annotation.GetMapping; |
|||
import org.springframework.web.bind.annotation.PathVariable; |
|||
import org.springframework.web.bind.annotation.PostMapping; |
|||
import org.springframework.web.bind.annotation.RequestBody; |
|||
import org.springframework.web.bind.annotation.RequestMapping; |
|||
import org.springframework.web.bind.annotation.RequestParam; |
|||
import org.springframework.web.bind.annotation.RestController; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.User; |
|||
import org.thingsboard.server.common.data.exception.ThingsboardException; |
|||
import org.thingsboard.server.common.data.id.CustomerId; |
|||
import org.thingsboard.server.common.data.id.NotificationTargetId; |
|||
import org.thingsboard.server.common.data.notification.targets.NotificationTarget; |
|||
import org.thingsboard.server.common.data.notification.targets.NotificationTargetConfig; |
|||
import org.thingsboard.server.common.data.notification.targets.NotificationTargetType; |
|||
import org.thingsboard.server.common.data.notification.targets.platform.CustomerUsersFilter; |
|||
import org.thingsboard.server.common.data.notification.targets.platform.PlatformUsersNotificationTargetConfig; |
|||
import org.thingsboard.server.common.data.notification.targets.platform.UsersFilter; |
|||
import org.thingsboard.server.common.data.notification.targets.platform.UsersFilterType; |
|||
import org.thingsboard.server.common.data.page.PageData; |
|||
import org.thingsboard.server.common.data.page.PageDataIterable; |
|||
import org.thingsboard.server.common.data.page.PageLink; |
|||
import org.thingsboard.server.dao.notification.NotificationTargetService; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.security.model.SecurityUser; |
|||
import org.thingsboard.server.service.security.permission.Operation; |
|||
|
|||
import javax.validation.Valid; |
|||
import java.util.Arrays; |
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
import java.util.stream.Collectors; |
|||
|
|||
import static org.thingsboard.server.controller.ControllerConstants.SYSTEM_OR_TENANT_AUTHORITY_PARAGRAPH; |
|||
import static org.thingsboard.server.service.security.permission.Resource.NOTIFICATION; |
|||
|
|||
@RestController |
|||
@TbCoreComponent |
|||
@RequestMapping("/api/notification") |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class NotificationTargetController extends BaseController { |
|||
|
|||
private final NotificationTargetService notificationTargetService; |
|||
|
|||
@ApiOperation(value = "Save notification target (saveNotificationTarget)", |
|||
notes = "Create or update notification target.\n\n" + |
|||
SYSTEM_OR_TENANT_AUTHORITY_PARAGRAPH) |
|||
@PostMapping("/target") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public NotificationTarget saveNotificationTarget(@RequestBody @Valid NotificationTarget notificationTarget, |
|||
@AuthenticationPrincipal SecurityUser user) throws Exception { |
|||
notificationTarget.setTenantId(user.getTenantId()); |
|||
checkEntity(notificationTarget.getId(), notificationTarget, NOTIFICATION); |
|||
|
|||
NotificationTargetConfig targetConfig = notificationTarget.getConfiguration(); |
|||
if (targetConfig.getType() == NotificationTargetType.PLATFORM_USERS) { |
|||
checkTargetUsers(user, targetConfig); |
|||
} |
|||
|
|||
return doSaveAndLog(EntityType.NOTIFICATION_TARGET, notificationTarget, notificationTargetService::saveNotificationTarget); |
|||
} |
|||
|
|||
@ApiOperation(value = "Get notification target by id (getNotificationTargetById)", |
|||
notes = "Fetch saved notification target by id." + |
|||
SYSTEM_OR_TENANT_AUTHORITY_PARAGRAPH) |
|||
@GetMapping("/target/{id}") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public NotificationTarget getNotificationTargetById(@PathVariable UUID id) throws ThingsboardException { |
|||
NotificationTargetId notificationTargetId = new NotificationTargetId(id); |
|||
return checkEntityId(notificationTargetId, notificationTargetService::findNotificationTargetById, Operation.READ); |
|||
} |
|||
|
|||
@ApiOperation(value = "Get recipients for notification target config (getRecipientsForNotificationTargetConfig)", |
|||
notes = "Get the list (page) of recipients (users) for such notification target configuration." + |
|||
SYSTEM_OR_TENANT_AUTHORITY_PARAGRAPH) |
|||
@PostMapping("/target/recipients") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public PageData<User> getRecipientsForNotificationTargetConfig(@RequestBody NotificationTarget notificationTarget, |
|||
@RequestParam int pageSize, |
|||
@RequestParam int page, |
|||
@AuthenticationPrincipal SecurityUser user) throws ThingsboardException { |
|||
// generic permission
|
|||
NotificationTargetConfig targetConfig = notificationTarget.getConfiguration(); |
|||
if (targetConfig.getType() == NotificationTargetType.PLATFORM_USERS) { |
|||
checkTargetUsers(user, targetConfig); |
|||
} else { |
|||
throw new IllegalArgumentException("Target type is not platform users"); |
|||
} |
|||
|
|||
PageLink pageLink = createPageLink(pageSize, page, null, null, null); |
|||
return notificationTargetService.findRecipientsForNotificationTargetConfig(user.getTenantId(), null, notificationTarget.getConfiguration(), pageLink); |
|||
} |
|||
|
|||
@GetMapping(value = "/targets", params = {"ids"}) |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public List<NotificationTarget> getNotificationTargetsByIds(@RequestParam("ids") UUID[] ids, |
|||
@AuthenticationPrincipal SecurityUser user) { |
|||
// generic permission
|
|||
List<NotificationTargetId> targetsIds = Arrays.stream(ids).map(NotificationTargetId::new).collect(Collectors.toList()); |
|||
return notificationTargetService.findNotificationTargetsByTenantIdAndIds(user.getTenantId(), targetsIds); |
|||
} |
|||
|
|||
@ApiOperation(value = "Get notification targets (getNotificationTargets)", |
|||
notes = "Fetch the page of notification targets owned by sysadmin or tenant." + |
|||
SYSTEM_OR_TENANT_AUTHORITY_PARAGRAPH) |
|||
@GetMapping("/targets") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public PageData<NotificationTarget> getNotificationTargets(@RequestParam int pageSize, |
|||
@RequestParam int page, |
|||
@RequestParam(required = false) String textSearch, |
|||
@RequestParam(required = false) String sortProperty, |
|||
@RequestParam(required = false) String sortOrder, |
|||
@AuthenticationPrincipal SecurityUser user) throws ThingsboardException { |
|||
// generic permission
|
|||
PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder); |
|||
return notificationTargetService.findNotificationTargetsByTenantId(user.getTenantId(), pageLink); |
|||
} |
|||
|
|||
@ApiOperation(value = "Delete notification target by id (deleteNotificationTargetById)", |
|||
notes = "Delete notification target by its id.\n\n" + |
|||
"This target cannot be referenced by existing scheduled notification requests or any notification rules." + |
|||
SYSTEM_OR_TENANT_AUTHORITY_PARAGRAPH) |
|||
@DeleteMapping("/target/{id}") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public void deleteNotificationTargetById(@PathVariable UUID id) throws Exception { |
|||
NotificationTargetId notificationTargetId = new NotificationTargetId(id); |
|||
NotificationTarget notificationTarget = checkEntityId(notificationTargetId, notificationTargetService::findNotificationTargetById, Operation.DELETE); |
|||
doDeleteAndLog(EntityType.NOTIFICATION_TARGET, notificationTarget, notificationTargetService::deleteNotificationTargetById); |
|||
} |
|||
|
|||
private void checkTargetUsers(SecurityUser user, NotificationTargetConfig targetConfig) throws ThingsboardException { |
|||
if (user.isSystemAdmin()) { |
|||
return; |
|||
} |
|||
// generic permission for users
|
|||
UsersFilter usersFilter = ((PlatformUsersNotificationTargetConfig) targetConfig).getUsersFilter(); |
|||
if (usersFilter.getType() == UsersFilterType.USER_LIST) { |
|||
PageDataIterable<User> recipients = new PageDataIterable<>(pageLink -> { |
|||
return notificationTargetService.findRecipientsForNotificationTargetConfig(user.getTenantId(), null, targetConfig, pageLink); |
|||
}, 200); |
|||
for (User recipient : recipients) { |
|||
checkEntity(user, recipient, Operation.READ); |
|||
} |
|||
} else if (usersFilter.getType() == UsersFilterType.CUSTOMER_USERS) { |
|||
CustomerId customerId = new CustomerId(((CustomerUsersFilter) usersFilter).getCustomerId()); |
|||
checkEntityId(customerId, Operation.READ); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,151 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.controller; |
|||
|
|||
import io.swagger.annotations.ApiOperation; |
|||
import lombok.RequiredArgsConstructor; |
|||
import org.springframework.security.access.prepost.PreAuthorize; |
|||
import org.springframework.security.core.annotation.AuthenticationPrincipal; |
|||
import org.springframework.web.bind.annotation.DeleteMapping; |
|||
import org.springframework.web.bind.annotation.GetMapping; |
|||
import org.springframework.web.bind.annotation.PathVariable; |
|||
import org.springframework.web.bind.annotation.PostMapping; |
|||
import org.springframework.web.bind.annotation.RequestBody; |
|||
import org.springframework.web.bind.annotation.RequestMapping; |
|||
import org.springframework.web.bind.annotation.RequestParam; |
|||
import org.springframework.web.bind.annotation.RestController; |
|||
import org.thingsboard.rule.engine.api.slack.SlackService; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.exception.ThingsboardException; |
|||
import org.thingsboard.server.common.data.id.NotificationTemplateId; |
|||
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod; |
|||
import org.thingsboard.server.common.data.notification.NotificationType; |
|||
import org.thingsboard.server.common.data.notification.settings.NotificationSettings; |
|||
import org.thingsboard.server.common.data.notification.settings.SlackNotificationDeliveryMethodConfig; |
|||
import org.thingsboard.server.common.data.notification.targets.slack.SlackConversationType; |
|||
import org.thingsboard.server.common.data.notification.template.NotificationTemplate; |
|||
import org.thingsboard.server.common.data.notification.targets.slack.SlackConversation; |
|||
import org.thingsboard.server.common.data.page.PageData; |
|||
import org.thingsboard.server.common.data.page.PageLink; |
|||
import org.thingsboard.server.dao.notification.NotificationSettingsService; |
|||
import org.thingsboard.server.dao.notification.NotificationTemplateService; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.security.model.SecurityUser; |
|||
import org.thingsboard.server.service.security.permission.Operation; |
|||
|
|||
import javax.validation.Valid; |
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
|
|||
import static org.thingsboard.server.controller.ControllerConstants.SYSTEM_OR_TENANT_AUTHORITY_PARAGRAPH; |
|||
import static org.thingsboard.server.service.security.permission.Resource.NOTIFICATION; |
|||
|
|||
@RestController |
|||
@TbCoreComponent |
|||
@RequiredArgsConstructor |
|||
@RequestMapping("/api/notification") |
|||
public class NotificationTemplateController extends BaseController { |
|||
|
|||
private final NotificationTemplateService notificationTemplateService; |
|||
private final NotificationSettingsService notificationSettingsService; |
|||
private final SlackService slackService; |
|||
|
|||
@ApiOperation(value = "Save notification template (saveNotificationTemplate)", |
|||
notes = "Create or update notification template.\n\n" + |
|||
"Example:\n" + |
|||
"```\n{\n \"name\": \"Hello to all my users\",\n" + |
|||
" \"notificationType\": \"Message from administrator\",\n" + |
|||
" \"configuration\": {\n" + |
|||
" \"defaultTextTemplate\": \"Hello everyone\", # required if any of the templates' bodies is not set\n" + |
|||
" \"templates\": {\n" + |
|||
" \"PUSH\": {\n \"method\": \"PUSH\",\n \"body\": null # defaultTextTemplate will be used if body is not set\n },\n" + |
|||
" \"SMS\": {\n \"method\": \"SMS\",\n \"body\": null\n },\n" + |
|||
" \"EMAIL\": {\n \"method\": \"EMAIL\",\n \"body\": \"Non-default value for email notification: <body>Hello everyone</body>\",\n \"subject\": \"Message from administrator\"\n },\n" + |
|||
" \"SLACK\": {\n \"method\": \"SLACK\",\n \"body\": null,\n \"conversationType\": \"PUBLIC_CHANNEL\",\n \"conversationId\": \"U02LD7BJOU2\" # received from listSlackConversations API method\n }\n" + |
|||
" }\n" + |
|||
" }\n}\n```" + |
|||
SYSTEM_OR_TENANT_AUTHORITY_PARAGRAPH) |
|||
@PostMapping("/template") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public NotificationTemplate saveNotificationTemplate(@RequestBody @Valid NotificationTemplate notificationTemplate) throws Exception { |
|||
notificationTemplate.setTenantId(getTenantId()); |
|||
checkEntity(notificationTemplate.getId(), notificationTemplate, NOTIFICATION); |
|||
return doSaveAndLog(EntityType.NOTIFICATION_TEMPLATE, notificationTemplate, notificationTemplateService::saveNotificationTemplate); |
|||
} |
|||
|
|||
@ApiOperation(value = "Get notification template by id (getNotificationTemplateById)", |
|||
notes = "Fetch notification template by id." + |
|||
SYSTEM_OR_TENANT_AUTHORITY_PARAGRAPH) |
|||
@GetMapping("/template/{id}") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public NotificationTemplate getNotificationTemplateById(@PathVariable UUID id) throws ThingsboardException { |
|||
NotificationTemplateId notificationTemplateId = new NotificationTemplateId(id); |
|||
return checkEntityId(notificationTemplateId, notificationTemplateService::findNotificationTemplateById, Operation.READ); |
|||
} |
|||
|
|||
@ApiOperation(value = "Get notification templates (getNotificationTemplates)", |
|||
notes = "Fetch the page of notification templates owned by sysadmin or tenant." + |
|||
SYSTEM_OR_TENANT_AUTHORITY_PARAGRAPH) |
|||
@GetMapping("/templates") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public PageData<NotificationTemplate> getNotificationTemplates(@RequestParam int pageSize, |
|||
@RequestParam int page, |
|||
@RequestParam(required = false) String textSearch, |
|||
@RequestParam(required = false) String sortProperty, |
|||
@RequestParam(required = false) String sortOrder, |
|||
@RequestParam(required = false) NotificationType[] notificationTypes, |
|||
@AuthenticationPrincipal SecurityUser user) throws ThingsboardException { |
|||
// generic permission
|
|||
PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder); |
|||
if (notificationTypes == null || notificationTypes.length == 0) { |
|||
notificationTypes = NotificationType.values(); |
|||
} |
|||
return notificationTemplateService.findNotificationTemplatesByTenantIdAndNotificationTypes(user.getTenantId(), |
|||
List.of(notificationTypes), pageLink); |
|||
} |
|||
|
|||
@ApiOperation(value = "Delete notification template by id (deleteNotificationTemplateById", |
|||
notes = "Delete notification template by its id.\n\n" + |
|||
"This template cannot be referenced by existing scheduled notification requests or any notification rules." + |
|||
SYSTEM_OR_TENANT_AUTHORITY_PARAGRAPH) |
|||
@DeleteMapping("/template/{id}") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public void deleteNotificationTemplateById(@PathVariable UUID id) throws Exception { |
|||
NotificationTemplateId notificationTemplateId = new NotificationTemplateId(id); |
|||
NotificationTemplate notificationTemplate = checkEntityId(notificationTemplateId, notificationTemplateService::findNotificationTemplateById, Operation.DELETE); |
|||
doDeleteAndLog(EntityType.NOTIFICATION_TEMPLATE, notificationTemplate, notificationTemplateService::deleteNotificationTemplateById); |
|||
} |
|||
|
|||
@ApiOperation(value = "List Slack conversations (listSlackConversations)", |
|||
notes = "List available Slack conversations by type to use in notification template.\n\n" + |
|||
"Slack must be configured in notification settings." + |
|||
SYSTEM_OR_TENANT_AUTHORITY_PARAGRAPH) |
|||
@GetMapping("/slack/conversations") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public List<SlackConversation> listSlackConversations(@RequestParam SlackConversationType type, |
|||
@AuthenticationPrincipal SecurityUser user) { |
|||
// generic permission
|
|||
NotificationSettings settings = notificationSettingsService.findNotificationSettings(user.getTenantId()); |
|||
SlackNotificationDeliveryMethodConfig slackConfig = (SlackNotificationDeliveryMethodConfig) |
|||
settings.getDeliveryMethodsConfigs().get(NotificationDeliveryMethod.SLACK); |
|||
if (slackConfig == null) { |
|||
throw new IllegalArgumentException("Slack is not configured"); |
|||
} |
|||
|
|||
return slackService.listConversations(user.getTenantId(), slackConfig.getBotToken(), type); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,33 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.executors; |
|||
|
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.common.util.AbstractListeningExecutor; |
|||
|
|||
@Component |
|||
public class NotificationExecutorService extends AbstractListeningExecutor { |
|||
|
|||
@Value("${notification_system.thread_pool_size}") |
|||
private int notificationSystemExecutorThreadPoolSize; |
|||
|
|||
@Override |
|||
protected int getThreadPollSize() { |
|||
return notificationSystemExecutorThreadPoolSize; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,414 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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 lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.common.util.DonAsynchron; |
|||
import org.thingsboard.rule.engine.api.NotificationCenter; |
|||
import org.thingsboard.server.common.data.User; |
|||
import org.thingsboard.server.common.data.id.NotificationId; |
|||
import org.thingsboard.server.common.data.id.NotificationRequestId; |
|||
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.AlreadySentException; |
|||
import org.thingsboard.server.common.data.notification.Notification; |
|||
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod; |
|||
import org.thingsboard.server.common.data.notification.NotificationProcessingContext; |
|||
import org.thingsboard.server.common.data.notification.NotificationRequest; |
|||
import org.thingsboard.server.common.data.notification.NotificationRequestConfig; |
|||
import org.thingsboard.server.common.data.notification.NotificationRequestStats; |
|||
import org.thingsboard.server.common.data.notification.NotificationRequestStatus; |
|||
import org.thingsboard.server.common.data.notification.NotificationStatus; |
|||
import org.thingsboard.server.common.data.notification.NotificationType; |
|||
import org.thingsboard.server.common.data.notification.settings.NotificationSettings; |
|||
import org.thingsboard.server.common.data.notification.targets.NotificationRecipient; |
|||
import org.thingsboard.server.common.data.notification.targets.NotificationTarget; |
|||
import org.thingsboard.server.common.data.notification.targets.slack.SlackNotificationTargetConfig; |
|||
import org.thingsboard.server.common.data.notification.template.DeliveryMethodNotificationTemplate; |
|||
import org.thingsboard.server.common.data.notification.template.NotificationTemplate; |
|||
import org.thingsboard.server.common.data.notification.template.PushDeliveryMethodNotificationTemplate; |
|||
import org.thingsboard.server.common.data.page.PageDataIterable; |
|||
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; |
|||
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.notification.NotificationRequestService; |
|||
import org.thingsboard.server.dao.notification.NotificationService; |
|||
import org.thingsboard.server.dao.notification.NotificationSettingsService; |
|||
import org.thingsboard.server.dao.notification.NotificationTargetService; |
|||
import org.thingsboard.server.dao.notification.NotificationTemplateService; |
|||
import org.thingsboard.server.gen.transport.TransportProtos; |
|||
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.queue.discovery.NotificationsTopicService; |
|||
import org.thingsboard.server.queue.provider.TbQueueProducerProvider; |
|||
import org.thingsboard.server.service.executors.DbCallbackExecutorService; |
|||
import org.thingsboard.server.service.executors.NotificationExecutorService; |
|||
import org.thingsboard.server.service.notification.channels.NotificationChannel; |
|||
import org.thingsboard.server.service.subscription.TbSubscriptionUtils; |
|||
import org.thingsboard.server.service.telemetry.AbstractSubscriptionService; |
|||
import org.thingsboard.server.service.ws.notification.sub.NotificationRequestUpdate; |
|||
import org.thingsboard.server.service.ws.notification.sub.NotificationUpdate; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.Collections; |
|||
import java.util.HashSet; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Set; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@Service |
|||
@Slf4j |
|||
@RequiredArgsConstructor |
|||
@SuppressWarnings({"UnstableApiUsage", "rawtypes"}) |
|||
public class DefaultNotificationCenter extends AbstractSubscriptionService implements NotificationCenter, NotificationChannel<User, PushDeliveryMethodNotificationTemplate> { |
|||
|
|||
private final NotificationTargetService notificationTargetService; |
|||
private final NotificationRequestService notificationRequestService; |
|||
private final NotificationService notificationService; |
|||
private final NotificationTemplateService notificationTemplateService; |
|||
private final NotificationSettingsService notificationSettingsService; |
|||
private final NotificationExecutorService notificationExecutor; |
|||
private final DbCallbackExecutorService dbCallbackExecutorService; |
|||
private final NotificationsTopicService notificationsTopicService; |
|||
private final TbQueueProducerProvider producerProvider; |
|||
private Map<NotificationDeliveryMethod, NotificationChannel> channels; |
|||
|
|||
|
|||
@Override |
|||
public NotificationRequest processNotificationRequest(TenantId tenantId, NotificationRequest notificationRequest) { |
|||
NotificationSettings settings = notificationSettingsService.findNotificationSettings(tenantId); |
|||
NotificationTemplate notificationTemplate; |
|||
if (notificationRequest.getTemplateId() != null) { |
|||
notificationTemplate = notificationTemplateService.findNotificationTemplateById(tenantId, notificationRequest.getTemplateId()); |
|||
} else { |
|||
notificationTemplate = notificationRequest.getTemplate(); |
|||
} |
|||
if (notificationTemplate == null) throw new IllegalArgumentException("Template is missing"); |
|||
|
|||
List<NotificationTarget> targets = notificationTargetService.findNotificationTargetsByTenantIdAndIds(tenantId, |
|||
notificationRequest.getTargets().stream().map(NotificationTargetId::new).collect(Collectors.toList())); |
|||
|
|||
notificationTemplate.getConfiguration().getDeliveryMethodsTemplates().forEach((deliveryMethod, template) -> { |
|||
if (!template.isEnabled()) return; |
|||
if (deliveryMethod == NotificationDeliveryMethod.SLACK) { |
|||
if (!settings.getDeliveryMethodsConfigs().containsKey(deliveryMethod)) { |
|||
throw new IllegalArgumentException("Slack must be configured in the settings"); |
|||
} |
|||
} |
|||
if (notificationRequest.getRuleId() == null) { |
|||
if (targets.stream().noneMatch(target -> target.getConfiguration().getType().getSupportedDeliveryMethods().contains(deliveryMethod))) { |
|||
throw new IllegalArgumentException("Target for " + deliveryMethod.getName() + " delivery method is missing"); |
|||
} |
|||
} |
|||
}); |
|||
|
|||
if (notificationRequest.getAdditionalConfig() != null) { |
|||
NotificationRequestConfig config = notificationRequest.getAdditionalConfig(); |
|||
if (config.getSendingDelayInSec() > 0 && notificationRequest.getId() == null) { |
|||
notificationRequest.setStatus(NotificationRequestStatus.SCHEDULED); |
|||
NotificationRequest savedNotificationRequest = notificationRequestService.saveNotificationRequest(tenantId, notificationRequest); |
|||
forwardToNotificationSchedulerService(tenantId, savedNotificationRequest.getId()); |
|||
return savedNotificationRequest; |
|||
} |
|||
} |
|||
|
|||
log.debug("Processing notification request (tenantId: {}, targets: {})", tenantId, notificationRequest.getTargets()); |
|||
notificationRequest.setStatus(NotificationRequestStatus.PROCESSING); |
|||
NotificationRequest savedNotificationRequest = notificationRequestService.saveNotificationRequest(tenantId, notificationRequest); |
|||
|
|||
NotificationProcessingContext ctx = NotificationProcessingContext.builder() |
|||
.tenantId(tenantId) |
|||
.request(savedNotificationRequest) |
|||
.settings(settings) |
|||
.template(notificationTemplate) |
|||
.build(); |
|||
|
|||
notificationExecutor.submit(() -> { |
|||
List<ListenableFuture<Void>> results = new ArrayList<>(); |
|||
|
|||
for (NotificationTarget target : targets) { |
|||
List<ListenableFuture<Void>> result = processForTarget(target, ctx); |
|||
results.addAll(result); |
|||
} |
|||
|
|||
Futures.whenAllComplete(results).run(() -> { |
|||
NotificationRequestId requestId = savedNotificationRequest.getId(); |
|||
log.debug("[{}] Notification request processing is finished", requestId); |
|||
NotificationRequestStats stats = ctx.getStats(); |
|||
try { |
|||
notificationRequestService.updateNotificationRequest(tenantId, requestId, NotificationRequestStatus.SENT, stats); |
|||
} catch (Exception e) { |
|||
log.error("[{}] Failed to update stats for notification request", requestId, e); |
|||
} |
|||
|
|||
UserId senderId = notificationRequest.getSenderId(); |
|||
if (senderId != null) { |
|||
if (stats.getErrors().isEmpty()) { |
|||
int sent = stats.getSent().values().stream().mapToInt(AtomicInteger::get).sum(); |
|||
sendBasicNotification(tenantId, senderId, "Notifications sent", |
|||
"All notifications were successfully sent (" + sent + ")"); |
|||
} else { |
|||
int failures = stats.getErrors().values().stream().mapToInt(Map::size).sum(); |
|||
sendBasicNotification(tenantId, senderId, "Notification failure", |
|||
"Some notifications were not sent (" + failures + ")"); // TODO: 'Go to' button
|
|||
} |
|||
} |
|||
}, dbCallbackExecutorService); |
|||
}); |
|||
|
|||
return savedNotificationRequest; |
|||
} |
|||
|
|||
private List<ListenableFuture<Void>> processForTarget(NotificationTarget target, NotificationProcessingContext ctx) { |
|||
Iterable<? extends NotificationRecipient> recipients; |
|||
switch (target.getConfiguration().getType()) { |
|||
case PLATFORM_USERS: { |
|||
recipients = new PageDataIterable<>(pageLink -> { |
|||
return notificationTargetService.findRecipientsForNotificationTargetConfig(ctx.getTenantId(), ctx.getCustomerId(), target.getConfiguration(), pageLink); |
|||
}, 200); |
|||
break; |
|||
} |
|||
case SLACK: { |
|||
SlackNotificationTargetConfig slackTargetConfig = (SlackNotificationTargetConfig) target.getConfiguration(); |
|||
recipients = List.of(slackTargetConfig.getConversation()); |
|||
break; |
|||
} |
|||
default: { |
|||
recipients = Collections.emptyList(); |
|||
} |
|||
} |
|||
|
|||
Set<NotificationDeliveryMethod> deliveryMethods = new HashSet<>(ctx.getDeliveryMethods()); |
|||
deliveryMethods.removeIf(deliveryMethod -> !target.getConfiguration().getType().getSupportedDeliveryMethods().contains(deliveryMethod)); |
|||
log.debug("[{}] Processing notification request for {} target ({}) for delivery methods {}", ctx.getRequest().getId(), target.getConfiguration().getType(), target.getId(), deliveryMethods); |
|||
|
|||
List<ListenableFuture<Void>> results = new ArrayList<>(); |
|||
if (!deliveryMethods.isEmpty()) { |
|||
for (NotificationRecipient recipient : recipients) { |
|||
for (NotificationDeliveryMethod deliveryMethod : deliveryMethods) { |
|||
ListenableFuture<Void> resultFuture = processForRecipient(deliveryMethod, recipient, ctx); |
|||
DonAsynchron.withCallback(resultFuture, result -> { |
|||
ctx.getStats().reportSent(deliveryMethod, recipient); |
|||
}, error -> { |
|||
ctx.getStats().reportError(deliveryMethod, error, recipient); |
|||
}); |
|||
results.add(resultFuture); |
|||
} |
|||
} |
|||
} |
|||
return results; |
|||
} |
|||
|
|||
private ListenableFuture<Void> processForRecipient(NotificationDeliveryMethod deliveryMethod, NotificationRecipient recipient, NotificationProcessingContext ctx) { |
|||
if (ctx.getStats().contains(deliveryMethod, recipient.getId())) { |
|||
return Futures.immediateFailedFuture(new AlreadySentException()); |
|||
} |
|||
Map<String, String> templateContext; |
|||
if (recipient instanceof User) { |
|||
templateContext = ctx.createTemplateContext(((User) recipient)); |
|||
} else { |
|||
templateContext = Collections.emptyMap(); |
|||
} |
|||
DeliveryMethodNotificationTemplate processedTemplate; |
|||
try { |
|||
processedTemplate = ctx.getProcessedTemplate(deliveryMethod, templateContext); |
|||
} catch (Exception e) { |
|||
return Futures.immediateFailedFuture(e); |
|||
} |
|||
|
|||
NotificationChannel notificationChannel = channels.get(deliveryMethod); |
|||
log.trace("[{}] Sending {} notification for recipient {}", ctx.getRequest().getId(), deliveryMethod, recipient); |
|||
return notificationChannel.sendNotification(recipient, processedTemplate, ctx); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> sendNotification(User recipient, PushDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) { |
|||
NotificationRequest request = ctx.getRequest(); |
|||
Notification notification = Notification.builder() |
|||
.requestId(request.getId()) |
|||
.recipientId(recipient.getId()) |
|||
.type(ctx.getNotificationTemplate().getNotificationType()) |
|||
.subject(processedTemplate.getSubject()) |
|||
.text(processedTemplate.getBody()) |
|||
.additionalConfig(processedTemplate.getAdditionalConfig()) |
|||
.info(request.getInfo()) |
|||
.status(NotificationStatus.SENT) |
|||
.build(); |
|||
try { |
|||
notification = notificationService.saveNotification(recipient.getTenantId(), notification); |
|||
} catch (Exception e) { |
|||
log.error("Failed to create notification for recipient {}", recipient.getId(), e); |
|||
return Futures.immediateFailedFuture(e); |
|||
} |
|||
|
|||
NotificationUpdate update = NotificationUpdate.builder() |
|||
.notification(notification) |
|||
.updateType(ComponentLifecycleEvent.CREATED) |
|||
.build(); |
|||
return onNotificationUpdate(recipient.getTenantId(), recipient.getId(), update); |
|||
} |
|||
|
|||
@Override |
|||
public void sendBasicNotification(TenantId tenantId, UserId recipientId, String subject, String text) { |
|||
Notification notification = Notification.builder() |
|||
.recipientId(recipientId) |
|||
.type(NotificationType.GENERAL) |
|||
.subject(subject) |
|||
.text(text) |
|||
.status(NotificationStatus.SENT) |
|||
.build(); |
|||
notification = notificationService.saveNotification(TenantId.SYS_TENANT_ID, notification); |
|||
|
|||
NotificationUpdate update = NotificationUpdate.builder() |
|||
.notification(notification) |
|||
.updateType(ComponentLifecycleEvent.CREATED) |
|||
.build(); |
|||
onNotificationUpdate(tenantId, recipientId, update); |
|||
} |
|||
|
|||
@Override |
|||
public void markNotificationAsRead(TenantId tenantId, UserId recipientId, NotificationId notificationId) { |
|||
boolean updated = notificationService.markNotificationAsRead(tenantId, recipientId, notificationId); |
|||
if (updated) { |
|||
log.trace("Marked notification {} as read (recipient id: {}, tenant id: {})", notificationId, recipientId, tenantId); |
|||
NotificationUpdate update = NotificationUpdate.builder() |
|||
.notificationId(notificationId) |
|||
.updatedStatus(NotificationStatus.READ) |
|||
.updateType(ComponentLifecycleEvent.UPDATED) |
|||
.build(); |
|||
onNotificationUpdate(tenantId, recipientId, update); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void markAllNotificationsAsRead(TenantId tenantId, UserId recipientId) { |
|||
int updatedCount = notificationService.markAllNotificationsAsRead(tenantId, recipientId); |
|||
if (updatedCount > 0) { |
|||
log.trace("Marked all notifications as read (recipient id: {}, tenant id: {})", recipientId, tenantId); |
|||
NotificationUpdate update = NotificationUpdate.builder() |
|||
.allNotifications(true) |
|||
.updatedStatus(NotificationStatus.READ) |
|||
.updateType(ComponentLifecycleEvent.UPDATED) |
|||
.build(); |
|||
onNotificationUpdate(tenantId, recipientId, update); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void deleteNotification(TenantId tenantId, UserId recipientId, NotificationId notificationId) { |
|||
Notification notification = notificationService.findNotificationById(tenantId, notificationId); |
|||
boolean deleted = notificationService.deleteNotification(tenantId, recipientId, notificationId); |
|||
if (deleted) { |
|||
NotificationUpdate update = NotificationUpdate.builder() |
|||
.notification(notification) |
|||
.updateType(ComponentLifecycleEvent.DELETED) |
|||
.build(); |
|||
onNotificationUpdate(tenantId, recipientId, update); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public NotificationRequest updateNotificationRequest(TenantId tenantId, NotificationRequest notificationRequest) { |
|||
log.debug("Updating notification request {}", notificationRequest.getId()); |
|||
notificationRequest = notificationRequestService.saveNotificationRequest(tenantId, notificationRequest); |
|||
// marking related notifications as unread: TODO: causes each subscription to fetch notifications on each request update
|
|||
notificationService.updateNotificationsStatusByRequestId(tenantId, notificationRequest.getId(), NotificationStatus.SENT); |
|||
|
|||
// TODO: no need to send request update for other than PLATFORM_USERS target type
|
|||
onNotificationRequestUpdate(tenantId, NotificationRequestUpdate.builder() |
|||
.notificationRequestId(notificationRequest.getId()) |
|||
.notificationInfo(notificationRequest.getInfo()) |
|||
.deleted(false) |
|||
.build()); |
|||
return notificationRequest; |
|||
} |
|||
|
|||
@Override |
|||
public void deleteNotificationRequest(TenantId tenantId, NotificationRequestId notificationRequestId) { |
|||
log.debug("Deleting notification request {}", notificationRequestId); |
|||
NotificationRequest notificationRequest = notificationRequestService.findNotificationRequestById(tenantId, notificationRequestId); |
|||
notificationRequestService.deleteNotificationRequest(tenantId, notificationRequest); |
|||
|
|||
// TODO: no need to send request update for other than PLATFORM_USERS target type
|
|||
if (notificationRequest.isSent()) { |
|||
onNotificationRequestUpdate(tenantId, NotificationRequestUpdate.builder() |
|||
.notificationRequestId(notificationRequestId) |
|||
.deleted(true) |
|||
.build()); |
|||
} |
|||
clusterService.broadcastEntityStateChangeEvent(tenantId, notificationRequestId, ComponentLifecycleEvent.DELETED); |
|||
} |
|||
|
|||
private void forwardToNotificationSchedulerService(TenantId tenantId, NotificationRequestId notificationRequestId) { |
|||
TransportProtos.NotificationSchedulerServiceMsg.Builder msg = TransportProtos.NotificationSchedulerServiceMsg.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setRequestIdMSB(notificationRequestId.getId().getMostSignificantBits()) |
|||
.setRequestIdLSB(notificationRequestId.getId().getLeastSignificantBits()) |
|||
.setTs(System.currentTimeMillis()); |
|||
TransportProtos.ToCoreMsg toCoreMsg = TransportProtos.ToCoreMsg.newBuilder() |
|||
.setNotificationSchedulerServiceMsg(msg) |
|||
.build(); |
|||
clusterService.pushMsgToCore(tenantId, notificationRequestId, toCoreMsg, null); |
|||
} |
|||
|
|||
private ListenableFuture<Void> onNotificationUpdate(TenantId tenantId, UserId recipientId, NotificationUpdate update) { |
|||
log.trace("Submitting notification update for recipient {}: {}", recipientId, update); |
|||
return Futures.submit(() -> { |
|||
forwardToSubscriptionManagerService(tenantId, recipientId, subscriptionManagerService -> { |
|||
subscriptionManagerService.onNotificationUpdate(tenantId, recipientId, update, TbCallback.EMPTY); |
|||
}, () -> TbSubscriptionUtils.notificationUpdateToProto(tenantId, recipientId, update)); |
|||
}, wsCallBackExecutor); |
|||
} |
|||
|
|||
private void onNotificationRequestUpdate(TenantId tenantId, NotificationRequestUpdate update) { |
|||
log.trace("Submitting notification request update: {}", update); |
|||
wsCallBackExecutor.submit(() -> { |
|||
TransportProtos.ToCoreNotificationMsg notificationRequestUpdateProto = 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); |
|||
producerProvider.getTbCoreNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), notificationRequestUpdateProto), null); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationDeliveryMethod getDeliveryMethod() { |
|||
return NotificationDeliveryMethod.PUSH; |
|||
} |
|||
|
|||
@Override |
|||
protected String getExecutorPrefix() { |
|||
return "notification"; |
|||
} |
|||
|
|||
@Autowired |
|||
public void setChannels(List<NotificationChannel> channels, NotificationCenter websocketNotificationChannel) { |
|||
this.channels = channels.stream().collect(Collectors.toMap(NotificationChannel::getDeliveryMethod, c -> c)); |
|||
this.channels.put(NotificationDeliveryMethod.PUSH, (NotificationChannel) websocketNotificationChannel); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,176 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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 lombok.Data; |
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.context.event.EventListener; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.rule.engine.api.NotificationCenter; |
|||
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.notification.NotificationRequest; |
|||
import org.thingsboard.server.common.data.notification.NotificationRequestConfig; |
|||
import org.thingsboard.server.common.data.page.PageDataIterable; |
|||
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; |
|||
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; |
|||
import org.thingsboard.server.common.msg.queue.ServiceType; |
|||
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|||
import org.thingsboard.server.dao.notification.NotificationRequestService; |
|||
import org.thingsboard.server.queue.scheduler.SchedulerComponent; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.executors.NotificationExecutorService; |
|||
import org.thingsboard.server.service.partition.AbstractPartitionBasedService; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import java.util.Collections; |
|||
import java.util.HashSet; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Optional; |
|||
import java.util.Set; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ScheduledFuture; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
@TbCoreComponent |
|||
@Service |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
@SuppressWarnings("UnstableApiUsage") |
|||
public class DefaultNotificationSchedulerService extends AbstractPartitionBasedService<NotificationRequestId> implements NotificationSchedulerService { |
|||
|
|||
private final NotificationCenter notificationCenter; |
|||
private final NotificationRequestService notificationRequestService; |
|||
private final SchedulerComponent scheduler; |
|||
private final NotificationExecutorService notificationExecutor; |
|||
|
|||
private final Map<NotificationRequestId, ScheduledRequestMetadata> scheduledNotificationRequests = new ConcurrentHashMap<>(); |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
super.init(); |
|||
} |
|||
|
|||
@Override |
|||
protected Map<TopicPartitionInfo, List<ListenableFuture<?>>> onAddedPartitions(Set<TopicPartitionInfo> addedPartitions) { |
|||
PageDataIterable<NotificationRequest> notificationRequests = new PageDataIterable<>(pageLink -> { |
|||
return notificationRequestService.findScheduledNotificationRequests(pageLink); |
|||
}, 1000); |
|||
for (NotificationRequest notificationRequest : notificationRequests) { |
|||
TopicPartitionInfo requestPartition = partitionService.resolve(ServiceType.TB_CORE, notificationRequest.getTenantId(), notificationRequest.getId()); |
|||
if (addedPartitions.contains(requestPartition)) { |
|||
partitionedEntities.computeIfAbsent(requestPartition, k -> ConcurrentHashMap.newKeySet()).add(notificationRequest.getId()); |
|||
if (!scheduledNotificationRequests.containsKey(notificationRequest.getId())) { |
|||
scheduleNotificationRequest(notificationRequest.getTenantId(), notificationRequest, notificationRequest.getCreatedTime()); |
|||
} |
|||
} |
|||
} |
|||
return Collections.emptyMap(); |
|||
} |
|||
|
|||
@Override |
|||
public void scheduleNotificationRequest(TenantId tenantId, NotificationRequestId notificationRequestId, long requestTs) { |
|||
NotificationRequest notificationRequest = notificationRequestService.findNotificationRequestById(tenantId, notificationRequestId); |
|||
scheduleNotificationRequest(tenantId, notificationRequest, requestTs); |
|||
} |
|||
|
|||
private void scheduleNotificationRequest(TenantId tenantId, NotificationRequest request, long requestTs) { |
|||
int delayInSec = Optional.ofNullable(request) |
|||
.map(NotificationRequest::getAdditionalConfig) |
|||
.map(NotificationRequestConfig::getSendingDelayInSec) |
|||
.orElse(0); |
|||
if (delayInSec <= 0) return; |
|||
long delayInMs = TimeUnit.SECONDS.toMillis(delayInSec) - (System.currentTimeMillis() - requestTs); |
|||
if (delayInMs < 0) { |
|||
delayInMs = 0; |
|||
} |
|||
|
|||
ScheduledFuture<?> scheduledTask = scheduler.schedule(() -> { |
|||
NotificationRequest notificationRequest = notificationRequestService.findNotificationRequestById(tenantId, request.getId()); |
|||
if (notificationRequest == null) return; |
|||
|
|||
notificationExecutor.executeAsync(() -> { |
|||
try { |
|||
notificationCenter.processNotificationRequest(tenantId, notificationRequest); |
|||
} catch (Exception e) { |
|||
log.error("Failed to process scheduled notification request {}", notificationRequest.getId(), e); |
|||
UserId senderId = notificationRequest.getSenderId(); |
|||
if (senderId != null) { |
|||
notificationCenter.sendBasicNotification(tenantId, senderId, "Notification failure", |
|||
"Failed to process scheduled notification (request " + notificationRequest.getId() + "): " + e.getMessage()); |
|||
} |
|||
} |
|||
}); |
|||
scheduledNotificationRequests.remove(notificationRequest.getId()); |
|||
}, delayInMs, TimeUnit.MILLISECONDS); |
|||
scheduledNotificationRequests.put(request.getId(), new ScheduledRequestMetadata(tenantId, scheduledTask)); |
|||
} |
|||
|
|||
@EventListener(ComponentLifecycleMsg.class) |
|||
public void handleComponentLifecycleEvent(ComponentLifecycleMsg event) { |
|||
if (event.getEvent() == ComponentLifecycleEvent.DELETED) { |
|||
EntityId entityId = event.getEntityId(); |
|||
switch (entityId.getEntityType()) { |
|||
case NOTIFICATION_REQUEST: |
|||
cancelAndRemove((NotificationRequestId) entityId); |
|||
break; |
|||
case TENANT: |
|||
Set<NotificationRequestId> toCancel = new HashSet<>(); |
|||
scheduledNotificationRequests.forEach((notificationRequestId, scheduledRequestMetadata) -> { |
|||
if (scheduledRequestMetadata.getTenantId().equals(entityId)) { |
|||
toCancel.add(notificationRequestId); |
|||
} |
|||
}); |
|||
toCancel.forEach(this::cancelAndRemove); |
|||
break; |
|||
} |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
protected void cleanupEntityOnPartitionRemoval(NotificationRequestId notificationRequestId) { |
|||
cancelAndRemove(notificationRequestId); |
|||
} |
|||
|
|||
private void cancelAndRemove(NotificationRequestId notificationRequestId) { |
|||
ScheduledRequestMetadata md = scheduledNotificationRequests.remove(notificationRequestId); |
|||
if (md != null) { |
|||
md.getFuture().cancel(false); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
protected String getServiceName() { |
|||
return "Notifications scheduler"; |
|||
} |
|||
|
|||
@Override |
|||
protected String getSchedulerExecutorName() { |
|||
return "notifications-scheduler"; |
|||
} |
|||
|
|||
@Data |
|||
private static class ScheduledRequestMetadata { |
|||
private final TenantId tenantId; |
|||
private final ScheduledFuture<?> future; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,25 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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 org.thingsboard.server.common.data.id.NotificationRequestId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
public interface NotificationSchedulerService { |
|||
|
|||
void scheduleNotificationRequest(TenantId tenantId, NotificationRequestId notificationRequestId, long requestTs); |
|||
|
|||
} |
|||
@ -0,0 +1,48 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.channels; |
|||
|
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import lombok.RequiredArgsConstructor; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.rule.engine.api.MailService; |
|||
import org.thingsboard.server.common.data.User; |
|||
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod; |
|||
import org.thingsboard.server.common.data.notification.template.EmailDeliveryMethodNotificationTemplate; |
|||
import org.thingsboard.server.service.mail.MailExecutorService; |
|||
import org.thingsboard.server.common.data.notification.NotificationProcessingContext; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
public class EmailNotificationChannel implements NotificationChannel<User, EmailDeliveryMethodNotificationTemplate> { |
|||
|
|||
private final MailService mailService; |
|||
private final MailExecutorService executor; |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> sendNotification(User recipient, EmailDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) { |
|||
return executor.submit(() -> { |
|||
mailService.sendEmail(recipient.getTenantId(), recipient.getEmail(), processedTemplate.getSubject(), processedTemplate.getBody()); |
|||
return null; |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationDeliveryMethod getDeliveryMethod() { |
|||
return NotificationDeliveryMethod.EMAIL; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,30 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.channels; |
|||
|
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod; |
|||
import org.thingsboard.server.common.data.notification.NotificationProcessingContext; |
|||
import org.thingsboard.server.common.data.notification.targets.NotificationRecipient; |
|||
import org.thingsboard.server.common.data.notification.template.DeliveryMethodNotificationTemplate; |
|||
|
|||
public interface NotificationChannel<R extends NotificationRecipient, T extends DeliveryMethodNotificationTemplate> { |
|||
|
|||
ListenableFuture<Void> sendNotification(R recipient, T processedTemplate, NotificationProcessingContext ctx); |
|||
|
|||
NotificationDeliveryMethod getDeliveryMethod(); |
|||
|
|||
} |
|||
@ -0,0 +1,50 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.channels; |
|||
|
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import lombok.RequiredArgsConstructor; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.rule.engine.api.slack.SlackService; |
|||
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod; |
|||
import org.thingsboard.server.common.data.notification.NotificationProcessingContext; |
|||
import org.thingsboard.server.common.data.notification.settings.SlackNotificationDeliveryMethodConfig; |
|||
import org.thingsboard.server.common.data.notification.targets.slack.SlackConversation; |
|||
import org.thingsboard.server.common.data.notification.template.SlackDeliveryMethodNotificationTemplate; |
|||
import org.thingsboard.server.service.executors.ExternalCallExecutorService; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
public class SlackNotificationChannel implements NotificationChannel<SlackConversation, SlackDeliveryMethodNotificationTemplate> { |
|||
|
|||
private final SlackService slackService; |
|||
private final ExternalCallExecutorService executor; |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> sendNotification(SlackConversation conversation, SlackDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) { |
|||
SlackNotificationDeliveryMethodConfig config = ctx.getDeliveryMethodConfig(NotificationDeliveryMethod.SLACK); |
|||
return executor.submit(() -> { |
|||
slackService.sendMessage(ctx.getTenantId(), config.getBotToken(), conversation.getId(), processedTemplate.getBody()); |
|||
return null; |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationDeliveryMethod getDeliveryMethod() { |
|||
return NotificationDeliveryMethod.SLACK; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,55 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.channels; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import lombok.RequiredArgsConstructor; |
|||
import org.apache.commons.lang3.StringUtils; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.rule.engine.api.SmsService; |
|||
import org.thingsboard.server.common.data.User; |
|||
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod; |
|||
import org.thingsboard.server.common.data.notification.template.SmsDeliveryMethodNotificationTemplate; |
|||
import org.thingsboard.server.common.data.notification.NotificationProcessingContext; |
|||
import org.thingsboard.server.service.sms.SmsExecutorService; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
public class SmsNotificationChannel implements NotificationChannel<User, SmsDeliveryMethodNotificationTemplate> { |
|||
|
|||
private final SmsService smsService; |
|||
private final SmsExecutorService executor; |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> sendNotification(User recipient, SmsDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) { |
|||
String phone = recipient.getPhone(); |
|||
if (StringUtils.isBlank(phone)) { |
|||
return Futures.immediateFailedFuture(new RuntimeException("User does not have phone number")); |
|||
} |
|||
|
|||
return executor.submit(() -> { |
|||
smsService.sendSms(recipient.getTenantId(), recipient.getCustomerId(), new String[]{phone}, processedTemplate.getBody()); |
|||
return null; |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationDeliveryMethod getDeliveryMethod() { |
|||
return NotificationDeliveryMethod.SMS; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,237 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.rule; |
|||
|
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.context.annotation.Lazy; |
|||
import org.springframework.context.event.EventListener; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.common.util.DonAsynchron; |
|||
import org.thingsboard.rule.engine.api.NotificationCenter; |
|||
import org.thingsboard.server.common.data.DataConstants; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
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.NotificationRuleId; |
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
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.info.NotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.rule.NotificationRule; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerConfig; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType; |
|||
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; |
|||
import org.thingsboard.server.dao.notification.NotificationRequestService; |
|||
import org.thingsboard.server.dao.notification.NotificationRuleService; |
|||
import org.thingsboard.server.service.executors.DbCallbackExecutorService; |
|||
import org.thingsboard.server.service.executors.NotificationExecutorService; |
|||
import org.thingsboard.server.service.notification.rule.trigger.AlarmTriggerProcessor.AlarmTriggerObject; |
|||
import org.thingsboard.server.service.notification.rule.trigger.NotificationRuleTriggerProcessor; |
|||
import org.thingsboard.server.service.notification.rule.trigger.RuleEngineComponentLifecycleEventTriggerProcessor.RuleEngineComponentLifecycleEventTriggerObject; |
|||
|
|||
import java.util.Collection; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.UUID; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@Service |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class DefaultNotificationRuleProcessingService implements NotificationRuleProcessingService { |
|||
|
|||
private final NotificationRuleService notificationRuleService; |
|||
private final NotificationRequestService notificationRequestService; |
|||
@Autowired @Lazy |
|||
private NotificationCenter notificationCenter; |
|||
private Map<NotificationRuleTriggerType, NotificationRuleTriggerProcessor> triggerProcessors; |
|||
|
|||
private final NotificationExecutorService notificationExecutor; |
|||
private final DbCallbackExecutorService dbCallbackExecutor; |
|||
|
|||
private final Map<String, NotificationRuleTriggerType> msgTypeToTriggerType = Map.of( |
|||
DataConstants.INACTIVITY_EVENT, NotificationRuleTriggerType.DEVICE_INACTIVITY, |
|||
DataConstants.ENTITY_CREATED, NotificationRuleTriggerType.ENTITY_ACTION, |
|||
DataConstants.ENTITY_UPDATED, NotificationRuleTriggerType.ENTITY_ACTION, |
|||
DataConstants.ENTITY_DELETED, NotificationRuleTriggerType.ENTITY_ACTION, |
|||
DataConstants.COMMENT_CREATED, NotificationRuleTriggerType.ALARM_COMMENT, |
|||
DataConstants.COMMENT_UPDATED, NotificationRuleTriggerType.ALARM_COMMENT |
|||
); |
|||
|
|||
@Override |
|||
public void process(TenantId tenantId, TbMsg ruleEngineMsg) { |
|||
String msgType = ruleEngineMsg.getType(); |
|||
NotificationRuleTriggerType triggerType = msgTypeToTriggerType.get(msgType); |
|||
if (triggerType == null) { |
|||
return; |
|||
} |
|||
|
|||
processTrigger(tenantId, triggerType, ruleEngineMsg.getOriginator(), ruleEngineMsg); |
|||
} |
|||
|
|||
@Override |
|||
public void process(TenantId tenantId, Alarm alarm, boolean deleted) { |
|||
AlarmTriggerObject triggerObject = AlarmTriggerObject.builder() |
|||
.alarm(alarm) |
|||
.deleted(deleted) |
|||
.build(); |
|||
processTrigger(tenantId, NotificationRuleTriggerType.ALARM, alarm.getId(), triggerObject); |
|||
} |
|||
|
|||
@Override |
|||
public void process(TenantId tenantId, RuleChainId ruleChainId, String ruleChainName, EntityId componentId, String componentName, ComponentLifecycleEvent eventType, Exception error) { |
|||
RuleEngineComponentLifecycleEventTriggerObject triggerObject = RuleEngineComponentLifecycleEventTriggerObject.builder() |
|||
.ruleChainId(ruleChainId) |
|||
.ruleChainName(ruleChainName) |
|||
.componentId(componentId) |
|||
.componentName(componentName) |
|||
.eventType(eventType) |
|||
.error(error) |
|||
.build(); |
|||
processTrigger(tenantId, NotificationRuleTriggerType.RULE_ENGINE_COMPONENT_LIFECYCLE_EVENT, componentId, triggerObject); |
|||
} |
|||
|
|||
private void processTrigger(TenantId tenantId, NotificationRuleTriggerType triggerType, EntityId originatorEntityId, Object triggerObject) { |
|||
ListenableFuture<List<NotificationRule>> rulesFuture = dbCallbackExecutor.submit(() -> { |
|||
return notificationRuleService.findNotificationRulesByTenantIdAndTriggerType(tenantId, triggerType); |
|||
}); |
|||
DonAsynchron.withCallback(rulesFuture, rules -> { |
|||
for (NotificationRule rule : rules) { |
|||
notificationExecutor.submit(() -> { |
|||
processNotificationRule(rule, originatorEntityId, triggerObject); |
|||
}); |
|||
} |
|||
}, e -> { |
|||
log.error("Failed to find notification rules by trigger type {}", triggerType, e); |
|||
}); |
|||
} |
|||
|
|||
private void processNotificationRule(NotificationRule rule, EntityId originatorEntityId, Object triggerObject) { |
|||
NotificationRuleTriggerConfig triggerConfig = rule.getTriggerConfig(); |
|||
log.debug("Processing notification rule '{}' for trigger type {}", rule.getName(), rule.getTriggerType()); |
|||
|
|||
if (triggerConfig.getTriggerType().isUpdatable()) { |
|||
List<NotificationRequest> notificationRequests = notificationRequestService.findNotificationRequestsByRuleIdAndOriginatorEntityId(rule.getTenantId(), rule.getId(), originatorEntityId); |
|||
if (!notificationRequests.isEmpty()) { |
|||
if (matchesClearRule(triggerObject, triggerConfig)) { |
|||
notificationRequests = notificationRequests.stream() |
|||
.filter(notificationRequest -> { |
|||
if (!notificationRequest.isSent()) { |
|||
dbCallbackExecutor.submit(() -> { |
|||
notificationCenter.deleteNotificationRequest(rule.getTenantId(), notificationRequest.getId()); |
|||
}); |
|||
return false; |
|||
} else { |
|||
return true; |
|||
} |
|||
}) |
|||
.collect(Collectors.toList()); |
|||
// not returning because we need to update notifications if any
|
|||
} |
|||
|
|||
NotificationInfo notificationInfo = constructNotificationInfo(triggerObject, triggerConfig); |
|||
for (NotificationRequest notificationRequest : notificationRequests) { |
|||
NotificationInfo previousNotificationInfo = notificationRequest.getInfo(); |
|||
if (!notificationInfo.equals(previousNotificationInfo)) { |
|||
notificationRequest.setInfo(notificationInfo); |
|||
dbCallbackExecutor.submit(() -> { |
|||
notificationCenter.updateNotificationRequest(rule.getTenantId(), notificationRequest); |
|||
}); |
|||
} |
|||
} |
|||
return; |
|||
} |
|||
} |
|||
|
|||
if (!matchesFilter(triggerObject, triggerConfig)) { |
|||
return; |
|||
} |
|||
|
|||
NotificationInfo notificationInfo = constructNotificationInfo(triggerObject, triggerConfig); |
|||
rule.getRecipientsConfig().getTargetsTable().forEach((delay, targets) -> { |
|||
notificationExecutor.submit(() -> { |
|||
try { |
|||
log.debug("Submitting notification request for rule '{}' with delay of {} sec to targets {}", rule.getName(), delay, targets); |
|||
submitNotificationRequest(targets, rule, originatorEntityId, notificationInfo, delay); |
|||
} catch (Exception e) { |
|||
log.error("Failed to submit notification request for rule {}", rule.getId(), e); |
|||
} |
|||
}); |
|||
}); |
|||
} |
|||
|
|||
private boolean matchesFilter(Object triggerObject, NotificationRuleTriggerConfig triggerConfig) { |
|||
return triggerProcessors.get(triggerConfig.getTriggerType()).matchesFilter(triggerObject, triggerConfig); |
|||
} |
|||
|
|||
private boolean matchesClearRule(Object triggerObject, NotificationRuleTriggerConfig triggerConfig) { |
|||
return triggerProcessors.get(triggerConfig.getTriggerType()).matchesClearRule(triggerObject, triggerConfig); |
|||
} |
|||
|
|||
private NotificationInfo constructNotificationInfo(Object triggerObject, NotificationRuleTriggerConfig triggerConfig) { |
|||
return triggerProcessors.get(triggerConfig.getTriggerType()).constructNotificationInfo(triggerObject, triggerConfig); |
|||
} |
|||
|
|||
private void submitNotificationRequest(List<UUID> targets, NotificationRule rule, |
|||
EntityId originatorEntityId, NotificationInfo notificationInfo, int delayInSec) { |
|||
NotificationRequestConfig config = new NotificationRequestConfig(); |
|||
if (delayInSec > 0) { |
|||
config.setSendingDelayInSec(delayInSec); |
|||
} |
|||
NotificationRequest notificationRequest = NotificationRequest.builder() |
|||
.tenantId(rule.getTenantId()) |
|||
.targets(targets) |
|||
.templateId(rule.getTemplateId()) |
|||
.additionalConfig(config) |
|||
.info(notificationInfo) |
|||
.ruleId(rule.getId()) |
|||
.originatorEntityId(originatorEntityId) |
|||
.build(); |
|||
notificationCenter.processNotificationRequest(rule.getTenantId(), notificationRequest); |
|||
} |
|||
|
|||
@EventListener(ComponentLifecycleMsg.class) |
|||
public void onNotificationRuleDeleted(ComponentLifecycleMsg componentLifecycleMsg) { |
|||
if (componentLifecycleMsg.getEvent() != ComponentLifecycleEvent.DELETED || |
|||
componentLifecycleMsg.getEntityId().getEntityType() != EntityType.NOTIFICATION_RULE) { |
|||
return; |
|||
} |
|||
|
|||
TenantId tenantId = componentLifecycleMsg.getTenantId(); |
|||
NotificationRuleId notificationRuleId = (NotificationRuleId) componentLifecycleMsg.getEntityId(); |
|||
dbCallbackExecutor.submit(() -> { |
|||
List<NotificationRequestId> scheduledForRule = notificationRequestService.findNotificationRequestsIdsByStatusAndRuleId(tenantId, NotificationRequestStatus.SCHEDULED, notificationRuleId); |
|||
for (NotificationRequestId notificationRequestId : scheduledForRule) { |
|||
notificationCenter.deleteNotificationRequest(tenantId, notificationRequestId); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
@Autowired |
|||
public void setTriggerProcessors(Collection<NotificationRuleTriggerProcessor> processors) { |
|||
this.triggerProcessors = processors.stream() |
|||
.collect(Collectors.toMap(NotificationRuleTriggerProcessor::getTriggerType, p -> p)); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,34 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.rule; |
|||
|
|||
import org.thingsboard.server.common.data.alarm.Alarm; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
|
|||
public interface NotificationRuleProcessingService { |
|||
|
|||
void process(TenantId tenantId, TbMsg ruleEngineMsg); |
|||
|
|||
void process(TenantId tenantId, Alarm alarm, boolean deleted); |
|||
|
|||
void process(TenantId tenantId, RuleChainId ruleChainId, String ruleChainName, |
|||
EntityId componentId, String componentName, ComponentLifecycleEvent eventType, Exception error); |
|||
|
|||
} |
|||
@ -0,0 +1,52 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.rule.trigger; |
|||
|
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.server.common.data.alarm.Alarm; |
|||
import org.thingsboard.server.common.data.alarm.AlarmComment; |
|||
import org.thingsboard.server.common.data.notification.info.AlarmCommentNotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.info.NotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.AlarmCommentNotificationRuleTriggerConfig; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
|
|||
@Service |
|||
public class AlarmCommentTriggerProcessor implements NotificationRuleTriggerProcessor<TbMsg, AlarmCommentNotificationRuleTriggerConfig> { |
|||
|
|||
@Override |
|||
public boolean matchesFilter(TbMsg ruleEngineMsg, AlarmCommentNotificationRuleTriggerConfig triggerConfig) { |
|||
return ruleEngineMsg.getMetaData().getValue("comment") != null; |
|||
} |
|||
|
|||
@Override |
|||
public NotificationInfo constructNotificationInfo(TbMsg ruleEngineMsg, AlarmCommentNotificationRuleTriggerConfig triggerConfig) { |
|||
AlarmComment comment = JacksonUtil.fromString(ruleEngineMsg.getMetaData().getValue("comment"), AlarmComment.class); |
|||
Alarm alarm = JacksonUtil.fromString(ruleEngineMsg.getData(), Alarm.class); |
|||
return AlarmCommentNotificationInfo.builder() |
|||
.comment(comment.getComment().get("text").asText()) |
|||
.alarmType(alarm.getType()) |
|||
.alarmId(comment.getAlarmId().getId()) |
|||
.build(); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationRuleTriggerType getTriggerType() { |
|||
return NotificationRuleTriggerType.ALARM_COMMENT; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,80 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.rule.trigger; |
|||
|
|||
import lombok.Builder; |
|||
import lombok.Data; |
|||
import org.apache.commons.collections.CollectionUtils; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.alarm.Alarm; |
|||
import org.thingsboard.server.common.data.notification.info.AlarmNotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.info.NotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.AlarmNotificationRuleTriggerConfig; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.AlarmNotificationRuleTriggerConfig.ClearRule; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType; |
|||
import org.thingsboard.server.service.notification.rule.trigger.AlarmTriggerProcessor.AlarmTriggerObject; |
|||
|
|||
@Service |
|||
public class AlarmTriggerProcessor implements NotificationRuleTriggerProcessor<AlarmTriggerObject, AlarmNotificationRuleTriggerConfig> { |
|||
|
|||
@Override |
|||
public boolean matchesFilter(AlarmTriggerObject triggerObject, AlarmNotificationRuleTriggerConfig triggerConfig) { |
|||
Alarm alarm = triggerObject.getAlarm(); |
|||
return (CollectionUtils.isEmpty(triggerConfig.getAlarmTypes()) || triggerConfig.getAlarmTypes().contains(alarm.getType())) && |
|||
(CollectionUtils.isEmpty(triggerConfig.getAlarmSeverities()) || triggerConfig.getAlarmSeverities().contains(alarm.getSeverity())); |
|||
} |
|||
|
|||
@Override |
|||
public boolean matchesClearRule(AlarmTriggerObject triggerObject, AlarmNotificationRuleTriggerConfig triggerConfig) { |
|||
if (triggerObject.isDeleted()) { |
|||
return true; |
|||
} |
|||
Alarm alarm = triggerObject.getAlarm(); |
|||
ClearRule clearRule = triggerConfig.getClearRule(); |
|||
if (clearRule != null) { |
|||
if (clearRule.getAlarmStatus() != null) { |
|||
return clearRule.getAlarmStatus().equals(alarm.getStatus()); |
|||
} |
|||
} |
|||
return false; |
|||
} |
|||
|
|||
@Override |
|||
public NotificationInfo constructNotificationInfo(AlarmTriggerObject triggerObject, AlarmNotificationRuleTriggerConfig triggerConfig) { |
|||
Alarm alarm = triggerObject.getAlarm(); |
|||
return AlarmNotificationInfo.builder() |
|||
.alarmId(alarm.getUuidId()) |
|||
.alarmType(alarm.getType()) |
|||
.alarmOriginator(alarm.getOriginator()) |
|||
.alarmSeverity(alarm.getSeverity()) |
|||
.alarmStatus(alarm.getStatus()) |
|||
.alarmCustomerId(alarm.getCustomerId()) |
|||
.build(); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationRuleTriggerType getTriggerType() { |
|||
return NotificationRuleTriggerType.ALARM; |
|||
} |
|||
|
|||
@Data |
|||
@Builder |
|||
public static class AlarmTriggerObject { |
|||
private final Alarm alarm; |
|||
private final boolean deleted; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,65 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.rule.trigger; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import org.apache.commons.collections.CollectionUtils; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.DeviceProfile; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.notification.info.DeviceInactivityNotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.info.NotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.DeviceInactivityNotificationRuleTriggerConfig; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
import org.thingsboard.server.service.profile.TbDeviceProfileCache; |
|||
|
|||
@Service |
|||
@RequiredArgsConstructor |
|||
public class DeviceInactivityTriggerProcessor implements NotificationRuleTriggerProcessor<TbMsg, DeviceInactivityNotificationRuleTriggerConfig> { |
|||
|
|||
private final TbDeviceProfileCache deviceProfileCache; |
|||
|
|||
@Override |
|||
public boolean matchesFilter(TbMsg ruleEngineMsg, DeviceInactivityNotificationRuleTriggerConfig triggerConfig) { |
|||
DeviceId deviceId = (DeviceId) ruleEngineMsg.getOriginator(); |
|||
if (CollectionUtils.isNotEmpty(triggerConfig.getDevices())) { |
|||
return triggerConfig.getDevices().contains(deviceId.getId()); |
|||
} else if (CollectionUtils.isNotEmpty(triggerConfig.getDeviceProfiles())) { |
|||
DeviceProfile deviceProfile = deviceProfileCache.get(TenantId.SYS_TENANT_ID, deviceId); |
|||
return deviceProfile != null && triggerConfig.getDeviceProfiles().contains(deviceProfile.getUuidId()); |
|||
} else { |
|||
return true; |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public NotificationInfo constructNotificationInfo(TbMsg ruleEngineMsg, DeviceInactivityNotificationRuleTriggerConfig triggerConfig) { |
|||
return DeviceInactivityNotificationInfo.builder() |
|||
.deviceId(ruleEngineMsg.getOriginator().getId()) |
|||
.deviceName(ruleEngineMsg.getMetaData().getValue("deviceName")) |
|||
.deviceType(ruleEngineMsg.getMetaData().getValue("deviceType")) |
|||
.deviceCustomerId(ruleEngineMsg.getCustomerId()) |
|||
.build(); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationRuleTriggerType getTriggerType() { |
|||
return NotificationRuleTriggerType.DEVICE_INACTIVITY; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,77 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.rule.trigger; |
|||
|
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.DataConstants; |
|||
import org.thingsboard.server.common.data.audit.ActionType; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.notification.info.EntityActionNotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.info.NotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.EntityActionNotificationRuleTriggerConfig; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
@Service |
|||
public class EntityActionTriggerProcessor implements NotificationRuleTriggerProcessor<TbMsg, EntityActionNotificationRuleTriggerConfig> { |
|||
|
|||
@Override |
|||
public boolean matchesFilter(TbMsg ruleEngineMsg, EntityActionNotificationRuleTriggerConfig triggerConfig) { |
|||
String msgType = ruleEngineMsg.getType(); |
|||
if (msgType.equals(DataConstants.ENTITY_CREATED)) { |
|||
if (!triggerConfig.isCreated()) { |
|||
return false; |
|||
} |
|||
} else if (msgType.equals(DataConstants.ENTITY_UPDATED)) { |
|||
if (!triggerConfig.isUpdated()) { |
|||
return false; |
|||
} |
|||
} else if (msgType.equals(DataConstants.ENTITY_DELETED)) { |
|||
if (!triggerConfig.isDeleted()) { |
|||
return false; |
|||
} |
|||
} else { |
|||
return false; |
|||
} |
|||
return triggerConfig.getEntityType() == null || ruleEngineMsg.getOriginator().getEntityType() == triggerConfig.getEntityType(); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationInfo constructNotificationInfo(TbMsg ruleEngineMsg, EntityActionNotificationRuleTriggerConfig triggerConfig) { |
|||
EntityId entityId = ruleEngineMsg.getOriginator(); |
|||
String msgType = ruleEngineMsg.getType(); |
|||
ActionType actionType = msgType.equals(DataConstants.ENTITY_CREATED) ? ActionType.ADDED : |
|||
msgType.equals(DataConstants.ENTITY_UPDATED) ? ActionType.UPDATED : |
|||
msgType.equals(DataConstants.ENTITY_DELETED) ? ActionType.DELETED : null; |
|||
return EntityActionNotificationInfo.builder() |
|||
.entityType(entityId.getEntityType()) |
|||
.entityId(entityId.getId()) |
|||
.entityName(ruleEngineMsg.getMetaData().getValue("entityName")) |
|||
.actionType(actionType) |
|||
.originatorUserId(UUID.fromString(ruleEngineMsg.getMetaData().getValue("userId"))) |
|||
.originatorUserName(ruleEngineMsg.getMetaData().getValue("userName")) |
|||
.entityCustomerId(ruleEngineMsg.getCustomerId()) |
|||
.build(); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationRuleTriggerType getTriggerType() { |
|||
return NotificationRuleTriggerType.ENTITY_ACTION; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,34 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.rule.trigger; |
|||
|
|||
import org.thingsboard.server.common.data.notification.info.NotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerConfig; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType; |
|||
|
|||
public interface NotificationRuleTriggerProcessor<T, C extends NotificationRuleTriggerConfig> { |
|||
|
|||
boolean matchesFilter(T triggerObject, C triggerConfig); |
|||
|
|||
default boolean matchesClearRule(T triggerObject, C triggerConfig) { |
|||
return false; |
|||
} |
|||
|
|||
NotificationInfo constructNotificationInfo(T triggerObject, C triggerConfig); |
|||
|
|||
NotificationRuleTriggerType getTriggerType(); |
|||
|
|||
} |
|||
@ -0,0 +1,110 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.rule.trigger; |
|||
|
|||
import com.google.common.base.Strings; |
|||
import lombok.Builder; |
|||
import lombok.Data; |
|||
import org.apache.commons.collections.CollectionUtils; |
|||
import org.apache.commons.lang3.StringUtils; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
import org.thingsboard.server.common.data.notification.info.NotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.info.RuleEngineComponentLifecycleEventNotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType; |
|||
import org.thingsboard.server.common.data.notification.rule.trigger.RuleEngineComponentLifecycleEventNotificationRuleTriggerConfig; |
|||
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; |
|||
import org.thingsboard.server.service.notification.rule.trigger.RuleEngineComponentLifecycleEventTriggerProcessor.RuleEngineComponentLifecycleEventTriggerObject; |
|||
|
|||
import java.util.Set; |
|||
|
|||
@Service |
|||
public class RuleEngineComponentLifecycleEventTriggerProcessor implements NotificationRuleTriggerProcessor<RuleEngineComponentLifecycleEventTriggerObject, RuleEngineComponentLifecycleEventNotificationRuleTriggerConfig> { |
|||
|
|||
@Override |
|||
public boolean matchesFilter(RuleEngineComponentLifecycleEventTriggerObject triggerObject, RuleEngineComponentLifecycleEventNotificationRuleTriggerConfig triggerConfig) { |
|||
if (CollectionUtils.isNotEmpty(triggerConfig.getRuleChains())) { |
|||
if (!triggerConfig.getRuleChains().contains(triggerObject.getRuleChainId().getId())) { |
|||
return false; |
|||
} |
|||
} |
|||
|
|||
EntityType componentType = triggerObject.getComponentId().getEntityType(); |
|||
Set<ComponentLifecycleEvent> trackedEvents; |
|||
boolean onlyFailures; |
|||
if (componentType == EntityType.RULE_CHAIN) { |
|||
trackedEvents = triggerConfig.getRuleChainEvents(); |
|||
onlyFailures = triggerConfig.isOnlyRuleChainLifecycleFailures(); |
|||
} else if (componentType == EntityType.RULE_NODE && triggerConfig.isTrackRuleNodeEvents()) { |
|||
trackedEvents = triggerConfig.getRuleNodeEvents(); |
|||
onlyFailures = triggerConfig.isOnlyRuleNodeLifecycleFailures(); |
|||
} else { |
|||
return false; |
|||
} |
|||
if (CollectionUtils.isEmpty(trackedEvents)) { |
|||
trackedEvents = Set.of(ComponentLifecycleEvent.STARTED, ComponentLifecycleEvent.UPDATED, ComponentLifecycleEvent.STOPPED); |
|||
} |
|||
|
|||
if (!trackedEvents.contains(triggerObject.getEventType())) { |
|||
return false; |
|||
} |
|||
if (onlyFailures) { |
|||
return triggerObject.getError() != null; |
|||
} |
|||
return true; |
|||
} |
|||
|
|||
@Override |
|||
public NotificationInfo constructNotificationInfo(RuleEngineComponentLifecycleEventTriggerObject triggerObject, RuleEngineComponentLifecycleEventNotificationRuleTriggerConfig triggerConfig) { |
|||
return RuleEngineComponentLifecycleEventNotificationInfo.builder() |
|||
.ruleChainId(triggerObject.getRuleChainId()) |
|||
.ruleChainName(triggerObject.getRuleChainName()) |
|||
.componentId(triggerObject.getComponentId()) |
|||
.componentName(triggerObject.getComponentName()) |
|||
.eventType(triggerObject.getEventType()) |
|||
.error(getErrorMsg(triggerObject.getError())) |
|||
.build(); |
|||
} |
|||
|
|||
private String getErrorMsg(Exception error) { |
|||
String errorMsg = error != null ? error.getMessage() : null; |
|||
errorMsg = Strings.nullToEmpty(errorMsg); |
|||
int lengthLimit = 150; |
|||
if (errorMsg.length() > lengthLimit) { |
|||
errorMsg = StringUtils.substring(errorMsg, 0, lengthLimit + 1).trim() + "[...]"; |
|||
} |
|||
return errorMsg; |
|||
} |
|||
|
|||
@Override |
|||
public NotificationRuleTriggerType getTriggerType() { |
|||
return NotificationRuleTriggerType.RULE_ENGINE_COMPONENT_LIFECYCLE_EVENT; |
|||
} |
|||
|
|||
@Data |
|||
@Builder |
|||
public static class RuleEngineComponentLifecycleEventTriggerObject { |
|||
private final RuleChainId ruleChainId; |
|||
private final String ruleChainName; |
|||
private final EntityId componentId; |
|||
private final String componentName; |
|||
private final ComponentLifecycleEvent eventType; |
|||
private final Exception error; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,154 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.slack; |
|||
|
|||
import com.github.benmanes.caffeine.cache.Cache; |
|||
import com.github.benmanes.caffeine.cache.Caffeine; |
|||
import com.slack.api.Slack; |
|||
import com.slack.api.methods.MethodsClient; |
|||
import com.slack.api.methods.SlackApiRequest; |
|||
import com.slack.api.methods.SlackApiTextResponse; |
|||
import com.slack.api.methods.request.chat.ChatPostMessageRequest; |
|||
import com.slack.api.methods.request.conversations.ConversationsListRequest; |
|||
import com.slack.api.methods.request.users.UsersListRequest; |
|||
import com.slack.api.methods.response.conversations.ConversationsListResponse; |
|||
import com.slack.api.methods.response.users.UsersListResponse; |
|||
import com.slack.api.model.ConversationType; |
|||
import lombok.RequiredArgsConstructor; |
|||
import org.apache.commons.lang3.StringUtils; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.notification.targets.slack.SlackConversation; |
|||
import org.thingsboard.rule.engine.api.slack.SlackService; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod; |
|||
import org.thingsboard.server.common.data.notification.settings.NotificationSettings; |
|||
import org.thingsboard.server.common.data.notification.settings.SlackNotificationDeliveryMethodConfig; |
|||
import org.thingsboard.server.common.data.notification.targets.slack.SlackConversationType; |
|||
import org.thingsboard.server.common.data.util.ThrowingBiFunction; |
|||
import org.thingsboard.server.dao.notification.NotificationSettingsService; |
|||
|
|||
import java.util.List; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@Service |
|||
@RequiredArgsConstructor |
|||
public class DefaultSlackService implements SlackService { |
|||
|
|||
private final NotificationSettingsService notificationSettingsService; |
|||
|
|||
private final Slack slack = Slack.getInstance(); |
|||
private final Cache<String, List<SlackConversation>> cache = Caffeine.newBuilder() |
|||
.expireAfterWrite(20, TimeUnit.SECONDS) |
|||
.maximumSize(100) |
|||
.build(); |
|||
private static final int CONVERSATIONS_LOAD_LIMIT = 1000; |
|||
|
|||
@Override |
|||
public void sendMessage(TenantId tenantId, String token, String conversationId, String message) { |
|||
ChatPostMessageRequest request = ChatPostMessageRequest.builder() |
|||
.channel(conversationId) |
|||
.text(message) |
|||
.build(); |
|||
sendRequest(token, request, MethodsClient::chatPostMessage); |
|||
} |
|||
|
|||
@Override |
|||
public List<SlackConversation> listConversations(TenantId tenantId, String token, SlackConversationType conversationType) { |
|||
return cache.get(conversationType + ":" + token, k -> { |
|||
if (conversationType == SlackConversationType.DIRECT) { |
|||
UsersListRequest request = UsersListRequest.builder() |
|||
.limit(CONVERSATIONS_LOAD_LIMIT) |
|||
.build(); |
|||
|
|||
UsersListResponse response = sendRequest(token, request, MethodsClient::usersList); |
|||
return response.getMembers().stream() |
|||
.filter(user -> !user.isDeleted() && !user.isStranger() && !user.isBot()) |
|||
.map(user -> { |
|||
SlackConversation conversation = new SlackConversation(); |
|||
conversation.setId(user.getId()); |
|||
conversation.setName(String.format("@%s (%s)", user.getName(), user.getRealName())); |
|||
return conversation; |
|||
}) |
|||
.collect(Collectors.toList()); |
|||
} else { |
|||
ConversationsListRequest request = ConversationsListRequest.builder() |
|||
.types(List.of(conversationType == SlackConversationType.PUBLIC_CHANNEL ? |
|||
ConversationType.PUBLIC_CHANNEL : |
|||
ConversationType.PRIVATE_CHANNEL)) |
|||
.limit(CONVERSATIONS_LOAD_LIMIT) |
|||
.excludeArchived(true) |
|||
.build(); |
|||
|
|||
ConversationsListResponse response = sendRequest(token, request, MethodsClient::conversationsList); |
|||
return response.getChannels().stream() |
|||
.filter(channel -> !channel.isArchived()) |
|||
.map(channel -> { |
|||
SlackConversation conversation = new SlackConversation(); |
|||
conversation.setId(channel.getId()); |
|||
conversation.setName("#" + channel.getName()); |
|||
return conversation; |
|||
}) |
|||
.collect(Collectors.toList()); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
public SlackConversation findConversation(TenantId tenantId, String token, SlackConversationType conversationType, String namePattern) { |
|||
List<SlackConversation> conversations = listConversations(tenantId, token, conversationType); |
|||
return conversations.stream() |
|||
.filter(conversation -> StringUtils.containsIgnoreCase(conversation.getName(), namePattern)) |
|||
.findFirst().orElse(null); |
|||
} |
|||
|
|||
@Override |
|||
public String getToken(TenantId tenantId) { |
|||
NotificationSettings settings = notificationSettingsService.findNotificationSettings(tenantId); |
|||
SlackNotificationDeliveryMethodConfig slackConfig = (SlackNotificationDeliveryMethodConfig) |
|||
settings.getDeliveryMethodsConfigs().get(NotificationDeliveryMethod.SLACK); |
|||
if (slackConfig != null) { |
|||
return slackConfig.getBotToken(); |
|||
} else { |
|||
return null; |
|||
} |
|||
} |
|||
|
|||
private <T extends SlackApiRequest, R extends SlackApiTextResponse> R sendRequest(String token, T request, ThrowingBiFunction<MethodsClient, T, R> method) { |
|||
MethodsClient client = slack.methods(token); |
|||
R response; |
|||
try { |
|||
response = method.apply(client, request); |
|||
} catch (Exception e) { |
|||
throw new RuntimeException(e.getMessage(), e); |
|||
} |
|||
|
|||
if (!response.isOk()) { |
|||
String error = response.getError(); |
|||
if (error == null) { |
|||
error = "unknown error"; |
|||
} |
|||
if (error.contains("missing_scope")) { |
|||
String neededScope = response.getNeeded(); |
|||
throw new RuntimeException("Bot token scope '" + neededScope + "' is needed"); |
|||
} |
|||
throw new RuntimeException("Failed to send message via Slack: " + error); |
|||
} |
|||
|
|||
return response; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,71 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.ttl; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Value; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.scheduling.annotation.Scheduled; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.notification.NotificationRequestConfig; |
|||
import org.thingsboard.server.dao.notification.NotificationRequestDao; |
|||
import org.thingsboard.server.dao.sqlts.insert.sql.SqlPartitioningRepository; |
|||
import org.thingsboard.server.queue.discovery.PartitionService; |
|||
|
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
import static org.thingsboard.server.dao.model.ModelConstants.NOTIFICATION_TABLE_NAME; |
|||
|
|||
@Service |
|||
@ConditionalOnExpression("${sql.ttl.notifications.enabled:true} && ${sql.ttl.notifications.ttl:0} > 0") |
|||
@Slf4j |
|||
public class NotificationsCleanUpService extends AbstractCleanUpService { |
|||
|
|||
private final SqlPartitioningRepository partitioningRepository; |
|||
private final NotificationRequestDao notificationRequestDao; |
|||
|
|||
@Value("${sql.ttl.notifications.ttl:2592000}") |
|||
private long ttlInSec; |
|||
@Value("${sql.notifications.partition_size:168}") |
|||
private int partitionSizeInHours; |
|||
|
|||
public NotificationsCleanUpService(PartitionService partitionService, SqlPartitioningRepository partitioningRepository, |
|||
NotificationRequestDao notificationRequestDao) { |
|||
super(partitionService); |
|||
this.partitioningRepository = partitioningRepository; |
|||
this.notificationRequestDao = notificationRequestDao; |
|||
} |
|||
|
|||
@Scheduled(initialDelayString = "#{T(org.apache.commons.lang3.RandomUtils).nextLong(0, ${sql.ttl.notifications.checking_interval_ms:86400000})}", |
|||
fixedDelayString = "${sql.ttl.notifications.checking_interval_ms:86400000}") |
|||
public void cleanUp() { |
|||
long expTime = System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(ttlInSec); |
|||
long partitionDurationMs = TimeUnit.HOURS.toMillis(partitionSizeInHours); |
|||
if (!isSystemTenantPartitionMine()) { |
|||
partitioningRepository.cleanupPartitionsCache(NOTIFICATION_TABLE_NAME, expTime, partitionDurationMs); |
|||
return; |
|||
} |
|||
|
|||
long lastRemovedNotificationTs = partitioningRepository.dropPartitionsBefore(NOTIFICATION_TABLE_NAME, expTime, partitionDurationMs); |
|||
if (lastRemovedNotificationTs > 0) { |
|||
long gap = TimeUnit.MINUTES.toMillis(10); |
|||
long requestExpTime = lastRemovedNotificationTs - TimeUnit.SECONDS.toMillis(NotificationRequestConfig.MAX_SENDING_DELAY) - gap; |
|||
// TODO: double-check this
|
|||
notificationRequestDao.removeAllByCreatedTimeBefore(requestExpTime); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,38 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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; |
|||
|
|||
import lombok.Getter; |
|||
import lombok.RequiredArgsConstructor; |
|||
|
|||
import java.util.Arrays; |
|||
import java.util.Optional; |
|||
|
|||
@RequiredArgsConstructor |
|||
@Getter |
|||
public enum WebSocketSessionType { |
|||
TELEMETRY("telemetry"), |
|||
NOTIFICATIONS("notifications"); |
|||
|
|||
private final String name; |
|||
|
|||
public static Optional<WebSocketSessionType> forName(String name) { |
|||
return Arrays.stream(values()) |
|||
.filter(sessionType -> sessionType.getName().equals(name)) |
|||
.findFirst(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,256 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.context.annotation.Lazy; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.rule.engine.api.NotificationCenter; |
|||
import org.thingsboard.server.common.data.id.IdBased; |
|||
import org.thingsboard.server.common.data.id.NotificationId; |
|||
import org.thingsboard.server.common.data.id.UserId; |
|||
import org.thingsboard.server.common.data.notification.Notification; |
|||
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.service.security.model.SecurityUser; |
|||
import org.thingsboard.server.service.subscription.TbLocalSubscriptionService; |
|||
import org.thingsboard.server.service.ws.WebSocketService; |
|||
import org.thingsboard.server.service.ws.WebSocketSessionRef; |
|||
import org.thingsboard.server.service.ws.notification.cmd.MarkAllNotificationsAsReadCmd; |
|||
import org.thingsboard.server.service.ws.notification.cmd.MarkNotificationsAsReadCmd; |
|||
import org.thingsboard.server.service.ws.notification.cmd.NotificationsCountSubCmd; |
|||
import org.thingsboard.server.service.ws.notification.cmd.NotificationsSubCmd; |
|||
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; |
|||
import org.thingsboard.server.service.ws.telemetry.cmd.v2.CmdUpdate; |
|||
import org.thingsboard.server.service.ws.telemetry.cmd.v2.UnsubscribeCmd; |
|||
|
|||
import java.util.Set; |
|||
import java.util.UUID; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@Service |
|||
@TbCoreComponent |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class DefaultNotificationCommandsHandler implements NotificationCommandsHandler { |
|||
|
|||
private final NotificationService notificationService; |
|||
private final TbLocalSubscriptionService localSubscriptionService; |
|||
private final NotificationCenter notificationCenter; |
|||
private final TbServiceInfoProvider serviceInfoProvider; |
|||
@Autowired @Lazy |
|||
private WebSocketService wsService; |
|||
|
|||
@Override |
|||
public void handleUnreadNotificationsSubCmd(WebSocketSessionRef sessionRef, NotificationsSubCmd cmd) { |
|||
log.debug("[{}] Handling unread notifications subscription cmd (cmdId: {})", sessionRef.getSessionId(), cmd.getCmdId()); |
|||
SecurityUser securityCtx = sessionRef.getSecurityCtx(); |
|||
NotificationsSubscription subscription = NotificationsSubscription.builder() |
|||
.serviceId(serviceInfoProvider.getServiceId()) |
|||
.sessionId(sessionRef.getSessionId()) |
|||
.subscriptionId(cmd.getCmdId()) |
|||
.tenantId(securityCtx.getTenantId()) |
|||
.entityId(securityCtx.getId()) |
|||
.updateProcessor(this::handleNotificationsSubscriptionUpdate) |
|||
.limit(cmd.getLimit()) |
|||
.build(); |
|||
localSubscriptionService.addSubscription(subscription); |
|||
|
|||
fetchUnreadNotifications(subscription); |
|||
sendUpdate(sessionRef.getSessionId(), subscription.createFullUpdate()); |
|||
} |
|||
|
|||
@Override |
|||
public void handleUnreadNotificationsCountSubCmd(WebSocketSessionRef sessionRef, NotificationsCountSubCmd cmd) { |
|||
log.debug("[{}] Handling unread notifications count subscription cmd (cmdId: {})", sessionRef.getSessionId(), cmd.getCmdId()); |
|||
SecurityUser securityCtx = sessionRef.getSecurityCtx(); |
|||
NotificationsCountSubscription subscription = NotificationsCountSubscription.builder() |
|||
.serviceId(serviceInfoProvider.getServiceId()) |
|||
.sessionId(sessionRef.getSessionId()) |
|||
.subscriptionId(cmd.getCmdId()) |
|||
.tenantId(securityCtx.getTenantId()) |
|||
.entityId(securityCtx.getId()) |
|||
.updateProcessor(this::handleNotificationsCountSubscriptionUpdate) |
|||
.build(); |
|||
localSubscriptionService.addSubscription(subscription); |
|||
|
|||
fetchUnreadNotificationsCount(subscription); |
|||
sendUpdate(sessionRef.getSessionId(), subscription.createUpdate()); |
|||
} |
|||
|
|||
private void fetchUnreadNotifications(NotificationsSubscription subscription) { |
|||
log.trace("[{}, subId: {}] Fetching unread notifications from DB", subscription.getSessionId(), subscription.getSubscriptionId()); |
|||
PageData<Notification> notifications = notificationService.findLatestUnreadNotificationsByRecipientId(subscription.getTenantId(), |
|||
(UserId) subscription.getEntityId(), subscription.getLimit()); |
|||
subscription.getLatestUnreadNotifications().clear(); |
|||
notifications.getData().forEach(notification -> { |
|||
subscription.getLatestUnreadNotifications().put(notification.getUuidId(), notification); |
|||
}); |
|||
subscription.getTotalUnreadCounter().set((int) notifications.getTotalElements()); |
|||
} |
|||
|
|||
private void fetchUnreadNotificationsCount(NotificationsCountSubscription subscription) { |
|||
log.trace("[{}, subId: {}] Fetching unread notifications count from DB", subscription.getSessionId(), subscription.getSubscriptionId()); |
|||
int unreadCount = notificationService.countUnreadNotificationsByRecipientId(subscription.getTenantId(), (UserId) subscription.getEntityId()); |
|||
subscription.getUnreadCounter().set(unreadCount); |
|||
} |
|||
|
|||
|
|||
/* Notifications subscription update handling */ |
|||
private void handleNotificationsSubscriptionUpdate(NotificationsSubscription subscription, NotificationsSubscriptionUpdate subscriptionUpdate) { |
|||
if (subscriptionUpdate.getNotificationUpdate() != null) { |
|||
handleNotificationUpdate(subscription, subscriptionUpdate.getNotificationUpdate()); |
|||
} else if (subscriptionUpdate.getNotificationRequestUpdate() != null) { |
|||
handleNotificationRequestUpdate(subscription, subscriptionUpdate.getNotificationRequestUpdate()); |
|||
} |
|||
} |
|||
|
|||
private void handleNotificationUpdate(NotificationsSubscription subscription, NotificationUpdate update) { |
|||
log.trace("[{}, subId: {}] Handling notification update: {}", subscription.getSessionId(), subscription.getSubscriptionId(), update); |
|||
Notification notification = update.getNotification(); |
|||
UUID notificationId = update.getNotificationId(); |
|||
switch (update.getUpdateType()) { |
|||
case CREATED: { |
|||
subscription.getLatestUnreadNotifications().put(notificationId, notification); |
|||
subscription.getTotalUnreadCounter().incrementAndGet(); |
|||
if (subscription.getLatestUnreadNotifications().size() > subscription.getLimit()) { |
|||
Set<UUID> beyondLimit = subscription.getSortedNotifications().stream().skip(subscription.getLimit()) |
|||
.map(IdBased::getUuidId).collect(Collectors.toSet()); |
|||
beyondLimit.forEach(id -> subscription.getLatestUnreadNotifications().remove(id)); |
|||
} |
|||
sendUpdate(subscription.getSessionId(), subscription.createPartialUpdate(notification)); |
|||
break; |
|||
} |
|||
case UPDATED: { |
|||
if (update.getUpdatedStatus() == NotificationStatus.READ) { |
|||
if (update.isAllNotifications() || subscription.getLatestUnreadNotifications().containsKey(notificationId)) { |
|||
fetchUnreadNotifications(subscription); |
|||
sendUpdate(subscription.getSessionId(), subscription.createFullUpdate()); |
|||
} else { |
|||
subscription.getTotalUnreadCounter().decrementAndGet(); |
|||
sendUpdate(subscription.getSessionId(), subscription.createCountUpdate()); |
|||
} |
|||
} else if (notification.getStatus() != NotificationStatus.READ) { |
|||
if (subscription.getLatestUnreadNotifications().containsKey(notificationId)) { |
|||
subscription.getLatestUnreadNotifications().put(notificationId, notification); |
|||
sendUpdate(subscription.getSessionId(), subscription.createPartialUpdate(notification)); |
|||
} |
|||
} |
|||
break; |
|||
} |
|||
case DELETED: { |
|||
if (subscription.getLatestUnreadNotifications().containsKey(notificationId)) { |
|||
fetchUnreadNotifications(subscription); |
|||
sendUpdate(subscription.getSessionId(), subscription.createFullUpdate()); |
|||
} else if (notification.getStatus() != NotificationStatus.READ) { |
|||
subscription.getTotalUnreadCounter().decrementAndGet(); |
|||
sendUpdate(subscription.getSessionId(), subscription.createCountUpdate()); |
|||
} |
|||
break; |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void handleNotificationRequestUpdate(NotificationsSubscription subscription, NotificationRequestUpdate update) { |
|||
log.trace("[{}, subId: {}] Handling notification request update: {}", subscription.getSessionId(), subscription.getSubscriptionId(), update); |
|||
fetchUnreadNotifications(subscription); // FIXME: figure out how not to fetch notifications on each request update...
|
|||
sendUpdate(subscription.getSessionId(), subscription.createFullUpdate()); |
|||
} |
|||
|
|||
|
|||
/* Notifications count subscription update handling */ |
|||
private void handleNotificationsCountSubscriptionUpdate(NotificationsCountSubscription subscription, NotificationsSubscriptionUpdate subscriptionUpdate) { |
|||
if (subscriptionUpdate.getNotificationUpdate() != null) { |
|||
handleNotificationUpdate(subscription, subscriptionUpdate.getNotificationUpdate()); |
|||
} else if (subscriptionUpdate.getNotificationRequestUpdate() != null) { |
|||
handleNotificationRequestUpdate(subscription, subscriptionUpdate.getNotificationRequestUpdate()); |
|||
} |
|||
} |
|||
|
|||
private void handleNotificationUpdate(NotificationsCountSubscription subscription, NotificationUpdate update) { |
|||
log.trace("[{}, subId: {}] Handling notification update for count sub: {}", subscription.getSessionId(), subscription.getSubscriptionId(), update); |
|||
Notification notification = update.getNotification(); |
|||
switch (update.getUpdateType()) { |
|||
case CREATED: { |
|||
subscription.getUnreadCounter().incrementAndGet(); |
|||
sendUpdate(subscription.getSessionId(), subscription.createUpdate()); |
|||
break; |
|||
} |
|||
case UPDATED: { |
|||
if (update.getUpdatedStatus() == NotificationStatus.READ) { |
|||
if (update.isAllNotifications()) { |
|||
fetchUnreadNotificationsCount(subscription); |
|||
} else { |
|||
subscription.getUnreadCounter().decrementAndGet(); |
|||
} |
|||
sendUpdate(subscription.getSessionId(), subscription.createUpdate()); |
|||
} |
|||
break; |
|||
} |
|||
case DELETED: { |
|||
if (notification.getStatus() != NotificationStatus.READ) { |
|||
subscription.getUnreadCounter().decrementAndGet(); |
|||
sendUpdate(subscription.getSessionId(), subscription.createUpdate()); |
|||
} |
|||
break; |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void handleNotificationRequestUpdate(NotificationsCountSubscription subscription, NotificationRequestUpdate update) { |
|||
log.trace("[{}, subId: {}] Handling notification request update for count sub: {}", subscription.getSessionId(), subscription.getSubscriptionId(), update); |
|||
fetchUnreadNotificationsCount(subscription); // FIXME: figure out how not to fetch notifications on each request update...
|
|||
sendUpdate(subscription.getSessionId(), subscription.createUpdate()); |
|||
} |
|||
|
|||
|
|||
private void sendUpdate(String sessionId, CmdUpdate update) { |
|||
log.trace("[{}, cmdId: {}] Sending WS update: {}", sessionId, update.getCmdId(), update); |
|||
wsService.sendWsMsg(sessionId, update); |
|||
} |
|||
|
|||
|
|||
@Override |
|||
public void handleMarkAsReadCmd(WebSocketSessionRef sessionRef, MarkNotificationsAsReadCmd cmd) { |
|||
SecurityUser securityCtx = sessionRef.getSecurityCtx(); |
|||
cmd.getNotifications().stream() |
|||
.map(NotificationId::new) |
|||
.forEach(notificationId -> { |
|||
notificationCenter.markNotificationAsRead(securityCtx.getTenantId(), securityCtx.getId(), notificationId); |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
public void handleMarkAllAsReadCmd(WebSocketSessionRef sessionRef, MarkAllNotificationsAsReadCmd cmd) { |
|||
SecurityUser securityCtx = sessionRef.getSecurityCtx(); |
|||
notificationCenter.markAllNotificationsAsRead(securityCtx.getTenantId(), securityCtx.getId()); |
|||
} |
|||
|
|||
@Override |
|||
public void handleUnsubCmd(WebSocketSessionRef sessionRef, UnsubscribeCmd cmd) { |
|||
localSubscriptionService.cancelSubscription(sessionRef.getSessionId(), cmd.getCmdId()); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,37 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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; |
|||
|
|||
import org.thingsboard.server.service.ws.WebSocketSessionRef; |
|||
import org.thingsboard.server.service.ws.notification.cmd.MarkAllNotificationsAsReadCmd; |
|||
import org.thingsboard.server.service.ws.notification.cmd.MarkNotificationsAsReadCmd; |
|||
import org.thingsboard.server.service.ws.notification.cmd.NotificationsSubCmd; |
|||
import org.thingsboard.server.service.ws.notification.cmd.NotificationsCountSubCmd; |
|||
import org.thingsboard.server.service.ws.telemetry.cmd.v2.UnsubscribeCmd; |
|||
|
|||
public interface NotificationCommandsHandler { |
|||
|
|||
void handleUnreadNotificationsSubCmd(WebSocketSessionRef sessionRef, NotificationsSubCmd cmd); |
|||
|
|||
void handleUnreadNotificationsCountSubCmd(WebSocketSessionRef sessionRef, NotificationsCountSubCmd cmd); |
|||
|
|||
void handleMarkAsReadCmd(WebSocketSessionRef sessionRef, MarkNotificationsAsReadCmd cmd); |
|||
|
|||
void handleMarkAllAsReadCmd(WebSocketSessionRef sessionRef, MarkAllNotificationsAsReadCmd cmd); |
|||
|
|||
void handleUnsubCmd(WebSocketSessionRef sessionRef, UnsubscribeCmd cmd); |
|||
|
|||
} |
|||
@ -0,0 +1,27 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.cmd; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
public class MarkAllNotificationsAsReadCmd implements WsCmd { |
|||
private int cmdId; |
|||
} |
|||
@ -0,0 +1,31 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.cmd; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
public class MarkNotificationsAsReadCmd implements WsCmd { |
|||
private int cmdId; |
|||
private List<UUID> notifications; |
|||
} |
|||
@ -0,0 +1,33 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.cmd; |
|||
|
|||
import lombok.Data; |
|||
|
|||
@Data |
|||
public class NotificationCmdsWrapper { |
|||
|
|||
private NotificationsCountSubCmd unreadCountSubCmd; |
|||
|
|||
private NotificationsSubCmd unreadSubCmd; |
|||
|
|||
private MarkNotificationsAsReadCmd markAsReadCmd; |
|||
|
|||
private MarkAllNotificationsAsReadCmd markAllAsReadCmd; |
|||
|
|||
private NotificationsUnsubCmd unsubCmd; |
|||
|
|||
} |
|||
@ -0,0 +1,27 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.cmd; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
public class NotificationsCountSubCmd implements WsCmd { |
|||
private int cmdId; |
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue