public CompletableFuture < RaftSessionState > openSession ( String serviceName , PrimitiveType primitiveType , ServiceConfig config , ReadConsistency readConsistency , CommunicationStrategy communicationStrategy , Duration minTimeout , Duration maxTimeout ) { checkNotNull ( serviceName , "serviceName cannot be null" ) ; checkNotNull ( primitiveType , "serviceType cannot be null" ) ; checkNotNull ( communicationStrategy , "communicationStrategy cannot be null" ) ; checkNotNull ( maxTimeout , "timeout cannot be null" ) ; log . debug ( "Opening session; name: {}, type: {}" , serviceName , primitiveType ) ; OpenSessionRequest request = OpenSessionRequest . builder ( ) . withMemberId ( memberId ) . withServiceName ( serviceName ) . withServiceType ( primitiveType ) . withServiceConfig ( Serializer . using ( primitiveType . namespace ( ) ) . encode ( config ) ) . withReadConsistency ( readConsistency ) . withMinTimeout ( minTimeout . toMillis ( ) ) . withMaxTimeout ( maxTimeout . toMillis ( ) ) . build ( ) ; CompletableFuture < RaftSessionState > future = new CompletableFuture <> ( ) ; ThreadContext proxyContext = threadContextFactory . createContext ( ) ; connection . openSession ( request ) . whenCompleteAsync ( ( response , error ) -> { if ( error == null ) { if ( response . status ( ) == RaftResponse . Status . OK ) { // Create and store the proxy state. RaftSessionState state = new RaftSessionState ( clientId , SessionId . from ( response . session ( ) ) , serviceName , primitiveType , response . timeout ( ) ) ; sessions . put ( state . getSessionId ( ) . id ( ) , state ) ; state . addStateChangeListener ( s -> { if ( s == PrimitiveState . EXPIRED || s == PrimitiveState . CLOSED ) { sessions . remove ( state . getSessionId ( ) . id ( ) ) ; } } ) ; // Ensure the proxy session info is reset and the session is kept alive. keepAliveSessions ( System . currentTimeMillis ( ) , state . getSessionTimeout ( ) ) ; future . complete ( state ) ; } else { future . completeExceptionally ( new RaftException . Unavailable ( response . error ( ) . message ( ) ) ) ; } } else { future . completeExceptionally ( new RaftException . Unavailable ( error . getMessage ( ) ) ) ; } } , proxyContext ) ; return future ; }
Opens a new session .