|
|
|
@ -35,10 +35,12 @@ import org.thingsboard.server.common.data.id.UserId; |
|
|
|
import org.thingsboard.server.common.data.rpc.RpcError; |
|
|
|
import org.thingsboard.server.common.msg.MsgType; |
|
|
|
import org.thingsboard.server.common.msg.TbActorMsg; |
|
|
|
import org.thingsboard.server.common.msg.notification.trigger.NotificationRuleTrigger; |
|
|
|
import org.thingsboard.server.common.msg.queue.ServiceType; |
|
|
|
import org.thingsboard.server.common.msg.queue.TbCallback; |
|
|
|
import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse; |
|
|
|
import org.thingsboard.server.common.stats.StatsFactory; |
|
|
|
import org.thingsboard.server.dao.notification.NotificationRuleProcessingService; |
|
|
|
import org.thingsboard.server.dao.tenant.TbTenantProfileCache; |
|
|
|
import org.thingsboard.server.gen.transport.TransportProtos; |
|
|
|
import org.thingsboard.server.gen.transport.TransportProtos.DeviceStateServiceMsgProto; |
|
|
|
@ -129,6 +131,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore |
|
|
|
private final OtaPackageStateService firmwareStateService; |
|
|
|
private final GitVersionControlQueueService vcQueueService; |
|
|
|
private final NotificationSchedulerService notificationSchedulerService; |
|
|
|
private final NotificationRuleProcessingService notificationRuleProcessingService; |
|
|
|
private final TbCoreConsumerStats stats; |
|
|
|
protected final TbQueueConsumer<TbProtoQueueMsg<ToUsageStatsServiceMsg>> usageStatsConsumer; |
|
|
|
private final TbQueueConsumer<TbProtoQueueMsg<ToOtaPackageStateServiceMsg>> firmwareStatesConsumer; |
|
|
|
@ -156,7 +159,8 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore |
|
|
|
PartitionService partitionService, |
|
|
|
ApplicationEventPublisher eventPublisher, |
|
|
|
Optional<JwtSettingsService> jwtSettingsService, |
|
|
|
NotificationSchedulerService notificationSchedulerService) { |
|
|
|
NotificationSchedulerService notificationSchedulerService, |
|
|
|
NotificationRuleProcessingService notificationRuleProcessingService) { |
|
|
|
super(actorContext, encodingService, tenantProfileCache, deviceProfileCache, assetProfileCache, apiUsageStateService, partitionService, eventPublisher, tbCoreQueueFactory.createToCoreNotificationsMsgConsumer(), jwtSettingsService); |
|
|
|
this.mainConsumer = tbCoreQueueFactory.createToCoreMsgConsumer(); |
|
|
|
this.usageStatsConsumer = tbCoreQueueFactory.createToUsageStatsServiceMsgConsumer(); |
|
|
|
@ -171,6 +175,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore |
|
|
|
this.firmwareStateService = firmwareStateService; |
|
|
|
this.vcQueueService = vcQueueService; |
|
|
|
this.notificationSchedulerService = notificationSchedulerService; |
|
|
|
this.notificationRuleProcessingService = notificationRuleProcessingService; |
|
|
|
} |
|
|
|
|
|
|
|
@PostConstruct |
|
|
|
@ -269,6 +274,9 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore |
|
|
|
TransportProtos.NotificationSchedulerServiceMsg notificationSchedulerServiceMsg = toCoreMsg.getNotificationSchedulerServiceMsg(); |
|
|
|
log.trace("[{}] Forwarding message to notification scheduler service {}", id, toCoreMsg.getNotificationSchedulerServiceMsg()); |
|
|
|
forwardToNotificationSchedulerService(notificationSchedulerServiceMsg, callback); |
|
|
|
} else if (toCoreMsg.hasNotificationRuleProcessingServiceMsg()) { |
|
|
|
Optional<NotificationRuleTrigger> notificationRuleTrigger = encodingService.decode(toCoreMsg.getNotificationRuleProcessingServiceMsg().getTrigger().toByteArray()); |
|
|
|
notificationRuleTrigger.ifPresent(notificationRuleProcessingService::process); |
|
|
|
} |
|
|
|
} catch (Throwable e) { |
|
|
|
log.warn("[{}] Failed to process message: {}", id, msg, e); |
|
|
|
|
ViacheslavKlimov
3 years ago