@ -22,13 +22,12 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired ;
import org.springframework.stereotype.Service ;
import org.thingsboard.common.util.DonAsynchron ;
import org.thingsboard.rule.engine.api.MailService ;
import org.thingsboard.rule.engine.api.NotificationCenter ;
import org.thingsboard.rule.engine.api.SmsService ;
import org.thingsboard.server.common.data.EntityType ;
import org.thingsboard.server.common.data.User ;
import org.thingsboard.server.common.data.id.NotificationId ;
import org.thingsboard.server.common.data.id.NotificationRequestId ;
import org.thingsboard.server.common.data.id.NotificationRuleId ;
import org.thingsboard.server.common.data.id.NotificationTargetId ;
import org.thingsboard.server.common.data.id.TenantId ;
import org.thingsboard.server.common.data.id.UserId ;
@ -46,7 +45,6 @@ import org.thingsboard.server.common.data.notification.settings.NotificationSett
import org.thingsboard.server.common.data.notification.targets.NotificationRecipient ;
import org.thingsboard.server.common.data.notification.targets.NotificationTarget ;
import org.thingsboard.server.common.data.notification.targets.platform.PlatformUsersNotificationTargetConfig ;
import org.thingsboard.server.common.data.notification.targets.platform.UsersFilterType ;
import org.thingsboard.server.common.data.notification.targets.slack.SlackNotificationTargetConfig ;
import org.thingsboard.server.common.data.notification.template.DeliveryMethodNotificationTemplate ;
import org.thingsboard.server.common.data.notification.template.NotificationTemplate ;
@ -84,6 +82,7 @@ import java.util.List;
import java.util.Map ;
import java.util.Set ;
import java.util.UUID ;
import java.util.function.Consumer ;
import java.util.stream.Collectors ;
@Service
@ -103,61 +102,59 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
private final NotificationsTopicService notificationsTopicService ;
private final TbQueueProducerProvider producerProvider ;
private final RateLimitService rateLimitService ;
private final MailService mailService ;
private final SmsService smsService ;
private Map < NotificationDeliveryMethod , NotificationChannel > channels ;
@Override
public NotificationRequest processNotificationRequest ( TenantId tenantId , NotificationRequest n otificationRequest) {
public NotificationRequest processNotificationRequest ( TenantId tenantId , NotificationRequest request , Consumer < N otificationRequestStats > callback ) {
if ( ! rateLimitService . checkRateLimit ( tenantId , LimitedApi . NOTIFICATION_REQUEST ) ) {
throw new TbRateLimitsException ( EntityType . TENANT ) ;
}
NotificationSettings settings = notificationSettingsService . findNotificationSettings ( tenantId ) ;
NotificationTemplate notificationTemplate ;
if ( notificationR equest. getTemplateId ( ) ! = null ) {
notificationTemplate = notificationTemplateService . findNotificationTemplateById ( tenantId , notificationR equest. getTemplateId ( ) ) ;
if ( r equest. getTemplateId ( ) ! = null ) {
notificationTemplate = notificationTemplateService . findNotificationTemplateById ( tenantId , r equest. getTemplateId ( ) ) ;
} else {
notificationTemplate = notificationR equest. getTemplate ( ) ;
notificationTemplate = r equest. getTemplate ( ) ;
}
if ( notificationTemplate = = null ) throw new IllegalArgumentException ( "Template is missing" ) ;
List < NotificationTarget > targets = notificationR equest. getTargets ( ) . stream ( ) . map ( NotificationTargetId : : new )
List < NotificationTarget > targets = r equest. getTargets ( ) . stream ( ) . map ( NotificationTargetId : : new )
. map ( id - > notificationTargetService . findNotificationTargetById ( tenantId , id ) ) . collect ( Collectors . toList ( ) ) ;
Set < NotificationDeliveryMethod > availableDeliveryMethods = getAvailableDeliveryMethods ( tenantId ) ;
NotificationRuleId ruleId = request . getRuleId ( ) ;
notificationTemplate . getConfiguration ( ) . getDeliveryMethodsTemplates ( ) . forEach ( ( deliveryMethod , template ) - > {
if ( ! template . isEnabled ( ) ) return ;
if ( ! availableDeliveryMethods . contains ( deliveryMetho d) ) {
throw new IllegalArgumentException ( "Settings for " + deliveryMethod . getName ( ) + " are miss ing" ) ;
if ( ! channels . get ( deliveryMethod ) . check ( tenantI d) ) {
throw new IllegalArgumentException ( "Unable to send notification via " + deliveryMethod . getName ( ) + ": not configured or not work ing" ) ;
}
if ( notificationRequest . getRuleId ( ) = = null ) {
if ( ruleId = = null ) {
if ( targets . stream ( ) . noneMatch ( target - > target . getConfiguration ( ) . getType ( ) . getSupportedDeliveryMethods ( ) . contains ( deliveryMethod ) ) ) {
throw new IllegalArgumentException ( "Target for " + deliveryMethod . getName ( ) + " delivery method is missing" ) ;
}
}
} ) ;
if ( notificationR equest. getAdditionalConfig ( ) ! = null ) {
NotificationRequestConfig config = notificationR equest. getAdditionalConfig ( ) ;
if ( config . getSendingDelayInSec ( ) > 0 & & notificationR equest. getId ( ) = = null ) {
notificationR equest. setStatus ( NotificationRequestStatus . SCHEDULED ) ;
NotificationRequest savedNotificationR equest= notificationRequestService . saveNotificationRequest ( tenantId , notificationR equest) ;
forwardToNotificationSchedulerService ( tenantId , savedNotificationR equest. getId ( ) ) ;
return savedNotificationR equest;
if ( r equest. getAdditionalConfig ( ) ! = null ) {
NotificationRequestConfig config = r equest. getAdditionalConfig ( ) ;
if ( config . getSendingDelayInSec ( ) > 0 & & r equest. getId ( ) = = null ) {
r equest. setStatus ( NotificationRequestStatus . SCHEDULED ) ;
r equest= notificationRequestService . saveNotificationRequest ( tenantId , r equest) ;
forwardToNotificationSchedulerService ( tenantId , r equest. getId ( ) ) ;
return r equest;
}
}
NotificationSettings settings = notificationSettingsService . findNotificationSettings ( tenantId ) ;
log . debug ( "Processing notification request (tenantId: {}, targets: {})" , tenantId , notificationR equest. getTargets ( ) ) ;
notificationR equest. setStatus ( NotificationRequestStatus . PROCESSING ) ;
NotificationRequest savedNotificationR equest= notificationRequestService . saveNotificationRequest ( tenantId , notificationR equest) ;
log . debug ( "Processing notification request (tenantId: {}, targets: {})" , tenantId , r equest. getTargets ( ) ) ;
r equest. setStatus ( NotificationRequestStatus . PROCESSING ) ;
r equest= notificationRequestService . saveNotificationRequest ( tenantId , r equest) ;
NotificationProcessingContext ctx = NotificationProcessingContext . builder ( )
. tenantId ( tenantId )
. request ( savedNotificationRequest )
. settings ( settings )
. request ( request )
. template ( notificationTemplate )
. settings ( settings )
. build ( ) ;
notificationExecutor . submit ( ( ) - > {
@ -169,7 +166,7 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
}
Futures . whenAllComplete ( results ) . run ( ( ) - > {
NotificationRequestId requestId = savedNotificationRequest . getId ( ) ;
NotificationRequestId requestId = ctx . getRequest ( ) . getId ( ) ;
log . debug ( "[{}] Notification request processing is finished" , requestId ) ;
NotificationRequestStats stats = ctx . getStats ( ) ;
try {
@ -177,36 +174,39 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
} catch ( Exception e ) {
log . error ( "[{}] Failed to update stats for notification request" , requestId , e ) ;
}
if ( callback ! = null ) {
try {
callback . accept ( stats ) ;
} catch ( Exception e ) {
log . error ( "Failed to process callback for notification request {}" , requestId , e ) ;
}
}
} , dbCallbackExecutorService ) ;
} ) ;
return savedNotificationRequest ;
return r equest;
}
private List < ListenableFuture < Void > > processForTarget ( NotificationTarget target , NotificationProcessingContext ctx ) {
Iterable < ? extends NotificationRecipient > recipients ;
switch ( target . getConfiguration ( ) . getType ( ) ) {
case PLATFORM_USERS : {
PlatformUsersNotificationTargetConfig platformUsersTargetConfig = ( PlatformUsersNotificationTargetConfig ) target . getConfiguration ( ) ;
if ( platformUsersTargetConfig . getUsersFilter ( ) . getType ( ) = = UsersFilterType . AFFECTED_USER ) {
if ( ctx . getRequest ( ) . getInfo ( ) instanceof RuleOriginatedNotificationInfo ) {
UserId targetUserId = ( ( RuleOriginatedNotificationInfo ) ctx . getRequest ( ) . getInfo ( ) ) . getTargetUserId ( ) ;
if ( targetUserId ! = null ) {
recipients = List . of ( userService . findUserById ( ctx . getTenantId ( ) , targetUserId ) ) ;
break ;
}
}
recipients = Collections . emptyList ( ) ;
PlatformUsersNotificationTargetConfig targetConfig = ( PlatformUsersNotificationTargetConfig ) target . getConfiguration ( ) ;
if ( targetConfig . getUsersFilter ( ) . getType ( ) . isForRules ( ) ) {
recipients = new PageDataIterable < > ( pageLink - > {
return notificationTargetService . findRecipientsForRuleNotificationTargetConfig ( ctx . getTenantId ( ) , targetConfig , ( RuleOriginatedNotificationInfo ) ctx . getRequest ( ) . getInfo ( ) , pageLink ) ;
} , 500 ) ;
} else {
recipients = new PageDataIterable < > ( pageLink - > {
return notificationTargetService . findRecipientsForNotificationTargetConfig ( ctx . getTenantId ( ) , c tx . getCustomerId ( ) , platformUsersT argetConfig, pageLink ) ;
return notificationTargetService . findRecipientsForNotificationTargetConfig ( ctx . getTenantId ( ) , targetConfig , pageLink ) ;
} , 500 ) ;
}
break ;
}
case SLACK : {
SlackNotificationTargetConfig slackT argetConfig= ( SlackNotificationTargetConfig ) target . getConfiguration ( ) ;
recipients = List . of ( slackT argetConfig. getConversation ( ) ) ;
SlackNotificationTargetConfig t argetConfig= ( SlackNotificationTargetConfig ) target . getConfiguration ( ) ;
recipients = List . of ( t argetConfig. getConversation ( ) ) ;
break ;
}
default : {
@ -239,15 +239,10 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
if ( ctx . getStats ( ) . contains ( deliveryMethod , recipient . getId ( ) ) ) {
return Futures . immediateFailedFuture ( new AlreadySentException ( ) ) ;
}
Map < String , String > templateContext ;
if ( recipient instanceof User ) {
templateContext = ctx . createTemplateContext ( ( ( User ) recipient ) ) ;
} else {
templateContext = Collections . emptyMap ( ) ;
}
DeliveryMethodNotificationTemplate processedTemplate ;
try {
processedTemplate = ctx . getProcessedTemplate ( deliveryMethod , templateContex t) ;
processedTemplate = ctx . getProcessedTemplate ( deliveryMethod , recipient ) ;
} catch ( Exception e ) {
return Futures . immediateFailedFuture ( e ) ;
}
@ -309,7 +304,7 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
log . trace ( "Marked notification {} as read (recipient id: {}, tenant id: {})" , notificationId , recipientId , tenantId ) ;
NotificationUpdate update = NotificationUpdate . builder ( )
. updated ( true )
. notificationId ( notificationId )
. notificationId ( notificationId . getId ( ) )
. newStatus ( NotificationStatus . READ )
. build ( ) ;
onNotificationUpdate ( tenantId , recipientId , update ) ;
@ -345,20 +340,15 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
@Override
public Set < NotificationDeliveryMethod > getAvailableDeliveryMethods ( TenantId tenantId ) {
Set < NotificationDeliveryMethod > deliveryMethods = new HashSet < > ( ) ;
deliveryMethods . add ( NotificationDeliveryMethod . WEB ) ;
NotificationSettings notificationSettings = notificationSettingsService . findNotificationSettings ( tenantId ) ;
if ( notificationSettings . getDeliveryMethodsConfigs ( ) . containsKey ( NotificationDeliveryMethod . SLACK ) ) {
deliveryMethods . add ( NotificationDeliveryMethod . SLACK ) ;
}
try {
mailService . testConnection ( tenantId ) ;
deliveryMethods . add ( NotificationDeliveryMethod . EMAIL ) ;
} catch ( Exception e ) { }
if ( smsService . isConfigured ( tenantId ) ) {
deliveryMethods . add ( NotificationDeliveryMethod . SMS ) ;
}
return deliveryMethods ;
return channels . values ( ) . stream ( )
. filter ( channel - > channel . check ( tenantId ) )
. map ( NotificationChannel : : getDeliveryMethod )
. collect ( Collectors . toSet ( ) ) ;
}
@Override
public boolean check ( TenantId tenantId ) {
return true ;
}
@Override
@ -374,6 +364,7 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
. deleted ( true )
. build ( ) ) ;
} else if ( notificationRequest . isScheduled ( ) ) {
// TODO: just forward to scheduler service
clusterService . broadcastEntityStateChangeEvent ( tenantId , notificationRequestId , ComponentLifecycleEvent . DELETED ) ;
}
}
rusikv
4 years ago