Andrii Shvaika
4 years ago
245 changed files with 8383 additions and 856 deletions
@ -0,0 +1,80 @@ |
|||
-- |
|||
-- Copyright © 2016-2022 The Thingsboard Authors |
|||
-- |
|||
-- Licensed under the Apache License, Version 2.0 (the "License"); |
|||
-- you may not use this file except in compliance with the License. |
|||
-- You may obtain a copy of the License at |
|||
-- |
|||
-- http://www.apache.org/licenses/LICENSE-2.0 |
|||
-- |
|||
-- Unless required by applicable law or agreed to in writing, software |
|||
-- distributed under the License is distributed on an "AS IS" BASIS, |
|||
-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
-- See the License for the specific language governing permissions and |
|||
-- limitations under the License. |
|||
-- |
|||
|
|||
CREATE TABLE IF NOT EXISTS notification_target ( |
|||
id UUID NOT NULL CONSTRAINT notification_target_pkey PRIMARY KEY, |
|||
created_time BIGINT NOT NULL, |
|||
tenant_id UUID NULL CONSTRAINT fk_notification_target_tenant_id REFERENCES tenant(id) ON DELETE CASCADE, |
|||
name VARCHAR(255) NOT NULL, |
|||
configuration VARCHAR(10000) NOT NULL |
|||
); |
|||
CREATE INDEX IF NOT EXISTS idx_notification_target_tenant_id_created_time ON notification_target(tenant_id, created_time DESC); |
|||
|
|||
CREATE TABLE IF NOT EXISTS notification_template ( |
|||
id UUID NOT NULL CONSTRAINT notification_template_pkey PRIMARY KEY, |
|||
created_time BIGINT NOT NULL, |
|||
tenant_id UUID NULL CONSTRAINT fk_notification_template_tenant_id REFERENCES tenant(id) ON DELETE CASCADE, |
|||
name VARCHAR(255) NOT NULL, |
|||
notification_type VARCHAR(255) NOT NULL, |
|||
configuration VARCHAR(10000) NOT NULL |
|||
); |
|||
|
|||
CREATE TABLE IF NOT EXISTS notification_rule ( |
|||
id UUID NOT NULL CONSTRAINT notification_rule_pkey PRIMARY KEY, |
|||
created_time BIGINT NOT NULL, |
|||
tenant_id UUID NULL CONSTRAINT fk_notification_rule_tenant_id REFERENCES tenant(id) ON DELETE CASCADE, |
|||
name VARCHAR(255) NOT NULL, |
|||
template_id UUID NOT NULL CONSTRAINT fk_notification_rule_template_id REFERENCES notification_template(id), |
|||
delivery_methods VARCHAR(255) NOT NULL, |
|||
configuration VARCHAR(2000) NOT NULL |
|||
); |
|||
|
|||
CREATE TABLE IF NOT EXISTS notification_request ( |
|||
id UUID NOT NULL CONSTRAINT notification_request_pkey PRIMARY KEY, |
|||
created_time BIGINT NOT NULL, |
|||
tenant_id UUID NULL CONSTRAINT fk_notification_request_tenant_id REFERENCES tenant(id) ON DELETE CASCADE, |
|||
targets VARCHAR(255) NOT NULL, |
|||
template_id UUID NOT NULL, |
|||
info VARCHAR(1000), |
|||
delivery_methods VARCHAR(255), |
|||
additional_config VARCHAR(1000), |
|||
originator_type VARCHAR(32) NOT NULL, |
|||
originator_entity_id UUID, |
|||
originator_entity_type VARCHAR(32), |
|||
rule_id UUID NULL, |
|||
status VARCHAR(32), |
|||
stats VARCHAR(1000) |
|||
); |
|||
CREATE INDEX IF NOT EXISTS idx_notification_request_tenant_id_originator_type_created_time ON notification_request(tenant_id, originator_type, created_time DESC); |
|||
|
|||
CREATE TABLE IF NOT EXISTS notification ( |
|||
id UUID NOT NULL, |
|||
created_time BIGINT NOT NULL, |
|||
request_id UUID NOT NULL CONSTRAINT fk_notification_request_id REFERENCES notification_request(id) ON DELETE CASCADE, |
|||
recipient_id UUID NOT NULL CONSTRAINT fk_notification_recipient_id REFERENCES tb_user(id) ON DELETE CASCADE, |
|||
type VARCHAR(255) NOT NULL, |
|||
text VARCHAR(1000) NOT NULL, |
|||
info VARCHAR(1000), |
|||
originator_type VARCHAR(32) NOT NULL, |
|||
status VARCHAR(32) |
|||
) PARTITION BY RANGE (created_time); |
|||
CREATE INDEX IF NOT EXISTS idx_notification_id ON notification(id); |
|||
CREATE INDEX IF NOT EXISTS idx_notification_recipient_id_created_time ON notification(recipient_id, created_time DESC); |
|||
CREATE INDEX IF NOT EXISTS idx_notification_notification_request_id ON notification(request_id); |
|||
|
|||
ALTER TABLE alarm ADD COLUMN IF NOT EXISTS notification_rule_id UUID; |
|||
|
|||
ALTER TABLE tb_user ADD COLUMN IF NOT EXISTS phone VARCHAR(255); |
|||
@ -0,0 +1,174 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.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.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.NotificationManager; |
|||
import org.thingsboard.server.common.data.notification.template.SlackConversation; |
|||
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.NotificationId; |
|||
import org.thingsboard.server.common.data.id.NotificationRequestId; |
|||
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.NotificationOriginatorType; |
|||
import org.thingsboard.server.common.data.notification.NotificationRequest; |
|||
import org.thingsboard.server.common.data.notification.settings.NotificationSettings; |
|||
import org.thingsboard.server.common.data.notification.settings.SlackNotificationDeliveryMethodConfig; |
|||
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.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.security.model.SecurityUser; |
|||
import org.thingsboard.server.service.security.permission.Operation; |
|||
import org.thingsboard.server.service.security.permission.Resource; |
|||
|
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
|
|||
@RestController |
|||
@TbCoreComponent |
|||
@RequestMapping("/api") |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class NotificationController extends BaseController { |
|||
|
|||
private final NotificationService notificationService; |
|||
private final NotificationRequestService notificationRequestService; |
|||
private final NotificationManager notificationManager; |
|||
private final NotificationSettingsService notificationSettingsService; |
|||
private final SlackService slackService; |
|||
|
|||
@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 { |
|||
PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder); |
|||
return notificationService.findNotificationsByUserIdAndReadStatus(user.getTenantId(), user.getId(), unreadOnly, pageLink); |
|||
} |
|||
|
|||
@PutMapping("/notification/{id}/read") // or maybe to NotificationUpdateRequest for the future
|
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") |
|||
public void markNotificationAsRead(@PathVariable UUID id, |
|||
@AuthenticationPrincipal SecurityUser user) { |
|||
NotificationId notificationId = new NotificationId(id); |
|||
notificationManager.markNotificationAsRead(user.getTenantId(), user.getId(), notificationId); |
|||
} |
|||
|
|||
// delete notification?
|
|||
|
|||
@PostMapping("/notification/request") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public NotificationRequest createNotificationRequest(@RequestBody 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"); |
|||
} |
|||
checkEntity(notificationRequest.getId(), notificationRequest, Resource.NOTIFICATION_REQUEST); |
|||
|
|||
notificationRequest.setOriginatorType(NotificationOriginatorType.ADMIN); |
|||
notificationRequest.setOriginatorEntityId(user.getId()); |
|||
notificationRequest.setOriginatorEntity(user); |
|||
if (notificationRequest.getInfo() != null && notificationRequest.getInfo().getOriginatorType() != null) { |
|||
throw new IllegalArgumentException("Unsupported notification info type"); |
|||
} |
|||
notificationRequest.setRuleId(null); |
|||
notificationRequest.setStatus(null); |
|||
notificationRequest.setStats(null); |
|||
|
|||
return doSaveAndLog(EntityType.NOTIFICATION_REQUEST, notificationRequest, notificationManager::processNotificationRequest); |
|||
} |
|||
|
|||
@GetMapping("/notification/request/{id}") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public NotificationRequest getNotificationRequestById(@PathVariable UUID id) throws ThingsboardException { |
|||
NotificationRequestId notificationRequestId = new NotificationRequestId(id); |
|||
return checkEntityId(notificationRequestId, notificationRequestService::findNotificationRequestById, Operation.READ); |
|||
} |
|||
|
|||
@GetMapping("/notification/requests") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public PageData<NotificationRequest> 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 { |
|||
PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder); |
|||
return notificationRequestService.findNotificationRequestsByTenantId(user.getTenantId(), 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, notificationManager::deleteNotificationRequest); |
|||
} |
|||
|
|||
|
|||
@PostMapping("/notification/settings") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public NotificationSettings saveNotificationSettings(@RequestBody NotificationSettings notificationSettings, |
|||
@AuthenticationPrincipal SecurityUser user) { |
|||
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) { |
|||
TenantId tenantId = user.isSystemAdmin() ? TenantId.SYS_TENANT_ID : user.getTenantId(); |
|||
return notificationSettingsService.findNotificationSettings(tenantId); |
|||
} |
|||
|
|||
@GetMapping("/notification/slack/conversations") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public List<SlackConversation> listSlackConversations(@RequestParam SlackConversation.Type type, |
|||
@AuthenticationPrincipal SecurityUser user) { |
|||
NotificationSettings settings = getNotificationSettings(user); |
|||
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,90 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.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.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 org.thingsboard.server.service.security.permission.Resource; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
@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 NotificationRule notificationRule) throws Exception { |
|||
checkEntity(notificationRule.getId(), notificationRule, Resource.NOTIFICATION_RULE); |
|||
return doSaveAndLog(EntityType.NOTIFICATION_RULE, notificationRule, notificationRuleService::saveNotificationRule); |
|||
} |
|||
|
|||
@GetMapping("/rule/{id}") |
|||
@PreAuthorize("hasAnyAuthority('TENANT_ADMIN')") |
|||
public NotificationRule getNotificationRuleById(@PathVariable UUID id) throws ThingsboardException { |
|||
NotificationRuleId notificationRuleId = new NotificationRuleId(id); |
|||
return checkEntityId(notificationRuleId, notificationRuleService::findNotificationRuleById, Operation.READ); |
|||
} |
|||
|
|||
@GetMapping("/rules") |
|||
@PreAuthorize("hasAnyAuthority('TENANT_ADMIN')") |
|||
public PageData<NotificationRule> 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 { |
|||
PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder); |
|||
return notificationRuleService.findNotificationRulesByTenantId(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,118 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.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.User; |
|||
import org.thingsboard.server.common.data.exception.ThingsboardException; |
|||
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.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 org.thingsboard.server.service.security.permission.Resource; |
|||
|
|||
import javax.validation.Valid; |
|||
import java.util.UUID; |
|||
|
|||
@RestController |
|||
@TbCoreComponent |
|||
@RequestMapping("/api/notification") |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class NotificationTargetController extends BaseController { |
|||
|
|||
private final NotificationTargetService notificationTargetService; |
|||
|
|||
@PostMapping("/target") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public NotificationTarget saveNotificationTarget(@RequestBody @Valid NotificationTarget notificationTarget, |
|||
@AuthenticationPrincipal SecurityUser user) throws Exception { |
|||
checkEntity(notificationTarget.getId(), notificationTarget, Resource.NOTIFICATION_TARGET); |
|||
if (!user.isSystemAdmin()) { |
|||
NotificationTargetConfig targetConfig = notificationTarget.getConfiguration(); |
|||
PageDataIterable<User> recipients = new PageDataIterable<>(pageLink -> { |
|||
return notificationTargetService.findRecipientsForNotificationTargetConfig(user.getTenantId(), null, targetConfig, pageLink); |
|||
}, 200); |
|||
for (User recipient : recipients) { |
|||
accessControlService.checkPermission(user, Resource.USER, Operation.READ, recipient.getId(), recipient); |
|||
} |
|||
} |
|||
|
|||
return doSaveAndLog(EntityType.NOTIFICATION_TARGET, notificationTarget, notificationTargetService::saveNotificationTarget); |
|||
} |
|||
|
|||
@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); |
|||
} |
|||
|
|||
@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 { |
|||
PageLink pageLink = createPageLink(pageSize, page, null, null, null); |
|||
PageData<User> recipients = notificationTargetService.findRecipientsForNotificationTargetConfig(user.getTenantId(), null, notificationTarget.getConfiguration(), pageLink); |
|||
if (!user.isSystemAdmin()) { |
|||
for (User recipient : recipients.getData()) { |
|||
accessControlService.checkPermission(user, Resource.USER, Operation.READ, recipient.getId(), recipient); |
|||
} |
|||
} |
|||
return recipients; |
|||
} |
|||
|
|||
@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 { |
|||
PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder); |
|||
return notificationTargetService.findNotificationTargetsByTenantId(user.getTenantId(), pageLink); |
|||
} |
|||
|
|||
@DeleteMapping("/target/{id}") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public void deleteNotificationTarget(@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); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,69 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.controller; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import org.springframework.security.access.prepost.PreAuthorize; |
|||
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.RestController; |
|||
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.template.NotificationTemplate; |
|||
import org.thingsboard.server.dao.notification.NotificationTemplateService; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.security.permission.Operation; |
|||
import org.thingsboard.server.service.security.permission.Resource; |
|||
|
|||
import javax.validation.Valid; |
|||
import java.util.UUID; |
|||
|
|||
@RestController |
|||
@TbCoreComponent |
|||
@RequiredArgsConstructor |
|||
@RequestMapping("/api/notification") |
|||
public class NotificationTemplateController extends BaseController { |
|||
|
|||
private final NotificationTemplateService notificationTemplateService; |
|||
|
|||
@PostMapping("/template") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public NotificationTemplate saveNotificationTemplate(@RequestBody @Valid NotificationTemplate notificationTemplate) throws Exception { |
|||
checkEntity(notificationTemplate.getId(), notificationTemplate, Resource.NOTIFICATION_TEMPLATE); |
|||
return doSaveAndLog(EntityType.NOTIFICATION_TEMPLATE, notificationTemplate, notificationTemplateService::saveNotificationTemplate); |
|||
} |
|||
|
|||
@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); |
|||
} |
|||
|
|||
@DeleteMapping("/template/{id}") |
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") |
|||
public void deleteNotificationTemplate(@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); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,29 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.executors; |
|||
|
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.common.util.AbstractListeningExecutor; |
|||
|
|||
@Component |
|||
public class NotificationExecutorService extends AbstractListeningExecutor { |
|||
|
|||
@Override |
|||
protected int getThreadPollSize() { |
|||
return 10; // FIXME [viacheslav]
|
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,278 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.notification; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import lombok.RequiredArgsConstructor; |
|||
import 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.NotificationManager; |
|||
import org.thingsboard.rule.engine.api.util.TbNodeUtils; |
|||
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.NotificationRequest; |
|||
import org.thingsboard.server.common.data.notification.NotificationRequestConfig; |
|||
import org.thingsboard.server.common.data.notification.NotificationRequestStatus; |
|||
import org.thingsboard.server.common.data.notification.NotificationStatus; |
|||
import org.thingsboard.server.common.data.notification.settings.NotificationSettings; |
|||
import org.thingsboard.server.common.data.notification.template.DeliveryMethodNotificationTemplate; |
|||
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.DaoUtil; |
|||
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.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.HashSet; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Set; |
|||
import java.util.UUID; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@Service |
|||
@Slf4j |
|||
@RequiredArgsConstructor |
|||
@SuppressWarnings("UnstableApiUsage") |
|||
public class DefaultNotificationManager extends AbstractSubscriptionService implements NotificationManager, NotificationChannel { |
|||
|
|||
private final NotificationTargetService notificationTargetService; |
|||
private final NotificationRequestService notificationRequestService; |
|||
private final NotificationService notificationService; |
|||
private final NotificationTemplateService notificationTemplateService; |
|||
private final NotificationSettingsService notificationSettingsService; |
|||
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) { |
|||
log.debug("Processing notification request (tenant id: {}, notification targets: {})", tenantId, notificationRequest.getTargets()); |
|||
notificationRequest.setTenantId(tenantId); |
|||
NotificationSettings settings = notificationSettingsService.findNotificationSettings(tenantId); |
|||
notificationRequest.getDeliveryMethods().forEach(deliveryMethod -> { |
|||
if (!settings.getDeliveryMethodsConfigs().containsKey(deliveryMethod) || !settings.getDeliveryMethodsConfigs().get(deliveryMethod).isEnabled()) { |
|||
throw new IllegalArgumentException("Delivery method " + deliveryMethod + " is not enabled or configured"); |
|||
} |
|||
}); |
|||
|
|||
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; |
|||
} |
|||
} |
|||
|
|||
notificationRequest.setStatus(NotificationRequestStatus.PROCESSED); |
|||
NotificationRequest savedNotificationRequest = notificationRequestService.saveNotificationRequest(tenantId, notificationRequest); |
|||
|
|||
NotificationProcessingContext ctx = NotificationProcessingContext.builder() |
|||
.tenantId(tenantId) |
|||
.request(savedNotificationRequest) |
|||
.settings(settings) |
|||
.build(); |
|||
ctx.init(notificationTemplateService); |
|||
|
|||
for (NotificationTargetId targetId : notificationRequest.getTargets()) { |
|||
DaoUtil.processBatches(pageLink -> { |
|||
return notificationTargetService.findRecipientsForNotificationTarget(tenantId, ctx.getOriginatorCustomerId(), targetId, pageLink); |
|||
}, 200, recipientsBatch -> { |
|||
List<ListenableFuture<Void>> results = new ArrayList<>(); |
|||
for (NotificationDeliveryMethod deliveryMethod : savedNotificationRequest.getDeliveryMethods()) { |
|||
NotificationChannel notificationChannel = channels.get(deliveryMethod); |
|||
log.debug("Sending {} notifications for request {} to recipients batch", deliveryMethod, savedNotificationRequest.getId()); |
|||
|
|||
List<User> recipients = recipientsBatch.getData(); |
|||
for (User recipient : recipients) { |
|||
ListenableFuture<Void> resultFuture = processForRecipient(notificationChannel, recipient, ctx); |
|||
DonAsynchron.withCallback(resultFuture, result -> { |
|||
ctx.getStats().reportSent(deliveryMethod, recipient); |
|||
}, error -> { |
|||
ctx.getStats().reportError(deliveryMethod, recipient, error); |
|||
}); |
|||
results.add(resultFuture); |
|||
} |
|||
} |
|||
|
|||
Futures.allAsList(results).addListener(() -> { |
|||
try { |
|||
notificationRequestService.updateNotificationRequestStats(tenantId, savedNotificationRequest.getId(), ctx.getStats()); |
|||
} catch (Exception e) { |
|||
log.error("Failed to update stats for notification request {}", savedNotificationRequest.getId(), e); |
|||
} |
|||
}, dbCallbackExecutorService); |
|||
}); |
|||
} |
|||
|
|||
return savedNotificationRequest; |
|||
} |
|||
|
|||
private ListenableFuture<Void> processForRecipient(NotificationChannel notificationChannel, User recipient, NotificationProcessingContext ctx) { |
|||
NotificationDeliveryMethod deliveryMethod = notificationChannel.getDeliveryMethod(); |
|||
if (ctx.getStats().contains(deliveryMethod, recipient.getId())) { |
|||
return Futures.immediateFailedFuture(new AlreadySentException()); |
|||
} |
|||
String text; |
|||
try { |
|||
DeliveryMethodNotificationTemplate template = ctx.getTemplate(deliveryMethod); |
|||
text = TbNodeUtils.processTemplate(template.getBody(), ctx.createTemplateContext(recipient)); |
|||
} catch (Exception e) { |
|||
return Futures.immediateFailedFuture(e); |
|||
} |
|||
return notificationChannel.sendNotification(recipient, text, ctx); |
|||
} |
|||
|
|||
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); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> sendNotification(User recipient, String text, NotificationProcessingContext ctx) { |
|||
NotificationRequest request = ctx.getRequest(); |
|||
log.trace("Creating notification for recipient {} (notification request id: {})", recipient.getId(), request.getId()); |
|||
Notification notification = Notification.builder() |
|||
.requestId(request.getId()) |
|||
.recipientId(recipient.getId()) |
|||
.type(ctx.getNotificationTemplate().getNotificationType()) |
|||
.text(text) |
|||
.info(request.getInfo()) |
|||
.originatorType(request.getOriginatorType()) |
|||
.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); |
|||
} |
|||
return onNotificationUpdate(recipient.getTenantId(), recipient.getId(), notification, true); |
|||
} |
|||
|
|||
@Override |
|||
public void markNotificationAsRead(TenantId tenantId, UserId recipientId, NotificationId notificationId) { |
|||
boolean updated = notificationService.markNotificationAsRead(tenantId, recipientId, notificationId); |
|||
if (updated) { |
|||
log.debug("Marking notification {} as read (recipient id: {}, tenant id: {})", notificationId, recipientId, tenantId); |
|||
Notification notification = notificationService.findNotificationById(tenantId, notificationId); |
|||
onNotificationUpdate(tenantId, recipientId, notification, false); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void deleteNotificationRequest(TenantId tenantId, NotificationRequestId notificationRequestId) { |
|||
log.debug("Deleting notification request {}", notificationRequestId); |
|||
notificationRequestService.deleteNotificationRequestById(tenantId, notificationRequestId); |
|||
onNotificationRequestUpdate(tenantId, NotificationRequestUpdate.builder() |
|||
.notificationRequestId(notificationRequestId) |
|||
.deleted(true) |
|||
.build()); |
|||
clusterService.broadcastEntityStateChangeEvent(tenantId, notificationRequestId, ComponentLifecycleEvent.DELETED); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationRequest updateNotificationRequest(TenantId tenantId, NotificationRequest notificationRequest) { |
|||
log.debug("Updating notification request {}", notificationRequest.getId()); |
|||
notificationRequest = notificationRequestService.saveNotificationRequest(tenantId, notificationRequest); |
|||
onNotificationRequestUpdate(tenantId, NotificationRequestUpdate.builder() |
|||
.notificationRequestId(notificationRequest.getId()) |
|||
.notificationInfo(notificationRequest.getInfo()) |
|||
.deleted(false) |
|||
.build()); |
|||
return notificationRequest; |
|||
} |
|||
|
|||
private ListenableFuture<Void> onNotificationUpdate(TenantId tenantId, UserId recipientId, Notification notification, boolean isNew) { |
|||
NotificationUpdate update = NotificationUpdate.builder() |
|||
.notification(notification) |
|||
.isNew(isNew) |
|||
.build(); |
|||
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) { |
|||
// todo: check delivery method
|
|||
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.WEBSOCKET; |
|||
} |
|||
|
|||
@Override |
|||
protected String getExecutorPrefix() { |
|||
return "notification"; |
|||
} |
|||
|
|||
@Autowired |
|||
public void setChannels(List<NotificationChannel> channels, NotificationManager websocketNotificationChannel) { |
|||
this.channels = channels.stream().collect(Collectors.toMap(NotificationChannel::getDeliveryMethod, c -> c)); |
|||
this.channels.put(NotificationDeliveryMethod.WEBSOCKET, (NotificationChannel) websocketNotificationChannel); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,180 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.notification; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import lombok.RequiredArgsConstructor; |
|||
import 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.rule.engine.api.NotificationManager; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.alarm.Alarm; |
|||
import org.thingsboard.server.common.data.id.NotificationRequestId; |
|||
import org.thingsboard.server.common.data.id.NotificationRuleId; |
|||
import org.thingsboard.server.common.data.id.NotificationTargetId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.notification.AlarmOriginatedNotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.NotificationInfo; |
|||
import org.thingsboard.server.common.data.notification.NotificationOriginatorType; |
|||
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.rule.NonConfirmedNotificationEscalation; |
|||
import org.thingsboard.server.common.data.notification.rule.NotificationRule; |
|||
import org.thingsboard.server.common.data.notification.rule.NotificationRuleConfig; |
|||
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; |
|||
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.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.executors.NotificationExecutorService; |
|||
|
|||
import java.util.List; |
|||
import java.util.Map; |
|||
|
|||
@Service |
|||
@TbCoreComponent |
|||
@RequiredArgsConstructor |
|||
@Slf4j |
|||
public class DefaultNotificationRuleProcessingService implements NotificationRuleProcessingService { |
|||
|
|||
private final NotificationRuleService notificationRuleService; |
|||
private final NotificationRequestService notificationRequestService; |
|||
@Autowired @Lazy |
|||
private NotificationManager notificationManager; |
|||
private final NotificationExecutorService notificationExecutor; |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> onAlarmCreatedOrUpdated(TenantId tenantId, Alarm alarm) { |
|||
return processAlarmUpdate(tenantId, alarm, false); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> onAlarmDeleted(TenantId tenantId, Alarm alarm) { |
|||
return processAlarmUpdate(tenantId, alarm, true); |
|||
} |
|||
|
|||
private ListenableFuture<Void> processAlarmUpdate(TenantId tenantId, Alarm alarm, boolean deleted) { |
|||
if (alarm.getNotificationRuleId() == null) return Futures.immediateFuture(null); |
|||
return notificationExecutor.submit(() -> { |
|||
onAlarmUpdate(tenantId, alarm.getNotificationRuleId(), alarm, deleted); |
|||
return null; |
|||
}); |
|||
} |
|||
|
|||
private void onAlarmUpdate(TenantId tenantId, NotificationRuleId notificationRuleId, Alarm alarm, boolean deleted) { |
|||
log.debug("Processing alarm update ({}) with notification rule {}", alarm.getId(), notificationRuleId); |
|||
List<NotificationRequest> notificationRequests = notificationRequestService.findNotificationRequestsByRuleIdAndOriginatorEntityId(tenantId, notificationRuleId, alarm.getId()); |
|||
NotificationRule notificationRule = notificationRuleService.findNotificationRuleById(tenantId, notificationRuleId); |
|||
if (notificationRule == null) return; |
|||
|
|||
if (alarmAcknowledged(alarm) || deleted) { |
|||
if (notificationRequests.isEmpty()) { |
|||
return; |
|||
} |
|||
for (NotificationRequest notificationRequest : notificationRequests) { |
|||
if (notificationRequest.getStatus() == NotificationRequestStatus.SCHEDULED) { |
|||
notificationManager.deleteNotificationRequest(tenantId, notificationRequest.getId()); |
|||
} |
|||
} |
|||
} |
|||
|
|||
if (notificationRequests.isEmpty()) { |
|||
NotificationRuleConfig config = notificationRule.getConfiguration(); |
|||
NotificationTargetId initialNotificationTargetId = config.getInitialNotificationTargetId(); |
|||
if (initialNotificationTargetId != null) { |
|||
submitNotificationRequest(tenantId, initialNotificationTargetId, notificationRule, alarm, 0); |
|||
} |
|||
if (config.getEscalationConfig() != null) { |
|||
for (NonConfirmedNotificationEscalation escalation : config.getEscalationConfig().getEscalations()) { |
|||
submitNotificationRequest(tenantId, escalation.getNotificationTargetId(), notificationRule, alarm, escalation.getDelayInSec()); |
|||
} |
|||
} |
|||
} else { |
|||
NotificationInfo newNotificationInfo = constructNotificationInfo(alarm); |
|||
for (NotificationRequest notificationRequest : notificationRequests) { |
|||
NotificationInfo previousNotificationInfo = notificationRequest.getInfo(); |
|||
if (!previousNotificationInfo.equals(newNotificationInfo)) { |
|||
notificationRequest.setInfo(newNotificationInfo); |
|||
notificationManager.updateNotificationRequest(tenantId, notificationRequest); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
private boolean alarmAcknowledged(Alarm alarm) { |
|||
return alarm.getStatus().isAck() && alarm.getStatus().isCleared(); |
|||
} |
|||
|
|||
private void submitNotificationRequest(TenantId tenantId, NotificationTargetId targetId, NotificationRule notificationRule, Alarm alarm, int delayInSec) { |
|||
NotificationRequestConfig config = new NotificationRequestConfig(); |
|||
if (delayInSec > 0) { |
|||
config.setSendingDelayInSec(delayInSec); |
|||
} |
|||
NotificationInfo notificationInfo = constructNotificationInfo(alarm); |
|||
Map<String, String> templateContext = Map.of( |
|||
"alarmType", alarm.getType(), |
|||
"alarmId", alarm.getId().toString(), |
|||
"alarmOriginatorEntityType", alarm.getOriginator().getEntityType().toString(), |
|||
"alarmOriginatorId", alarm.getOriginator().getId().toString() |
|||
); |
|||
NotificationRequest notificationRequest = NotificationRequest.builder() |
|||
.tenantId(tenantId) |
|||
.targets(List.of(targetId)) |
|||
.templateId(notificationRule.getTemplateId()) |
|||
.deliveryMethods(notificationRule.getDeliveryMethods()) |
|||
.additionalConfig(config) |
|||
.info(notificationInfo) |
|||
.ruleId(notificationRule.getId()) |
|||
.originatorType(NotificationOriginatorType.ALARM) |
|||
.originatorEntityId(alarm.getId()) |
|||
.originatorEntity(alarm) |
|||
.templateContext(templateContext) |
|||
.build(); |
|||
notificationManager.processNotificationRequest(tenantId, notificationRequest); |
|||
} |
|||
|
|||
private NotificationInfo constructNotificationInfo(Alarm alarm) { |
|||
// TODO: add info about assignee
|
|||
return AlarmOriginatedNotificationInfo.builder() |
|||
.alarmId(alarm.getId()) |
|||
.alarmType(alarm.getType()) |
|||
.alarmOriginator(alarm.getOriginator()) |
|||
.alarmSeverity(alarm.getSeverity()) |
|||
.alarmStatus(alarm.getStatus()) |
|||
.build(); |
|||
} |
|||
|
|||
@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(); |
|||
List<NotificationRequestId> scheduledForRule = notificationRequestService.findNotificationRequestsIdsByStatusAndRuleId(tenantId, NotificationRequestStatus.SCHEDULED, notificationRuleId); |
|||
for (NotificationRequestId notificationRequestId : scheduledForRule) { |
|||
notificationManager.deleteNotificationRequest(tenantId, notificationRequestId); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,166 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.notification; |
|||
|
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import 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.NotificationManager; |
|||
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.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 NotificationManager notificationManager; |
|||
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(() -> { |
|||
notificationManager.processNotificationRequest(tenantId, notificationRequest); |
|||
}); |
|||
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,101 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.notification; |
|||
|
|||
import com.google.common.base.Strings; |
|||
import lombok.Builder; |
|||
import lombok.Getter; |
|||
import org.apache.commons.lang3.StringUtils; |
|||
import org.thingsboard.server.common.data.HasCustomerId; |
|||
import org.thingsboard.server.common.data.User; |
|||
import org.thingsboard.server.common.data.id.CustomerId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod; |
|||
import org.thingsboard.server.common.data.notification.NotificationRequest; |
|||
import org.thingsboard.server.common.data.notification.NotificationRequestStats; |
|||
import org.thingsboard.server.common.data.notification.settings.NotificationDeliveryMethodConfig; |
|||
import org.thingsboard.server.common.data.notification.settings.NotificationSettings; |
|||
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.NotificationTemplateConfig; |
|||
import org.thingsboard.server.dao.notification.NotificationTemplateService; |
|||
|
|||
import java.util.HashMap; |
|||
import java.util.Map; |
|||
|
|||
@SuppressWarnings("unchecked") |
|||
public class NotificationProcessingContext { |
|||
|
|||
@Getter |
|||
private final TenantId tenantId; |
|||
private final HasCustomerId originatorEntity; |
|||
private final NotificationSettings settings; |
|||
@Getter |
|||
private final NotificationRequest request; |
|||
private final Map<String, String> additionalTemplateContext; |
|||
|
|||
@Getter |
|||
private NotificationTemplate notificationTemplate; |
|||
private Map<NotificationDeliveryMethod, DeliveryMethodNotificationTemplate> templates; |
|||
@Getter |
|||
private final NotificationRequestStats stats; |
|||
|
|||
@Builder |
|||
public NotificationProcessingContext(TenantId tenantId, NotificationRequest request, NotificationSettings settings) { |
|||
this.tenantId = tenantId; |
|||
this.originatorEntity = request.getOriginatorEntity(); |
|||
this.request = request; |
|||
this.settings = settings; |
|||
this.additionalTemplateContext = request.getTemplateContext(); |
|||
this.stats = new NotificationRequestStats(); |
|||
} |
|||
|
|||
public void init(NotificationTemplateService templateService) { |
|||
notificationTemplate = templateService.findNotificationTemplateById(tenantId, request.getTemplateId()); |
|||
NotificationTemplateConfig config = notificationTemplate.getConfiguration(); |
|||
for (NotificationDeliveryMethod deliveryMethod : request.getDeliveryMethods()) { |
|||
DeliveryMethodNotificationTemplate template = config.getTemplates().get(deliveryMethod); |
|||
if (StringUtils.isEmpty(template.getBody())) { |
|||
template.setBody(config.getDefaultTextTemplate()); |
|||
} |
|||
} |
|||
templates = config.getTemplates(); |
|||
} |
|||
|
|||
public <T extends DeliveryMethodNotificationTemplate> T getTemplate(NotificationDeliveryMethod deliveryMethod) { |
|||
return (T) templates.get(deliveryMethod); |
|||
} |
|||
|
|||
public <C extends NotificationDeliveryMethodConfig> C getDeliveryMethodConfig(NotificationDeliveryMethod deliveryMethod) { |
|||
return (C) settings.getDeliveryMethodsConfigs().get(deliveryMethod); |
|||
} |
|||
|
|||
public Map<String, String> createTemplateContext(User recipient) { |
|||
Map<String, String> templateContext = new HashMap<>(); |
|||
templateContext.put("email", recipient.getEmail()); |
|||
templateContext.put("firstName", Strings.nullToEmpty(recipient.getFirstName())); |
|||
templateContext.put("lastName", Strings.nullToEmpty(recipient.getLastName())); |
|||
if (additionalTemplateContext != null) { |
|||
templateContext.putAll(additionalTemplateContext); |
|||
} |
|||
return templateContext; |
|||
} |
|||
|
|||
public CustomerId getOriginatorCustomerId() { |
|||
return originatorEntity != null ? originatorEntity.getCustomerId() : null; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,28 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.notification; |
|||
|
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import org.thingsboard.server.common.data.alarm.Alarm; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
public interface NotificationRuleProcessingService { |
|||
|
|||
ListenableFuture<Void> onAlarmCreatedOrUpdated(TenantId tenantId, Alarm alarm); |
|||
|
|||
ListenableFuture<Void> onAlarmDeleted(TenantId tenantId, Alarm alarm); |
|||
|
|||
} |
|||
@ -0,0 +1,25 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.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,49 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.notification.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.service.notification.NotificationProcessingContext; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
public class EmailNotificationChannel implements NotificationChannel { |
|||
|
|||
private final MailService mailService; |
|||
private final MailExecutorService executor; |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> sendNotification(User recipient, String text, NotificationProcessingContext ctx) { |
|||
EmailDeliveryMethodNotificationTemplate template = ctx.getTemplate(NotificationDeliveryMethod.EMAIL); |
|||
return executor.submit(() -> { |
|||
mailService.sendEmail(recipient.getTenantId(), recipient.getEmail(), text, template.getSubject()); |
|||
return null; |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationDeliveryMethod getDeliveryMethod() { |
|||
return NotificationDeliveryMethod.EMAIL; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,29 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.notification.channels; |
|||
|
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import org.thingsboard.server.common.data.User; |
|||
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod; |
|||
import org.thingsboard.server.service.notification.NotificationProcessingContext; |
|||
|
|||
public interface NotificationChannel { |
|||
|
|||
ListenableFuture<Void> sendNotification(User recipient, String text, NotificationProcessingContext ctx); |
|||
|
|||
NotificationDeliveryMethod getDeliveryMethod(); |
|||
|
|||
} |
|||
@ -0,0 +1,77 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.notification.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.server.common.data.notification.template.SlackConversation; |
|||
import org.thingsboard.rule.engine.api.slack.SlackService; |
|||
import org.thingsboard.server.common.data.User; |
|||
import org.thingsboard.server.common.data.notification.AlreadySentException; |
|||
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod; |
|||
import org.thingsboard.server.common.data.notification.settings.SlackNotificationDeliveryMethodConfig; |
|||
import org.thingsboard.server.common.data.notification.template.SlackDeliveryMethodNotificationTemplate; |
|||
import org.thingsboard.server.service.executors.ExternalCallExecutorService; |
|||
import org.thingsboard.server.service.notification.NotificationProcessingContext; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
@SuppressWarnings("UnstableApiUsage") |
|||
public class SlackNotificationChannel implements NotificationChannel { |
|||
|
|||
private final SlackService slackService; |
|||
private final ExternalCallExecutorService executor; |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> sendNotification(User recipient, String text, NotificationProcessingContext ctx) { |
|||
SlackDeliveryMethodNotificationTemplate template = ctx.getTemplate(NotificationDeliveryMethod.SLACK); |
|||
SlackNotificationDeliveryMethodConfig config = ctx.getDeliveryMethodConfig(NotificationDeliveryMethod.SLACK); |
|||
|
|||
if (StringUtils.isNotEmpty(template.getConversationId())) { // if conversationId is set, we only need to send message once
|
|||
if (ctx.getStats().contains(NotificationDeliveryMethod.SLACK)) { |
|||
return Futures.immediateFailedFuture(new AlreadySentException()); |
|||
} else { |
|||
return executor.submit(() -> { |
|||
slackService.sendMessage(ctx.getTenantId(), config.getBotToken(), template.getConversationId(), text); |
|||
return null; |
|||
}); |
|||
} |
|||
} else { |
|||
if (StringUtils.isNoneEmpty(recipient.getFirstName(), recipient.getLastName())) { |
|||
String username = StringUtils.join(new String[]{recipient.getFirstName(), recipient.getLastName()}, ' '); |
|||
return executor.submit(() -> { |
|||
SlackConversation conversation = slackService.findConversation(recipient.getTenantId(), config.getBotToken(), SlackConversation.Type.USER, username); |
|||
if (conversation == null) { |
|||
throw new IllegalArgumentException("Slack user not found for given name '" + username + "'"); |
|||
} |
|||
slackService.sendMessage(ctx.getTenantId(), config.getBotToken(), conversation.getId(), text); |
|||
return null; |
|||
}); |
|||
} else { |
|||
return Futures.immediateFailedFuture(new IllegalArgumentException("Couldn't determine Slack username for the user")); |
|||
} |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public NotificationDeliveryMethod getDeliveryMethod() { |
|||
return NotificationDeliveryMethod.SLACK; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,51 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.notification.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.service.notification.NotificationProcessingContext; |
|||
import org.thingsboard.server.service.sms.SmsExecutorService; |
|||
|
|||
@Component |
|||
@RequiredArgsConstructor |
|||
public class SmsNotificationChannel implements NotificationChannel { |
|||
|
|||
private final SmsService smsService; |
|||
private final SmsExecutorService executor; |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> sendNotification(User recipient, String text, 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}, text); |
|||
return null; |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
public NotificationDeliveryMethod getDeliveryMethod() { |
|||
return NotificationDeliveryMethod.SMS; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,153 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.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.template.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.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_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, SlackConversation.Type conversationType) { |
|||
return cache.get(conversationType + ":" + token, k -> { |
|||
if (conversationType == SlackConversation.Type.USER) { |
|||
UsersListRequest request = UsersListRequest.builder() |
|||
.limit(CONVERSATIONS_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 == SlackConversation.Type.PUBLIC_CHANNEL ? |
|||
ConversationType.PUBLIC_CHANNEL : |
|||
ConversationType.PRIVATE_CHANNEL)) |
|||
.limit(CONVERSATIONS_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, SlackConversation.Type 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-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.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-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.ws; |
|||
|
|||
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,232 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.ws.notification; |
|||
|
|||
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.NotificationManager; |
|||
import org.thingsboard.server.common.data.id.IdBased; |
|||
import org.thingsboard.server.common.data.id.NotificationId; |
|||
import org.thingsboard.server.common.data.id.NotificationRequestId; |
|||
import org.thingsboard.server.common.data.id.UserId; |
|||
import org.thingsboard.server.common.data.notification.Notification; |
|||
import org.thingsboard.server.common.data.notification.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.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 NotificationManager notificationManager; |
|||
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.findLatestUnreadNotificationsByUserId(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.countUnreadNotificationsByUserId(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(); |
|||
if (update.isNew()) { |
|||
subscription.getLatestUnreadNotifications().put(notification.getUuidId(), 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(notificationId -> subscription.getLatestUnreadNotifications().remove(notificationId)); |
|||
} |
|||
sendUpdate(subscription.getSessionId(), subscription.createPartialUpdate(notification)); |
|||
} else { |
|||
if (notification.getStatus() != NotificationStatus.READ) { |
|||
if (subscription.getLatestUnreadNotifications().containsKey(notification.getUuidId())) { |
|||
subscription.getLatestUnreadNotifications().put(notification.getUuidId(), notification); |
|||
sendUpdate(subscription.getSessionId(), subscription.createPartialUpdate(notification)); |
|||
} |
|||
} else { |
|||
if (subscription.getLatestUnreadNotifications().containsKey(notification.getUuidId())) { |
|||
fetchUnreadNotifications(subscription); |
|||
sendUpdate(subscription.getSessionId(), subscription.createFullUpdate()); |
|||
} else { |
|||
subscription.getTotalUnreadCounter().decrementAndGet(); |
|||
sendUpdate(subscription.getSessionId(), subscription.createCountUpdate()); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void handleNotificationRequestUpdate(NotificationsSubscription subscription, NotificationRequestUpdate update) { |
|||
log.trace("[{}, subId: {}] Handling notification request update: {}", subscription.getSessionId(), subscription.getSubscriptionId(), update); |
|||
NotificationRequestId notificationRequestId = update.getNotificationRequestId(); |
|||
if (update.isDeleted()) { |
|||
if (subscription.getLatestUnreadNotifications().values().stream() |
|||
.anyMatch(notification -> notification.getRequestId().equals(notificationRequestId))) { |
|||
fetchUnreadNotifications(subscription); |
|||
sendUpdate(subscription.getSessionId(), subscription.createFullUpdate()); |
|||
} |
|||
} else { |
|||
subscription.getLatestUnreadNotifications().values().stream() |
|||
.filter(notification -> notification.getRequestId().equals(notificationRequestId)) |
|||
.forEach(notification -> { |
|||
notification.setInfo(update.getNotificationInfo()); |
|||
sendUpdate(subscription.getSessionId(), subscription.createPartialUpdate(notification)); |
|||
}); |
|||
} |
|||
} |
|||
|
|||
|
|||
/* 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(); |
|||
if (update.isNew()) { |
|||
subscription.getUnreadCounter().incrementAndGet(); |
|||
} else if (notification.getStatus() == NotificationStatus.READ) { |
|||
// for now this can only happen when user marks notification as read
|
|||
subscription.getUnreadCounter().decrementAndGet(); |
|||
} |
|||
sendUpdate(subscription.getSessionId(), subscription.createUpdate()); |
|||
} |
|||
|
|||
private void handleNotificationRequestUpdate(NotificationsCountSubscription subscription, NotificationRequestUpdate update) { |
|||
log.trace("[{}, subId: {}] Handling notification request update for count sub: {}", subscription.getSessionId(), subscription.getSubscriptionId(), update); |
|||
if (update.isDeleted()) { |
|||
fetchUnreadNotificationsCount(subscription); |
|||
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 -> { |
|||
notificationManager.markNotificationAsRead(securityCtx.getTenantId(), securityCtx.getId(), notificationId); |
|||
// fixme: should send bulk update event, not a separate event for each notification
|
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
public void handleUnsubCmd(WebSocketSessionRef sessionRef, UnsubscribeCmd cmd) { |
|||
localSubscriptionService.cancelSubscription(sessionRef.getSessionId(), cmd.getCmdId()); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,34 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.ws.notification; |
|||
|
|||
import org.thingsboard.server.service.ws.WebSocketSessionRef; |
|||
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 handleUnsubCmd(WebSocketSessionRef sessionRef, UnsubscribeCmd cmd); |
|||
|
|||
} |
|||
@ -0,0 +1,31 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.ws.notification.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,31 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.ws.notification.cmd; |
|||
|
|||
import lombok.Data; |
|||
|
|||
@Data |
|||
public class NotificationCmdsWrapper { |
|||
|
|||
private NotificationsCountSubCmd unreadCountSubCmd; |
|||
|
|||
private NotificationsSubCmd unreadSubCmd; |
|||
|
|||
private MarkNotificationsAsReadCmd markAsReadCmd; |
|||
|
|||
private NotificationsUnsubCmd unsubCmd; |
|||
|
|||
} |
|||
@ -0,0 +1,27 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.ws.notification.cmd; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
public class NotificationsCountSubCmd implements WsCmd { |
|||
private int cmdId; |
|||
} |
|||
@ -0,0 +1,28 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.ws.notification.cmd; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
public class NotificationsSubCmd implements WsCmd { |
|||
private int cmdId; |
|||
private int limit; |
|||
} |
|||
@ -0,0 +1,28 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.ws.notification.cmd; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
import org.thingsboard.server.service.ws.telemetry.cmd.v2.UnsubscribeCmd; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
public class NotificationsUnsubCmd implements UnsubscribeCmd, WsCmd { |
|||
private int cmdId; |
|||
} |
|||
@ -0,0 +1,46 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.ws.notification.cmd; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonCreator; |
|||
import com.fasterxml.jackson.annotation.JsonProperty; |
|||
import lombok.Builder; |
|||
import lombok.Getter; |
|||
import lombok.ToString; |
|||
import org.thingsboard.server.service.ws.telemetry.cmd.v2.CmdUpdate; |
|||
import org.thingsboard.server.service.ws.telemetry.cmd.v2.CmdUpdateType; |
|||
|
|||
@Getter |
|||
@ToString |
|||
public class UnreadNotificationsCountUpdate extends CmdUpdate { |
|||
|
|||
private final int totalUnreadCount; |
|||
|
|||
@Builder |
|||
@JsonCreator |
|||
public UnreadNotificationsCountUpdate(@JsonProperty("cmdId") int cmdId, @JsonProperty("errorCode") int errorCode, |
|||
@JsonProperty("errorMsg") String errorMsg, |
|||
@JsonProperty("totalUnreadCount") int totalUnreadCount) { |
|||
super(cmdId, errorCode, errorMsg); |
|||
this.totalUnreadCount = totalUnreadCount; |
|||
} |
|||
|
|||
@Override |
|||
public CmdUpdateType getCmdUpdateType() { |
|||
return CmdUpdateType.NOTIFICATIONS_COUNT; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,55 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.ws.notification.cmd; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonCreator; |
|||
import com.fasterxml.jackson.annotation.JsonProperty; |
|||
import lombok.Builder; |
|||
import lombok.Getter; |
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.data.notification.Notification; |
|||
import org.thingsboard.server.service.ws.telemetry.cmd.v2.CmdUpdate; |
|||
import org.thingsboard.server.service.ws.telemetry.cmd.v2.CmdUpdateType; |
|||
|
|||
import java.util.Collection; |
|||
|
|||
@Getter |
|||
@ToString(exclude = "notifications") |
|||
public class UnreadNotificationsUpdate extends CmdUpdate { |
|||
|
|||
private final Collection<Notification> notifications; |
|||
private final Notification update; |
|||
private final int totalUnreadCount; |
|||
|
|||
@Builder |
|||
@JsonCreator |
|||
public UnreadNotificationsUpdate(@JsonProperty("cmdId") int cmdId, @JsonProperty("errorCode") int errorCode, |
|||
@JsonProperty("errorMsg") String errorMsg, |
|||
@JsonProperty("notifications") Collection<Notification> notifications, |
|||
@JsonProperty("update") Notification update, |
|||
@JsonProperty("totalUnreadCount") int totalUnreadCount) { |
|||
super(cmdId, errorCode, errorMsg); |
|||
this.notifications = notifications; |
|||
this.update = update; |
|||
this.totalUnreadCount = totalUnreadCount; |
|||
} |
|||
|
|||
@Override |
|||
public CmdUpdateType getCmdUpdateType() { |
|||
return CmdUpdateType.NOTIFICATIONS; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,20 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.ws.notification.cmd; |
|||
|
|||
public interface WsCmd { |
|||
int getCmdId(); |
|||
} |
|||
@ -0,0 +1,34 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.ws.notification.sub; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Builder; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
import org.thingsboard.server.common.data.id.NotificationRequestId; |
|||
import org.thingsboard.server.common.data.notification.NotificationInfo; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
@Builder |
|||
public class NotificationRequestUpdate { |
|||
private NotificationRequestId notificationRequestId; |
|||
private String notificationReason; |
|||
private NotificationInfo notificationInfo; |
|||
private boolean deleted; |
|||
} |
|||
@ -0,0 +1,31 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.ws.notification.sub; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Builder; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
import org.thingsboard.server.common.data.notification.Notification; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
@Builder |
|||
public class NotificationUpdate { |
|||
private Notification notification; |
|||
private boolean isNew; |
|||
} |
|||
@ -0,0 +1,47 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.ws.notification.sub; |
|||
|
|||
import lombok.Builder; |
|||
import lombok.Getter; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.service.subscription.TbSubscription; |
|||
import org.thingsboard.server.service.subscription.TbSubscriptionType; |
|||
import org.thingsboard.server.service.ws.notification.cmd.UnreadNotificationsCountUpdate; |
|||
|
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
import java.util.function.BiConsumer; |
|||
|
|||
@Getter |
|||
public class NotificationsCountSubscription extends TbSubscription<NotificationsSubscriptionUpdate> { |
|||
|
|||
private final AtomicInteger unreadCounter = new AtomicInteger(); |
|||
|
|||
@Builder |
|||
public NotificationsCountSubscription(String serviceId, String sessionId, int subscriptionId, TenantId tenantId, EntityId entityId, |
|||
BiConsumer<NotificationsCountSubscription, NotificationsSubscriptionUpdate> updateProcessor) { |
|||
super(serviceId, sessionId, subscriptionId, tenantId, entityId, TbSubscriptionType.NOTIFICATIONS_COUNT, updateProcessor); |
|||
} |
|||
|
|||
public UnreadNotificationsCountUpdate createUpdate() { |
|||
return UnreadNotificationsCountUpdate.builder() |
|||
.cmdId(getSubscriptionId()) |
|||
.totalUnreadCount(unreadCounter.get()) |
|||
.build(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,81 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.ws.notification.sub; |
|||
|
|||
import lombok.Builder; |
|||
import lombok.Getter; |
|||
import org.thingsboard.server.common.data.BaseData; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.notification.Notification; |
|||
import org.thingsboard.server.service.subscription.TbSubscription; |
|||
import org.thingsboard.server.service.subscription.TbSubscriptionType; |
|||
import org.thingsboard.server.service.ws.notification.cmd.UnreadNotificationsUpdate; |
|||
|
|||
import java.util.Comparator; |
|||
import java.util.HashMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
import java.util.function.BiConsumer; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@Getter |
|||
public class NotificationsSubscription extends TbSubscription<NotificationsSubscriptionUpdate> { |
|||
|
|||
private final Map<UUID, Notification> latestUnreadNotifications = new HashMap<>(); |
|||
private final int limit; |
|||
private final AtomicInteger totalUnreadCounter = new AtomicInteger(); |
|||
|
|||
@Builder |
|||
public NotificationsSubscription(String serviceId, String sessionId, int subscriptionId, TenantId tenantId, EntityId entityId, |
|||
BiConsumer<NotificationsSubscription, NotificationsSubscriptionUpdate> updateProcessor, |
|||
int limit) { |
|||
super(serviceId, sessionId, subscriptionId, tenantId, entityId, TbSubscriptionType.NOTIFICATIONS, updateProcessor); |
|||
this.limit = limit; |
|||
} |
|||
|
|||
public UnreadNotificationsUpdate createFullUpdate() { |
|||
return UnreadNotificationsUpdate.builder() |
|||
.cmdId(getSubscriptionId()) |
|||
.notifications(getSortedNotifications()) |
|||
.totalUnreadCount(totalUnreadCounter.get()) |
|||
.build(); |
|||
} |
|||
|
|||
public List<Notification> getSortedNotifications() { |
|||
return latestUnreadNotifications.values().stream() |
|||
.sorted(Comparator.comparing(BaseData::getCreatedTime, Comparator.reverseOrder())) |
|||
.collect(Collectors.toList()); |
|||
} |
|||
|
|||
public UnreadNotificationsUpdate createPartialUpdate(Notification notification) { |
|||
return UnreadNotificationsUpdate.builder() |
|||
.cmdId(getSubscriptionId()) |
|||
.update(notification) |
|||
.totalUnreadCount(totalUnreadCounter.get()) |
|||
.build(); |
|||
} |
|||
|
|||
public UnreadNotificationsUpdate createCountUpdate() { |
|||
return UnreadNotificationsUpdate.builder() |
|||
.cmdId(getSubscriptionId()) |
|||
.totalUnreadCount(totalUnreadCounter.get()) |
|||
.build(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,36 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.ws.notification.sub; |
|||
|
|||
import lombok.Data; |
|||
|
|||
@Data |
|||
public class NotificationsSubscriptionUpdate { |
|||
|
|||
private final NotificationUpdate notificationUpdate; |
|||
private final NotificationRequestUpdate notificationRequestUpdate; |
|||
|
|||
public NotificationsSubscriptionUpdate(NotificationUpdate notificationUpdate) { |
|||
this.notificationUpdate = notificationUpdate; |
|||
this.notificationRequestUpdate = null; |
|||
} |
|||
|
|||
public NotificationsSubscriptionUpdate(NotificationRequestUpdate notificationRequestUpdate) { |
|||
this.notificationUpdate = null; |
|||
this.notificationRequestUpdate = notificationRequestUpdate; |
|||
} |
|||
|
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue