@ -76,6 +76,8 @@ import org.thingsboard.server.common.transport.TransportTenantProfileCache;
import org.thingsboard.server.common.transport.auth.GetOrCreateDeviceFromGatewayResponse ;
import org.thingsboard.server.common.transport.auth.TransportDeviceInfo ;
import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse ;
import org.thingsboard.server.common.transport.limits.EntityLimitKey ;
import org.thingsboard.server.common.transport.limits.EntityLimitsCache ;
import org.thingsboard.server.common.transport.limits.TransportRateLimitService ;
import org.thingsboard.server.common.transport.util.JsonUtils ;
import org.thingsboard.server.gen.transport.TransportProtos ;
@ -161,6 +163,7 @@ public class DefaultTransportService implements TransportService {
@Value ( "${transport.stats.enabled:false}" )
private boolean statsEnabled ;
@Autowired
@Lazy
private TbApiUsageReportClient apiUsageClient ;
@ -184,6 +187,8 @@ public class DefaultTransportService implements TransportService {
private final TransportResourceCache transportResourceCache ;
private final NotificationRuleProcessor notificationRuleProcessor ;
private final EntityLimitsCache entityLimitsCache ;
protected TbQueueRequestTemplate < TbProtoQueueMsg < TransportApiRequestMsg > , TbProtoQueueMsg < TransportApiResponseMsg > > transportApiRequestTemplate ;
protected TbQueueProducer < TbProtoQueueMsg < ToRuleEngineMsg > > ruleEngineMsgProducer ;
protected TbQueueProducer < TbProtoQueueMsg < ToCoreMsg > > tbCoreMsgProducer ;
@ -212,7 +217,8 @@ public class DefaultTransportService implements TransportService {
TransportTenantProfileCache tenantProfileCache ,
TransportRateLimitService rateLimitService ,
DataDecodingEncodingService dataDecodingEncodingService , SchedulerComponent scheduler , TransportResourceCache transportResourceCache ,
ApplicationEventPublisher eventPublisher , NotificationRuleProcessor notificationRuleProcessor ) {
ApplicationEventPublisher eventPublisher , NotificationRuleProcessor notificationRuleProcessor ,
EntityLimitsCache entityLimitsCache ) {
this . partitionService = partitionService ;
this . serviceInfoProvider = serviceInfoProvider ;
this . queueProvider = queueProvider ;
@ -227,6 +233,7 @@ public class DefaultTransportService implements TransportService {
this . transportResourceCache = transportResourceCache ;
this . eventPublisher = eventPublisher ;
this . notificationRuleProcessor = notificationRuleProcessor ;
this . entityLimitsCache = entityLimitsCache ;
}
@PostConstruct
@ -249,7 +256,7 @@ public class DefaultTransportService implements TransportService {
}
@AfterStartUp ( order = AfterStartUp . TRANSPORT_SERVICE )
private void start ( ) {
public void start ( ) {
mainConsumerExecutor . execute ( ( ) - > {
while ( ! stopped ) {
try {
@ -473,24 +480,33 @@ public class DefaultTransportService implements TransportService {
AsyncCallbackTemplate . withCallback ( response , callback : : onSuccess , callback : : onError , transportCallbackExecutor ) ;
}
@Override
public void process ( TransportProtos . GetOrCreateDeviceFromGatewayRequestMsg requestMsg , TransportServiceCallback < GetOrCreateDeviceFromGatewayResponse > callback ) {
public void process ( TenantId tenantId , T ransportProtos . GetOrCreateDeviceFromGatewayRequestMsg requestMsg , TransportServiceCallback < GetOrCreateDeviceFromGatewayResponse > callback ) {
TbProtoQueueMsg < TransportApiRequestMsg > protoMsg = new TbProtoQueueMsg < > ( UUID . randomUUID ( ) , TransportApiRequestMsg . newBuilder ( ) . setGetOrCreateDeviceRequestMsg ( requestMsg ) . build ( ) ) ;
log . trace ( "Processing msg: {}" , requestMsg ) ;
ListenableFuture < GetOrCreateDeviceFromGatewayResponse > response = Futures . transform ( transportApiRequestTemplate . send ( protoMsg ) , tmp - > {
TransportProtos . GetOrCreateDeviceFromGatewayResponseMsg msg = tmp . getValue ( ) . getGetOrCreateDeviceResponseMsg ( ) ;
GetOrCreateDeviceFromGatewayResponse . GetOrCreateDeviceFromGatewayResponseBuilder result = GetOrCreateDeviceFromGatewayResponse . builder ( ) ;
if ( msg . hasDeviceInfo ( ) ) {
TransportDeviceInfo tdi = getTransportDeviceInfo ( msg . getDeviceInfo ( ) ) ;
result . deviceInfo ( tdi ) ;
ByteString profileBody = msg . getProfileBody ( ) ;
if ( profileBody ! = null & & ! profileBody . isEmpty ( ) ) {
result . deviceProfile ( deviceProfileCache . getOrCreate ( tdi . getDeviceProfileId ( ) , profileBody ) ) ;
var key = new EntityLimitKey ( tenantId , StringUtils . truncate ( requestMsg . getDeviceName ( ) , 256 ) ) ;
if ( entityLimitsCache . get ( key ) ) {
transportCallbackExecutor . submit ( ( ) - > callback . onError ( new RuntimeException ( DataConstants . MAXIMUM_NUMBER_OF_DEVICES_REACHED ) ) ) ;
} else {
ListenableFuture < GetOrCreateDeviceFromGatewayResponse > response = Futures . transform ( transportApiRequestTemplate . send ( protoMsg ) , tmp - > {
TransportProtos . GetOrCreateDeviceFromGatewayResponseMsg msg = tmp . getValue ( ) . getGetOrCreateDeviceResponseMsg ( ) ;
GetOrCreateDeviceFromGatewayResponse . GetOrCreateDeviceFromGatewayResponseBuilder result = GetOrCreateDeviceFromGatewayResponse . builder ( ) ;
if ( msg . hasDeviceInfo ( ) ) {
TransportDeviceInfo tdi = getTransportDeviceInfo ( msg . getDeviceInfo ( ) ) ;
result . deviceInfo ( tdi ) ;
ByteString profileBody = msg . getProfileBody ( ) ;
if ( ! profileBody . isEmpty ( ) ) {
result . deviceProfile ( deviceProfileCache . getOrCreate ( tdi . getDeviceProfileId ( ) , profileBody ) ) ;
}
} else if ( TransportProtos . TransportApiRequestErrorCode . ENTITY_LIMIT . equals ( msg . getError ( ) ) ) {
entityLimitsCache . put ( key , true ) ;
throw new RuntimeException ( DataConstants . MAXIMUM_NUMBER_OF_DEVICES_REACHED ) ;
}
}
return result . build ( ) ;
} , MoreExecutors . directExecutor ( ) ) ;
AsyncCallbackTemplate . withCallback ( response , callback : : onSuccess , callback : : onError , transportCallbackExecutor ) ;
return result . build ( ) ;
} , MoreExecutors . directExecutor ( ) ) ;
AsyncCallbackTemplate . withCallback ( response , callback : : onSuccess , callback : : onError , transportCallbackExecutor ) ;
}
}
@Override