|
|
|
@ -15,13 +15,17 @@ |
|
|
|
*/ |
|
|
|
package org.thingsboard.server.service.notification.rule; |
|
|
|
|
|
|
|
import lombok.Data; |
|
|
|
import lombok.RequiredArgsConstructor; |
|
|
|
import lombok.extern.slf4j.Slf4j; |
|
|
|
import org.springframework.beans.factory.annotation.Autowired; |
|
|
|
import org.springframework.cache.Cache; |
|
|
|
import org.springframework.cache.CacheManager; |
|
|
|
import org.springframework.context.annotation.Lazy; |
|
|
|
import org.springframework.context.event.EventListener; |
|
|
|
import org.springframework.stereotype.Service; |
|
|
|
import org.thingsboard.rule.engine.api.NotificationCenter; |
|
|
|
import org.thingsboard.server.common.data.CacheConstants; |
|
|
|
import org.thingsboard.server.common.data.EntityType; |
|
|
|
import org.thingsboard.server.common.data.id.EntityId; |
|
|
|
import org.thingsboard.server.common.data.id.NotificationRequestId; |
|
|
|
@ -49,6 +53,8 @@ import org.thingsboard.server.service.notification.rule.cache.NotificationRulesC |
|
|
|
import org.thingsboard.server.service.notification.rule.trigger.NotificationRuleTriggerProcessor; |
|
|
|
import org.thingsboard.server.service.notification.rule.trigger.RuleEngineMsgNotificationRuleTriggerProcessor; |
|
|
|
|
|
|
|
import javax.annotation.PostConstruct; |
|
|
|
import java.io.Serializable; |
|
|
|
import java.util.Collection; |
|
|
|
import java.util.EnumMap; |
|
|
|
import java.util.HashMap; |
|
|
|
@ -71,9 +77,19 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess |
|
|
|
@Autowired @Lazy |
|
|
|
private NotificationCenter notificationCenter; |
|
|
|
private final NotificationExecutorService notificationExecutor; |
|
|
|
private final CacheManager cacheManager; |
|
|
|
private Cache sentNotifications; |
|
|
|
|
|
|
|
private final Map<NotificationRuleTriggerType, NotificationRuleTriggerProcessor> triggerProcessors = new EnumMap<>(NotificationRuleTriggerType.class); |
|
|
|
|
|
|
|
@PostConstruct |
|
|
|
private void init() { |
|
|
|
sentNotifications = cacheManager.getCache(CacheConstants.SENT_NOTIFICATIONS_CACHE); |
|
|
|
if (sentNotifications == null) { |
|
|
|
throw new IllegalStateException("Sent notifications cache is not set up"); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
|
public void process(NotificationRuleTrigger trigger) { |
|
|
|
NotificationRuleTriggerType triggerType = trigger.getType(); |
|
|
|
@ -126,6 +142,9 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess |
|
|
|
log.debug("[{}] Rate limit for notification requests per rule was exceeded (rule '{}')", rule.getTenantId(), rule.getName()); |
|
|
|
return; |
|
|
|
} |
|
|
|
if (trigger.getType().isDeduplicate() && alreadySent(rule.getId(), trigger)) { |
|
|
|
return; |
|
|
|
} |
|
|
|
|
|
|
|
NotificationInfo notificationInfo = constructNotificationInfo(trigger, triggerConfig); |
|
|
|
rule.getRecipientsConfig().getTargetsTable().forEach((delay, targets) -> { |
|
|
|
@ -175,6 +194,23 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess |
|
|
|
return triggerProcessors.get(triggerConfig.getTriggerType()).constructNotificationInfo(trigger); |
|
|
|
} |
|
|
|
|
|
|
|
private boolean alreadySent(NotificationRuleId ruleId, NotificationRuleTrigger trigger) { |
|
|
|
String key = ruleId + "_" + trigger.getOriginatorEntityId(); |
|
|
|
SentNotification sent = sentNotifications.get(key, SentNotification.class); |
|
|
|
boolean alreadySent; |
|
|
|
if (sent != null && sent.getTrigger().equals(trigger)) { |
|
|
|
alreadySent = true; |
|
|
|
log.debug("Notification for {} trigger was already sent, ignoring", trigger.getType()); |
|
|
|
// updating cache anyway so that the value is not removed by ttl
|
|
|
|
} else { |
|
|
|
alreadySent = false; |
|
|
|
sent = new SentNotification(trigger); |
|
|
|
} |
|
|
|
log.trace("[{}] Putting to sentNotifications cache: {}", ruleId, trigger); |
|
|
|
sentNotifications.put(key, sent); |
|
|
|
return alreadySent; |
|
|
|
} |
|
|
|
|
|
|
|
@EventListener(ComponentLifecycleMsg.class) |
|
|
|
public void onNotificationRuleDeleted(ComponentLifecycleMsg componentLifecycleMsg) { |
|
|
|
if (componentLifecycleMsg.getEvent() != ComponentLifecycleEvent.DELETED || |
|
|
|
@ -209,4 +245,11 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess |
|
|
|
RuleEngineMsgTrigger.msgTypeToTriggerType = ruleEngineMsgTypeToTriggerType; |
|
|
|
} |
|
|
|
|
|
|
|
@Data |
|
|
|
private static class SentNotification implements Serializable { |
|
|
|
private static final long serialVersionUID = 38973480405095422L; |
|
|
|
|
|
|
|
private final NotificationRuleTrigger trigger; |
|
|
|
} |
|
|
|
|
|
|
|
} |
|
|
|
|
Andrew Shvayka
3 years ago
GitHub