|
|
@ -21,6 +21,7 @@ import lombok.RequiredArgsConstructor; |
|
|
import lombok.extern.slf4j.Slf4j; |
|
|
import lombok.extern.slf4j.Slf4j; |
|
|
import org.springframework.context.event.EventListener; |
|
|
import org.springframework.context.event.EventListener; |
|
|
import org.springframework.stereotype.Service; |
|
|
import org.springframework.stereotype.Service; |
|
|
|
|
|
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|
|
import org.thingsboard.rule.engine.api.NotificationCenter; |
|
|
import org.thingsboard.rule.engine.api.NotificationCenter; |
|
|
import org.thingsboard.server.common.data.id.EntityId; |
|
|
import org.thingsboard.server.common.data.id.EntityId; |
|
|
import org.thingsboard.server.common.data.id.NotificationRequestId; |
|
|
import org.thingsboard.server.common.data.id.NotificationRequestId; |
|
|
@ -35,12 +36,12 @@ import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; |
|
|
import org.thingsboard.server.common.msg.queue.ServiceType; |
|
|
import org.thingsboard.server.common.msg.queue.ServiceType; |
|
|
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|
|
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|
|
import org.thingsboard.server.dao.notification.NotificationRequestService; |
|
|
import org.thingsboard.server.dao.notification.NotificationRequestService; |
|
|
import org.thingsboard.server.queue.scheduler.SchedulerComponent; |
|
|
|
|
|
import org.thingsboard.server.queue.util.TbCoreComponent; |
|
|
import org.thingsboard.server.queue.util.TbCoreComponent; |
|
|
import org.thingsboard.server.service.executors.NotificationExecutorService; |
|
|
import org.thingsboard.server.service.executors.NotificationExecutorService; |
|
|
import org.thingsboard.server.service.partition.AbstractPartitionBasedService; |
|
|
import org.thingsboard.server.service.partition.AbstractPartitionBasedService; |
|
|
|
|
|
|
|
|
import javax.annotation.PostConstruct; |
|
|
import javax.annotation.PostConstruct; |
|
|
|
|
|
import javax.annotation.PreDestroy; |
|
|
import java.util.Collections; |
|
|
import java.util.Collections; |
|
|
import java.util.HashSet; |
|
|
import java.util.HashSet; |
|
|
import java.util.List; |
|
|
import java.util.List; |
|
|
@ -48,6 +49,8 @@ import java.util.Map; |
|
|
import java.util.Optional; |
|
|
import java.util.Optional; |
|
|
import java.util.Set; |
|
|
import java.util.Set; |
|
|
import java.util.concurrent.ConcurrentHashMap; |
|
|
import java.util.concurrent.ConcurrentHashMap; |
|
|
|
|
|
import java.util.concurrent.Executors; |
|
|
|
|
|
import java.util.concurrent.ScheduledExecutorService; |
|
|
import java.util.concurrent.ScheduledFuture; |
|
|
import java.util.concurrent.ScheduledFuture; |
|
|
import java.util.concurrent.TimeUnit; |
|
|
import java.util.concurrent.TimeUnit; |
|
|
|
|
|
|
|
|
@ -55,16 +58,16 @@ import java.util.concurrent.TimeUnit; |
|
|
@Service |
|
|
@Service |
|
|
@RequiredArgsConstructor |
|
|
@RequiredArgsConstructor |
|
|
@Slf4j |
|
|
@Slf4j |
|
|
@SuppressWarnings("UnstableApiUsage") |
|
|
|
|
|
public class DefaultNotificationSchedulerService extends AbstractPartitionBasedService<NotificationRequestId> implements NotificationSchedulerService { |
|
|
public class DefaultNotificationSchedulerService extends AbstractPartitionBasedService<NotificationRequestId> implements NotificationSchedulerService { |
|
|
|
|
|
|
|
|
private final NotificationCenter notificationCenter; |
|
|
private final NotificationCenter notificationCenter; |
|
|
private final NotificationRequestService notificationRequestService; |
|
|
private final NotificationRequestService notificationRequestService; |
|
|
private final SchedulerComponent scheduler; |
|
|
|
|
|
private final NotificationExecutorService notificationExecutor; |
|
|
private final NotificationExecutorService notificationExecutor; |
|
|
|
|
|
private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("notification-scheduler")); |
|
|
|
|
|
|
|
|
private final Map<NotificationRequestId, ScheduledRequestMetadata> scheduledNotificationRequests = new ConcurrentHashMap<>(); |
|
|
private final Map<NotificationRequestId, ScheduledRequestMetadata> scheduledNotificationRequests = new ConcurrentHashMap<>(); |
|
|
|
|
|
|
|
|
|
|
|
@Override |
|
|
@PostConstruct |
|
|
@PostConstruct |
|
|
public void init() { |
|
|
public void init() { |
|
|
super.init(); |
|
|
super.init(); |
|
|
@ -166,6 +169,13 @@ public class DefaultNotificationSchedulerService extends AbstractPartitionBasedS |
|
|
return "notifications-scheduler"; |
|
|
return "notifications-scheduler"; |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
@Override |
|
|
|
|
|
@PreDestroy |
|
|
|
|
|
public void stop() { |
|
|
|
|
|
super.stop(); |
|
|
|
|
|
scheduler.shutdownNow(); |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
@Data |
|
|
@Data |
|
|
private static class ScheduledRequestMetadata { |
|
|
private static class ScheduledRequestMetadata { |
|
|
private final TenantId tenantId; |
|
|
private final TenantId tenantId; |
|
|
|