From 3734a0da49db4180045155bf90b562f6071ed4ad Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Fri, 4 Nov 2022 15:29:15 +0200 Subject: [PATCH] Notifications api permission checks, WS improvements, new target configs --- .../main/data/upgrade/3.4.2/schema_update.sql | 1 + .../server/controller/BaseController.java | 18 +- .../controller/NotificationController.java | 27 ++- .../NotificationTargetController.java | 76 +++++-- .../controller/plugin/TbWebSocketHandler.java | 3 +- .../DefaultNotificationProcessingService.java | 44 ++-- .../queue/DefaultTbCoreConsumerService.java | 6 +- .../service/security/permission/Resource.java | 3 +- .../permission/SysAdminPermissions.java | 2 + .../permission/TenantAdminPermissions.java | 3 +- .../DefaultSubscriptionManagerService.java | 9 +- .../DefaultTbLocalSubscriptionService.java | 3 +- .../SubscriptionManagerService.java | 2 +- .../subscription/TbSubscriptionType.java | 2 +- .../subscription/TbSubscriptionUtils.java | 22 +- .../telemetry/DefaultWebSocketService.java | 193 +++++++++--------- .../thingsboard/server/service/ws/WsCmd.java | 20 ++ .../DefaultNotificationCommandsHandler.java | 77 +++++-- .../NotificationCommandsHandler.java | 3 + .../cmd/MarkNotificationAsReadCmd.java | 3 +- .../cmd/NotificationCmdsWrapper.java | 10 +- .../cmd/NotificationsCountSubCmd.java | 29 +++ .../notification/cmd/NotificationsSubCmd.java | 4 +- .../cmd/NotificationsUnsubCmd.java | 7 +- .../cmd/UnreadNotificationsCountUpdate.java | 44 ++++ .../sub/NotificationsCountSubscription.java | 47 +++++ .../sub}/NotificationsSubscription.java | 11 +- .../sub/NotificationsSubscriptionUpdate.java | 1 + .../ws/telemetry/cmd/v2/CmdUpdateType.java | 3 +- .../controller/TbTestWebSocketClient.java | 1 + .../NotificationsWebSocketClient.java | 50 +++-- .../notification/NotificationsWsApiTest.java | 67 +++++- common/cluster-api/src/main/proto/queue.proto | 11 +- .../dao/notification/NotificationService.java | 6 +- .../NotificationTargetService.java | 9 +- .../server/dao/user/UserService.java | 2 + .../notification/NotificationRequest.java | 9 +- .../AllUsersNotificationTargetConfig.java | 28 +++ ...CustomerUsersNotificationTargetConfig.java | 32 +++ .../targets/NotificationTargetConfig.java | 4 +- .../targets/NotificationTargetConfigType.java | 3 + .../SingleUserNotificationTargetConfig.java | 4 +- .../UserListNotificationTargetConfig.java | 5 +- .../org/thingsboard/server/dao/DaoUtil.java | 6 +- .../DefaultNotificationService.java | 15 +- .../DefaultNotificationTargetService.java | 53 +++-- .../dao/notification/NotificationDao.java | 4 +- .../sql/notification/JpaNotificationDao.java | 9 +- .../notification/NotificationRepository.java | 9 +- .../server/dao/sql/user/JpaUserDao.java | 5 + .../thingsboard/server/dao/user/UserDao.java | 3 + .../server/dao/user/UserServiceImpl.java | 5 + .../resources/sql/schema-entities-idx.sql | 2 + 53 files changed, 752 insertions(+), 263 deletions(-) create mode 100644 application/src/main/java/org/thingsboard/server/service/ws/WsCmd.java create mode 100644 application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/NotificationsCountSubCmd.java create mode 100644 application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/UnreadNotificationsCountUpdate.java create mode 100644 application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsCountSubscription.java rename application/src/main/java/org/thingsboard/server/service/{subscription => ws/notification/sub}/NotificationsSubscription.java (85%) create mode 100644 common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/AllUsersNotificationTargetConfig.java create mode 100644 common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/CustomerUsersNotificationTargetConfig.java diff --git a/application/src/main/data/upgrade/3.4.2/schema_update.sql b/application/src/main/data/upgrade/3.4.2/schema_update.sql index 17c75773ad..77f078e67f 100644 --- a/application/src/main/data/upgrade/3.4.2/schema_update.sql +++ b/application/src/main/data/upgrade/3.4.2/schema_update.sql @@ -49,5 +49,6 @@ CREATE TABLE IF NOT EXISTS notification ( severity VARCHAR(32), 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_and_created_time ON notification(recipient_id, created_time DESC); CREATE INDEX IF NOT EXISTS idx_notification_recipient_id_and_status_and_created_time ON notification(recipient_id, status, created_time DESC); diff --git a/application/src/main/java/org/thingsboard/server/controller/BaseController.java b/application/src/main/java/org/thingsboard/server/controller/BaseController.java index ffc3fcf3cd..8f5fee218f 100644 --- a/application/src/main/java/org/thingsboard/server/controller/BaseController.java +++ b/application/src/main/java/org/thingsboard/server/controller/BaseController.java @@ -43,6 +43,7 @@ import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityView; import org.thingsboard.server.common.data.EntityViewInfo; +import org.thingsboard.server.common.data.HasName; import org.thingsboard.server.common.data.HasTenantId; import org.thingsboard.server.common.data.OtaPackage; import org.thingsboard.server.common.data.OtaPackageInfo; @@ -58,6 +59,7 @@ import org.thingsboard.server.common.data.alarm.AlarmInfo; import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.asset.AssetInfo; import org.thingsboard.server.common.data.asset.AssetProfile; +import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.EdgeEventActionType; import org.thingsboard.server.common.data.edge.EdgeEventType; @@ -75,6 +77,7 @@ import org.thingsboard.server.common.data.id.EdgeId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityIdFactory; import org.thingsboard.server.common.data.id.EntityViewId; +import org.thingsboard.server.common.data.id.HasId; import org.thingsboard.server.common.data.id.OtaPackageId; import org.thingsboard.server.common.data.id.QueueId; import org.thingsboard.server.common.data.id.RpcId; @@ -130,8 +133,8 @@ import org.thingsboard.server.exception.ThingsboardErrorResponseHandler; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; import org.thingsboard.server.queue.util.TbCoreComponent; +import org.thingsboard.server.service.action.EntityActionService; import org.thingsboard.server.service.component.ComponentDiscoveryService; -import org.thingsboard.server.service.edge.EdgeNotificationService; import org.thingsboard.server.service.edge.rpc.EdgeRpcService; import org.thingsboard.server.service.entitiy.TbNotificationEntityService; import org.thingsboard.server.service.ota.OtaPackageStateService; @@ -284,6 +287,9 @@ public abstract class BaseController { @Autowired protected TbNotificationEntityService notificationEntityService; + @Autowired + protected EntityActionService entityActionService; + @Autowired protected QueueService queueService; @@ -912,6 +918,16 @@ public abstract class BaseController { return error != null ? (Exception.class.isInstance(error) ? (Exception) error : new Exception(error)) : null; } + protected > void logEntityAction(SecurityUser user, EntityType entityType, E savedEntity, ActionType actionType) { + logEntityAction(user, entityType, null, savedEntity, actionType, null); + } + + protected > void logEntityAction(SecurityUser user, EntityType entityType, E entity, E savedEntity, ActionType actionType, Exception e) { + EntityId entityId = savedEntity != null ? savedEntity.getId() : emptyId(entityType); + entityActionService.logEntityAction(user, entityId, savedEntity != null ? savedEntity : entity, + user.getCustomerId(), actionType, e); + } + protected void sendEntityNotificationMsg(TenantId tenantId, EntityId entityId, EdgeEventActionType action) { sendNotificationMsgToEdge(tenantId, null, entityId, null, null, action); } diff --git a/application/src/main/java/org/thingsboard/server/controller/NotificationController.java b/application/src/main/java/org/thingsboard/server/controller/NotificationController.java index c3cfdc24e8..1bf771218a 100644 --- a/application/src/main/java/org/thingsboard/server/controller/NotificationController.java +++ b/application/src/main/java/org/thingsboard/server/controller/NotificationController.java @@ -28,6 +28,8 @@ 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.audit.ActionType; import org.thingsboard.server.common.data.exception.ThingsboardException; import org.thingsboard.server.common.data.id.NotificationId; import org.thingsboard.server.common.data.id.NotificationRequestId; @@ -81,9 +83,15 @@ public class NotificationController extends BaseController { @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") public NotificationRequest createNotificationRequest(@RequestBody NotificationRequest notificationRequest, @AuthenticationPrincipal SecurityUser user) throws ThingsboardException { - accessControlService.checkPermission(user, Resource.NOTIFICATION, Operation.CREATE); - // read permission for target's users - return notificationProcessingService.processNotificationRequest(user.getTenantId(), notificationRequest); + accessControlService.checkPermission(user, Resource.NOTIFICATION_REQUEST, Operation.CREATE, null, notificationRequest); + try { + NotificationRequest savedNotificationRequest = notificationProcessingService.processNotificationRequest(user.getTenantId(), notificationRequest); + logEntityAction(user, EntityType.NOTIFICATION_REQUEST, savedNotificationRequest, ActionType.ADDED); + return savedNotificationRequest; + } catch (Exception e) { + logEntityAction(user, EntityType.NOTIFICATION_REQUEST, notificationRequest, null, ActionType.ADDED, e); + throw e; + } } @GetMapping("/notification/request/{id}") @@ -102,16 +110,23 @@ public class NotificationController extends BaseController { @RequestParam(required = false) String sortProperty, @RequestParam(required = false) String sortOrder, @AuthenticationPrincipal SecurityUser user) throws ThingsboardException { - accessControlService.checkPermission(user, Resource.NOTIFICATION, Operation.CREATE); PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder); return notificationService.findNotificationRequestsByTenantIdAndPageLink(user.getTenantId(), pageLink); } @DeleteMapping("/notification/request/{id}") public void deleteNotificationRequest(@PathVariable UUID id, - @AuthenticationPrincipal SecurityUser user) { + @AuthenticationPrincipal SecurityUser user) throws ThingsboardException { NotificationRequestId notificationRequestId = new NotificationRequestId(id); - notificationProcessingService.deleteNotificationRequest(user.getTenantId(), notificationRequestId); + NotificationRequest notificationRequest = notificationService.findNotificationRequestById(user.getTenantId(), notificationRequestId); + accessControlService.checkPermission(user, Resource.NOTIFICATION_REQUEST, Operation.DELETE, notificationRequestId, notificationRequest); + try { + notificationProcessingService.deleteNotificationRequest(user.getTenantId(), notificationRequestId); + logEntityAction(user, EntityType.NOTIFICATION_REQUEST, notificationRequest, ActionType.DELETED); + } catch (Exception e) { + logEntityAction(user, EntityType.NOTIFICATION_REQUEST, notificationRequest, notificationRequest, ActionType.DELETED, e); + throw e; + } } } diff --git a/application/src/main/java/org/thingsboard/server/controller/NotificationTargetController.java b/application/src/main/java/org/thingsboard/server/controller/NotificationTargetController.java index 3eec3b6ffc..34451ddede 100644 --- a/application/src/main/java/org/thingsboard/server/controller/NotificationTargetController.java +++ b/application/src/main/java/org/thingsboard/server/controller/NotificationTargetController.java @@ -27,24 +27,26 @@ 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.audit.ActionType; import org.thingsboard.server.common.data.exception.ThingsboardException; 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.targets.NotificationTarget; +import org.thingsboard.server.common.data.notification.targets.NotificationTargetConfig; +import org.thingsboard.server.common.data.notification.targets.NotificationTargetConfigType; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; +import org.thingsboard.server.dao.DaoUtil; import org.thingsboard.server.dao.notification.NotificationTargetService; -import org.thingsboard.server.dao.user.UserService; 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.ArrayList; import java.util.List; import java.util.UUID; -import java.util.stream.Collectors; @RestController @TbCoreComponent @@ -52,37 +54,59 @@ import java.util.stream.Collectors; @RequiredArgsConstructor @Slf4j public class NotificationTargetController extends BaseController { - // fixme: permission check, log action, events private final NotificationTargetService notificationTargetService; - private final UserService userService; @PostMapping("/target") @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") public NotificationTarget saveNotificationTarget(@RequestBody NotificationTarget notificationTarget, - @AuthenticationPrincipal SecurityUser user) { - // fixme: read permission check for users in target - return notificationTargetService.saveNotificationTarget(user.getTenantId(), notificationTarget); + @AuthenticationPrincipal SecurityUser user) throws ThingsboardException { + accessControlService.checkPermission(user, Resource.NOTIFICATION_TARGET, Operation.CREATE, null, notificationTarget); + if (!user.isSystemAdmin()) { + NotificationTargetConfig targetConfig = notificationTarget.getConfiguration(); + if (targetConfig.getType() == NotificationTargetConfigType.SINGLE_USER || + targetConfig.getType() == NotificationTargetConfigType.USER_LIST) { + PageData recipients = notificationTargetService.findRecipientsForNotificationTargetConfig(user.getTenantId(), notificationTarget.getConfiguration(), null); + for (User recipient : recipients.getData()) { + accessControlService.checkPermission(user, Resource.USER, Operation.READ, recipient.getId(), recipient); + } + } + } + + try { + NotificationTarget savedNotificationTarget = notificationTargetService.saveNotificationTarget(user.getTenantId(), notificationTarget); + logEntityAction(user, EntityType.NOTIFICATION_TARGET, savedNotificationTarget, + notificationTarget.getId() == null ? ActionType.ADDED : ActionType.UPDATED); + return savedNotificationTarget; + } catch (Exception e) { + logEntityAction(user, EntityType.NOTIFICATION_TARGET, notificationTarget, null, + notificationTarget.getId() == null ? ActionType.ADDED : ActionType.UPDATED, e); + throw e; + } } @GetMapping("/target/{id}") @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") public NotificationTarget getNotificationTargetById(@PathVariable UUID id, - @AuthenticationPrincipal SecurityUser user) { + @AuthenticationPrincipal SecurityUser user) throws ThingsboardException { NotificationTargetId notificationTargetId = new NotificationTargetId(id); - return notificationTargetService.findNotificationTargetById(user.getTenantId(), notificationTargetId); + NotificationTarget notificationTarget = notificationTargetService.findNotificationTargetById(user.getTenantId(), notificationTargetId); + accessControlService.checkPermission(user, Resource.NOTIFICATION_TARGET, Operation.READ, notificationTargetId, notificationTarget); + return notificationTarget; } - @GetMapping("/target/{id}/recipients") + @PostMapping("/target/recipients") @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") - public List getRecipientsForNotificationTarget(@PathVariable UUID id, - @AuthenticationPrincipal SecurityUser user) throws ThingsboardException { - NotificationTargetId notificationTargetId = new NotificationTargetId(id); - // fixme: to page data - // todo: check read permission for recipients - List recipients = new ArrayList<>(); - for (UserId userId : notificationTargetService.findRecipientsForNotificationTarget(user.getTenantId(), notificationTargetId)) { - recipients.add(checkUserId(userId, Operation.READ)); + public PageData getRecipientsForNotificationTargetConfig(@RequestBody NotificationTarget notificationTarget, + @RequestParam int pageSize, + @RequestParam int page, + @AuthenticationPrincipal SecurityUser user) throws ThingsboardException { + PageLink pageLink = createPageLink(pageSize, page, null, null, null); + PageData recipients = notificationTargetService.findRecipientsForNotificationTargetConfig(user.getTenantId(), notificationTarget.getConfiguration(), pageLink); + if (!user.isSystemAdmin()) { + for (User recipient : recipients.getData()) { + accessControlService.checkPermission(user, Resource.USER, Operation.READ, recipient.getId(), recipient); + } } return recipients; } @@ -102,9 +126,17 @@ public class NotificationTargetController extends BaseController { @DeleteMapping("/target/{id}") @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')") public void deleteNotificationTarget(@PathVariable UUID id, - @AuthenticationPrincipal SecurityUser user) { + @AuthenticationPrincipal SecurityUser user) throws ThingsboardException { NotificationTargetId notificationTargetId = new NotificationTargetId(id); - notificationTargetService.deleteNotificationTarget(user.getTenantId(), notificationTargetId); + NotificationTarget notificationTarget = checkNotNull(notificationTargetService.findNotificationTargetById(user.getTenantId(), notificationTargetId)); + accessControlService.checkPermission(user, Resource.NOTIFICATION_TARGET, Operation.DELETE, notificationTargetId, notificationTarget); + + try { + notificationTargetService.deleteNotificationTarget(user.getTenantId(), notificationTargetId); + logEntityAction(user, EntityType.NOTIFICATION_TARGET, notificationTarget, ActionType.DELETED); + } catch (Exception e) { + logEntityAction(user, EntityType.NOTIFICATION_TARGET, null, notificationTarget, ActionType.DELETED, e); + } } } diff --git a/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java b/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java index 187603afc5..7cb0a8258e 100644 --- a/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java +++ b/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java @@ -19,6 +19,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.BeanCreationNotAllowedException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Lazy; import org.springframework.security.core.Authentication; import org.springframework.stereotype.Service; import org.springframework.web.socket.CloseStatus; @@ -69,7 +70,7 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements WebSocke private static final ConcurrentMap externalSessionMap = new ConcurrentHashMap<>(); - @Autowired + @Autowired @Lazy private WebSocketService webSocketService; @Autowired diff --git a/application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationProcessingService.java b/application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationProcessingService.java index 57b295bddd..99ede1fba9 100644 --- a/application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationProcessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationProcessingService.java @@ -32,10 +32,10 @@ import org.thingsboard.server.common.data.notification.NotificationStatus; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.dao.DaoUtil; import org.thingsboard.server.dao.notification.NotificationProcessingService; import org.thingsboard.server.dao.notification.NotificationService; import org.thingsboard.server.dao.notification.NotificationTargetService; -import org.thingsboard.server.dao.user.UserService; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.discovery.NotificationsTopicService; import org.thingsboard.server.queue.discovery.PartitionService; @@ -43,9 +43,7 @@ import org.thingsboard.server.service.executors.DbCallbackExecutorService; import org.thingsboard.server.service.subscription.TbSubscriptionUtils; import org.thingsboard.server.service.telemetry.AbstractSubscriptionService; -import java.util.ArrayList; import java.util.HashSet; -import java.util.List; import java.util.Map; import java.util.Set; import java.util.UUID; @@ -56,32 +54,23 @@ public class DefaultNotificationProcessingService extends AbstractSubscriptionSe private final NotificationTargetService notificationTargetService; private final NotificationService notificationService; - private final UserService userService; private final DbCallbackExecutorService dbCallbackExecutorService; private final NotificationsTopicService notificationsTopicService; public DefaultNotificationProcessingService(TbClusterService clusterService, PartitionService partitionService, NotificationTargetService notificationTargetService, - NotificationService notificationService, UserService userService, + NotificationService notificationService, DbCallbackExecutorService dbCallbackExecutorService, NotificationsTopicService notificationsTopicService) { super(clusterService, partitionService); this.notificationTargetService = notificationTargetService; this.notificationService = notificationService; - this.userService = userService; this.dbCallbackExecutorService = dbCallbackExecutorService; this.notificationsTopicService = notificationsTopicService; } @Override public NotificationRequest processNotificationRequest(TenantId tenantId, NotificationRequest notificationRequest) { - List recipientsIds = notificationTargetService.findRecipientsForNotificationTarget(tenantId, notificationRequest.getTargetId()); - List recipients = new ArrayList<>(); - for (UserId recipientId : recipientsIds) { - User recipient = userService.findUserById(tenantId, recipientId); // todo: add caching - recipients.add(recipient); - } - notificationRequest.setTenantId(tenantId); NotificationRequest savedNotificationRequest = notificationService.createNotificationRequest(tenantId, notificationRequest); @@ -90,20 +79,31 @@ public class DefaultNotificationProcessingService extends AbstractSubscriptionSe // todo: delayed sending; check all delayed notification requests on start up, schedule send } - for (User recipient : recipients) { + DaoUtil.processBatches(pageLink -> { + return notificationTargetService.findRecipientsForNotificationTarget(tenantId, notificationRequest.getTargetId(), pageLink); + }, 100, recipients -> { dbCallbackExecutorService.submit(() -> { - Notification notification = createNotification(recipient, savedNotificationRequest); - onNotificationUpdate(recipient.getTenantId(), recipient.getId(), notification); + for (User recipient : recipients) { + try { + Notification notification = createNotification(recipient, savedNotificationRequest); + onNotificationUpdate(recipient.getTenantId(), recipient.getId(), notification, true); + } catch (Exception e) { + log.error("Failed to create notification for recipient {}", recipient.getId(), e); + } + } }); - } + }); return savedNotificationRequest; } @Override public void markNotificationAsRead(TenantId tenantId, UserId recipientId, NotificationId notificationId) { - Notification notification = notificationService.updateNotificationStatus(tenantId, notificationId, NotificationStatus.READ); - onNotificationUpdate(tenantId, recipientId, notification); + boolean updated = notificationService.updateNotificationStatus(tenantId, recipientId, notificationId, NotificationStatus.READ); + if (updated) { + Notification notification = notificationService.findNotificationById(tenantId, notificationId); + onNotificationUpdate(tenantId, recipientId, notification, false); + } } @Override @@ -134,11 +134,11 @@ public class DefaultNotificationProcessingService extends AbstractSubscriptionSe return TbNodeUtils.processTemplate(template, context); } - private void onNotificationUpdate(TenantId tenantId, UserId recipientId, Notification notification) { + private void onNotificationUpdate(TenantId tenantId, UserId recipientId, Notification notification, boolean isNew) { forwardToSubscriptionManagerServiceOrSendToCore(tenantId, recipientId, subscriptionManagerService -> { - subscriptionManagerService.onNotificationUpdate(tenantId, recipientId, notification, TbCallback.EMPTY); + subscriptionManagerService.onNotificationUpdate(tenantId, recipientId, notification, isNew, TbCallback.EMPTY); }, () -> { - return TbSubscriptionUtils.notificationUpdateToProto(tenantId, recipientId, notification); + return TbSubscriptionUtils.notificationUpdateToProto(tenantId, recipientId, notification, isNew); }); } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index 0a7abdab51..1ee7c3cd16 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -464,6 +464,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService subscriptions = subscriptionsByEntityId.get(recipientId); if (subscriptions != null) { NotificationsSubscriptionUpdate subscriptionUpdate = NotificationsSubscriptionUpdate.builder() .notification(notification) + .isNewNotification(isNew) .build(); subscriptions.stream() - .filter(subscription -> subscription.getType() == TbSubscriptionType.NOTIFICATIONS) + .filter(subscription -> subscription.getType() == TbSubscriptionType.NOTIFICATIONS + || subscription.getType() == TbSubscriptionType.NOTIFICATIONS_COUNT) .forEach(subscription -> { if (serviceId.equals(subscription.getServiceId())) { localSubscriptionService.onSubscriptionUpdate(subscription.getSessionId(), @@ -362,7 +364,8 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene subscriptionsByEntityId.entrySet().stream() .filter(subEntry -> subEntry.getKey().getEntityType() == EntityType.USER) .flatMap(subEntry -> subEntry.getValue().stream() - .filter(sub -> sub.getType() == TbSubscriptionType.NOTIFICATIONS) + .filter(sub -> sub.getType() == TbSubscriptionType.NOTIFICATIONS + || sub.getType() == TbSubscriptionType.NOTIFICATIONS_COUNT) .filter(sub -> sub.getServiceId().equals(serviceId))) .forEach(subscription -> { localSubscriptionService.onSubscriptionUpdate(subscription.getSessionId(), subscription.getSubscriptionId(), subscriptionUpdate, TbCallback.EMPTY); diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java index e9001f3be3..5a51d0acee 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java @@ -172,7 +172,8 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer @Override public void onSubscriptionUpdate(String sessionId, int subscriptionId, NotificationsSubscriptionUpdate update, TbCallback callback) { TbSubscription subscription = subscriptionsBySessionId.getOrDefault(sessionId, Collections.emptyMap()).get(subscriptionId); - if (subscription != null && subscription.getType() == TbSubscriptionType.NOTIFICATIONS) { + if (subscription != null && (subscription.getType() == TbSubscriptionType.NOTIFICATIONS + || subscription.getType() == TbSubscriptionType.NOTIFICATIONS_COUNT)) { subscriptionUpdateExecutor.submit(() -> subscription.getUpdateProcessor().accept(subscription, update)); } callback.onSuccess(); diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java b/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java index 6724674143..239f9c94f9 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java @@ -49,7 +49,7 @@ public interface SubscriptionManagerService extends ApplicationListener 0) { return new TelemetrySubscriptionUpdate(proto.getSubscriptionId(), SubscriptionErrorCode.forCode(proto.getErrorCode()), proto.getErrorMsg()); @@ -367,6 +385,7 @@ public class TbSubscriptionUtils { .setSessionId(subscription.getSessionId()) .setSubscriptionId(subscription.getSubscriptionId()) .setNotification(JacksonUtil.toString(update.getNotification())) + .setIsNewNotification(update.isNewNotification()) .build(); return TransportProtos.ToCoreNotificationMsg.newBuilder() .setToLocalSubscriptionServiceMsg(TransportProtos.LocalSubscriptionServiceMsgProto.newBuilder() @@ -375,13 +394,14 @@ public class TbSubscriptionUtils { .build(); } - public static ToCoreMsg notificationUpdateToProto(TenantId tenantId, UserId recipientId, Notification notification) { + public static ToCoreMsg notificationUpdateToProto(TenantId tenantId, UserId recipientId, Notification notification, boolean isNew) { TransportProtos.NotificationUpdateProto updateProto = TransportProtos.NotificationUpdateProto.newBuilder() .setTenantIdMSB(tenantId.getId().getMostSignificantBits()) .setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) .setRecipientIdMSB(recipientId.getId().getMostSignificantBits()) .setRecipientIdLSB(recipientId.getId().getLeastSignificantBits()) .setNotification(JacksonUtil.toString(notification)) + .setIsNew(isNew) .build(); return ToCoreMsg.newBuilder() .setToSubscriptionMgrMsg(SubscriptionMgrMsgProto.newBuilder() diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultWebSocketService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultWebSocketService.java index f6df94b46f..e03e0fc3fb 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultWebSocketService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultWebSocketService.java @@ -22,8 +22,8 @@ import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; +import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.springframework.web.socket.CloseStatus; @@ -65,13 +65,10 @@ import org.thingsboard.server.service.subscription.TbTimeseriesSubscription; import org.thingsboard.server.service.ws.SessionEvent; import org.thingsboard.server.service.ws.WebSocketMsgEndpoint; import org.thingsboard.server.service.ws.WebSocketSessionRef; +import org.thingsboard.server.service.ws.WsCmd; import org.thingsboard.server.service.ws.WsSessionMetaData; -import org.thingsboard.server.service.ws.notification.DefaultNotificationCommandsHandler; import org.thingsboard.server.service.ws.notification.NotificationCommandsHandler; import org.thingsboard.server.service.ws.notification.cmd.NotificationCmdsWrapper; -import org.thingsboard.server.service.ws.notification.cmd.MarkNotificationAsReadCmd; -import org.thingsboard.server.service.ws.notification.cmd.NotificationsSubCmd; -import org.thingsboard.server.service.ws.notification.cmd.NotificationsUnsubCmd; import org.thingsboard.server.service.ws.telemetry.WebSocketService; import org.thingsboard.server.service.ws.telemetry.cmd.TelemetryPluginCmdsWrapper; import org.thingsboard.server.service.ws.telemetry.cmd.v1.AttributesSubscriptionCmd; @@ -105,6 +102,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.function.BiConsumer; import java.util.function.Consumer; import java.util.stream.Collectors; @@ -114,6 +112,7 @@ import java.util.stream.Collectors; @Service @TbCoreComponent @Slf4j +@RequiredArgsConstructor public class DefaultWebSocketService implements WebSocketService { public static final int NUMBER_OF_PING_ATTEMPTS = 3; @@ -130,53 +129,57 @@ public class DefaultWebSocketService implements WebSocketService { private final ConcurrentMap wsSessionsMap = new ConcurrentHashMap<>(); - @Autowired - private TbLocalSubscriptionService oldSubService; - - @Autowired - private TbEntityDataSubscriptionService entityDataSubService; - - @Autowired - private NotificationCommandsHandler notificationCmdsHandler; - - @Autowired - private WebSocketMsgEndpoint msgEndpoint; - - @Autowired - private AccessValidator accessValidator; - - @Autowired - private AttributesService attributesService; - - @Autowired - private TimeseriesService tsService; - - @Autowired - private TbServiceInfoProvider serviceInfoProvider; - - @Autowired - private TbTenantProfileCache tenantProfileCache; + private final TbLocalSubscriptionService oldSubService; + private final TbEntityDataSubscriptionService entityDataSubService; + private final NotificationCommandsHandler notificationCmdsHandler; + private final WebSocketMsgEndpoint msgEndpoint; + private final AccessValidator accessValidator; + private final AttributesService attributesService; + private final TimeseriesService tsService; + private final TbServiceInfoProvider serviceInfoProvider; + private final TbTenantProfileCache tenantProfileCache; @Value("${server.ws.ping_timeout:30000}") private long pingTimeout; - private ConcurrentMap> tenantSubscriptionsMap = new ConcurrentHashMap<>(); - private ConcurrentMap> customerSubscriptionsMap = new ConcurrentHashMap<>(); - private ConcurrentMap> regularUserSubscriptionsMap = new ConcurrentHashMap<>(); - private ConcurrentMap> publicUserSubscriptionsMap = new ConcurrentHashMap<>(); + private final ConcurrentMap> tenantSubscriptionsMap = new ConcurrentHashMap<>(); + private final ConcurrentMap> customerSubscriptionsMap = new ConcurrentHashMap<>(); + private final ConcurrentMap> regularUserSubscriptionsMap = new ConcurrentHashMap<>(); + private final ConcurrentMap> publicUserSubscriptionsMap = new ConcurrentHashMap<>(); private ExecutorService executor; + private ScheduledExecutorService pingExecutor; private String serviceId; - private ScheduledExecutorService pingExecutor; + private List> telemetryCmdsHandlers; + private List> notificationCmdsHandlers; @PostConstruct - public void initExecutor() { + public void init() { serviceId = serviceInfoProvider.getServiceId(); executor = ThingsBoardExecutors.newWorkStealingPool(50, getClass()); pingExecutor = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("telemetry-web-socket-ping")); pingExecutor.scheduleWithFixedDelay(this::sendPing, pingTimeout / NUMBER_OF_PING_ATTEMPTS, pingTimeout / NUMBER_OF_PING_ATTEMPTS, TimeUnit.MILLISECONDS); + + telemetryCmdsHandlers = List.of( + WsCmdListHandler.of(TelemetryPluginCmdsWrapper::getAttrSubCmds, this::handleWsAttributesSubscriptionCmd), + WsCmdListHandler.of(TelemetryPluginCmdsWrapper::getTsSubCmds, this::handleWsTimeseriesSubscriptionCmd), + WsCmdListHandler.of(TelemetryPluginCmdsWrapper::getHistoryCmds, this::handleWsHistoryCmd), + WsCmdListHandler.of(TelemetryPluginCmdsWrapper::getEntityDataCmds, this::handleWsEntityDataCmd), + WsCmdListHandler.of(TelemetryPluginCmdsWrapper::getAlarmDataCmds, this::handleWsAlarmDataCmd), + WsCmdListHandler.of(TelemetryPluginCmdsWrapper::getEntityCountCmds, this::handleWsEntityCountCmd), + WsCmdListHandler.of(TelemetryPluginCmdsWrapper::getEntityDataUnsubscribeCmds, this::handleWsDataUnsubscribeCmd), + WsCmdListHandler.of(TelemetryPluginCmdsWrapper::getAlarmDataUnsubscribeCmds, this::handleWsDataUnsubscribeCmd), + WsCmdListHandler.of(TelemetryPluginCmdsWrapper::getAlarmDataUnsubscribeCmds, this::handleWsDataUnsubscribeCmd), + WsCmdListHandler.of(TelemetryPluginCmdsWrapper::getEntityCountUnsubscribeCmds, this::handleWsDataUnsubscribeCmd) + ); + notificationCmdsHandlers = List.of( + WsCmdHandler.of(NotificationCmdsWrapper::getUnreadSubCmd, notificationCmdsHandler::handleUnreadNotificationsSubCmd), + WsCmdHandler.of(NotificationCmdsWrapper::getUnreadCountSubCmd, notificationCmdsHandler::handleUnreadNotificationsCountSubCmd), + WsCmdHandler.of(NotificationCmdsWrapper::getMarkAsReadCmd, notificationCmdsHandler::handleMarkAsReadCmd), + WsCmdHandler.of(NotificationCmdsWrapper::getUnsubCmd, notificationCmdsHandler::handleUnsubCmd) + ); } @PreDestroy @@ -231,77 +234,30 @@ public class DefaultWebSocketService implements WebSocketService { } } + private void processTelemetryCmds(WebSocketSessionRef sessionRef, String msg) throws JsonProcessingException { TelemetryPluginCmdsWrapper cmdsWrapper = jsonMapper.readValue(msg, TelemetryPluginCmdsWrapper.class); if (cmdsWrapper == null) { return; } - if (cmdsWrapper.getAttrSubCmds() != null) { - cmdsWrapper.getAttrSubCmds().forEach(cmd -> { - if (processSubscription(sessionRef, cmd)) { - handleWsAttributesSubscriptionCmd(sessionRef, cmd); - } - }); - } - if (cmdsWrapper.getTsSubCmds() != null) { - cmdsWrapper.getTsSubCmds().forEach(cmd -> { - if (processSubscription(sessionRef, cmd)) { - handleWsTimeseriesSubscriptionCmd(sessionRef, cmd); - } - }); - } - if (cmdsWrapper.getHistoryCmds() != null) { - cmdsWrapper.getHistoryCmds().forEach(cmd -> handleWsHistoryCmd(sessionRef, cmd)); - } - if (cmdsWrapper.getEntityDataCmds() != null) { - cmdsWrapper.getEntityDataCmds().forEach(cmd -> handleWsEntityDataCmd(sessionRef, cmd)); - } - if (cmdsWrapper.getAlarmDataCmds() != null) { - cmdsWrapper.getAlarmDataCmds().forEach(cmd -> handleWsAlarmDataCmd(sessionRef, cmd)); - } - if (cmdsWrapper.getEntityCountCmds() != null) { - cmdsWrapper.getEntityCountCmds().forEach(cmd -> handleWsEntityCountCmd(sessionRef, cmd)); - } - if (cmdsWrapper.getEntityDataUnsubscribeCmds() != null) { - cmdsWrapper.getEntityDataUnsubscribeCmds().forEach(cmd -> handleWsDataUnsubscribeCmd(sessionRef, cmd)); - } - if (cmdsWrapper.getAlarmDataUnsubscribeCmds() != null) { - cmdsWrapper.getAlarmDataUnsubscribeCmds().forEach(cmd -> handleWsDataUnsubscribeCmd(sessionRef, cmd)); - } - if (cmdsWrapper.getEntityCountUnsubscribeCmds() != null) { - cmdsWrapper.getEntityCountUnsubscribeCmds().forEach(cmd -> handleWsDataUnsubscribeCmd(sessionRef, cmd)); + for (WsCmdListHandler cmdHandler : telemetryCmdsHandlers) { + List cmds = cmdHandler.extractCmds(cmdsWrapper); + if (cmds != null) { + cmdHandler.handle(sessionRef, cmds); + } } } private void processNotificationCmds(WebSocketSessionRef sessionRef, String msg) throws IOException { NotificationCmdsWrapper cmdsWrapper = jsonMapper.readValue(msg, NotificationCmdsWrapper.class); - if (cmdsWrapper.getUnreadSubCmd() != null) { - handleUnreadNotificationsSubCmd(sessionRef, cmdsWrapper.getUnreadSubCmd()); - } else if (cmdsWrapper.getUnreadUnsubCmd() != null) { - handleUnreadNotificationsUnsubCmd(sessionRef, cmdsWrapper.getUnreadUnsubCmd()); - } else if (cmdsWrapper.getMarkAsReadCmd() != null) { - handleMarkNotificationAsReadCmd(sessionRef, cmdsWrapper.getMarkAsReadCmd()); - } - } - - private void handleUnreadNotificationsSubCmd(WebSocketSessionRef sessionRef, NotificationsSubCmd cmd) { - String sessionId = sessionRef.getSessionId(); - if (validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId)) { - notificationCmdsHandler.handleUnreadNotificationsSubCmd(sessionRef, cmd); - } - } - - private void handleUnreadNotificationsUnsubCmd(WebSocketSessionRef sessionRef, NotificationsUnsubCmd cmd) { - String sessionId = sessionRef.getSessionId(); - if (validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId)) { - notificationCmdsHandler.handleUnsubCmd(sessionRef, cmd); - } - } - - private void handleMarkNotificationAsReadCmd(WebSocketSessionRef sessionRef, MarkNotificationAsReadCmd cmd) { - String sessionId = sessionRef.getSessionId(); - if (validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId)) { - notificationCmdsHandler.handleMarkAsReadCmd(sessionRef, cmd); + for (WsCmdHandler cmdHandler : notificationCmdsHandlers) { + WsCmd cmd = cmdHandler.extractCmd(cmdsWrapper); + if (cmd != null) { + String sessionId = sessionRef.getSessionId(); + if (validateSessionMetadata(sessionRef, cmd.getCmdId(), sessionId)) { + cmdHandler.handle(sessionRef, cmd); // todo: handle exceptions + } + } } } @@ -479,6 +435,10 @@ public class DefaultWebSocketService implements WebSocketService { } private void handleWsAttributesSubscriptionCmd(WebSocketSessionRef sessionRef, AttributesSubscriptionCmd cmd) { + if (!processSubscription(sessionRef, cmd)) { + return; + } + String sessionId = sessionRef.getSessionId(); log.debug("[{}] Processing: {}", sessionId, cmd); @@ -644,6 +604,10 @@ public class DefaultWebSocketService implements WebSocketService { } private void handleWsTimeseriesSubscriptionCmd(WebSocketSessionRef sessionRef, TimeseriesSubscriptionCmd cmd) { + if (!processSubscription(sessionRef, cmd)) { + return; + } + String sessionId = sessionRef.getSessionId(); log.debug("[{}] Processing: {}", sessionId, cmd); @@ -990,4 +954,37 @@ public class DefaultWebSocketService implements WebSocketService { private int getLimit(int limit) { return limit == 0 ? DEFAULT_LIMIT : limit; } + + @RequiredArgsConstructor(staticName = "of") + public static class WsCmdHandler { + private final java.util.function.Function cmdExtractor; + private final BiConsumer handler; + + public C extractCmd(W cmdsWrapper) { + return cmdExtractor.apply(cmdsWrapper); + } + + @SuppressWarnings("unchecked") + public void handle(WebSocketSessionRef sessionRef, Object cmd) { + handler.accept(sessionRef, (C) cmd); + } + } + + @RequiredArgsConstructor(staticName = "of") + public static class WsCmdListHandler { + private final java.util.function.Function> cmdExtractor; + private final BiConsumer handler; + + public List extractCmds(W cmdsWrapper) { + return cmdExtractor.apply(cmdsWrapper); + } + + @SuppressWarnings("unchecked") + public void handle(WebSocketSessionRef sessionRef, List cmds) { + cmds.forEach(cmd -> { + handler.accept(sessionRef, (C) cmd); + }); + } + } + } diff --git a/application/src/main/java/org/thingsboard/server/service/ws/WsCmd.java b/application/src/main/java/org/thingsboard/server/service/ws/WsCmd.java new file mode 100644 index 0000000000..096c5930a5 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/ws/WsCmd.java @@ -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; + +public interface WsCmd { + int getCmdId(); +} diff --git a/application/src/main/java/org/thingsboard/server/service/ws/notification/DefaultNotificationCommandsHandler.java b/application/src/main/java/org/thingsboard/server/service/ws/notification/DefaultNotificationCommandsHandler.java index ad3f9ca4b0..ee0afd3c79 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/notification/DefaultNotificationCommandsHandler.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/notification/DefaultNotificationCommandsHandler.java @@ -16,6 +16,8 @@ package org.thingsboard.server.service.ws.notification; import lombok.RequiredArgsConstructor; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.thingsboard.server.common.data.id.IdBased; import org.thingsboard.server.common.data.id.NotificationId; @@ -28,14 +30,16 @@ import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.dao.notification.NotificationProcessingService; import org.thingsboard.server.service.security.model.SecurityUser; -import org.thingsboard.server.service.subscription.NotificationsSubscription; +import org.thingsboard.server.service.ws.notification.cmd.NotificationsCountSubCmd; +import org.thingsboard.server.service.ws.notification.sub.NotificationsSubscription; import org.thingsboard.server.service.subscription.TbLocalSubscriptionService; import org.thingsboard.server.service.ws.WebSocketSessionRef; import org.thingsboard.server.service.ws.notification.cmd.MarkNotificationAsReadCmd; import org.thingsboard.server.service.ws.notification.cmd.NotificationsSubCmd; -import org.thingsboard.server.service.ws.notification.cmd.UnreadNotificationsUpdate; import org.thingsboard.server.service.ws.notification.sub.NotificationsSubscriptionUpdate; +import org.thingsboard.server.service.ws.notification.sub.NotificationsCountSubscription; import org.thingsboard.server.service.ws.telemetry.WebSocketService; +import org.thingsboard.server.service.ws.telemetry.cmd.v2.CmdUpdate; import org.thingsboard.server.service.ws.telemetry.cmd.v2.UnsubscribeCmd; import java.util.Set; @@ -48,10 +52,11 @@ import java.util.stream.Collectors; public class DefaultNotificationCommandsHandler implements NotificationCommandsHandler { private final NotificationService notificationService; - private final WebSocketService wsService; private final TbLocalSubscriptionService localSubscriptionService; private final NotificationProcessingService notificationProcessingService; private final TbServiceInfoProvider serviceInfoProvider; + @Autowired @Lazy + private WebSocketService wsService; @Override public void handleUnreadNotificationsSubCmd(WebSocketSessionRef sessionRef, NotificationsSubCmd cmd) { @@ -62,7 +67,7 @@ public class DefaultNotificationCommandsHandler implements NotificationCommandsH .subscriptionId(cmd.getCmdId()) .tenantId(user.getTenantId()) .entityId(user.getId()) - .updateProcessor(this::handleSubscriptionUpdate) + .updateProcessor(this::handleNotificationsSubscriptionUpdate) .limit(cmd.getLimit()) .build(); localSubscriptionService.addSubscription(subscription); @@ -72,14 +77,20 @@ public class DefaultNotificationCommandsHandler implements NotificationCommandsH } @Override - public void handleMarkAsReadCmd(WebSocketSessionRef sessionRef, MarkNotificationAsReadCmd cmd) { - NotificationId notificationId = new NotificationId(cmd.getNotificationId()); - notificationProcessingService.markNotificationAsRead(sessionRef.getSecurityCtx().getTenantId(), sessionRef.getSecurityCtx().getId(), notificationId); - } + public void handleUnreadNotificationsCountSubCmd(WebSocketSessionRef sessionRef, NotificationsCountSubCmd cmd) { + SecurityUser user = sessionRef.getSecurityCtx(); + NotificationsCountSubscription subscription = NotificationsCountSubscription.builder() + .serviceId(serviceInfoProvider.getServiceId()) + .sessionId(sessionRef.getSessionId()) + .subscriptionId(cmd.getCmdId()) + .tenantId(user.getTenantId()) + .entityId(user.getId()) + .updateProcessor(this::handleNotificationsCountSubscriptionUpdate) + .build(); + localSubscriptionService.addSubscription(subscription); - @Override - public void handleUnsubCmd(WebSocketSessionRef sessionRef, UnsubscribeCmd cmd) { - localSubscriptionService.cancelSubscription(sessionRef.getSessionId(), cmd.getCmdId()); + fetchUnreadNotificationsCount(subscription); + sendUpdate(sessionRef.getSessionId(), subscription.createUpdate()); } private void fetchUnreadNotifications(NotificationsSubscription subscription) { @@ -87,19 +98,24 @@ public class DefaultNotificationCommandsHandler implements NotificationCommandsH (UserId) subscription.getEntityId(), subscription.getLimit()); subscription.getUnreadNotifications().clear(); subscription.getUnreadNotifications().putAll(notifications.getData().stream().collect(Collectors.toMap(IdBased::getUuidId, n -> n))); - subscription.getTotalUnreadCount().set((int) notifications.getTotalElements()); + subscription.getTotalUnreadCounter().set((int) notifications.getTotalElements()); + } + + private void fetchUnreadNotificationsCount(NotificationsCountSubscription subscription) { + int unreadCount = notificationService.countUnreadNotificationsByUserId(subscription.getTenantId(), (UserId) subscription.getEntityId()); + subscription.getUnreadCounter().set(unreadCount); } - private void handleSubscriptionUpdate(NotificationsSubscription subscription, NotificationsSubscriptionUpdate subscriptionUpdate) { + private void handleNotificationsSubscriptionUpdate(NotificationsSubscription subscription, NotificationsSubscriptionUpdate subscriptionUpdate) { if (subscriptionUpdate.getNotification() != null) { Notification notification = subscriptionUpdate.getNotification(); if (notification.getStatus() == NotificationStatus.READ) { fetchUnreadNotifications(subscription); sendUpdate(subscription.getSessionId(), subscription.createFullUpdate()); } else { - Notification previous = subscription.getUnreadNotifications().put(notification.getUuidId(), notification); - if (previous == null) { - subscription.getTotalUnreadCount().incrementAndGet(); + subscription.getUnreadNotifications().put(notification.getUuidId(), notification); + if (subscriptionUpdate.isNewNotification()) { + subscription.getTotalUnreadCounter().incrementAndGet(); Set beyondLimit = subscription.getUnreadNotifications().keySet().stream() .skip(subscription.getLimit()) .collect(Collectors.toSet()); @@ -116,8 +132,35 @@ public class DefaultNotificationCommandsHandler implements NotificationCommandsH } } - private void sendUpdate(String sessionId, UnreadNotificationsUpdate update) { + private void handleNotificationsCountSubscriptionUpdate(NotificationsCountSubscription subscription, NotificationsSubscriptionUpdate subscriptionUpdate) { + if (subscriptionUpdate.getNotification() != null) { + Notification notification = subscriptionUpdate.getNotification(); + if (subscriptionUpdate.isNewNotification()) { + subscription.getUnreadCounter().incrementAndGet(); + } else if (notification.getStatus() == NotificationStatus.READ) { + // for now this can only happen when user marks notification as read + subscription.getUnreadCounter().decrementAndGet(); + } + } else if (subscriptionUpdate.isNotificationRequestDeleted()) { + fetchUnreadNotificationsCount(subscription); + } + sendUpdate(subscription.getSessionId(), subscription.createUpdate()); + } + + private void sendUpdate(String sessionId, CmdUpdate update) { wsService.sendWsMsg(sessionId, update); } + + @Override + public void handleMarkAsReadCmd(WebSocketSessionRef sessionRef, MarkNotificationAsReadCmd cmd) { + NotificationId notificationId = new NotificationId(cmd.getNotificationId()); + notificationProcessingService.markNotificationAsRead(sessionRef.getSecurityCtx().getTenantId(), sessionRef.getSecurityCtx().getId(), notificationId); + } + + @Override + public void handleUnsubCmd(WebSocketSessionRef sessionRef, UnsubscribeCmd cmd) { + localSubscriptionService.cancelSubscription(sessionRef.getSessionId(), cmd.getCmdId()); + } + } diff --git a/application/src/main/java/org/thingsboard/server/service/ws/notification/NotificationCommandsHandler.java b/application/src/main/java/org/thingsboard/server/service/ws/notification/NotificationCommandsHandler.java index b45e512045..4cf1fb3809 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/notification/NotificationCommandsHandler.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/notification/NotificationCommandsHandler.java @@ -18,12 +18,15 @@ package org.thingsboard.server.service.ws.notification; import org.thingsboard.server.service.ws.WebSocketSessionRef; import org.thingsboard.server.service.ws.notification.cmd.MarkNotificationAsReadCmd; 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, MarkNotificationAsReadCmd cmd); void handleUnsubCmd(WebSocketSessionRef sessionRef, UnsubscribeCmd cmd); diff --git a/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/MarkNotificationAsReadCmd.java b/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/MarkNotificationAsReadCmd.java index 50844623eb..cfc6e8e38b 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/MarkNotificationAsReadCmd.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/MarkNotificationAsReadCmd.java @@ -18,13 +18,14 @@ package org.thingsboard.server.service.ws.notification.cmd; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; +import org.thingsboard.server.service.ws.WsCmd; import java.util.UUID; @Data @NoArgsConstructor @AllArgsConstructor -public class MarkNotificationAsReadCmd { +public class MarkNotificationAsReadCmd implements WsCmd { private int cmdId; private UUID notificationId; } diff --git a/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/NotificationCmdsWrapper.java b/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/NotificationCmdsWrapper.java index 367a6cc450..011e8cecca 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/NotificationCmdsWrapper.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/NotificationCmdsWrapper.java @@ -16,13 +16,15 @@ package org.thingsboard.server.service.ws.notification.cmd; import lombok.Data; -import org.thingsboard.server.service.ws.notification.cmd.MarkNotificationAsReadCmd; -import org.thingsboard.server.service.ws.notification.cmd.NotificationsSubCmd; @Data public class NotificationCmdsWrapper { - private NotificationsSubCmd unreadSubCmd; - private NotificationsUnsubCmd unreadUnsubCmd; + private NotificationsCountSubCmd unreadCountSubCmd; + + private NotificationsSubCmd unreadSubCmd; private MarkNotificationAsReadCmd markAsReadCmd; + + private NotificationsUnsubCmd unsubCmd; + } diff --git a/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/NotificationsCountSubCmd.java b/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/NotificationsCountSubCmd.java new file mode 100644 index 0000000000..2722d6ae7d --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/NotificationsCountSubCmd.java @@ -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.ws.notification.cmd; + +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.NoArgsConstructor; +import org.thingsboard.server.service.ws.WsCmd; + +@Data +@NoArgsConstructor +@AllArgsConstructor +public class NotificationsCountSubCmd implements WsCmd { + private int cmdId; +} diff --git a/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/NotificationsSubCmd.java b/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/NotificationsSubCmd.java index 376b34b18e..087b250314 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/NotificationsSubCmd.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/NotificationsSubCmd.java @@ -17,12 +17,14 @@ package org.thingsboard.server.service.ws.notification.cmd; import lombok.AllArgsConstructor; import lombok.Data; +import lombok.EqualsAndHashCode; import lombok.NoArgsConstructor; +import org.thingsboard.server.service.ws.WsCmd; @Data @NoArgsConstructor @AllArgsConstructor -public class NotificationsSubCmd { +public class NotificationsSubCmd implements WsCmd { private int cmdId; private int limit; } diff --git a/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/NotificationsUnsubCmd.java b/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/NotificationsUnsubCmd.java index 1daeb570cc..b81e76e81d 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/NotificationsUnsubCmd.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/NotificationsUnsubCmd.java @@ -15,10 +15,15 @@ */ package org.thingsboard.server.service.ws.notification.cmd; +import lombok.AllArgsConstructor; import lombok.Data; +import lombok.NoArgsConstructor; +import org.thingsboard.server.service.ws.WsCmd; import org.thingsboard.server.service.ws.telemetry.cmd.v2.UnsubscribeCmd; @Data -public class NotificationsUnsubCmd implements UnsubscribeCmd { +@NoArgsConstructor +@AllArgsConstructor +public class NotificationsUnsubCmd implements UnsubscribeCmd, WsCmd { private int cmdId; } diff --git a/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/UnreadNotificationsCountUpdate.java b/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/UnreadNotificationsCountUpdate.java new file mode 100644 index 0000000000..c839b36e8d --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/ws/notification/cmd/UnreadNotificationsCountUpdate.java @@ -0,0 +1,44 @@ +/** + * 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 org.thingsboard.server.service.ws.telemetry.cmd.v2.CmdUpdate; +import org.thingsboard.server.service.ws.telemetry.cmd.v2.CmdUpdateType; + +@Getter +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; + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsCountSubscription.java b/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsCountSubscription.java new file mode 100644 index 0000000000..ab06d90001 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsCountSubscription.java @@ -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 { + + private final AtomicInteger unreadCounter = new AtomicInteger(); + + @Builder + public NotificationsCountSubscription(String serviceId, String sessionId, int subscriptionId, TenantId tenantId, EntityId entityId, + BiConsumer updateProcessor) { + super(serviceId, sessionId, subscriptionId, tenantId, entityId, TbSubscriptionType.NOTIFICATIONS_COUNT, updateProcessor); + } + + public UnreadNotificationsCountUpdate createUpdate() { + return UnreadNotificationsCountUpdate.builder() + .cmdId(getSubscriptionId()) + .totalUnreadCount(unreadCounter.get()) + .build(); + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/NotificationsSubscription.java b/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsSubscription.java similarity index 85% rename from application/src/main/java/org/thingsboard/server/service/subscription/NotificationsSubscription.java rename to application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsSubscription.java index 00d23770e1..25d6e8725e 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/NotificationsSubscription.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsSubscription.java @@ -13,15 +13,16 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.service.subscription; +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.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 org.thingsboard.server.service.ws.notification.sub.NotificationsSubscriptionUpdate; import java.util.LinkedHashMap; import java.util.Map; @@ -34,7 +35,7 @@ public class NotificationsSubscription extends TbSubscription unreadNotifications = new LinkedHashMap<>(); private final int limit; - private final AtomicInteger totalUnreadCount = new AtomicInteger(); + private final AtomicInteger totalUnreadCounter = new AtomicInteger(); @Builder public NotificationsSubscription(String serviceId, String sessionId, int subscriptionId, TenantId tenantId, EntityId entityId, @@ -48,7 +49,7 @@ public class NotificationsSubscription extends TbSubscription currentNotifications = new LinkedHashMap<>(); @Getter - private int totalUnreadCount; + private UnreadNotificationsUpdate lastDataUpdate; @Getter - private UnreadNotificationsUpdate lastUpdate; + private UnreadNotificationsCountUpdate lastCountUpdate; public NotificationsWebSocketClient(String wsUrl, String token) throws URISyntaxException { super(new URI(wsUrl + "/api/ws/plugins/notifications?token=" + token)); @@ -47,38 +47,36 @@ public class NotificationsWebSocketClient extends TbTestWebSocketClient { public void subscribeForUnreadNotifications(int limit) { NotificationCmdsWrapper cmdsWrapper = new NotificationCmdsWrapper(); - cmdsWrapper.setUnreadSubCmd(new NotificationsSubCmd(newCmdId(), limit)); + cmdsWrapper.setUnreadSubCmd(new NotificationsSubCmd(1, limit)); sendCmd(cmdsWrapper); } - public void markNotificationAsRead(UUID notificationId) { + public void subscribeForUnreadNotificationsCount() { NotificationCmdsWrapper cmdsWrapper = new NotificationCmdsWrapper(); - cmdsWrapper.setMarkAsReadCmd(new MarkNotificationAsReadCmd(newCmdId(), notificationId)); + cmdsWrapper.setUnreadCountSubCmd(new NotificationsCountSubCmd(2)); sendCmd(cmdsWrapper); } - - private void handleUpdate(UnreadNotificationsUpdate update) { - totalUnreadCount = update.getTotalUnreadCount(); - if (update.getNotifications() != null) { - currentNotifications.clear(); - currentNotifications.putAll(update.getNotifications().stream().collect(Collectors.toMap(IdBased::getUuidId, n -> n))); - } else if (update.getUpdate() != null) { - Notification notification = update.getUpdate(); - currentNotifications.put(notification.getUuidId(), notification); - } + public void markNotificationAsRead(UUID notificationId) { + NotificationCmdsWrapper cmdsWrapper = new NotificationCmdsWrapper(); + cmdsWrapper.setMarkAsReadCmd(new MarkNotificationAsReadCmd(newCmdId(), notificationId)); + sendCmd(cmdsWrapper); } - public void sendCmd(NotificationCmdsWrapper cmdsWrapper) { - send(JacksonUtil.toString(cmdsWrapper)); + String cmd = JacksonUtil.toString(cmdsWrapper); + send(cmd); } @Override public void onMessage(String s) { - UnreadNotificationsUpdate update = JacksonUtil.fromString(s, UnreadNotificationsUpdate.class); - lastUpdate = update; - handleUpdate(update); + JsonNode update = JacksonUtil.toJsonNode(s); + CmdUpdateType updateType = CmdUpdateType.valueOf(update.get("cmdUpdateType").asText()); + if (updateType == CmdUpdateType.NOTIFICATIONS) { + lastDataUpdate = JacksonUtil.treeToValue(update, UnreadNotificationsUpdate.class); + } else if (updateType == CmdUpdateType.NOTIFICATIONS_COUNT) { + lastCountUpdate = JacksonUtil.treeToValue(update, UnreadNotificationsCountUpdate.class); + } super.onMessage(s); } diff --git a/application/src/test/java/org/thingsboard/server/service/notification/NotificationsWsApiTest.java b/application/src/test/java/org/thingsboard/server/service/notification/NotificationsWsApiTest.java index 8ff88b7e3c..6738b37a70 100644 --- a/application/src/test/java/org/thingsboard/server/service/notification/NotificationsWsApiTest.java +++ b/application/src/test/java/org/thingsboard/server/service/notification/NotificationsWsApiTest.java @@ -26,6 +26,7 @@ import org.thingsboard.server.common.data.notification.targets.SingleUserNotific import org.thingsboard.server.controller.AbstractControllerTest; import org.thingsboard.server.controller.TbTestWebSocketClient; import org.thingsboard.server.dao.service.DaoSqlTest; +import org.thingsboard.server.service.ws.notification.cmd.UnreadNotificationsCountUpdate; import org.thingsboard.server.service.ws.notification.cmd.UnreadNotificationsUpdate; import java.net.URISyntaxException; @@ -41,6 +42,41 @@ public class NotificationsWsApiTest extends AbstractControllerTest { loginTenantAdmin(); } + @Test + public void testSubscribingToUnreadNotificationsCount() { + NotificationTarget notificationTarget = createNotificationTarget(tenantAdminUserId); + String notificationText1 = "Notification 1"; + submitNotificationRequest(notificationTarget.getId(), "Just a test", notificationText1); + String notificationText2 = "Notification 2"; + submitNotificationRequest(notificationTarget.getId(), "Just a test", notificationText2); + + getWsClient().subscribeForUnreadNotificationsCount(); + getWsClient().waitForReply(); + + UnreadNotificationsCountUpdate update = getWsClient().getLastCountUpdate(); + assertThat(update.getTotalUnreadCount()).isEqualTo(2); + } + + @Test + public void testReceivingCountUpdates_multipleSessions() { + getWsClient().subscribeForUnreadNotificationsCount(); + getAnotherWsClient().subscribeForUnreadNotificationsCount(); + getWsClient().waitForReply(); + getAnotherWsClient().waitForReply(); + assertThat(getWsClient().getLastCountUpdate().getTotalUnreadCount()).isZero(); + + getWsClient().registerWaitForUpdate(); + getAnotherWsClient().registerWaitForUpdate(); + NotificationTarget notificationTarget = createNotificationTarget(tenantAdminUserId); + String notificationText = "Notification"; + submitNotificationRequest(notificationTarget.getId(), "Just a test", notificationText); + getWsClient().waitForUpdate(); + getAnotherWsClient().waitForUpdate(); + + assertThat(getWsClient().getLastCountUpdate().getTotalUnreadCount()).isOne(); + assertThat(getAnotherWsClient().getLastCountUpdate().getTotalUnreadCount()).isOne(); + } + @Test public void testSubscribingToUnreadNotifications_multipleSessions() throws Exception { NotificationTarget notificationTarget = createNotificationTarget(tenantAdminUserId); @@ -54,8 +90,8 @@ public class NotificationsWsApiTest extends AbstractControllerTest { getWsClient().waitForReply(); getAnotherWsClient().waitForReply(); - checkFullNotificationsUpdate(getWsClient().getLastUpdate(), notificationText1, notificationText2); - checkFullNotificationsUpdate(getAnotherWsClient().getLastUpdate(), notificationText1, notificationText2); + checkFullNotificationsUpdate(getWsClient().getLastDataUpdate(), notificationText1, notificationText2); + checkFullNotificationsUpdate(getAnotherWsClient().getLastDataUpdate(), notificationText1, notificationText2); } @Test @@ -64,7 +100,7 @@ public class NotificationsWsApiTest extends AbstractControllerTest { getAnotherWsClient().subscribeForUnreadNotifications(10); getWsClient().waitForReply(); getAnotherWsClient().waitForReply(); - UnreadNotificationsUpdate notificationsUpdate = getWsClient().getLastUpdate(); + UnreadNotificationsUpdate notificationsUpdate = getWsClient().getLastDataUpdate(); assertThat(notificationsUpdate.getTotalUnreadCount()).isZero(); getWsClient().registerWaitForUpdate(); @@ -75,8 +111,8 @@ public class NotificationsWsApiTest extends AbstractControllerTest { getWsClient().waitForUpdate(); getAnotherWsClient().waitForUpdate(); - checkPartialNotificationsUpdate(getWsClient().getLastUpdate(), notificationText, 1); - checkPartialNotificationsUpdate(getAnotherWsClient().getLastUpdate(), notificationText, 1); + checkPartialNotificationsUpdate(getWsClient().getLastDataUpdate(), notificationText, 1); + checkPartialNotificationsUpdate(getAnotherWsClient().getLastDataUpdate(), notificationText, 1); } @Test @@ -85,28 +121,37 @@ public class NotificationsWsApiTest extends AbstractControllerTest { getAnotherWsClient().subscribeForUnreadNotifications(10); getWsClient().waitForReply(); getAnotherWsClient().waitForReply(); + getAnotherWsClient().subscribeForUnreadNotificationsCount(); + getAnotherWsClient().waitForReply(); NotificationTarget notificationTarget = createNotificationTarget(tenantAdminUserId); getWsClient().registerWaitForUpdate(); + getAnotherWsClient().registerWaitForUpdate(2); String notificationText1 = "Notification 1"; submitNotificationRequest(notificationTarget.getId(), "Just a test", notificationText1); getWsClient().waitForUpdate(); - Notification notification1 = getWsClient().getLastUpdate().getUpdate(); + getAnotherWsClient().waitForUpdate(); + Notification notification1 = getWsClient().getLastDataUpdate().getUpdate(); getWsClient().registerWaitForUpdate(); + getAnotherWsClient().registerWaitForUpdate(2); String notificationText2 = "Notification 2"; submitNotificationRequest(notificationTarget.getId(), "Just a test", notificationText2); getWsClient().waitForUpdate(); - assertThat(getWsClient().getLastUpdate().getTotalUnreadCount()).isEqualTo(2); + getAnotherWsClient().waitForUpdate(); + assertThat(getWsClient().getLastDataUpdate().getTotalUnreadCount()).isEqualTo(2); + assertThat(getAnotherWsClient().getLastDataUpdate().getTotalUnreadCount()).isEqualTo(2); + assertThat(getAnotherWsClient().getLastCountUpdate().getTotalUnreadCount()).isEqualTo(2); getWsClient().registerWaitForUpdate(); - getAnotherWsClient().registerWaitForUpdate(); + getAnotherWsClient().registerWaitForUpdate(2); getWsClient().markNotificationAsRead(notification1.getUuidId()); getWsClient().waitForUpdate(); getAnotherWsClient().waitForUpdate(); - checkFullNotificationsUpdate(getWsClient().getLastUpdate(), notificationText2); - checkFullNotificationsUpdate(getAnotherWsClient().getLastUpdate(), notificationText2); + checkFullNotificationsUpdate(getWsClient().getLastDataUpdate(), notificationText2); + checkFullNotificationsUpdate(getAnotherWsClient().getLastDataUpdate(), notificationText2); + assertThat(getAnotherWsClient().getLastCountUpdate().getTotalUnreadCount()).isOne(); } public void testReceivingUpdatesWhenSubscriptionAtAnotherInstance() {} @@ -127,7 +172,7 @@ public class NotificationsWsApiTest extends AbstractControllerTest { notificationTarget.setTenantId(tenantId); notificationTarget.setName("User " + userId); SingleUserNotificationTargetConfig config = new SingleUserNotificationTargetConfig(); - config.setUserId(userId); + config.setUserId(userId.getId()); notificationTarget.setConfiguration(config); return doPost("/api/notification/target", notificationTarget, NotificationTarget.class); } diff --git a/common/cluster-api/src/main/proto/queue.proto b/common/cluster-api/src/main/proto/queue.proto index 7fb764579d..c1a6dd342e 100644 --- a/common/cluster-api/src/main/proto/queue.proto +++ b/common/cluster-api/src/main/proto/queue.proto @@ -552,6 +552,10 @@ message NotificationsSubscriptionProto { int32 limit = 2; } +message NotificationsCountSubscriptionProto { + TbSubscriptionProto sub = 1; +} + message TbSubscriptionUpdateProto { string sessionId = 1; int32 subscriptionId = 2; @@ -573,6 +577,7 @@ message NotificationsSubscriptionUpdateProto { string sessionId = 1; int32 subscriptionId = 2; string notification = 3; + bool isNewNotification = 4; } message NotificationUpdateProto { @@ -581,6 +586,7 @@ message NotificationUpdateProto { int64 recipientIdMSB = 3; int64 recipientIdLSB = 4; string notification = 5; + bool isNew = 6; } message NotificationRequestDeleteProto { @@ -692,8 +698,9 @@ message SubscriptionMgrMsgProto { TbAlarmDeleteProto alarmDelete = 9; TbTimeSeriesDeleteProto tsDelete = 10; NotificationsSubscriptionProto notificationsSub = 11; - NotificationUpdateProto notificationUpdate = 12; - NotificationRequestDeleteProto notificationRequestDelete = 13; + NotificationsCountSubscriptionProto notificationsCountSub = 12; + NotificationUpdateProto notificationUpdate = 13; + NotificationRequestDeleteProto notificationRequestDelete = 14; } message LocalSubscriptionServiceMsgProto { diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationService.java index c8e1a7ea08..a4bebfd135 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationService.java @@ -38,10 +38,14 @@ public interface NotificationService { Notification createNotification(TenantId tenantId, Notification notification); - Notification updateNotificationStatus(TenantId tenantId, NotificationId notificationId, NotificationStatus status); + Notification findNotificationById(TenantId tenantId, NotificationId notificationId); + + boolean updateNotificationStatus(TenantId tenantId, UserId userId, NotificationId notificationId, NotificationStatus status); PageData findNotificationsByUserIdAndReadStatusAndPageLink(TenantId tenantId, UserId userId, boolean unreadOnly, PageLink pageLink); PageData findLatestUnreadNotificationsByUserId(TenantId tenantId, UserId userId, int limit); + int countUnreadNotificationsByUserId(TenantId tenantId, UserId userId); + } diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationTargetService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationTargetService.java index f6cdf780e5..10c7155d97 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationTargetService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/notification/NotificationTargetService.java @@ -15,15 +15,14 @@ */ package org.thingsboard.server.dao.notification; +import org.thingsboard.server.common.data.User; 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.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.PageLink; -import java.util.List; - public interface NotificationTargetService { NotificationTarget saveNotificationTarget(TenantId tenantId, NotificationTarget notificationTarget); @@ -32,7 +31,9 @@ public interface NotificationTargetService { PageData findNotificationTargetsByTenantIdAndPageLink(TenantId tenantId, PageLink pageLink); - List findRecipientsForNotificationTarget(TenantId tenantId, NotificationTargetId notificationTargetId); + PageData findRecipientsForNotificationTarget(TenantId tenantId, NotificationTargetId notificationTargetId, PageLink pageLink); + + PageData findRecipientsForNotificationTargetConfig(TenantId tenantId, NotificationTargetConfig targetConfig, PageLink pageLink); void deleteNotificationTarget(TenantId tenantId, NotificationTargetId notificationTargetId); diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/user/UserService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/user/UserService.java index 93da1ecd6c..41864b327d 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/user/UserService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/user/UserService.java @@ -57,6 +57,8 @@ public interface UserService { PageData findTenantAdmins(TenantId tenantId, PageLink pageLink); + PageData findUsers(TenantId tenantId, PageLink pageLink); + void deleteTenantAdmins(TenantId tenantId); PageData findCustomerUsers(TenantId tenantId, CustomerId customerId, PageLink pageLink); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequest.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequest.java index 42cff008d0..dd800ef3fc 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequest.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequest.java @@ -21,11 +21,11 @@ import lombok.Data; import lombok.EqualsAndHashCode; import lombok.NoArgsConstructor; import org.thingsboard.server.common.data.BaseData; +import org.thingsboard.server.common.data.HasName; import org.thingsboard.server.common.data.HasTenantId; import org.thingsboard.server.common.data.id.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.validation.NoXss; import javax.validation.Valid; @@ -37,7 +37,7 @@ import javax.validation.constraints.NotNull; @NoArgsConstructor @AllArgsConstructor @Builder -public class NotificationRequest extends BaseData implements HasTenantId { +public class NotificationRequest extends BaseData implements HasTenantId, HasName { private TenantId tenantId; @NotNull(message = "Target is not specified") @@ -55,4 +55,9 @@ public class NotificationRequest extends BaseData impleme public static final String GENERAL_NOTIFICATION_REASON = "General"; public static final String ALARM_NOTIFICATION_REASON = "Alarm"; + @Override + public String getName() { + return notificationReason; + } + } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/AllUsersNotificationTargetConfig.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/AllUsersNotificationTargetConfig.java new file mode 100644 index 0000000000..8e380cafa5 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/AllUsersNotificationTargetConfig.java @@ -0,0 +1,28 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.notification.targets; + +import lombok.Data; + +@Data +public class AllUsersNotificationTargetConfig implements NotificationTargetConfig { + + @Override + public NotificationTargetConfigType getType() { + return NotificationTargetConfigType.ALL_USERS; + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/CustomerUsersNotificationTargetConfig.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/CustomerUsersNotificationTargetConfig.java new file mode 100644 index 0000000000..90e913c17e --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/CustomerUsersNotificationTargetConfig.java @@ -0,0 +1,32 @@ +/** + * Copyright © 2016-2022 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.notification.targets; + +import lombok.Data; + +import java.util.UUID; + +@Data +public class CustomerUsersNotificationTargetConfig implements NotificationTargetConfig { + + private UUID customerId; + + @Override + public NotificationTargetConfigType getType() { + return NotificationTargetConfigType.CUSTOMER_USERS; + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/NotificationTargetConfig.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/NotificationTargetConfig.java index 0623bf07b9..e3fb02bf4a 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/NotificationTargetConfig.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/NotificationTargetConfig.java @@ -24,7 +24,9 @@ import com.fasterxml.jackson.annotation.JsonTypeInfo; @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type") @JsonSubTypes({ @Type(value = SingleUserNotificationTargetConfig.class, name = "SINGLE_USER"), - @Type(value = UserListNotificationTargetConfig.class, name = "USER_LIST") + @Type(value = UserListNotificationTargetConfig.class, name = "USER_LIST"), + @Type(value = CustomerUsersNotificationTargetConfig.class, name = "CUSTOMER_USERS"), + @Type(value = AllUsersNotificationTargetConfig.class, name = "ALL_USERS") }) public interface NotificationTargetConfig { diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/NotificationTargetConfigType.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/NotificationTargetConfigType.java index 2413eb9326..b7694c9de8 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/NotificationTargetConfigType.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/NotificationTargetConfigType.java @@ -19,6 +19,9 @@ public enum NotificationTargetConfigType { SINGLE_USER, USER_LIST, + CUSTOMER_USERS, + ALL_USERS + // USER_GROUP, // USERS_WITH_ROLE, // QUERY // ? diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/SingleUserNotificationTargetConfig.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/SingleUserNotificationTargetConfig.java index 9fbb6ca600..12dc7c59ce 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/SingleUserNotificationTargetConfig.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/SingleUserNotificationTargetConfig.java @@ -16,15 +16,15 @@ package org.thingsboard.server.common.data.notification.targets; import lombok.Data; -import org.thingsboard.server.common.data.id.UserId; import javax.validation.constraints.NotNull; +import java.util.UUID; @Data public class SingleUserNotificationTargetConfig implements NotificationTargetConfig { @NotNull - private UserId userId; + private UUID userId; @Override public NotificationTargetConfigType getType() { diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/UserListNotificationTargetConfig.java b/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/UserListNotificationTargetConfig.java index 22bc26703d..ea05948ff6 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/UserListNotificationTargetConfig.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/notification/targets/UserListNotificationTargetConfig.java @@ -18,12 +18,15 @@ package org.thingsboard.server.common.data.notification.targets; import lombok.Data; import org.thingsboard.server.common.data.id.UserId; +import javax.validation.constraints.NotEmpty; import java.util.List; +import java.util.UUID; @Data public class UserListNotificationTargetConfig implements NotificationTargetConfig { - private List usersIds; + @NotEmpty + private List usersIds; @Override public NotificationTargetConfigType getType() { diff --git a/dao/src/main/java/org/thingsboard/server/dao/DaoUtil.java b/dao/src/main/java/org/thingsboard/server/dao/DaoUtil.java index 4be3f8e6ac..4e82583ba2 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/DaoUtil.java +++ b/dao/src/main/java/org/thingsboard/server/dao/DaoUtil.java @@ -110,13 +110,17 @@ public abstract class DaoUtil { } public static void processInBatches(Function> finder, int batchSize, Consumer processor) { + processBatches(finder, batchSize, batch -> batch.forEach(processor)); + } + + public static void processBatches(Function> finder, int batchSize, Consumer> processor) { PageLink pageLink = new PageLink(batchSize); PageData batch; boolean hasNextBatch; do { batch = finder.apply(pageLink); - batch.getData().forEach(processor); + processor.accept(batch.getData()); hasNextBatch = batch.hasNext(); pageLink = pageLink.nextPageLink(); diff --git a/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationService.java b/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationService.java index e6e69fed22..c15480c747 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationService.java @@ -81,13 +81,17 @@ public class DefaultNotificationService implements NotificationService { return notificationDao.save(tenantId, notification); } - @Transactional @Override - public Notification updateNotificationStatus(TenantId tenantId, NotificationId notificationId, NotificationStatus status) { - notificationDao.updateStatus(tenantId, notificationId, status); + public Notification findNotificationById(TenantId tenantId, NotificationId notificationId) { return notificationDao.findById(tenantId, notificationId.getId()); } + @Transactional + @Override + public boolean updateNotificationStatus(TenantId tenantId, UserId userId, NotificationId notificationId, NotificationStatus status) { + return notificationDao.updateStatusByIdAndUserId(tenantId, userId, notificationId, status); + } + @Override public PageData findNotificationsByUserIdAndReadStatusAndPageLink(TenantId tenantId, UserId userId, boolean unreadOnly, PageLink pageLink) { if (unreadOnly) { @@ -104,6 +108,11 @@ public class DefaultNotificationService implements NotificationService { return findNotificationsByUserIdAndReadStatusAndPageLink(tenantId, userId, true, pageLink); } + @Override + public int countUnreadNotificationsByUserId(TenantId tenantId, UserId userId) { + return notificationDao.countUnreadByUserId(tenantId, userId); + } + private static class NotificationRequestValidator extends DataValidator { @Override diff --git a/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationTargetService.java b/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationTargetService.java index 52f9f8b61b..00892576e4 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationTargetService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/notification/DefaultNotificationTargetService.java @@ -18,9 +18,12 @@ package org.thingsboard.server.dao.notification; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; +import org.thingsboard.server.common.data.User; +import org.thingsboard.server.common.data.id.CustomerId; 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.targets.CustomerUsersNotificationTargetConfig; import org.thingsboard.server.common.data.notification.targets.NotificationTarget; import org.thingsboard.server.common.data.notification.targets.NotificationTargetConfig; import org.thingsboard.server.common.data.notification.targets.SingleUserNotificationTargetConfig; @@ -28,9 +31,10 @@ import org.thingsboard.server.common.data.notification.targets.UserListNotificat import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; import org.thingsboard.server.dao.service.DataValidator; +import org.thingsboard.server.dao.user.UserService; -import java.util.ArrayList; import java.util.List; +import java.util.stream.Collectors; @Service @Slf4j @@ -38,11 +42,11 @@ import java.util.List; public class DefaultNotificationTargetService implements NotificationTargetService { private final NotificationTargetDao notificationTargetDao; + private final UserService userService; private final NotificationTargetValidator validator = new NotificationTargetValidator(); @Override public NotificationTarget saveNotificationTarget(TenantId tenantId, NotificationTarget notificationTarget) { - notificationTarget.setTenantId(tenantId); validator.validate(notificationTarget, NotificationTarget::getTenantId); return notificationTargetDao.save(tenantId, notificationTarget); } @@ -58,21 +62,42 @@ public class DefaultNotificationTargetService implements NotificationTargetServi } @Override - public List findRecipientsForNotificationTarget(TenantId tenantId, NotificationTargetId notificationTargetId) { + public PageData findRecipientsForNotificationTarget(TenantId tenantId, NotificationTargetId notificationTargetId, PageLink pageLink) { NotificationTarget notificationTarget = findNotificationTargetById(tenantId, notificationTargetId); NotificationTargetConfig configuration = notificationTarget.getConfiguration(); - List recipients = new ArrayList<>(); - switch (configuration.getType()) { - case SINGLE_USER: - SingleUserNotificationTargetConfig singleUserNotificationTargetConfig = (SingleUserNotificationTargetConfig) configuration; - recipients.add(singleUserNotificationTargetConfig.getUserId()); - break; - case USER_LIST: - UserListNotificationTargetConfig userListNotificationTargetConfig = (UserListNotificationTargetConfig) configuration; - recipients.addAll(userListNotificationTargetConfig.getUsersIds()); - break; + return findRecipientsForNotificationTargetConfig(tenantId, configuration, pageLink); + } + + @Override + public PageData findRecipientsForNotificationTargetConfig(TenantId tenantId, NotificationTargetConfig targetConfig, PageLink pageLink) { + switch (targetConfig.getType()) { + case SINGLE_USER: { + UserId userId = new UserId(((SingleUserNotificationTargetConfig) targetConfig).getUserId()); + User user = userService.findUserById(tenantId, userId); + return new PageData<>(List.of(user), 1, 1, false); + } + case USER_LIST: { + List users = ((UserListNotificationTargetConfig) targetConfig).getUsersIds().stream() + .map(UserId::new).map(userId -> userService.findUserById(tenantId, userId)) + .collect(Collectors.toList()); + return new PageData<>(users, 1, users.size(), false); + } + case CUSTOMER_USERS: { + if (tenantId.equals(TenantId.SYS_TENANT_ID)) { + throw new IllegalArgumentException("Customer users target is not supported for system administrator"); + } + CustomerId customerId = new CustomerId(((CustomerUsersNotificationTargetConfig) targetConfig).getCustomerId()); + return userService.findCustomerUsers(tenantId, customerId, pageLink); + } + case ALL_USERS: { + if (!tenantId.equals(TenantId.SYS_TENANT_ID)) { + return userService.findUsersByTenantId(tenantId, pageLink); + } else { + return userService.findUsers(TenantId.SYS_TENANT_ID, pageLink); + } + } } - return recipients; + return new PageData<>(); } @Override diff --git a/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationDao.java b/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationDao.java index b088d75c35..c16df8819a 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/notification/NotificationDao.java @@ -30,6 +30,8 @@ public interface NotificationDao extends Dao { PageData findByUserIdAndPageLink(TenantId tenantId, UserId userId, PageLink pageLink); - void updateStatus(TenantId tenantId, NotificationId notificationId, NotificationStatus status); + boolean updateStatusByIdAndUserId(TenantId tenantId, UserId userId, NotificationId notificationId, NotificationStatus status); + + int countUnreadByUserId(TenantId tenantId, UserId userId); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationDao.java index bb9c30a71b..2611c32c90 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationDao.java @@ -73,8 +73,13 @@ public class JpaNotificationDao extends JpaAbstractDao :status") + int updateStatusByIdAndRecipientId(@Param("id") UUID id, + @Param("recipientId") UUID recipientId, + @Param("status") NotificationStatus status); + + int countByRecipientIdAndStatusNot(UUID recipientId, NotificationStatus status); void deleteByRequestId(UUID requestId); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/user/JpaUserDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/user/JpaUserDao.java index ad8cac6a28..b566395eb8 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/user/JpaUserDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/user/JpaUserDao.java @@ -95,6 +95,11 @@ public class JpaUserDao extends JpaAbstractSearchTextDao imple } + @Override + public PageData findAll(TenantId tenantId, PageLink pageLink) { + return DaoUtil.toPageData(userRepository.findAll(DaoUtil.toPageable(pageLink))); + } + @Override public Long countByTenantId(TenantId tenantId) { return userRepository.countByTenantId(tenantId.getId()); diff --git a/dao/src/main/java/org/thingsboard/server/dao/user/UserDao.java b/dao/src/main/java/org/thingsboard/server/dao/user/UserDao.java index b906b3278d..081e30b525 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/user/UserDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/user/UserDao.java @@ -69,4 +69,7 @@ public interface UserDao extends Dao, TenantEntityDao { * @return the list of user entities */ PageData findCustomerUsers(UUID tenantId, UUID customerId, PageLink pageLink); + + PageData findAll(TenantId tenantId, PageLink pageLink); + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/user/UserServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/user/UserServiceImpl.java index fc0de8a4fa..dce9f772bf 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/user/UserServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/user/UserServiceImpl.java @@ -230,6 +230,11 @@ public class UserServiceImpl extends AbstractEntityService implements UserServic return userDao.findTenantAdmins(tenantId.getId(), pageLink); } + @Override + public PageData findUsers(TenantId tenantId, PageLink pageLink) { + return userDao.findAll(tenantId, pageLink); + } + @Override public void deleteTenantAdmins(TenantId tenantId) { log.trace("Executing deleteTenantAdmins, tenantId [{}]", tenantId); diff --git a/dao/src/main/resources/sql/schema-entities-idx.sql b/dao/src/main/resources/sql/schema-entities-idx.sql index 7bd4d69f5a..8f43b1ac4e 100644 --- a/dao/src/main/resources/sql/schema-entities-idx.sql +++ b/dao/src/main/resources/sql/schema-entities-idx.sql @@ -78,6 +78,8 @@ CREATE INDEX IF NOT EXISTS idx_notification_target_tenant_id_and_created_time ON CREATE INDEX IF NOT EXISTS idx_notification_request_tenant_id_and_created_time ON notification_request(tenant_id, created_time DESC); +CREATE INDEX IF NOT EXISTS idx_notification_id ON notification(id); + CREATE INDEX IF NOT EXISTS idx_notification_recipient_id_and_created_time ON notification(recipient_id, created_time DESC); CREATE INDEX IF NOT EXISTS idx_notification_recipient_id_and_status_and_created_time ON notification(recipient_id, status, created_time DESC);