@ -26,6 +26,7 @@ import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty ;
import org.springframework.stereotype.Service ;
import org.thingsboard.common.util.ThingsBoardThreadFactory ;
import org.thingsboard.server.cluster.TbClusterService ;
import org.thingsboard.server.common.data.DataConstants ;
import org.thingsboard.server.common.data.ResourceUtils ;
import org.thingsboard.server.common.data.edge.Edge ;
@ -34,6 +35,10 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry ;
import org.thingsboard.server.common.data.kv.BooleanDataEntry ;
import org.thingsboard.server.common.data.kv.LongDataEntry ;
import org.thingsboard.server.common.msg.edge.EdgeEventUpdateMsg ;
import org.thingsboard.server.common.msg.edge.EdgeSessionMsg ;
import org.thingsboard.server.common.msg.edge.FromEdgeSyncResponse ;
import org.thingsboard.server.common.msg.edge.ToEdgeSyncRequest ;
import org.thingsboard.server.gen.edge.v1.EdgeRpcServiceGrpc ;
import org.thingsboard.server.gen.edge.v1.RequestMsg ;
import org.thingsboard.server.gen.edge.v1.ResponseMsg ;
@ -59,6 +64,7 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit ;
import java.util.concurrent.locks.Lock ;
import java.util.concurrent.locks.ReentrantLock ;
import java.util.function.Consumer ;
@Service
@Slf4j
@ -71,6 +77,8 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
private final Map < EdgeId , Boolean > sessionNewEvents = new HashMap < > ( ) ;
private final ConcurrentMap < EdgeId , ScheduledFuture < ? > > sessionEdgeEventChecks = new ConcurrentHashMap < > ( ) ;
private final ConcurrentMap < UUID , Consumer < FromEdgeSyncResponse > > localSyncEdgeRequests = new ConcurrentHashMap < > ( ) ;
@Value ( "${edges.rpc.port}" )
private int rpcPort ;
@Value ( "${edges.rpc.ssl.enabled}" )
@ -98,12 +106,17 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
@Autowired
private TelemetrySubscriptionService tsSubService ;
@Autowired
private TbClusterService clusterService ;
private Server server ;
private ScheduledExecutorService edgeEventProcessingExecutorService ;
private ScheduledExecutorService sendDownlinkExecutorService ;
private ScheduledExecutorService executorService ;
@PostConstruct
public void init ( ) {
log . info ( "Initializing Edge RPC service!" ) ;
@ -129,8 +142,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
log . error ( "Failed to start Edge RPC server!" , e ) ;
throw new RuntimeException ( "Failed to start Edge RPC server!" ) ;
}
this . edgeEventProcessingExecutorService = Executors . newScheduledThreadPool ( schedulerPoolSize , ThingsBoardThreadFactory . forName ( "edge-scheduler" ) ) ;
this . edgeEventProcessingExecutorService = Executors . newScheduledThreadPool ( schedulerPoolSize , ThingsBoardThreadFactory . forName ( "edge-event-check- scheduler" ) ) ;
this . sendDownlinkExecutorService = Executors . newScheduledThreadPool ( sendSchedulerPoolSize , ThingsBoardThreadFactory . forName ( "edge-send-scheduler" ) ) ;
this . executorService = Executors . newSingleThreadScheduledExecutor ( ThingsBoardThreadFactory . forName ( "edge-service" ) ) ;
log . info ( "Edge RPC service initialized!" ) ;
}
@ -153,6 +167,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
if ( sendDownlinkExecutorService ! = null ) {
sendDownlinkExecutorService . shutdownNow ( ) ;
}
if ( executorService ! = null ) {
executorService . shutdownNow ( ) ;
}
}
@Override
@ -160,47 +177,76 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
return new EdgeGrpcSession ( ctx , outputStream , this : : onEdgeConnect , this : : onEdgeDisconnect , sendDownlinkExecutorService ) . getInputStream ( ) ;
}
@Override
public void onToEdgeSessionMsg ( TenantId tenantId , EdgeSessionMsg msg ) {
executorService . execute ( ( ) - > {
switch ( msg . getMsgType ( ) ) {
case EDGE_EVENT_UPDATE_TO_EDGE_SESSION_MSG :
EdgeEventUpdateMsg edgeEventUpdateMsg = ( EdgeEventUpdateMsg ) msg ;
log . trace ( "[{}] onToEdgeSessionMsg [{}]" , edgeEventUpdateMsg . getTenantId ( ) , msg ) ;
onEdgeEvent ( tenantId , edgeEventUpdateMsg . getEdgeId ( ) ) ;
break ;
case EDGE_SYNC_REQUEST_TO_EDGE_SESSION_MSG :
ToEdgeSyncRequest toEdgeSyncRequest = ( ToEdgeSyncRequest ) msg ;
log . trace ( "[{}] toEdgeSyncRequest [{}]" , toEdgeSyncRequest . getTenantId ( ) , msg ) ;
startSyncProcess ( tenantId , toEdgeSyncRequest . getEdgeId ( ) , toEdgeSyncRequest . getId ( ) ) ;
break ;
case EDGE_SYNC_RESPONSE_FROM_EDGE_SESSION_MSG :
FromEdgeSyncResponse fromEdgeSyncResponse = ( FromEdgeSyncResponse ) msg ;
log . trace ( "[{}] fromEdgeSyncResponse [{}]" , fromEdgeSyncResponse . getTenantId ( ) , msg ) ;
processSyncResponse ( fromEdgeSyncResponse ) ;
break ;
}
} ) ;
}
@Override
public void updateEdge ( TenantId tenantId , Edge edge ) {
EdgeGrpcSession session = sessions . get ( edge . getId ( ) ) ;
if ( session ! = null & & session . isConnected ( ) ) {
log . debug ( "[{}] Updating configuration for edge [{}] [{}]" , tenantId , edge . getName ( ) , edge . getId ( ) ) ;
session . onConfigurationUpdate ( edge ) ;
} else {
log . debug ( "[{}] Session doesn't exist for edge [{}] [{}]" , tenantId , edge . getName ( ) , edge . getId ( ) ) ;
}
executorService . execute ( ( ) - > {
EdgeGrpcSession session = sessions . get ( edge . getId ( ) ) ;
if ( session ! = null & & session . isConnected ( ) ) {
log . debug ( "[{}] Updating configuration for edge [{}] [{}]" , tenantId , edge . getName ( ) , edge . getId ( ) ) ;
session . onConfigurationUpdate ( edge ) ;
} else {
log . debug ( "[{}] Session doesn't exist for edge [{}] [{}]" , tenantId , edge . getName ( ) , edge . getId ( ) ) ;
}
} ) ;
}
@Override
public void deleteEdge ( TenantId tenantId , EdgeId edgeId ) {
executorService . execute ( ( ) - > {
EdgeGrpcSession session = sessions . get ( edgeId ) ;
if ( session ! = null & & session . isConnected ( ) ) {
log . info ( "[{}] Closing and removing session for edge [{}]" , tenantId , edgeId ) ;
session . close ( ) ;
sessions . remove ( edgeId ) ;
final Lock newEventLock = sessionNewEventsLocks . computeIfAbsent ( edgeId , id - > new ReentrantLock ( ) ) ;
newEventLock . lock ( ) ;
try {
sessionNewEvents . remove ( edgeId ) ;
} finally {
newEventLock . unlock ( ) ;
}
cancelScheduleEdgeEventsCheck ( edgeId ) ;
}
} ) ;
}
private void onEdgeEvent ( TenantId tenantId , EdgeId edgeId ) {
EdgeGrpcSession session = sessions . get ( edgeId ) ;
if ( session ! = null & & session . isConnected ( ) ) {
log . info ( "[{}] Closing and removing session for edge [{}]" , tenantId , edgeId ) ;
session . close ( ) ;
sessions . remove ( edgeId ) ;
log . trace ( "[{}] onEdgeEvent [{}]" , tenantId , edgeId . getId ( ) ) ;
final Lock newEventLock = sessionNewEventsLocks . computeIfAbsent ( edgeId , id - > new ReentrantLock ( ) ) ;
newEventLock . lock ( ) ;
try {
sessionNewEvents . remove ( edgeId ) ;
if ( Boolean . FALSE . equals ( sessionNewEvents . get ( edgeId ) ) ) {
log . trace ( "[{}] set session new events flag to true [{}]" , tenantId , edgeId . getId ( ) ) ;
sessionNewEvents . put ( edgeId , true ) ;
}
} finally {
newEventLock . unlock ( ) ;
}
cancelScheduleEdgeEventsCheck ( edgeId ) ;
}
}
@Override
public void onEdgeEvent ( TenantId tenantId , EdgeId edgeId ) {
log . trace ( "[{}] onEdgeEvent [{}]" , tenantId , edgeId . getId ( ) ) ;
final Lock newEventLock = sessionNewEventsLocks . computeIfAbsent ( edgeId , id - > new ReentrantLock ( ) ) ;
newEventLock . lock ( ) ;
try {
if ( Boolean . FALSE . equals ( sessionNewEvents . get ( edgeId ) ) ) {
log . trace ( "[{}] set session new events flag to true [{}]" , tenantId , edgeId . getId ( ) ) ;
sessionNewEvents . put ( edgeId , true ) ;
}
} finally {
newEventLock . unlock ( ) ;
}
}
@ -220,14 +266,47 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
scheduleEdgeEventsCheck ( edgeGrpcSession ) ;
}
@Override
public void startSyncProcess ( TenantId tenantId , EdgeId edgeId ) {
private void startSyncProcess ( TenantId tenantId , EdgeId edgeId , UUID requestId ) {
EdgeGrpcSession session = sessions . get ( edgeId ) ;
if ( session ! = null & & session . isConnected ( ) ) {
session . startSyncProcess ( tenantId , edgeId ) ;
if ( session ! = null ) {
boolean success = false ;
if ( session . isConnected ( ) ) {
session . startSyncProcess ( tenantId , edgeId ) ;
success = true ;
}
clusterService . pushEdgeSyncResponseToCore ( new FromEdgeSyncResponse ( requestId , tenantId , edgeId , success ) ) ;
}
}
@Override
public void processSyncRequest ( ToEdgeSyncRequest request , Consumer < FromEdgeSyncResponse > responseConsumer ) {
log . trace ( "[{}][{}] Processing sync edge request [{}]" , request . getTenantId ( ) , request . getId ( ) , request . getEdgeId ( ) ) ;
UUID requestId = request . getId ( ) ;
localSyncEdgeRequests . put ( requestId , responseConsumer ) ;
clusterService . pushEdgeSyncRequestToCore ( request ) ;
scheduleSyncRequestTimeout ( request , requestId ) ;
}
private void scheduleSyncRequestTimeout ( ToEdgeSyncRequest request , UUID requestId ) {
log . trace ( "[{}] scheduling sync edge request" , requestId ) ;
executorService . schedule ( ( ) - > {
log . trace ( "[{}] checking if sync edge request is not processed..." , requestId ) ;
Consumer < FromEdgeSyncResponse > consumer = localSyncEdgeRequests . remove ( requestId ) ;
if ( consumer ! = null ) {
log . trace ( "[{}] timeout for processing sync edge request." , requestId ) ;
consumer . accept ( new FromEdgeSyncResponse ( requestId , request . getTenantId ( ) , request . getEdgeId ( ) , false ) ) ;
}
} , 20 , TimeUnit . SECONDS ) ;
}
private void processSyncResponse ( FromEdgeSyncResponse response ) {
log . trace ( "[{}] Received response from sync service: [{}]" , response . getId ( ) , response ) ;
UUID requestId = response . getId ( ) ;
Consumer < FromEdgeSyncResponse > consumer = localSyncEdgeRequests . remove ( requestId ) ;
if ( consumer ! = null ) {
consumer . accept ( response ) ;
} else {
log . error ( "[{}] Edge is not connected [{}]" , tenantId , edgeId ) ;
throw new RuntimeException ( "Edge is not connected" ) ;
log . trace ( "[{}] Unknown or stale sync response received [{}]" , requestId , response ) ;
}
}
Volodymyr Babak
4 years ago