@ -24,8 +24,15 @@ import org.thingsboard.rule.engine.api.RpcError;
import org.thingsboard.server.actors.ActorSystemContext ;
import org.thingsboard.server.actors.ActorSystemContext ;
import org.thingsboard.server.common.data.id.TenantId ;
import org.thingsboard.server.common.data.id.TenantId ;
import org.thingsboard.server.common.msg.TbMsg ;
import org.thingsboard.server.common.msg.TbMsg ;
import org.thingsboard.server.common.msg.queue.* ;
import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg ;
import org.thingsboard.server.common.msg.queue.RuleEngineException ;
import org.thingsboard.server.common.msg.queue.RuleNodeInfo ;
import org.thingsboard.server.common.msg.queue.ServiceQueue ;
import org.thingsboard.server.common.msg.queue.ServiceType ;
import org.thingsboard.server.common.msg.queue.TbCallback ;
import org.thingsboard.server.common.msg.queue.TbMsgCallback ;
import org.thingsboard.server.common.stats.StatsFactory ;
import org.thingsboard.server.common.stats.StatsFactory ;
import org.thingsboard.server.common.transport.util.DataDecodingEncodingService ;
import org.thingsboard.server.gen.transport.TransportProtos ;
import org.thingsboard.server.gen.transport.TransportProtos ;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg ;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg ;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg ;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg ;
@ -36,19 +43,33 @@ import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory;
import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings ;
import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings ;
import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration ;
import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration ;
import org.thingsboard.server.queue.util.TbRuleEngineComponent ;
import org.thingsboard.server.queue.util.TbRuleEngineComponent ;
import org.thingsboard.server.common.transport.util.DataDecodingEncodingService ;
import org.thingsboard.server.service.apiusage.TbApiUsageStateService ;
import org.thingsboard.server.service.apiusage.TbApiUsageStateService ;
import org.thingsboard.server.service.profile.TbDeviceProfileCache ;
import org.thingsboard.server.service.profile.TbDeviceProfileCache ;
import org.thingsboard.server.service.profile.TbTenantProfileCache ;
import org.thingsboard.server.service.profile.TbTenantProfileCache ;
import org.thingsboard.server.service.queue.processing.* ;
import org.thingsboard.server.service.queue.processing.AbstractConsumerService ;
import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingDecision ;
import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingResult ;
import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategy ;
import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingStrategyFactory ;
import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategy ;
import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategyFactory ;
import org.thingsboard.server.service.rpc.FromDeviceRpcResponse ;
import org.thingsboard.server.service.rpc.FromDeviceRpcResponse ;
import org.thingsboard.server.service.rpc.TbRuleEngineDeviceRpcService ;
import org.thingsboard.server.service.rpc.TbRuleEngineDeviceRpcService ;
import org.thingsboard.server.service.stats.RuleEngineStatisticsService ;
import org.thingsboard.server.service.stats.RuleEngineStatisticsService ;
import javax.annotation.PostConstruct ;
import javax.annotation.PostConstruct ;
import javax.annotation.PreDestroy ;
import javax.annotation.PreDestroy ;
import java.util.* ;
import java.util.Collections ;
import java.util.concurrent.* ;
import java.util.HashSet ;
import java.util.List ;
import java.util.Map ;
import java.util.Set ;
import java.util.UUID ;
import java.util.concurrent.ConcurrentHashMap ;
import java.util.concurrent.ConcurrentMap ;
import java.util.concurrent.ExecutorService ;
import java.util.concurrent.Executors ;
import java.util.concurrent.TimeUnit ;
@Service
@Service
@TbRuleEngineComponent
@TbRuleEngineComponent
@ -79,12 +100,15 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
public DefaultTbRuleEngineConsumerService ( TbRuleEngineProcessingStrategyFactory processingStrategyFactory ,
public DefaultTbRuleEngineConsumerService ( TbRuleEngineProcessingStrategyFactory processingStrategyFactory ,
TbRuleEngineSubmitStrategyFactory submitStrategyFactory ,
TbRuleEngineSubmitStrategyFactory submitStrategyFactory ,
TbQueueRuleEngineSettings ruleEngineSettings ,
TbQueueRuleEngineSettings ruleEngineSettings ,
TbTenantProfileCache tenantProfileCache ,
TbRuleEngineQueueFactory tbRuleEngineQueueFactory ,
TbApiUsageStateService apiUsageState Service,
RuleEngineStatisticsService statistics Service,
TbRuleEngineQueueFactory tbRuleEngineQueueFactory , RuleEngineStatisticsService statisticsService ,
ActorSystemContext actorContext ,
ActorSystemContext actorContext , DataDecodingEncodingService encodingService ,
DataDecodingEncodingService encodingService ,
TbRuleEngineDeviceRpcService tbDeviceRpcService ,
TbRuleEngineDeviceRpcService tbDeviceRpcService ,
StatsFactory statsFactory , TbDeviceProfileCache deviceProfileCache ) {
StatsFactory statsFactory ,
TbDeviceProfileCache deviceProfileCache ,
TbTenantProfileCache tenantProfileCache ,
TbApiUsageStateService apiUsageStateService ) {
super ( actorContext , encodingService , tenantProfileCache , deviceProfileCache , apiUsageStateService , tbRuleEngineQueueFactory . createToRuleEngineNotificationsMsgConsumer ( ) ) ;
super ( actorContext , encodingService , tenantProfileCache , deviceProfileCache , apiUsageStateService , tbRuleEngineQueueFactory . createToRuleEngineNotificationsMsgConsumer ( ) ) ;
this . statisticsService = statisticsService ;
this . statisticsService = statisticsService ;
this . ruleEngineSettings = ruleEngineSettings ;
this . ruleEngineSettings = ruleEngineSettings ;