@ -1,5 +1,5 @@
/ * *
* Copyright © 2016 - 2021 The Thingsboard Authors
* Copyright © 2016 - 2022 The Thingsboard Authors
*
* Licensed under the Apache License , Version 2 . 0 ( the "License" ) ;
* you may not use this file except in compliance with the License .
@ -97,6 +97,7 @@ import javax.annotation.Nullable;
import java.util.ArrayList ;
import java.util.Arrays ;
import java.util.Collections ;
import java.util.ConcurrentModificationException ;
import java.util.HashMap ;
import java.util.HashSet ;
import java.util.LinkedHashMap ;
@ -116,6 +117,7 @@ import java.util.stream.Collectors;
@Slf4j
class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
static final String SESSION_TIMEOUT_MESSAGE = "session timeout!" ;
final TenantId tenantId ;
final DeviceId deviceId ;
final LinkedHashMapRemoveEldest < UUID , SessionInfoMetaData > sessions ;
@ -450,6 +452,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
. setRequestId ( requestId )
. setSharedStateMsg ( true )
. addAllSharedAttributeList ( toTsKvProtos ( result ) )
. setIsMultipleAttributesRequest ( request . getSharedAttributeNamesCount ( ) > 1 )
. build ( ) ;
sendToTransport ( responseMsg , sessionInfo ) ;
}
@ -471,6 +474,8 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
. setRequestId ( requestId )
. addAllClientAttributeList ( toTsKvProtos ( result . get ( 0 ) ) )
. addAllSharedAttributeList ( toTsKvProtos ( result . get ( 1 ) ) )
. setIsMultipleAttributesRequest (
request . getSharedAttributeNamesCount ( ) + request . getClientAttributeNamesCount ( ) > 1 )
. build ( ) ;
sendToTransport ( responseMsg , sessionInfo ) ;
}
@ -939,7 +944,6 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
}
void init ( TbActorCtx ctx ) {
schedulePeriodicMsgWithDelay ( ctx , SessionTimeoutCheckMsg . instance ( ) , systemContext . getSessionReportTimeout ( ) , systemContext . getSessionReportTimeout ( ) ) ;
PageLink pageLink = new PageLink ( 1024 , 0 , null , new SortOrder ( "createdTime" ) ) ;
PageData < Rpc > pageData ;
do {
@ -961,19 +965,35 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
}
void checkSessionsTimeout ( ) {
log . debug ( "[{}] checkSessionsTimeout started. Size before check {}" , deviceId , sessions . size ( ) ) ;
long expTime = System . currentTimeMillis ( ) - systemContext . getSessionInactivityTimeout ( ) ;
Map < UUID , SessionInfoMetaData > sessionsToRemove = sessions . entrySet ( ) . stream ( ) . filter ( kv - > kv . getValue ( ) . getLastActivityTime ( ) < expTime ) . collect ( Collectors . toMap ( Map . Entry : : getKey , Map . Entry : : getValue ) ) ;
sessionsToRemove . forEach ( ( sessionId , sessionMD ) - > {
sessions . remove ( sessionId ) ;
rpcSubscriptions . remove ( sessionId ) ;
attributeSubscriptions . remove ( sessionId ) ;
notifyTransportAboutClosedSession ( sessionId , sessionMD , "session timeout!" ) ;
} ) ;
if ( ! sessionsToRemove . isEmpty ( ) ) {
dumpSessions ( ) ;
final long expTime = System . currentTimeMillis ( ) - systemContext . getSessionInactivityTimeout ( ) ;
List < UUID > expiredIds = null ;
for ( Map . Entry < UUID , SessionInfoMetaData > kv : sessions . entrySet ( ) ) { //entry set are cached for stable sessions
if ( kv . getValue ( ) . getLastActivityTime ( ) < expTime ) {
final UUID id = kv . getKey ( ) ;
if ( expiredIds = = null ) {
expiredIds = new ArrayList < > ( 1 ) ; //most of the expired sessions is a single event
}
expiredIds . add ( id ) ;
}
}
log . debug ( "[{}] checkSessionsTimeout finished. Size after check {}" , deviceId , sessions . size ( ) ) ;
if ( expiredIds ! = null ) {
int removed = 0 ;
for ( UUID id : expiredIds ) {
final SessionInfoMetaData session = sessions . remove ( id ) ;
rpcSubscriptions . remove ( id ) ;
attributeSubscriptions . remove ( id ) ;
if ( session ! = null ) {
removed + + ;
notifyTransportAboutClosedSession ( id , session , SESSION_TIMEOUT_MESSAGE ) ;
}
}
if ( removed ! = 0 ) {
dumpSessions ( ) ;
}
}
}
}
Viacheslav Klimov
4 years ago