signature:
public class InternalContext { /** * Sets the new current dependency & adds it to the state . */
Dependency < ? > pushDependency ( Dependency < ? > dependency , Object source ) { } }

implementation: Dependency < ? > previous = this . dependency ; this . dependency = dependency ; doPushState ( dependency , source ) ; return previous ;


signature:
public class InternalContext { /** * Adds to the state without setting the dependency . */
void pushState ( com . google . inject . Key < ? > key , Object source ) { } }

implementation: doPushState ( key , source ) ;


signature:
public class InternalContext { /** * Returns the current dependency chain ( all the state stored in the dependencyStack ) . */
java . util . List < com . google . inject . spi . DependencyAndSource > getDependencyChain ( ) { } }

implementation: com . google . common . collect . ImmutableList . Builder < com . google . inject . spi . DependencyAndSource > builder = com . google . common . collect . ImmutableList . builder ( ) ; for ( int i = 0 ; i < dependencyStackSize ; i += 2 ) { Object evenEntry = dependencyStack [ i ] ; Dependency < ? > dependency ; if ( evenEntry instanceof com . google . inject . Key ) { dependency = Dependency . get ( ( com . google . inject . Key < ? > ) evenEntry ) ; } else { dependency = ( Dependency < ? > ) evenEntry ; } builder . add ( new com . google . inject . spi . DependencyAndSource ( dependency , dependencyStack [ i + 1 ] ) ) ; } return builder . build ( ) ;


signature:
public class Initializables { /** * Returns an initializable for an instance that requires no initialization . */
static < T > Initializable < T > of ( final T instance ) { } }

implementation: return new Initializable < T > ( ) { @ Override public T get ( ) { return instance ; } @ Override public String toString ( ) { return String . valueOf ( instance ) ; } } ;


signature:
public class ServletDefinition { /** * Utility that delegates to the actual service method of the servlet wrapped with a contextual
 * request ( i . e . with correctly computed path info ) .
 * < p > We need to suppress deprecation coz we use HttpServletRequestWrapper , which implements
 * deprecated API for backwards compatibility . */
void doService ( final ServletRequest servletRequest , ServletResponse servletResponse ) throws ServletException , IOException { } }

implementation:
HttpServletRequest request = new HttpServletRequestWrapper ( ( HttpServletRequest ) servletRequest ) { private boolean pathComputed ; private String path ; private boolean pathInfoComputed ; private String pathInfo ; @ Override public String getPathInfo ( ) { if ( ! isPathInfoComputed ( ) ) { String servletPath = getServletPath ( ) ; int servletPathLength = servletPath . length ( ) ; String requestUri = getRequestURI ( ) ; pathInfo = requestUri . substring ( getContextPath ( ) . length ( ) ) . replaceAll ( "[/]{2,}" , "/" ) ; // See : https : / / github . com / google / guice / issues / 372
if ( pathInfo . startsWith ( servletPath ) ) { pathInfo = pathInfo . substring ( servletPathLength ) ; // Corner case : when servlet path & request path match exactly ( without trailing ' / ' ) ,
// then pathinfo is null .
if ( pathInfo . isEmpty ( ) && servletPathLength > 0 ) { pathInfo = null ; } else { try { pathInfo = new URI ( pathInfo ) . getPath ( ) ; } catch ( URISyntaxException e ) { // ugh , just leave it alone then
} } } else { pathInfo = null ; // we know nothing additional about the URI .
} pathInfoComputed = true ; } return pathInfo ; } // NOTE ( dhanji ) : These two are a bit of a hack to help ensure that request dispatcher - sent
// requests don ' t use the same path info that was memoized for the original request .
// NOTE ( iqshum ) : I don ' t think this is possible , since the dispatcher - sent request would
// perform its own wrapping .
private boolean isPathInfoComputed ( ) { return pathInfoComputed && servletRequest . getAttribute ( REQUEST_DISPATCHER_REQUEST ) == null ; } private boolean isPathComputed ( ) { return pathComputed && servletRequest . getAttribute ( REQUEST_DISPATCHER_REQUEST ) == null ; } @ Override public String getServletPath ( ) { return computePath ( ) ; } @ Override public String getPathTranslated ( ) { final String info = getPathInfo ( ) ; return ( null == info ) ? null : getRealPath ( info ) ; } // Memoizer pattern .
private String computePath ( ) { if ( ! isPathComputed ( ) ) { String servletPath = super . getServletPath ( ) ; path = patternMatcher . extractPath ( servletPath ) ; pathComputed = true ; if ( null == path ) { path = servletPath ; } } return path ; } } ; doServiceImpl ( request , ( HttpServletResponse ) servletResponse ) ;


signature:
public class ProviderInternalFactory { /** * Provisions a new instance . Subclasses should override this to catch exceptions & rethrow as
 * ErrorsExceptions . */
protected T provision ( Provider < ? extends T > provider , Dependency < ? > dependency , ConstructionContext < T > constructionContext ) throws InternalProvisionException { } }

implementation: T t = provider . get ( ) ; if ( t == null && ! dependency . isNullable ( ) ) { InternalProvisionException . onNullInjectedIntoNonNullableDependency ( source , dependency ) ; } constructionContext . setProxyDelegates ( t ) ; return t ;


signature:
public class ConnectionEventListener { /** * Dispatch events .
 * @ param type
 * @ param remoteAddr
 * @ param conn */
public void onEvent ( ConnectionEventType type , String remoteAddr , Connection conn ) { } }

implementation: List < ConnectionEventProcessor > processorList = this . processors . get ( type ) ; if ( processorList != null ) { for ( ConnectionEventProcessor processor : processorList ) { processor . onEvent ( remoteAddr , conn ) ; } }


signature:
public class ConnectionEventListener { /** * Add event processor .
 * @ param type
 * @ param processor */
public void addConnectionEventProcessor ( ConnectionEventType type , ConnectionEventProcessor processor ) { } }

implementation: List < ConnectionEventProcessor > processorList = this . processors . get ( type ) ; if ( processorList == null ) { this . processors . putIfAbsent ( type , new ArrayList < ConnectionEventProcessor > ( 1 ) ) ; processorList = this . processors . get ( type ) ; } processorList . add ( processor ) ;


signature:
public class FutureTaskUtil { /** * get the result of a future task
 * Notice : the run method of this task should have been called at first .
 * @ param task
 * @ param < T >
 * @ return */
public static < T > T getFutureTaskResult ( RunStateRecordedFutureTask < T > task , Logger logger ) { } }

implementation: T t = null ; if ( null != task ) { try { t = task . getAfterRun ( ) ; } catch ( InterruptedException e ) { logger . error ( "Future task interrupted!" , e ) ; } catch ( ExecutionException e ) { logger . error ( "Future task execute failed!" , e ) ; } catch ( FutureTaskNotRunYetException e ) { logger . error ( "Future task has not run yet!" , e ) ; } catch ( FutureTaskNotCompleted e ) { logger . error ( "Future task has not completed!" , e ) ; } } return t ;


signature:
public class FutureTaskUtil { /** * launder the throwable
 * @ param t */
public static void launderThrowable ( Throwable t ) { } }

implementation: if ( t instanceof RuntimeException ) { throw ( RuntimeException ) t ; } else if ( t instanceof Error ) { throw ( Error ) t ; } else { throw new IllegalStateException ( "Not unchecked!" , t ) ; }


signature:
public class ProcessorManager { /** * Register processor to process command that has the command code of cmdCode .
 * @ param cmdCode
 * @ param processor */
public void registerProcessor ( CommandCode cmdCode , RemotingProcessor < ? > processor ) { } }

implementation: if ( this . cmd2processors . containsKey ( cmdCode ) ) { logger . warn ( "Processor for cmd={} is already registered, the processor is {}, and changed to {}" , cmdCode , cmd2processors . get ( cmdCode ) . getClass ( ) . getName ( ) , processor . getClass ( ) . getName ( ) ) ; } this . cmd2processors . put ( cmdCode , processor ) ;


signature:
public class ProcessorManager { /** * Register the default processor to process command with no specific processor registered .
 * @ param processor */
public void registerDefaultProcessor ( RemotingProcessor < ? > processor ) { } }

implementation: if ( this . defaultProcessor == null ) { this . defaultProcessor = processor ; } else { throw new IllegalStateException ( "The defaultProcessor has already been registered: " + this . defaultProcessor . getClass ( ) ) ; }


signature:
public class ProcessorManager { /** * Get the specific processor with command code of cmdCode if registered , otherwise the default processor is returned .
 * @ param cmdCode
 * @ return */
public RemotingProcessor < ? > getProcessor ( CommandCode cmdCode ) { } }

implementation: RemotingProcessor < ? > processor = this . cmd2processors . get ( cmdCode ) ; if ( processor != null ) { return processor ; } return this . defaultProcessor ;


signature:
public class RpcAddressParser { /** * try get from cache
 * @ param url
 * @ return */
private Url tryGet ( String url ) { } }

implementation: SoftReference < Url > softRef = Url . parsedUrls . get ( url ) ; return ( null == softRef ) ? null : softRef . get ( ) ;


signature:
public class ProtocolCodeBasedDecoder { /** * decode the protocol code
 * @ param in input byte buf
 * @ return an instance of ProtocolCode */
protected ProtocolCode decodeProtocolCode ( ByteBuf in ) { } }

implementation: if ( in . readableBytes ( ) >= protocolCodeLength ) { byte [ ] protocolCodeBytes = new byte [ protocolCodeLength ] ; in . readBytes ( protocolCodeBytes ) ; return ProtocolCode . fromBytes ( protocolCodeBytes ) ; } return null ;


signature:
public class DefaultConnectionManager { /** * Get all connections of all poolKey .
 * @ return a map with poolKey as key and a list of connections in ConnectionPool as value */
@ Override public Map < String , List < Connection > > getAll ( ) { } }

implementation: Map < String , List < Connection > > allConnections = new HashMap < String , List < Connection > > ( ) ; Iterator < Map . Entry < String , RunStateRecordedFutureTask < ConnectionPool > > > iterator = this . getConnPools ( ) . entrySet ( ) . iterator ( ) ; while ( iterator . hasNext ( ) ) { Map . Entry < String , RunStateRecordedFutureTask < ConnectionPool > > entry = iterator . next ( ) ; ConnectionPool pool = FutureTaskUtil . getFutureTaskResult ( entry . getValue ( ) , logger ) ; if ( null != pool ) { allConnections . put ( entry . getKey ( ) , pool . getAll ( ) ) ; } } return allConnections ;


signature:
public class DefaultConnectionManager { /** * Warning ! This is weakly consistent implementation , to prevent lock the whole { @ link ConcurrentHashMap } .
 * @ see ConnectionManager # removeAll ( ) */
@ Override public void removeAll ( ) { } }

implementation: if ( null == this . connTasks || this . connTasks . isEmpty ( ) ) { return ; } if ( null != this . connTasks && ! this . connTasks . isEmpty ( ) ) { Iterator < String > iter = this . connTasks . keySet ( ) . iterator ( ) ; while ( iter . hasNext ( ) ) { String poolKey = iter . next ( ) ; this . removeTask ( poolKey ) ; iter . remove ( ) ; } logger . warn ( "All connection pool and connections have been removed!" ) ; }


signature:
public class DefaultConnectionManager { /** * in case of cache pollution and connection leak , to do schedule scan
 * @ see com . alipay . remoting . Scannable # scan ( ) */
@ Override public void scan ( ) { } }

implementation: if ( null != this . connTasks && ! this . connTasks . isEmpty ( ) ) { Iterator < String > iter = this . connTasks . keySet ( ) . iterator ( ) ; while ( iter . hasNext ( ) ) { String poolKey = iter . next ( ) ; ConnectionPool pool = this . getConnectionPool ( this . connTasks . get ( poolKey ) ) ; if ( null != pool ) { pool . scan ( ) ; if ( pool . isEmpty ( ) ) { if ( ( System . currentTimeMillis ( ) - pool . getLastAccessTimestamp ( ) ) > DEFAULT_EXPIRE_TIME ) { iter . remove ( ) ; logger . warn ( "Remove expired pool task of poolKey {} which is empty." , poolKey ) ; } } } } }


signature:
public class DefaultConnectionManager { /** * If no task cached , create one and initialize the connections .
 * @ see ConnectionManager # getAndCreateIfAbsent ( Url ) */
@ Override public Connection getAndCreateIfAbsent ( Url url ) throws InterruptedException , RemotingException { } }

implementation:
// get and create a connection pool with initialized connections .
ConnectionPool pool = this . getConnectionPoolAndCreateIfAbsent ( url . getUniqueKey ( ) , new ConnectionPoolCall ( url ) ) ; if ( null != pool ) { return pool . get ( ) ; } else { logger . error ( "[NOTIFYME] bug detected! pool here must not be null!" ) ; return null ; }


signature:
public class DefaultConnectionManager { /** * If no task cached , create one and initialize the connections .
 * If task cached , check whether the number of connections adequate , if not then heal it .
 * @ param url
 * @ return
 * @ throws InterruptedException
 * @ throws RemotingException */
@ Override public void createConnectionAndHealIfNeed ( Url url ) throws InterruptedException , RemotingException { } }

implementation:
// get and create a connection pool with initialized connections .
ConnectionPool pool = this . getConnectionPoolAndCreateIfAbsent ( url . getUniqueKey ( ) , new ConnectionPoolCall ( url ) ) ; if ( null != pool ) { healIfNeed ( pool , url ) ; } else { logger . error ( "[NOTIFYME] bug detected! pool here must not be null!" ) ; }


signature:
public class DefaultConnectionManager { /** * Get the mapping instance of { @ link ConnectionPool } with the specified poolKey ,
 * or create one if there is none mapping in connTasks .
 * @ param poolKey mapping key of { @ link ConnectionPool }
 * @ param callable the callable task
 * @ return a non - nullable instance of { @ link ConnectionPool }
 * @ throws RemotingException if there is no way to get an available { @ link ConnectionPool }
 * @ throws InterruptedException */
private ConnectionPool getConnectionPoolAndCreateIfAbsent ( String poolKey , Callable < ConnectionPool > callable ) throws RemotingException , InterruptedException { } }

implementation:
RunStateRecordedFutureTask < ConnectionPool > initialTask = null ; ConnectionPool pool = null ; int retry = DEFAULT_RETRY_TIMES ; int timesOfResultNull = 0 ; int timesOfInterrupt = 0 ; for ( int i = 0 ; ( i < retry ) && ( pool == null ) ; ++ i ) { initialTask = this . connTasks . get ( poolKey ) ; if ( null == initialTask ) { initialTask = new RunStateRecordedFutureTask < ConnectionPool > ( callable ) ; initialTask = this . connTasks . putIfAbsent ( poolKey , initialTask ) ; if ( null == initialTask ) { initialTask = this . connTasks . get ( poolKey ) ; initialTask . run ( ) ; } } try { pool = initialTask . get ( ) ; if ( null == pool ) { if ( i + 1 < retry ) { timesOfResultNull ++ ; continue ; } this . connTasks . remove ( poolKey ) ; String errMsg = "Get future task result null for poolKey [" + poolKey + "] after [" + ( timesOfResultNull + 1 ) + "] times try." ; throw new RemotingException ( errMsg ) ; } } catch ( InterruptedException e ) { if ( i + 1 < retry ) { timesOfInterrupt ++ ; continue ; // retry if interrupted
} this . connTasks . remove ( poolKey ) ; logger . warn ( "Future task of poolKey {} interrupted {} times. InterruptedException thrown and stop retry." , poolKey , ( timesOfInterrupt + 1 ) , e ) ; throw e ; } catch ( ExecutionException e ) { // DO NOT retry if ExecutionException occurred
this . connTasks . remove ( poolKey ) ; Throwable cause = e . getCause ( ) ; if ( cause instanceof RemotingException ) { throw ( RemotingException ) cause ; } else { FutureTaskUtil . launderThrowable ( cause ) ; } } } return pool ;


signature:
public class DefaultConnectionManager { /** * remove task and remove all connections
 * @ param poolKey */
private void removeTask ( String poolKey ) { } }

implementation: RunStateRecordedFutureTask < ConnectionPool > task = this . connTasks . remove ( poolKey ) ; if ( null != task ) { ConnectionPool pool = FutureTaskUtil . getFutureTaskResult ( task , logger ) ; if ( null != pool ) { pool . removeAllAndTryClose ( ) ; } }


signature:
public class DefaultConnectionManager { /** * execute heal connection tasks if the actual number of connections in pool is less than expected
 * @ param pool
 * @ param url */
private void healIfNeed ( ConnectionPool pool , Url url ) throws RemotingException , InterruptedException { } }

implementation:
String poolKey = url . getUniqueKey ( ) ; // only when async creating connections done
// and the actual size of connections less than expected , the healing task can be run .
if ( pool . isAsyncCreationDone ( ) && pool . size ( ) < url . getConnNum ( ) ) { FutureTask < Integer > task = this . healTasks . get ( poolKey ) ; if ( null == task ) { task = new FutureTask < Integer > ( new HealConnectionCall ( url , pool ) ) ; task = this . healTasks . putIfAbsent ( poolKey , task ) ; if ( null == task ) { task = this . healTasks . get ( poolKey ) ; task . run ( ) ; } } try { int numAfterHeal = task . get ( ) ; if ( logger . isDebugEnabled ( ) ) { logger . debug ( "[NOTIFYME] - conn num after heal {}, expected {}, warmup {}" , numAfterHeal , url . getConnNum ( ) , url . isConnWarmup ( ) ) ; } } catch ( InterruptedException e ) { this . healTasks . remove ( poolKey ) ; throw e ; } catch ( ExecutionException e ) { this . healTasks . remove ( poolKey ) ; Throwable cause = e . getCause ( ) ; if ( cause instanceof RemotingException ) { throw ( RemotingException ) cause ; } else { FutureTaskUtil . launderThrowable ( cause ) ; } } // heal task is one - off , remove from cache directly after run
this . healTasks . remove ( poolKey ) ; }


signature:
public class DefaultConnectionManager { /** * do create connections
 * @ param url
 * @ param pool
 * @ param taskName
 * @ param syncCreateNumWhenNotWarmup you can specify this param to ensure at least desired number of connections available in sync way
 * @ throws RemotingException */
private void doCreate ( final Url url , final ConnectionPool pool , final String taskName , final int syncCreateNumWhenNotWarmup ) throws RemotingException { } }

implementation:
final int actualNum = pool . size ( ) ; final int expectNum = url . getConnNum ( ) ; if ( actualNum < expectNum ) { if ( logger . isDebugEnabled ( ) ) { logger . debug ( "actual num {}, expect num {}, task name {}" , actualNum , expectNum , taskName ) ; } if ( url . isConnWarmup ( ) ) { for ( int i = actualNum ; i < expectNum ; ++ i ) { Connection connection = create ( url ) ; pool . add ( connection ) ; } } else { if ( syncCreateNumWhenNotWarmup < 0 || syncCreateNumWhenNotWarmup > url . getConnNum ( ) ) { throw new IllegalArgumentException ( "sync create number when not warmup should be [0," + url . getConnNum ( ) + "]" ) ; } // create connection in sync way
if ( syncCreateNumWhenNotWarmup > 0 ) { for ( int i = 0 ; i < syncCreateNumWhenNotWarmup ; ++ i ) { Connection connection = create ( url ) ; pool . add ( connection ) ; } if ( syncCreateNumWhenNotWarmup == url . getConnNum ( ) ) { return ; } } // initialize executor in lazy way
initializeExecutor ( ) ; pool . markAsyncCreationStart ( ) ; // mark the start of async
try { this . asyncCreateConnectionExecutor . execute ( new Runnable ( ) { @ Override public void run ( ) { try { for ( int i = pool . size ( ) ; i < url . getConnNum ( ) ; ++ i ) { Connection conn = null ; try { conn = create ( url ) ; } catch ( RemotingException e ) { logger . error ( "Exception occurred in async create connection thread for {}, taskName {}" , url . getUniqueKey ( ) , taskName , e ) ; } pool . add ( conn ) ; } } finally { pool . markAsyncCreationDone ( ) ; // mark the end of async
} } } ) ; } catch ( RejectedExecutionException e ) { pool . markAsyncCreationDone ( ) ; // mark the end of async when reject
throw e ; } } // end of NOT warm up
} // end of if


signature:
public class DefaultConnectionManager { /** * initialize executor */
private void initializeExecutor ( ) { } }

implementation: if ( ! this . executorInitialized ) { this . executorInitialized = true ; this . asyncCreateConnectionExecutor = new ThreadPoolExecutor ( minPoolSize , maxPoolSize , keepAliveTime , TimeUnit . SECONDS , new ArrayBlockingQueue < Runnable > ( queueSize ) , new NamedThreadFactory ( "Bolt-conn-warmup-executor" , true ) ) ; }


signature:
public class RpcClient { /** * Shutdown .
 * Notice : < br >
 * < li > Rpc client can not be used any more after shutdown .
 * < li > If you need , you should destroy it , and instantiate another one . */
public void shutdown ( ) { } }

implementation: this . connectionManager . removeAll ( ) ; logger . warn ( "Close all connections from client side!" ) ; this . taskScanner . shutdown ( ) ; logger . warn ( "Rpc client shutdown!" ) ; if ( reconnectManager != null ) { reconnectManager . stop ( ) ; } if ( connectionMonitor != null ) { connectionMonitor . destroy ( ) ; }


signature:
public class RpcClient { /** * One way invocation using a string address , address format example - 127.0.0.1:12200 ? key1 = value1 & key2 = value2 < br >
 * Notice : < br >
 * < ol >
 * < li > < b > DO NOT modify the request object concurrently when this method is called . < / b > < / li >
 * < li > When do invocation , use the string address to find a available connection , if none then create one . < / li >
 * < ul >
 * < li > You can use { @ link RpcConfigs # CONNECT _ TIMEOUT _ KEY } to specify connection timeout , time unit is milliseconds , e . g [ 127.0.0.1:12200 ? _ CONNECTTIMEOUT = 3000]
 * < li > You can use { @ link RpcConfigs # CONNECTION _ NUM _ KEY } to specify connection number for each ip and port , e . g [ 127.0.0.1:12200 ? _ CONNECTIONNUM = 30]
 * < li > You can use { @ link RpcConfigs # CONNECTION _ WARMUP _ KEY } to specify whether need warmup all connections for the first time you call this method , e . g [ 127.0.0.1:12200 ? _ CONNECTIONWARMUP = false ]
 * < / ul >
 * < li > You should use { @ link # closeConnection ( String addr ) } to close it if you want .
 * < / ol >
 * @ param addr
 * @ param request
 * @ throws RemotingException
 * @ throws InterruptedException */
public void oneway ( final String addr , final Object request ) throws RemotingException , InterruptedException { } }

implementation: this . rpcRemoting . oneway ( addr , request , null ) ;


signature:
public class RpcClient { /** * One way invocation using a { @ link Connection } < br >
 * Notice : < br >
 * < b > DO NOT modify the request object concurrently when this method is called . < / b >
 * @ param conn
 * @ param request
 * @ throws RemotingException */
public void oneway ( final Connection conn , final Object request ) throws RemotingException { } }

implementation: this . rpcRemoting . oneway ( conn , request , null ) ;


signature:
public class RpcClient { /** * Oneway invocation with a { @ link InvokeContext } , common api notice please see { @ link # oneway ( Connection , Object ) }
 * @ param conn
 * @ param request
 * @ param invokeContext
 * @ throws RemotingException */
public void oneway ( final Connection conn , final Object request , final InvokeContext invokeContext ) throws RemotingException { } }

implementation: this . rpcRemoting . oneway ( conn , request , invokeContext ) ;


signature:
public class RpcClient { /** * Synchronous invocation using a string address , address format example - 127.0.0.1:12200 ? key1 = value1 & key2 = value2 < br >
 * Notice : < br >
 * < ol >
 * < li > < b > DO NOT modify the request object concurrently when this method is called . < / b > < / li >
 * < li > When do invocation , use the string address to find a available connection , if none then create one . < / li >
 * < ul >
 * < li > You can use { @ link RpcConfigs # CONNECT _ TIMEOUT _ KEY } to specify connection timeout , time unit is milliseconds , e . g [ 127.0.0.1:12200 ? _ CONNECTTIMEOUT = 3000]
 * < li > You can use { @ link RpcConfigs # CONNECTION _ NUM _ KEY } to specify connection number for each ip and port , e . g [ 127.0.0.1:12200 ? _ CONNECTIONNUM = 30]
 * < li > You can use { @ link RpcConfigs # CONNECTION _ WARMUP _ KEY } to specify whether need warmup all connections for the first time you call this method , e . g [ 127.0.0.1:12200 ? _ CONNECTIONWARMUP = false ]
 * < / ul >
 * < li > You should use { @ link # closeConnection ( String addr ) } to close it if you want .
 * < / ol >
 * @ param addr
 * @ param request
 * @ param timeoutMillis
 * @ return Object
 * @ throws RemotingException
 * @ throws InterruptedException */
public Object invokeSync ( final String addr , final Object request , final int timeoutMillis ) throws RemotingException , InterruptedException { } }

implementation: return this . rpcRemoting . invokeSync ( addr , request , null , timeoutMillis ) ;


signature:
public class RpcClient { /** * Synchronous invocation with a { @ link InvokeContext } , common api notice please see { @ link # invokeSync ( String , Object , int ) }
 * @ param addr
 * @ param request
 * @ param invokeContext
 * @ param timeoutMillis
 * @ return
 * @ throws RemotingException
 * @ throws InterruptedException */
public Object invokeSync ( final String addr , final Object request , final InvokeContext invokeContext , final int timeoutMillis ) throws RemotingException , InterruptedException { } }

implementation: return this . rpcRemoting . invokeSync ( addr , request , invokeContext , timeoutMillis ) ;


signature:
public class RpcClient { /** * Synchronous invocation using a parsed { @ link Url } < br >
 * Notice : < br >
 * < ol >
 * < li > < b > DO NOT modify the request object concurrently when this method is called . < / b > < / li >
 * < li > When do invocation , use the parsed { @ link Url } to find a available connection , if none then create one . < / li >
 * < ul >
 * < li > You can use { @ link Url # setConnectTimeout } to specify connection timeout , time unit is milliseconds .
 * < li > You can use { @ link Url # setConnNum } to specify connection number for each ip and port .
 * < li > You can use { @ link Url # setConnWarmup } to specify whether need warmup all connections for the first time you call this method .
 * < / ul >
 * < li > You should use { @ link # closeConnection ( Url url ) } to close it if you want .
 * < / ol >
 * @ param url
 * @ param request
 * @ param timeoutMillis
 * @ return Object
 * @ throws RemotingException
 * @ throws InterruptedException */
public Object invokeSync ( final Url url , final Object request , final int timeoutMillis ) throws RemotingException , InterruptedException { } }

implementation: return this . invokeSync ( url , request , null , timeoutMillis ) ;


signature:
public class RpcClient { /** * Synchronous invocation with a { @ link InvokeContext } , common api notice please see { @ link # invokeSync ( Url , Object , int ) }
 * @ param url
 * @ param request
 * @ param invokeContext
 * @ param timeoutMillis
 * @ return
 * @ throws RemotingException
 * @ throws InterruptedException */
public Object invokeSync ( final Url url , final Object request , final InvokeContext invokeContext , final int timeoutMillis ) throws RemotingException , InterruptedException { } }

implementation: return this . rpcRemoting . invokeSync ( url , request , invokeContext , timeoutMillis ) ;


signature:
public class RpcClient { /** * Synchronous invocation using a { @ link Connection } < br >
 * Notice : < br >
 * < b > DO NOT modify the request object concurrently when this method is called . < / b >
 * @ param conn
 * @ param request
 * @ param timeoutMillis
 * @ return Object
 * @ throws RemotingException
 * @ throws InterruptedException */
public Object invokeSync ( final Connection conn , final Object request , final int timeoutMillis ) throws RemotingException , InterruptedException { } }

implementation: return this . rpcRemoting . invokeSync ( conn , request , null , timeoutMillis ) ;


signature:
public class RpcClient { /** * Synchronous invocation with a { @ link InvokeContext } , common api notice please see { @ link # invokeSync ( Connection , Object , int ) }
 * @ param conn
 * @ param request
 * @ param invokeContext
 * @ param timeoutMillis
 * @ return Object
 * @ throws RemotingException
 * @ throws InterruptedException */
public Object invokeSync ( final Connection conn , final Object request , final InvokeContext invokeContext , final int timeoutMillis ) throws RemotingException , InterruptedException { } }

implementation: return this . rpcRemoting . invokeSync ( conn , request , invokeContext , timeoutMillis ) ;


signature:
public class RpcClient { /** * Future invocation using a { @ link Connection } < br >
 * You can get result use the returned { @ link RpcResponseFuture } .
 * Notice : < br >
 * < b > DO NOT modify the request object concurrently when this method is called . < / b >
 * @ param conn
 * @ param request
 * @ param timeoutMillis
 * @ return
 * @ throws RemotingException */
public RpcResponseFuture invokeWithFuture ( final Connection conn , final Object request , int timeoutMillis ) throws RemotingException { } }

implementation: return this . rpcRemoting . invokeWithFuture ( conn , request , null , timeoutMillis ) ;


signature:
public class RpcClient { /** * Callback invocation with a { @ link InvokeContext } , common api notice please see { @ link # invokeWithCallback ( Url , Object , InvokeCallback , int ) }
 * @ param url
 * @ param request
 * @ param invokeContext
 * @ param invokeCallback
 * @ param timeoutMillis
 * @ throws RemotingException
 * @ throws InterruptedException */
public void invokeWithCallback ( final Url url , final Object request , final InvokeContext invokeContext , final InvokeCallback invokeCallback , final int timeoutMillis ) throws RemotingException , InterruptedException { } }

implementation: this . rpcRemoting . invokeWithCallback ( url , request , invokeContext , invokeCallback , timeoutMillis ) ;


signature:
public class RpcClient { /** * Callback invocation using a { @ link Connection } < br >
 * You can specify an implementation of { @ link InvokeCallback } to get the result .
 * Notice : < br >
 * < b > DO NOT modify the request object concurrently when this method is called . < / b >
 * @ param conn
 * @ param request
 * @ param invokeCallback
 * @ param timeoutMillis
 * @ throws RemotingException */
public void invokeWithCallback ( final Connection conn , final Object request , final InvokeCallback invokeCallback , final int timeoutMillis ) throws RemotingException { } }

implementation: this . rpcRemoting . invokeWithCallback ( conn , request , null , invokeCallback , timeoutMillis ) ;


signature:
public class RpcClient { /** * Callback invocation with a { @ link InvokeContext } , common api notice please see { @ link # invokeWithCallback ( Connection , Object , InvokeCallback , int ) }
 * @ param conn
 * @ param request
 * @ param invokeContext
 * @ param invokeCallback
 * @ param timeoutMillis
 * @ throws RemotingException */
public void invokeWithCallback ( final Connection conn , final Object request , final InvokeContext invokeContext , final InvokeCallback invokeCallback , final int timeoutMillis ) throws RemotingException { } }

implementation: this . rpcRemoting . invokeWithCallback ( conn , request , invokeContext , invokeCallback , timeoutMillis ) ;


signature:
public class RpcClient { /** * Create a stand alone connection using ip and port . < br >
 * Notice : < br >
 * < li > Each time you call this method , will create a new connection .
 * < li > Bolt will not control this connection .
 * < li > You should use { @ link # closeStandaloneConnection } to close it .
 * @ param ip
 * @ param port
 * @ param connectTimeout
 * @ return
 * @ throws RemotingException */
public Connection createStandaloneConnection ( String ip , int port , int connectTimeout ) throws RemotingException { } }

implementation: return this . connectionManager . create ( ip , port , connectTimeout ) ;


signature:
public class RpcClient { /** * Create a stand alone connection using address , address format example - 127.0.0.1:12200 < br >
 * Notice : < br >
 * < ol >
 * < li > Each time you can this method , will create a new connection .
 * < li > Bolt will not control this connection .
 * < li > You should use { @ link # closeStandaloneConnection } to close it .
 * < / ol >
 * @ param addr
 * @ param connectTimeout
 * @ return
 * @ throws RemotingException */
public Connection createStandaloneConnection ( String addr , int connectTimeout ) throws RemotingException { } }

implementation: return this . connectionManager . create ( addr , connectTimeout ) ;


signature:
public class RpcClient { /** * Get a connection using address , address format example - 127.0.0.1:12200 ? key1 = value1 & key2 = value2 < br >
 * Notice : < br >
 * < ol >
 * < li > Get a connection , if none then create . < / li >
 * < ul >
 * < li > You can use { @ link RpcConfigs # CONNECT _ TIMEOUT _ KEY } to specify connection timeout , time unit is milliseconds , e . g [ 127.0.0.1:12200 ? _ CONNECTTIMEOUT = 3000]
 * < li > You can use { @ link RpcConfigs # CONNECTION _ NUM _ KEY } to specify connection number for each ip and port , e . g [ 127.0.0.1:12200 ? _ CONNECTIONNUM = 30]
 * < li > You can use { @ link RpcConfigs # CONNECTION _ WARMUP _ KEY } to specify whether need warmup all connections for the first time you call this method , e . g [ 127.0.0.1:12200 ? _ CONNECTIONWARMUP = false ]
 * < / ul >
 * < li > Bolt will control this connection in { @ link ConnectionPool }
 * < li > You should use { @ link # closeConnection ( String addr ) } to close it .
 * < / ol >
 * @ param addr
 * @ param connectTimeout this is prior to url args { @ link RpcConfigs # CONNECT _ TIMEOUT _ KEY }
 * @ return
 * @ throws RemotingException */
public Connection getConnection ( String addr , int connectTimeout ) throws RemotingException , InterruptedException { } }

implementation: Url url = this . addressParser . parse ( addr ) ; return this . getConnection ( url , connectTimeout ) ;


signature:
public class RpcClient { /** * Get a connection using a { @ link Url } . < br >
 * Notice :
 * < ol >
 * < li > Get a connection , if none then create .
 * < li > Bolt will control this connection in { @ link com . alipay . remoting . ConnectionPool }
 * < li > You should use { @ link # closeConnection ( Url url ) } to close it .
 * < / ol >
 * @ param url
 * @ param connectTimeout this is prior to url args { @ link RpcConfigs # CONNECT _ TIMEOUT _ KEY }
 * @ return
 * @ throws RemotingException */
public Connection getConnection ( Url url , int connectTimeout ) throws RemotingException , InterruptedException { } }

implementation: url . setConnectTimeout ( connectTimeout ) ; return this . connectionManager . getAndCreateIfAbsent ( url ) ;


signature:
public class RpcClient { /** * check connection , the address format example - 127.0.0.1:12200 ? key1 = value1 & key2 = value2
 * @ param addr
 * @ throws RemotingException
 * @ return true if and only if there is a connection , and the connection is active and writable ; else return false */
public boolean checkConnection ( String addr ) { } }

implementation: Url url = this . addressParser . parse ( addr ) ; Connection conn = this . connectionManager . get ( url . getUniqueKey ( ) ) ; try { this . connectionManager . check ( conn ) ; } catch ( Exception e ) { return false ; } return true ;


signature:
public class RpcClient { /** * Close all connections of a address
 * @ param addr */
public void closeConnection ( String addr ) { } }

implementation: Url url = this . addressParser . parse ( addr ) ; this . connectionManager . remove ( url . getUniqueKey ( ) ) ;


signature:
public class RpcClient { /** * Enable heart beat for a certain connection .
 * If this address not connected , then do nothing .
 * Notice : this method takes no effect on a stand alone connection .
 * @ param addr */
public void enableConnHeartbeat ( String addr ) { } }

implementation: Url url = this . addressParser . parse ( addr ) ; this . enableConnHeartbeat ( url ) ;


signature:
public class RpcClient { /** * Enable heart beat for a certain connection .
 * If this { @ link Url } not connected , then do nothing .
 * Notice : this method takes no effect on a stand alone connection .
 * @ param url */
public void enableConnHeartbeat ( Url url ) { } }

implementation: if ( null != url ) { this . connectionManager . enableHeartbeat ( this . connectionManager . get ( url . getUniqueKey ( ) ) ) ; }


signature:
public class RpcClient { /** * Disable heart beat for a certain connection .
 * If this addr not connected , then do nothing .
 * Notice : this method takes no effect on a stand alone connection .
 * @ param addr */
public void disableConnHeartbeat ( String addr ) { } }

implementation: Url url = this . addressParser . parse ( addr ) ; this . disableConnHeartbeat ( url ) ;


signature:
public class RpcClient { /** * Disable heart beat for a certain connection .
 * If this { @ link Url } not connected , then do nothing .
 * Notice : this method takes no effect on a stand alone connection .
 * @ param url */
public void disableConnHeartbeat ( Url url ) { } }

implementation: if ( null != url ) { this . connectionManager . disableHeartbeat ( this . connectionManager . get ( url . getUniqueKey ( ) ) ) ; }


signature:
public class Connection { /** * Initialization . */
private void init ( ) { } }

implementation: this . channel . attr ( HEARTBEAT_COUNT ) . set ( new Integer ( 0 ) ) ; this . channel . attr ( PROTOCOL ) . set ( this . protocolCode ) ; this . channel . attr ( VERSION ) . set ( this . version ) ; this . channel . attr ( HEARTBEAT_SWITCH ) . set ( true ) ;


signature:
public class Connection { /** * Do something when closing . */
public void onClose ( ) { } }

implementation: Iterator < Entry < Integer , InvokeFuture > > iter = invokeFutureMap . entrySet ( ) . iterator ( ) ; while ( iter . hasNext ( ) ) { Entry < Integer , InvokeFuture > entry = iter . next ( ) ; iter . remove ( ) ; InvokeFuture future = entry . getValue ( ) ; if ( future != null ) { future . putResponse ( future . createConnectionClosedResponse ( this . getRemoteAddress ( ) ) ) ; future . cancelTimeout ( ) ; future . tryAsyncExecuteInvokeCallbackAbnormally ( ) ; } }


signature:
public class Connection { /** * Close the connection . */
public void close ( ) { } }

implementation: if ( closed . compareAndSet ( false , true ) ) { try { if ( this . getChannel ( ) != null ) { this . getChannel ( ) . close ( ) . addListener ( new ChannelFutureListener ( ) { @ Override public void operationComplete ( ChannelFuture future ) throws Exception { if ( logger . isInfoEnabled ( ) ) { logger . info ( "Close the connection to remote address={}, result={}, cause={}" , RemotingUtil . parseRemoteAddress ( Connection . this . getChannel ( ) ) , future . isSuccess ( ) , future . cause ( ) ) ; } } } ) ; } } catch ( Exception e ) { logger . warn ( "Exception caught when closing connection {}" , RemotingUtil . parseRemoteAddress ( Connection . this . getChannel ( ) ) , e ) ; } }


signature:
public class Connection { /** * set attribute if key absent .
 * @ param key
 * @ param value
 * @ return */
public Object setAttributeIfAbsent ( String key , Object value ) { } }

implementation: return attributes . putIfAbsent ( key , value ) ;


signature:
public class AbstractBatchDecoder { /** * This method has been modified to check the size of decoded msgs , which is represented by the
 * local variable { @ code RecyclableArrayList out } . If has decoded more than one msg ,
 * then construct an array list to submit all decoded msgs to the pipeline .
 * @ param ctx channel handler context
 * @ param msg data */
@ Override public void channelRead ( ChannelHandlerContext ctx , Object msg ) throws Exception { } }

implementation:
if ( msg instanceof ByteBuf ) { RecyclableArrayList out = RecyclableArrayList . newInstance ( ) ; try { ByteBuf data = ( ByteBuf ) msg ; first = cumulation == null ; if ( first ) { cumulation = data ; } else { cumulation = cumulator . cumulate ( ctx . alloc ( ) , cumulation , data ) ; } callDecode ( ctx , cumulation , out ) ; } catch ( DecoderException e ) { throw e ; } catch ( Throwable t ) { throw new DecoderException ( t ) ; } finally { if ( cumulation != null && ! cumulation . isReadable ( ) ) { numReads = 0 ; cumulation . release ( ) ; cumulation = null ; } else if ( ++ numReads >= discardAfterReads ) { // We did enough reads already try to discard some bytes so we not risk to see a OOME .
// See https : / / github . com / netty / netty / issues / 4275
numReads = 0 ; discardSomeReadBytes ( ) ; } int size = out . size ( ) ; if ( size == 0 ) { decodeWasNull = true ; } else if ( size == 1 ) { ctx . fireChannelRead ( out . get ( 0 ) ) ; } else { ArrayList < Object > ret = new ArrayList < Object > ( size ) ; for ( int i = 0 ; i < size ; i ++ ) { ret . add ( out . get ( i ) ) ; } ctx . fireChannelRead ( ret ) ; } out . recycle ( ) ; } } else { ctx . fireChannelRead ( msg ) ; }


signature:
public class AbstractBatchDecoder { /** * Called once data should be decoded from the given { @ link ByteBuf } . This method will call
 * { @ link # decode ( ChannelHandlerContext , ByteBuf , List ) } as long as decoding should take place .
 * @ param ctx the { @ link ChannelHandlerContext } which this { @ link ByteToMessageDecoder } belongs to
 * @ param in the { @ link ByteBuf } from which to read data
 * @ param out the { @ link List } to which decoded messages should be added */
protected void callDecode ( ChannelHandlerContext ctx , ByteBuf in , List < Object > out ) { } }

implementation:
try { while ( in . isReadable ( ) ) { int outSize = out . size ( ) ; int oldInputLength = in . readableBytes ( ) ; decode ( ctx , in , out ) ; // Check if this handler was removed before continuing the loop .
// If it was removed , it is not safe to continue to operate on the buffer .
// See https : / / github . com / netty / netty / issues / 1664
if ( ctx . isRemoved ( ) ) { break ; } if ( outSize == out . size ( ) ) { if ( oldInputLength == in . readableBytes ( ) ) { break ; } else { continue ; } } if ( oldInputLength == in . readableBytes ( ) ) { throw new DecoderException ( StringUtil . simpleClassName ( getClass ( ) ) + ".decode() did not read anything but decoded a message." ) ; } if ( isSingleDecode ( ) ) { break ; } } } catch ( DecoderException e ) { throw e ; } catch ( Throwable cause ) { throw new DecoderException ( cause ) ; }


signature:
public class UserProcessorRegisterHelper { /** * Help register single - interest user processor .
 * @ param processor the processor need to be registered
 * @ param userProcessors the map of user processors */
public static void registerUserProcessor ( UserProcessor < ? > processor , ConcurrentHashMap < String , UserProcessor < ? > > userProcessors ) { } }

implementation: if ( null == processor ) { throw new RuntimeException ( "User processor should not be null!" ) ; } if ( processor instanceof MultiInterestUserProcessor ) { registerUserProcessor ( ( MultiInterestUserProcessor ) processor , userProcessors ) ; } else { if ( StringUtils . isBlank ( processor . interest ( ) ) ) { throw new RuntimeException ( "Processor interest should not be blank!" ) ; } UserProcessor < ? > preProcessor = userProcessors . putIfAbsent ( processor . interest ( ) , processor ) ; if ( preProcessor != null ) { String errMsg = "Processor with interest key [" + processor . interest ( ) + "] has already been registered to rpc server, can not register again!" ; throw new RuntimeException ( errMsg ) ; } }


signature:
public class UserProcessorRegisterHelper { /** * Help register multi - interest user processor .
 * @ param processor the processor with multi - interest need to be registered
 * @ param userProcessors the map of user processors */
private static void registerUserProcessor ( MultiInterestUserProcessor < ? > processor , ConcurrentHashMap < String , UserProcessor < ? > > userProcessors ) { } }

implementation: if ( null == processor . multiInterest ( ) || processor . multiInterest ( ) . isEmpty ( ) ) { throw new RuntimeException ( "Processor interest should not be blank!" ) ; } for ( String interest : processor . multiInterest ( ) ) { UserProcessor < ? > preProcessor = userProcessors . putIfAbsent ( interest , processor ) ; if ( preProcessor != null ) { String errMsg = "Processor with interest key [" + interest + "] has already been registered to rpc server, can not register again!" ; throw new RuntimeException ( errMsg ) ; } }


signature:
public class RpcRequestProcessor { /** * Send response using remoting context if necessary . < br >
 * If request type is oneway , no need to send any response nor exception .
 * @ param ctx remoting context
 * @ param type type code
 * @ param response remoting command */
public void sendResponseIfNecessary ( final RemotingContext ctx , byte type , final RemotingCommand response ) { } }

implementation:
final int id = response . getId ( ) ; if ( type != RpcCommandType . REQUEST_ONEWAY ) { RemotingCommand serializedResponse = response ; try { response . serialize ( ) ; } catch ( SerializationException e ) { String errMsg = "SerializationException occurred when sendResponseIfNecessary in RpcRequestProcessor, id=" + id ; logger . error ( errMsg , e ) ; serializedResponse = this . getCommandFactory ( ) . createExceptionResponse ( id , ResponseStatus . SERVER_SERIAL_EXCEPTION , e ) ; try { serializedResponse . serialize ( ) ; // serialize again for exception response
} catch ( SerializationException e1 ) { // should not happen
logger . error ( "serialize SerializationException response failed!" ) ; } } catch ( Throwable t ) { String errMsg = "Serialize RpcResponseCommand failed when sendResponseIfNecessary in RpcRequestProcessor, id=" + id ; logger . error ( errMsg , t ) ; serializedResponse = this . getCommandFactory ( ) . createExceptionResponse ( id , t , errMsg ) ; } ctx . writeAndFlush ( serializedResponse ) . addListener ( new ChannelFutureListener ( ) { @ Override public void operationComplete ( ChannelFuture future ) throws Exception { if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Rpc response sent! requestId=" + id + ". The address is " + RemotingUtil . parseRemoteAddress ( ctx . getChannelContext ( ) . channel ( ) ) ) ; } if ( ! future . isSuccess ( ) ) { logger . error ( "Rpc response send failed! id=" + id + ". The address is " + RemotingUtil . parseRemoteAddress ( ctx . getChannelContext ( ) . channel ( ) ) , future . cause ( ) ) ; } } } ) ; } else { if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Oneway rpc request received, do not send response, id=" + id + ", the address is " + RemotingUtil . parseRemoteAddress ( ctx . getChannelContext ( ) . channel ( ) ) ) ; } }


signature:
public class RpcRequestProcessor { /** * dispatch request command to user processor
 * @ param ctx remoting context
 * @ param cmd rpc request command */
private void dispatchToUserProcessor ( RemotingContext ctx , RpcRequestCommand cmd ) { } }

implementation:
final int id = cmd . getId ( ) ; final byte type = cmd . getType ( ) ; // processor here must not be null , for it have been checked before
UserProcessor processor = ctx . getUserProcessor ( cmd . getRequestClass ( ) ) ; if ( processor instanceof AsyncUserProcessor ) { try { processor . handleRequest ( processor . preHandleRequest ( ctx , cmd . getRequestObject ( ) ) , new RpcAsyncContext ( ctx , cmd , this ) , cmd . getRequestObject ( ) ) ; } catch ( RejectedExecutionException e ) { logger . warn ( "RejectedExecutionException occurred when do ASYNC process in RpcRequestProcessor" ) ; sendResponseIfNecessary ( ctx , type , this . getCommandFactory ( ) . createExceptionResponse ( id , ResponseStatus . SERVER_THREADPOOL_BUSY ) ) ; } catch ( Throwable t ) { String errMsg = "AYSNC process rpc request failed in RpcRequestProcessor, id=" + id ; logger . error ( errMsg , t ) ; sendResponseIfNecessary ( ctx , type , this . getCommandFactory ( ) . createExceptionResponse ( id , t , errMsg ) ) ; } } else { try { Object responseObject = processor . handleRequest ( processor . preHandleRequest ( ctx , cmd . getRequestObject ( ) ) , cmd . getRequestObject ( ) ) ; sendResponseIfNecessary ( ctx , type , this . getCommandFactory ( ) . createResponse ( responseObject , cmd ) ) ; } catch ( RejectedExecutionException e ) { logger . warn ( "RejectedExecutionException occurred when do SYNC process in RpcRequestProcessor" ) ; sendResponseIfNecessary ( ctx , type , this . getCommandFactory ( ) . createExceptionResponse ( id , ResponseStatus . SERVER_THREADPOOL_BUSY ) ) ; } catch ( Throwable t ) { String errMsg = "SYNC process rpc request failed in RpcRequestProcessor, id=" + id ; logger . error ( errMsg , t ) ; sendResponseIfNecessary ( ctx , type , this . getCommandFactory ( ) . createExceptionResponse ( id , t , errMsg ) ) ; } }


signature:
public class RpcRequestProcessor { /** * deserialize request command
 * @ return true if deserialize success ; false if exception catched */
private boolean deserializeRequestCommand ( RemotingContext ctx , RpcRequestCommand cmd , int level ) { } }

implementation: boolean result ; try { cmd . deserialize ( level ) ; result = true ; } catch ( DeserializationException e ) { logger . error ( "DeserializationException occurred when process in RpcRequestProcessor, id={}, deserializeLevel={}" , cmd . getId ( ) , RpcDeserializeLevel . valueOf ( level ) , e ) ; sendResponseIfNecessary ( ctx , cmd . getType ( ) , this . getCommandFactory ( ) . createExceptionResponse ( cmd . getId ( ) , ResponseStatus . SERVER_DESERIAL_EXCEPTION , e ) ) ; result = false ; } catch ( Throwable t ) { String errMsg = "Deserialize RpcRequestCommand failed in RpcRequestProcessor, id=" + cmd . getId ( ) + ", deserializeLevel=" + level ; logger . error ( errMsg , t ) ; sendResponseIfNecessary ( ctx , cmd . getType ( ) , this . getCommandFactory ( ) . createExceptionResponse ( cmd . getId ( ) , t , errMsg ) ) ; result = false ; } return result ;


signature:
public class RpcRequestProcessor { /** * pre process remoting context , initial some useful infos and pass to biz
 * @ param ctx remoting context
 * @ param cmd rpc request command
 * @ param currentTimestamp current timestamp */
private void preProcessRemotingContext ( RemotingContext ctx , RpcRequestCommand cmd , long currentTimestamp ) { } }

implementation: ctx . setArriveTimestamp ( cmd . getArriveTime ( ) ) ; ctx . setTimeout ( cmd . getTimeout ( ) ) ; ctx . setRpcCommandType ( cmd . getType ( ) ) ; ctx . getInvokeContext ( ) . putIfAbsent ( InvokeContext . BOLT_PROCESS_WAIT_TIME , currentTimestamp - cmd . getArriveTime ( ) ) ;


signature:
public class RpcRequestProcessor { /** * print some log when request timeout and discarded in io thread . */
private void timeoutLog ( final RpcRequestCommand cmd , long currentTimestamp , RemotingContext ctx ) { } }

implementation: if ( logger . isDebugEnabled ( ) ) { logger . debug ( "request id [{}] currenTimestamp [{}] - arriveTime [{}] = server cost [{}] >= timeout value [{}]." , cmd . getId ( ) , currentTimestamp , cmd . getArriveTime ( ) , ( currentTimestamp - cmd . getArriveTime ( ) ) , cmd . getTimeout ( ) ) ; } String remoteAddr = "UNKNOWN" ; if ( null != ctx ) { ChannelHandlerContext channelCtx = ctx . getChannelContext ( ) ; Channel channel = channelCtx . channel ( ) ; if ( null != channel ) { remoteAddr = RemotingUtil . parseRemoteAddress ( channel ) ; } } logger . warn ( "Rpc request id[{}], from remoteAddr[{}] stop process, total wait time in queue is [{}], client timeout setting is [{}]." , cmd . getId ( ) , remoteAddr , ( currentTimestamp - cmd . getArriveTime ( ) ) , cmd . getTimeout ( ) ) ;


signature:
public class RpcRequestProcessor { /** * print some debug log when receive request */
private void debugLog ( RemotingContext ctx , RpcRequestCommand cmd , long currentTimestamp ) { } }

implementation: if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Rpc request received! requestId={}, from {}" , cmd . getId ( ) , RemotingUtil . parseRemoteAddress ( ctx . getChannelContext ( ) . channel ( ) ) ) ; logger . debug ( "request id {} currenTimestamp {} - arriveTime {} = server cost {} < timeout {}." , cmd . getId ( ) , currentTimestamp , cmd . getArriveTime ( ) , ( currentTimestamp - cmd . getArriveTime ( ) ) , cmd . getTimeout ( ) ) ; }


signature:
public class AbstractRemotingProcessor { /** * Process the remoting command with its own executor or with the defaultExecutor if its own if null .
 * @ param ctx
 * @ param msg
 * @ param defaultExecutor
 * @ throws Exception */
@ Override public void process ( RemotingContext ctx , T msg , ExecutorService defaultExecutor ) throws Exception { } }

implementation: ProcessTask task = new ProcessTask ( ctx , msg ) ; if ( this . getExecutor ( ) != null ) { this . getExecutor ( ) . execute ( task ) ; } else { defaultExecutor . execute ( task ) ; }


signature:
public class BaseRemoting { /** * Synchronous invocation
 * @ param conn
 * @ param request
 * @ param timeoutMillis
 * @ return
 * @ throws InterruptedException
 * @ throws RemotingException */
protected RemotingCommand invokeSync ( final Connection conn , final RemotingCommand request , final int timeoutMillis ) throws RemotingException , InterruptedException { } }

implementation: final InvokeFuture future = createInvokeFuture ( request , request . getInvokeContext ( ) ) ; conn . addInvokeFuture ( future ) ; final int requestId = request . getId ( ) ; try { conn . getChannel ( ) . writeAndFlush ( request ) . addListener ( new ChannelFutureListener ( ) { @ Override public void operationComplete ( ChannelFuture f ) throws Exception { if ( ! f . isSuccess ( ) ) { conn . removeInvokeFuture ( requestId ) ; future . putResponse ( commandFactory . createSendFailedResponse ( conn . getRemoteAddress ( ) , f . cause ( ) ) ) ; logger . error ( "Invoke send failed, id={}" , requestId , f . cause ( ) ) ; } } } ) ; } catch ( Exception e ) { conn . removeInvokeFuture ( requestId ) ; future . putResponse ( commandFactory . createSendFailedResponse ( conn . getRemoteAddress ( ) , e ) ) ; logger . error ( "Exception caught when sending invocation, id={}" , requestId , e ) ; } RemotingCommand response = future . waitResponse ( timeoutMillis ) ; if ( response == null ) { conn . removeInvokeFuture ( requestId ) ; response = this . commandFactory . createTimeoutResponse ( conn . getRemoteAddress ( ) ) ; logger . warn ( "Wait response, request id={} timeout!" , requestId ) ; } return response ;


signature:
public class BaseRemoting { /** * Invocation with callback .
 * @ param conn
 * @ param request
 * @ param invokeCallback
 * @ param timeoutMillis
 * @ throws InterruptedException */
protected void invokeWithCallback ( final Connection conn , final RemotingCommand request , final InvokeCallback invokeCallback , final int timeoutMillis ) { } }

implementation: final InvokeFuture future = createInvokeFuture ( conn , request , request . getInvokeContext ( ) , invokeCallback ) ; conn . addInvokeFuture ( future ) ; final int requestId = request . getId ( ) ; try { Timeout timeout = TimerHolder . getTimer ( ) . newTimeout ( new TimerTask ( ) { @ Override public void run ( Timeout timeout ) throws Exception { InvokeFuture future = conn . removeInvokeFuture ( requestId ) ; if ( future != null ) { future . putResponse ( commandFactory . createTimeoutResponse ( conn . getRemoteAddress ( ) ) ) ; future . tryAsyncExecuteInvokeCallbackAbnormally ( ) ; } } } , timeoutMillis , TimeUnit . MILLISECONDS ) ; future . addTimeout ( timeout ) ; conn . getChannel ( ) . writeAndFlush ( request ) . addListener ( new ChannelFutureListener ( ) { @ Override public void operationComplete ( ChannelFuture cf ) throws Exception { if ( ! cf . isSuccess ( ) ) { InvokeFuture f = conn . removeInvokeFuture ( requestId ) ; if ( f != null ) { f . cancelTimeout ( ) ; f . putResponse ( commandFactory . createSendFailedResponse ( conn . getRemoteAddress ( ) , cf . cause ( ) ) ) ; f . tryAsyncExecuteInvokeCallbackAbnormally ( ) ; } logger . error ( "Invoke send failed. The address is {}" , RemotingUtil . parseRemoteAddress ( conn . getChannel ( ) ) , cf . cause ( ) ) ; } } } ) ; } catch ( Exception e ) { InvokeFuture f = conn . removeInvokeFuture ( requestId ) ; if ( f != null ) { f . cancelTimeout ( ) ; f . putResponse ( commandFactory . createSendFailedResponse ( conn . getRemoteAddress ( ) , e ) ) ; f . tryAsyncExecuteInvokeCallbackAbnormally ( ) ; } logger . error ( "Exception caught when sending invocation. The address is {}" , RemotingUtil . parseRemoteAddress ( conn . getChannel ( ) ) , e ) ; }


signature:
public class BaseRemoting { /** * Oneway invocation .
 * @ param conn
 * @ param request
 * @ throws InterruptedException */
protected void oneway ( final Connection conn , final RemotingCommand request ) { } }

implementation: try { conn . getChannel ( ) . writeAndFlush ( request ) . addListener ( new ChannelFutureListener ( ) { @ Override public void operationComplete ( ChannelFuture f ) throws Exception { if ( ! f . isSuccess ( ) ) { logger . error ( "Invoke send failed. The address is {}" , RemotingUtil . parseRemoteAddress ( conn . getChannel ( ) ) , f . cause ( ) ) ; } } } ) ; } catch ( Exception e ) { if ( null == conn ) { logger . error ( "Conn is null" ) ; } else { logger . error ( "Exception caught when sending invocation. The address is {}" , RemotingUtil . parseRemoteAddress ( conn . getChannel ( ) ) , e ) ; } }


signature:
public class ProtocolSwitch { /** * create an instance of { @ link ProtocolSwitch } according to byte value
 * @ param value
 * @ return ProtocolSwitchStatus with initialized bit set . */
public static ProtocolSwitch create ( int value ) { } }

implementation: ProtocolSwitch status = new ProtocolSwitch ( ) ; status . setBs ( toBitSet ( value ) ) ; return status ;


signature:
public class ProtocolSwitch { /** * create an instance of { @ link ProtocolSwitch } according to switch index
 * @ param index the switch index which you want to set true
 * @ return ProtocolSwitchStatus with initialized bit set . */
public static ProtocolSwitch create ( int [ ] index ) { } }

implementation: ProtocolSwitch status = new ProtocolSwitch ( ) ; for ( int i = 0 ; i < index . length ; ++ i ) { status . turnOn ( index [ i ] ) ; } return status ;


signature:
public class ProtocolSwitch { /** * from bit set to byte
 * @ param bs
 * @ return byte represent the bit set */
public static byte toByte ( BitSet bs ) { } }

implementation: int value = 0 ; for ( int i = 0 ; i < bs . length ( ) ; ++ i ) { if ( bs . get ( i ) ) { value += 1 << i ; } } if ( bs . length ( ) > 7 ) { throw new IllegalArgumentException ( "The byte value " + value + " generated according to bit set " + bs + " is out of range, should be limited between [" + Byte . MIN_VALUE + "] to [" + Byte . MAX_VALUE + "]" ) ; } return ( byte ) value ;


signature:
public class ProtocolSwitch { /** * from byte to bit set
 * @ param value
 * @ return bit set represent the byte */
public static BitSet toBitSet ( int value ) { } }

implementation: if ( value > Byte . MAX_VALUE || value < Byte . MIN_VALUE ) { throw new IllegalArgumentException ( "The value " + value + " is out of byte range, should be limited between [" + Byte . MIN_VALUE + "] to [" + Byte . MAX_VALUE + "]" ) ; } BitSet bs = new BitSet ( ) ; int index = 0 ; while ( value != 0 ) { if ( value % 2 != 0 ) { bs . set ( index ) ; } ++ index ; value = ( byte ) ( value >> 1 ) ; } return bs ;


signature:
public class RpcClientRemoting { /** * Get connection and set init invokeContext if invokeContext not { @ code null }
 * @ param url target url
 * @ param invokeContext invoke context to set
 * @ return connection */
protected Connection getConnectionAndInitInvokeContext ( Url url , InvokeContext invokeContext ) throws RemotingException , InterruptedException { } }

implementation: long start = System . currentTimeMillis ( ) ; Connection conn ; try { conn = this . connectionManager . getAndCreateIfAbsent ( url ) ; } finally { if ( null != invokeContext ) { invokeContext . putIfAbsent ( InvokeContext . CLIENT_CONN_CREATETIME , ( System . currentTimeMillis ( ) - start ) ) ; } } return conn ;


signature:
public class ReconnectManager { /** * add reconnect task
 * @ param url */
public void addReconnectTask ( Url url ) { } }

implementation: ReconnectTask task = new ReconnectTask ( ) ; task . url = url ; tasks . add ( task ) ;


signature:
public class ReconnectManager { /** * stop reconnect thread */
public void stop ( ) { } }

implementation: if ( ! this . started ) { return ; } this . started = false ; healConnectionThreads . interrupt ( ) ; this . tasks . clear ( ) ; this . canceled . clear ( ) ;


signature:
public class RpcRemoting { /** * Oneway rpc invocation . < br >
 * Notice ! DO NOT modify the request object concurrently when this method is called .
 * @ param addr
 * @ param request
 * @ param invokeContext
 * @ throws RemotingException
 * @ throws InterruptedException */
public void oneway ( final String addr , final Object request , final InvokeContext invokeContext ) throws RemotingException , InterruptedException { } }

implementation: Url url = this . addressParser . parse ( addr ) ; this . oneway ( url , request , invokeContext ) ;


signature:
public class RpcRemoting { /** * Oneway rpc invocation . < br >
 * Notice ! DO NOT modify the request object concurrently when this method is called .
 * @ param conn
 * @ param request
 * @ param invokeContext
 * @ throws RemotingException */
public void oneway ( final Connection conn , final Object request , final InvokeContext invokeContext ) throws RemotingException { } }

implementation: RequestCommand requestCommand = ( RequestCommand ) toRemotingCommand ( request , conn , invokeContext , - 1 ) ; requestCommand . setType ( RpcCommandType . REQUEST_ONEWAY ) ; preProcessInvokeContext ( invokeContext , requestCommand , conn ) ; super . oneway ( conn , requestCommand ) ;


signature:
public class RpcRemoting { /** * Synchronous rpc invocation . < br >
 * Notice ! DO NOT modify the request object concurrently when this method is called .
 * @ param addr
 * @ param request
 * @ param invokeContext
 * @ param timeoutMillis
 * @ return
 * @ throws RemotingException
 * @ throws InterruptedException */
public Object invokeSync ( final String addr , final Object request , final InvokeContext invokeContext , final int timeoutMillis ) throws RemotingException , InterruptedException { } }

implementation: Url url = this . addressParser . parse ( addr ) ; return this . invokeSync ( url , request , invokeContext , timeoutMillis ) ;


signature:
public class RpcRemoting { /** * Synchronous rpc invocation . < br >
 * Notice ! DO NOT modify the request object concurrently when this method is called .
 * @ param conn
 * @ param request
 * @ param invokeContext
 * @ param timeoutMillis
 * @ return
 * @ throws RemotingException
 * @ throws InterruptedException */
public Object invokeSync ( final Connection conn , final Object request , final InvokeContext invokeContext , final int timeoutMillis ) throws RemotingException , InterruptedException { } }

implementation: RemotingCommand requestCommand = toRemotingCommand ( request , conn , invokeContext , timeoutMillis ) ; preProcessInvokeContext ( invokeContext , requestCommand , conn ) ; ResponseCommand responseCommand = ( ResponseCommand ) super . invokeSync ( conn , requestCommand , timeoutMillis ) ; responseCommand . setInvokeContext ( invokeContext ) ; Object responseObject = RpcResponseResolver . resolveResponseObject ( responseCommand , RemotingUtil . parseRemoteAddress ( conn . getChannel ( ) ) ) ; return responseObject ;


signature:
public class RpcRemoting { /** * Rpc invocation with future returned . < br >
 * Notice ! DO NOT modify the request object concurrently when this method is called .
 * @ param conn
 * @ param request
 * @ param invokeContext
 * @ param timeoutMillis
 * @ return
 * @ throws RemotingException */
public RpcResponseFuture invokeWithFuture ( final Connection conn , final Object request , final InvokeContext invokeContext , final int timeoutMillis ) throws RemotingException { } }

implementation: RemotingCommand requestCommand = toRemotingCommand ( request , conn , invokeContext , timeoutMillis ) ; preProcessInvokeContext ( invokeContext , requestCommand , conn ) ; InvokeFuture future = super . invokeWithFuture ( conn , requestCommand , timeoutMillis ) ; return new RpcResponseFuture ( RemotingUtil . parseRemoteAddress ( conn . getChannel ( ) ) , future ) ;


signature:
public class RpcRemoting { /** * Rpc invocation with callback . < br >
 * Notice ! DO NOT modify the request object concurrently when this method is called .
 * @ param addr
 * @ param request
 * @ param invokeContext
 * @ param invokeCallback
 * @ param timeoutMillis
 * @ throws RemotingException
 * @ throws InterruptedException */
public void invokeWithCallback ( String addr , Object request , final InvokeContext invokeContext , InvokeCallback invokeCallback , int timeoutMillis ) throws RemotingException , InterruptedException { } }

implementation: Url url = this . addressParser . parse ( addr ) ; this . invokeWithCallback ( url , request , invokeContext , invokeCallback , timeoutMillis ) ;


signature:
public class RpcRemoting { /** * Rpc invocation with callback . < br >
 * Notice ! DO NOT modify the request object concurrently when this method is called .
 * @ param conn
 * @ param request
 * @ param invokeContext
 * @ param invokeCallback
 * @ param timeoutMillis
 * @ throws RemotingException */
public void invokeWithCallback ( final Connection conn , final Object request , final InvokeContext invokeContext , final InvokeCallback invokeCallback , final int timeoutMillis ) throws RemotingException { } }

implementation: RemotingCommand requestCommand = toRemotingCommand ( request , conn , invokeContext , timeoutMillis ) ; preProcessInvokeContext ( invokeContext , requestCommand , conn ) ; super . invokeWithCallback ( conn , requestCommand , invokeCallback , timeoutMillis ) ;


signature:
public class RpcRemoting { /** * Convert application request object to remoting request command .
 * @ param request
 * @ param conn
 * @ param timeoutMillis
 * @ return
 * @ throws CodecException */
protected RemotingCommand toRemotingCommand ( Object request , Connection conn , InvokeContext invokeContext , int timeoutMillis ) throws SerializationException { } }

implementation:
RpcRequestCommand command = this . getCommandFactory ( ) . createRequestCommand ( request ) ; if ( null != invokeContext ) { // set client custom serializer for request command if not null
Object clientCustomSerializer = invokeContext . get ( InvokeContext . BOLT_CUSTOM_SERIALIZER ) ; if ( null != clientCustomSerializer ) { try { command . setSerializer ( ( Byte ) clientCustomSerializer ) ; } catch ( ClassCastException e ) { throw new IllegalArgumentException ( "Illegal custom serializer [" + clientCustomSerializer + "], the type of value should be [byte], but now is [" + clientCustomSerializer . getClass ( ) . getName ( ) + "]." ) ; } } // enable crc by default , user can disable by set invoke context ` false ` for key ` InvokeContext . BOLT _ CRC _ SWITCH `
Boolean crcSwitch = invokeContext . get ( InvokeContext . BOLT_CRC_SWITCH , ProtocolSwitch . CRC_SWITCH_DEFAULT_VALUE ) ; if ( null != crcSwitch && crcSwitch ) { command . setProtocolSwitch ( ProtocolSwitch . create ( new int [ ] { ProtocolSwitch . CRC_SWITCH_INDEX } ) ) ; } } else { // enable crc by default , if there is no invoke context .
command . setProtocolSwitch ( ProtocolSwitch . create ( new int [ ] { ProtocolSwitch . CRC_SWITCH_INDEX } ) ) ; } command . setTimeout ( timeoutMillis ) ; command . setRequestClass ( request . getClass ( ) . getName ( ) ) ; command . setInvokeContext ( invokeContext ) ; command . serialize ( ) ; logDebugInfo ( command ) ; return command ;


signature:
public class ScheduledDisconnectStrategy { /** * Filter connections to monitor
 * @ param connections */
@ Override public Map < String , List < Connection > > filter ( List < Connection > connections ) { } }

implementation: List < Connection > serviceOnConnections = new ArrayList < Connection > ( ) ; List < Connection > serviceOffConnections = new ArrayList < Connection > ( ) ; Map < String , List < Connection > > filteredConnections = new ConcurrentHashMap < String , List < Connection > > ( ) ; for ( Connection connection : connections ) { String serviceStatus = ( String ) connection . getAttribute ( Configs . CONN_SERVICE_STATUS ) ; if ( serviceStatus != null ) { if ( connection . isInvokeFutureMapFinish ( ) && ! freshSelectConnections . containsValue ( connection ) ) { serviceOffConnections . add ( connection ) ; } } else { serviceOnConnections . add ( connection ) ; } } filteredConnections . put ( Configs . CONN_SERVICE_STATUS_ON , serviceOnConnections ) ; filteredConnections . put ( Configs . CONN_SERVICE_STATUS_OFF , serviceOffConnections ) ; return filteredConnections ;


signature:
public class ScheduledDisconnectStrategy { /** * Monitor connections and close connections with status is off
 * @ param connPools */
@ Override public void monitor ( Map < String , RunStateRecordedFutureTask < ConnectionPool > > connPools ) { } }

implementation: try { if ( null != connPools && ! connPools . isEmpty ( ) ) { Iterator < Map . Entry < String , RunStateRecordedFutureTask < ConnectionPool > > > iter = connPools . entrySet ( ) . iterator ( ) ; while ( iter . hasNext ( ) ) { Map . Entry < String , RunStateRecordedFutureTask < ConnectionPool > > entry = iter . next ( ) ; String poolKey = entry . getKey ( ) ; ConnectionPool pool = FutureTaskUtil . getFutureTaskResult ( entry . getValue ( ) , logger ) ; List < Connection > connections = pool . getAll ( ) ; Map < String , List < Connection > > filteredConnectons = this . filter ( connections ) ; List < Connection > serviceOnConnections = filteredConnectons . get ( Configs . CONN_SERVICE_STATUS_ON ) ; List < Connection > serviceOffConnections = filteredConnectons . get ( Configs . CONN_SERVICE_STATUS_OFF ) ; if ( serviceOnConnections . size ( ) > CONNECTION_THRESHOLD ) { Connection freshSelectConnect = serviceOnConnections . get ( random . nextInt ( serviceOnConnections . size ( ) ) ) ; freshSelectConnect . setAttribute ( Configs . CONN_SERVICE_STATUS , Configs . CONN_SERVICE_STATUS_OFF ) ; Connection lastSelectConnect = freshSelectConnections . remove ( poolKey ) ; freshSelectConnections . put ( poolKey , freshSelectConnect ) ; closeFreshSelectConnections ( lastSelectConnect , serviceOffConnections ) ; } else { if ( freshSelectConnections . containsKey ( poolKey ) ) { Connection lastSelectConnect = freshSelectConnections . remove ( poolKey ) ; closeFreshSelectConnections ( lastSelectConnect , serviceOffConnections ) ; } if ( logger . isInfoEnabled ( ) ) { logger . info ( "the size of serviceOnConnections [{}] reached CONNECTION_THRESHOLD [{}]." , serviceOnConnections . size ( ) , CONNECTION_THRESHOLD ) ; } } for ( Connection offConn : serviceOffConnections ) { if ( offConn . isFine ( ) ) { offConn . close ( ) ; } } } } } catch ( Exception e ) { logger . error ( "ScheduledDisconnectStrategy monitor error" , e ) ; }


signature:
public class ScheduledDisconnectStrategy { /** * close the connection of the fresh select connections
 * @ param lastSelectConnect
 * @ param serviceOffConnections
 * @ throws InterruptedException */
private void closeFreshSelectConnections ( Connection lastSelectConnect , List < Connection > serviceOffConnections ) throws InterruptedException { } }

implementation: if ( null != lastSelectConnect ) { if ( lastSelectConnect . isInvokeFutureMapFinish ( ) ) { serviceOffConnections . add ( lastSelectConnect ) ; } else { Thread . sleep ( RETRY_DETECT_PERIOD ) ; if ( lastSelectConnect . isInvokeFutureMapFinish ( ) ) { serviceOffConnections . add ( lastSelectConnect ) ; } else { if ( logger . isInfoEnabled ( ) ) { logger . info ( "Address={} won't close at this schedule turn" , RemotingUtil . parseRemoteAddress ( lastSelectConnect . getChannel ( ) ) ) ; } } } }


signature:
public class RpcCommandHandler { /** * Handle the request ( s ) . */
private void handle ( final RemotingContext ctx , final Object msg ) { } }

implementation:
try { if ( msg instanceof List ) { final Runnable handleTask = new Runnable ( ) { @ Override public void run ( ) { if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Batch message! size={}" , ( ( List < ? > ) msg ) . size ( ) ) ; } for ( final Object m : ( List < ? > ) msg ) { RpcCommandHandler . this . process ( ctx , m ) ; } } } ; if ( RpcConfigManager . dispatch_msg_list_in_default_executor ( ) ) { // If msg is list , then the batch submission to biz threadpool can save io thread .
// See com . alipay . remoting . decoder . ProtocolDecoder
processorManager . getDefaultExecutor ( ) . execute ( handleTask ) ; } else { handleTask . run ( ) ; } } else { process ( ctx , msg ) ; } } catch ( final Throwable t ) { processException ( ctx , msg , t ) ; }


signature:
public class RpcCommandHandler { /** * Return error command if necessary . */
private void processExceptionForSingleCommand ( RemotingContext ctx , Object msg , Throwable t ) { } }

implementation:
final int id = ( ( RpcCommand ) msg ) . getId ( ) ; final String emsg = "Exception caught when processing " + ( ( msg instanceof RequestCommand ) ? "request, id=" : "response, id=" ) ; logger . warn ( emsg + id , t ) ; if ( msg instanceof RequestCommand ) { final RequestCommand cmd = ( RequestCommand ) msg ; if ( cmd . getType ( ) != RpcCommandType . REQUEST_ONEWAY ) { if ( t instanceof RejectedExecutionException ) { final ResponseCommand response = this . commandFactory . createExceptionResponse ( id , ResponseStatus . SERVER_THREADPOOL_BUSY ) ; // RejectedExecutionException here assures no response has been sent back
// Other exceptions should be processed where exception was caught , because here we don ' t known whether ack had been sent back .
ctx . getChannelContext ( ) . writeAndFlush ( response ) . addListener ( new ChannelFutureListener ( ) { @ Override public void operationComplete ( ChannelFuture future ) throws Exception { if ( future . isSuccess ( ) ) { if ( logger . isInfoEnabled ( ) ) { logger . info ( "Write back exception response done, requestId={}, status={}" , id , response . getResponseStatus ( ) ) ; } } else { logger . error ( "Write back exception response failed, requestId={}" , id , future . cause ( ) ) ; } } } ) ; } } }


signature:
public class ConfigManager { /** * ~ ~ ~ public helper methods to retrieve system property */
public static boolean getBool ( String key , String defaultValue ) { } }

implementation: return Boolean . parseBoolean ( System . getProperty ( key , defaultValue ) ) ;


signature:
public class RpcCommand { /** * Deserialize according to mask .
 * < ol >
 * < li > If mask < = { @ link RpcDeserializeLevel # DESERIALIZE _ CLAZZ } , only deserialize clazz - only one part . < / li >
 * < li > If mask < = { @ link RpcDeserializeLevel # DESERIALIZE _ HEADER } , deserialize clazz and header - two parts . < / li >
 * < li > If mask < = { @ link RpcDeserializeLevel # DESERIALIZE _ ALL } , deserialize clazz , header and content - all three parts . < / li >
 * < / ol >
 * @ param mask
 * @ throws CodecException */
public void deserialize ( long mask ) throws DeserializationException { } }

implementation: if ( mask <= RpcDeserializeLevel . DESERIALIZE_CLAZZ ) { this . deserializeClazz ( ) ; } else if ( mask <= RpcDeserializeLevel . DESERIALIZE_HEADER ) { this . deserializeClazz ( ) ; this . deserializeHeader ( this . getInvokeContext ( ) ) ; } else if ( mask <= RpcDeserializeLevel . DESERIALIZE_ALL ) { this . deserialize ( ) ; }


signature:
public class RpcRequestCommand { /** * Getter method for property < tt > customSerializer < / tt > .
 * @ return property value of customSerializer */
public CustomSerializer getCustomSerializer ( ) { } }

implementation: if ( this . customSerializer != null ) { return customSerializer ; } if ( this . requestClass != null ) { this . customSerializer = CustomSerializerManager . getCustomSerializer ( this . requestClass ) ; } if ( this . customSerializer == null ) { this . customSerializer = CustomSerializerManager . getCustomSerializer ( this . getCmdCode ( ) ) ; } return this . customSerializer ;


signature:
public class ConnectionEventHandler { /** * Setter method for property < tt > listener < / tt > .
 * @ param listener value to be assigned to property listener */
public void setConnectionEventListener ( ConnectionEventListener listener ) { } }

implementation: if ( listener != null ) { this . eventListener = listener ; if ( this . eventExecutor == null ) { this . eventExecutor = new ConnectionEventExecutor ( ) ; } }


signature:
public class ConnectionEventHandler { /** * print info log
 * @ param format
 * @ param addr */
private void infoLog ( String format , String addr ) { } }

implementation: if ( logger . isInfoEnabled ( ) ) { if ( StringUtils . isNotEmpty ( addr ) ) { logger . info ( format , addr ) ; } else { logger . info ( format , "UNKNOWN-ADDR" ) ; } }


signature:
public class RemotingContext { /** * whether this request already timeout
 * @ return */
public boolean isRequestTimeout ( ) { } }

implementation: if ( this . timeout > 0 && ( this . rpcCommandType != RpcCommandType . REQUEST_ONEWAY ) && ( System . currentTimeMillis ( ) - this . arriveTimestamp ) > this . timeout ) { return true ; } return false ;


signature:
public class RemotingContext { /** * Get user processor for class name .
 * @ param className
 * @ return */
public UserProcessor < ? > getUserProcessor ( String className ) { } }

implementation: return StringUtils . isBlank ( className ) ? null : this . userProcessors . get ( className ) ;


signature:
public class RandomSelectStrategy { /** * get one connection randomly
 * @ param conns
 * @ return */
private Connection randomGet ( List < Connection > conns ) { } }

implementation: if ( null == conns || conns . isEmpty ( ) ) { return null ; } int size = conns . size ( ) ; int tries = 0 ; Connection result = null ; while ( ( result == null || ! result . isFine ( ) ) && tries ++ < MAX_TIMES ) { result = conns . get ( this . random . nextInt ( size ) ) ; } if ( result != null && ! result . isFine ( ) ) { result = null ; } return result ;


signature:
public class RpcResponseCommand { /** * Getter method for property < tt > customSerializer < / tt > .
 * @ return property value of customSerializer */
public CustomSerializer getCustomSerializer ( ) { } }

implementation: if ( this . customSerializer != null ) { return customSerializer ; } if ( this . responseClass != null ) { this . customSerializer = CustomSerializerManager . getCustomSerializer ( this . responseClass ) ; } if ( this . customSerializer == null ) { this . customSerializer = CustomSerializerManager . getCustomSerializer ( this . getCmdCode ( ) ) ; } return this . customSerializer ;


signature:
public class InvokeContext { /** * get
 * @ param key
 * @ return */
@ SuppressWarnings ( "unchecked" ) public < T > T get ( String key ) { } }

implementation: return ( T ) this . context . get ( key ) ;


signature:
public class InvokeContext { /** * get and use default if not found
 * @ param key
 * @ param defaultIfNotFound
 * @ param < T >
 * @ return */
@ SuppressWarnings ( "unchecked" ) public < T > T get ( String key , T defaultIfNotFound ) { } }

implementation: return this . context . get ( key ) != null ? ( T ) this . context . get ( key ) : defaultIfNotFound ;


signature:
public class Url { /** * Get property value according to property key
 * @ param key
 * @ return property value */
public String getProperty ( String key ) { } }

implementation: if ( properties == null ) { return null ; } return properties . getProperty ( key ) ;


signature:
public class CrcUtil { /** * Compute CRC32 code for byte [ ] .
 * @ param array
 * @ param offset
 * @ param length
 * @ return */
public static final int crc32 ( byte [ ] array , int offset , int length ) { } }

implementation: CRC32 crc32 = CRC_32_THREAD_LOCAL . get ( ) ; crc32 . update ( array , offset , length ) ; int ret = ( int ) crc32 . getValue ( ) ; crc32 . reset ( ) ; return ret ;


signature:
public class RpcResponseResolver { /** * Analyze the response command and generate the response object .
 * @ param responseCommand response command
 * @ param addr response address
 * @ return response object */
public static Object resolveResponseObject ( ResponseCommand responseCommand , String addr ) throws RemotingException { } }

implementation: preProcess ( responseCommand , addr ) ; if ( responseCommand . getResponseStatus ( ) == ResponseStatus . SUCCESS ) { return toResponseObject ( responseCommand ) ; } else { String msg = String . format ( "Rpc invocation exception: %s, the address is %s, id=%s" , responseCommand . getResponseStatus ( ) , addr , responseCommand . getId ( ) ) ; logger . warn ( msg ) ; if ( responseCommand . getCause ( ) != null ) { throw new InvokeException ( msg , responseCommand . getCause ( ) ) ; } else { throw new InvokeException ( msg + ", please check the server log for more." ) ; } }


signature:
public class RpcResponseResolver { /** * Convert remoting response command to application response object . */
private static Object toResponseObject ( ResponseCommand responseCommand ) throws CodecException { } }

implementation: RpcResponseCommand response = ( RpcResponseCommand ) responseCommand ; response . deserialize ( ) ; return response . getResponseObject ( ) ;


signature:
public class RpcResponseResolver { /** * Convert remoting response command to throwable if it is a throwable , otherwise return null . */
private static Throwable toThrowable ( ResponseCommand responseCommand ) throws CodecException { } }

implementation: RpcResponseCommand resp = ( RpcResponseCommand ) responseCommand ; resp . deserialize ( ) ; Object ex = resp . getResponseObject ( ) ; if ( ex != null && ex instanceof Throwable ) { return ( Throwable ) ex ; } return null ;


signature:
public class RpcResponseResolver { /** * Detail your error msg with the error msg returned from response command */
private static String detailErrMsg ( String clientErrMsg , ResponseCommand responseCommand ) { } }

implementation: RpcResponseCommand resp = ( RpcResponseCommand ) responseCommand ; if ( StringUtils . isNotBlank ( resp . getErrorMsg ( ) ) ) { return String . format ( "%s, ServerErrorMsg:%s" , clientErrMsg , resp . getErrorMsg ( ) ) ; } else { return String . format ( "%s, ServerErrorMsg:null" , clientErrMsg ) ; }


signature:
public class RpcCommandFactory { /** * create server exception using error msg and fill the stack trace using the stack trace of throwable .
 * @ param t the origin throwable to fill the stack trace of rpc server exception
 * @ param errMsg additional error msg , < code > null < / code > is allowed
 * @ return an instance of RpcServerException */
private RpcServerException createServerException ( Throwable t , String errMsg ) { } }

implementation: String formattedErrMsg = String . format ( "[Server]OriginErrorMsg: %s: %s. AdditionalErrorMsg: %s" , t . getClass ( ) . getName ( ) , t . getMessage ( ) , errMsg ) ; RpcServerException e = new RpcServerException ( formattedErrMsg ) ; e . setStackTrace ( t . getStackTrace ( ) ) ; return e ;


signature:
public class TraceLogUtil { /** * print trace log
 * @ param traceId
 * @ param invokeContext */
public static void printConnectionTraceLog ( Logger logger , String traceId , InvokeContext invokeContext ) { } }

implementation: String sourceIp = invokeContext . get ( InvokeContext . CLIENT_LOCAL_IP ) ; Integer sourcePort = invokeContext . get ( InvokeContext . CLIENT_LOCAL_PORT ) ; String targetIp = invokeContext . get ( InvokeContext . CLIENT_REMOTE_IP ) ; Integer targetPort = invokeContext . get ( InvokeContext . CLIENT_REMOTE_PORT ) ; StringBuilder logMsg = new StringBuilder ( ) ; logMsg . append ( traceId ) . append ( "," ) ; logMsg . append ( sourceIp ) . append ( "," ) ; logMsg . append ( sourcePort ) . append ( "," ) ; logMsg . append ( targetIp ) . append ( "," ) ; logMsg . append ( targetPort ) ; if ( logger . isInfoEnabled ( ) ) { logger . info ( logMsg . toString ( ) ) ; }


signature:
public class NettyEventLoopUtil { /** * Create the right event loop according to current platform and system property , fallback to NIO when epoll not enabled .
 * @ param nThreads
 * @ param threadFactory
 * @ return an EventLoopGroup suitable for the current platform */
public static EventLoopGroup newEventLoopGroup ( int nThreads , ThreadFactory threadFactory ) { } }

implementation: return epollEnabled ? new EpollEventLoopGroup ( nThreads , threadFactory ) : new NioEventLoopGroup ( nThreads , threadFactory ) ;


signature:
public class NettyEventLoopUtil { /** * Use { @ link EpollMode # LEVEL _ TRIGGERED } for server bootstrap if level trigger enabled by system properties ,
 * otherwise use { @ link EpollMode # EDGE _ TRIGGERED } .
 * @ param serverBootstrap server bootstrap */
public static void enableTriggeredMode ( ServerBootstrap serverBootstrap ) { } }

implementation: if ( epollEnabled ) { if ( ConfigManager . netty_epoll_lt_enabled ( ) ) { serverBootstrap . childOption ( EpollChannelOption . EPOLL_MODE , EpollMode . LEVEL_TRIGGERED ) ; } else { serverBootstrap . childOption ( EpollChannelOption . EPOLL_MODE , EpollMode . EDGE_TRIGGERED ) ; } }


signature:
public class RpcTaskScanner { /** * Start ! */
public void start ( ) { } }

implementation: scheduledService . scheduleWithFixedDelay ( new Runnable ( ) { @ Override public void run ( ) { for ( Scannable scanned : scanList ) { try { scanned . scan ( ) ; } catch ( Throwable t ) { logger . error ( "Exception caught when scannings." , t ) ; } } } } , 10000 , 10000 , TimeUnit . MILLISECONDS ) ;


signature:
public class RemotingUtil { /** * Parse the remote address of the channel .
 * @ param channel
 * @ return */
public static String parseRemoteAddress ( final Channel channel ) { } }

implementation: if ( null == channel ) { return StringUtils . EMPTY ; } final SocketAddress remote = channel . remoteAddress ( ) ; return doParse ( remote != null ? remote . toString ( ) . trim ( ) : StringUtils . EMPTY ) ;


signature:
public class RemotingUtil { /** * Parse the local address of the channel .
 * @ param channel
 * @ return */
public static String parseLocalAddress ( final Channel channel ) { } }

implementation: if ( null == channel ) { return StringUtils . EMPTY ; } final SocketAddress local = channel . localAddress ( ) ; return doParse ( local != null ? local . toString ( ) . trim ( ) : StringUtils . EMPTY ) ;


signature:
public class RemotingUtil { /** * Parse the remote host ip of the channel .
 * @ param channel
 * @ return */
public static String parseRemoteIP ( final Channel channel ) { } }

implementation: if ( null == channel ) { return StringUtils . EMPTY ; } final InetSocketAddress remote = ( InetSocketAddress ) channel . remoteAddress ( ) ; if ( remote != null ) { return remote . getAddress ( ) . getHostAddress ( ) ; } return StringUtils . EMPTY ;


signature:
public class RemotingUtil { /** * Parse the remote hostname of the channel .
 * Note : take care to use this method , for a reverse name lookup takes uncertain time in { @ link InetAddress # getHostName } .
 * @ param channel
 * @ return */
public static String parseRemoteHostName ( final Channel channel ) { } }

implementation: if ( null == channel ) { return StringUtils . EMPTY ; } final InetSocketAddress remote = ( InetSocketAddress ) channel . remoteAddress ( ) ; if ( remote != null ) { return remote . getAddress ( ) . getHostName ( ) ; } return StringUtils . EMPTY ;


signature:
public class RemotingUtil { /** * Parse the local host ip of the channel .
 * @ param channel
 * @ return */
public static String parseLocalIP ( final Channel channel ) { } }

implementation: if ( null == channel ) { return StringUtils . EMPTY ; } final InetSocketAddress local = ( InetSocketAddress ) channel . localAddress ( ) ; if ( local != null ) { return local . getAddress ( ) . getHostAddress ( ) ; } return StringUtils . EMPTY ;


signature:
public class RemotingUtil { /** * Parse the remote host port of the channel .
 * @ param channel
 * @ return int */
public static int parseRemotePort ( final Channel channel ) { } }

implementation: if ( null == channel ) { return - 1 ; } final InetSocketAddress remote = ( InetSocketAddress ) channel . remoteAddress ( ) ; if ( remote != null ) { return remote . getPort ( ) ; } return - 1 ;


signature:
public class RemotingUtil { /** * Parse the local host port of the channel .
 * @ param channel
 * @ return int */
public static int parseLocalPort ( final Channel channel ) { } }

implementation: if ( null == channel ) { return - 1 ; } final InetSocketAddress local = ( InetSocketAddress ) channel . localAddress ( ) ; if ( local != null ) { return local . getPort ( ) ; } return - 1 ;


signature:
public class RemotingUtil { /** * Parse the socket address , omit the leading " / " if present .
 * e . g . 1 / 127.0.0.1:1234 - > 127.0.0.1:1234
 * e . g . 2 sofatest - 2 . stack . alipay . net / 10.209.155.54:12200 - > 10.209.155.54:12200
 * @ param socketAddress
 * @ return String */
public static String parseSocketAddressToString ( SocketAddress socketAddress ) { } }

implementation: if ( socketAddress != null ) { return doParse ( socketAddress . toString ( ) . trim ( ) ) ; } return StringUtils . EMPTY ;


signature:
public class RemotingUtil { /** * Parse the host ip of socket address .
 * e . g . / 127.0.0.1:1234 - > 127.0.0.1
 * @ param socketAddress
 * @ return String */
public static String parseSocketAddressToHostIp ( SocketAddress socketAddress ) { } }

implementation: final InetSocketAddress addrs = ( InetSocketAddress ) socketAddress ; if ( addrs != null ) { InetAddress addr = addrs . getAddress ( ) ; if ( null != addr ) { return addr . getHostAddress ( ) ; } } return StringUtils . EMPTY ;


signature:
public class RemotingUtil { /** * < ol >
 * < li > if an address starts with a ' / ' , skip it .
 * < li > if an address contains a ' / ' , substring it .
 * < / ol >
 * @ param addr
 * @ return */
private static String doParse ( String addr ) { } }

implementation: if ( StringUtils . isBlank ( addr ) ) { return StringUtils . EMPTY ; } if ( addr . charAt ( 0 ) == '/' ) { return addr . substring ( 1 ) ; } else { int len = addr . length ( ) ; for ( int i = 1 ; i < len ; ++ i ) { if ( addr . charAt ( i ) == '/' ) { return addr . substring ( i + 1 ) ; } } return addr ; }


signature:
public class ConnectionPool { /** * add a connection
 * @ param connection */
public void add ( Connection connection ) { } }

implementation: markAccess ( ) ; if ( null == connection ) { return ; } boolean res = this . conns . addIfAbsent ( connection ) ; if ( res ) { connection . increaseRef ( ) ; }


signature:
public class ConnectionPool { /** * removeAndTryClose a connection
 * @ param connection */
public void removeAndTryClose ( Connection connection ) { } }

implementation: if ( null == connection ) { return ; } boolean res = this . conns . remove ( connection ) ; if ( res ) { connection . decreaseRef ( ) ; } if ( connection . noRef ( ) ) { connection . close ( ) ; }


signature:
public class ConnectionPool { /** * get a connection
 * @ return */
public Connection get ( ) { } }

implementation: markAccess ( ) ; if ( null != this . conns ) { List < Connection > snapshot = new ArrayList < Connection > ( this . conns ) ; if ( snapshot . size ( ) > 0 ) { return this . strategy . select ( snapshot ) ; } else { return null ; } } else { return null ; }


signature:
public class CustomSerializerManager { /** * Register custom serializer for class name .
 * @ param className
 * @ param serializer
 * @ return */
public static void registerCustomSerializer ( String className , CustomSerializer serializer ) { } }

implementation: CustomSerializer prevSerializer = classCustomSerializer . putIfAbsent ( className , serializer ) ; if ( prevSerializer != null ) { throw new RuntimeException ( "CustomSerializer has been registered for class: " + className + ", the custom serializer is: " + prevSerializer . getClass ( ) . getName ( ) ) ; }


signature:
public class CustomSerializerManager { /** * Get the custom serializer for class name .
 * @ param className
 * @ return */
public static CustomSerializer getCustomSerializer ( String className ) { } }

implementation: if ( ! classCustomSerializer . isEmpty ( ) ) { return classCustomSerializer . get ( className ) ; } return null ;


signature:
public class CustomSerializerManager { /** * Register custom serializer for command code .
 * @ param code
 * @ param serializer
 * @ return */
public static void registerCustomSerializer ( CommandCode code , CustomSerializer serializer ) { } }

implementation: CustomSerializer prevSerializer = commandCustomSerializer . putIfAbsent ( code , serializer ) ; if ( prevSerializer != null ) { throw new RuntimeException ( "CustomSerializer has been registered for command code: " + code + ", the custom serializer is: " + prevSerializer . getClass ( ) . getName ( ) ) ; }


signature:
public class CustomSerializerManager { /** * Get the custom serializer for command code .
 * @ param code
 * @ return */
public static CustomSerializer getCustomSerializer ( CommandCode code ) { } }

implementation: if ( ! commandCustomSerializer . isEmpty ( ) ) { return commandCustomSerializer . get ( code ) ; } return null ;


signature:
public class RpcServer { /** * Notice : only { @ link GlobalSwitch # SERVER _ MANAGE _ CONNECTION _ SWITCH } switch on , will close all connections .
 * @ see AbstractRemotingServer # doStop ( ) */
@ Override protected boolean doStop ( ) { } }

implementation: if ( null != this . channelFuture ) { this . channelFuture . channel ( ) . close ( ) ; } if ( this . switches ( ) . isOn ( GlobalSwitch . SERVER_SYNC_STOP ) ) { this . bossGroup . shutdownGracefully ( ) . awaitUninterruptibly ( ) ; } else { this . bossGroup . shutdownGracefully ( ) ; } if ( this . switches ( ) . isOn ( GlobalSwitch . SERVER_MANAGE_CONNECTION_SWITCH ) && null != this . connectionManager ) { this . connectionManager . removeAll ( ) ; logger . warn ( "Close all connections from server side!" ) ; } logger . warn ( "Rpc Server stopped!" ) ; return true ;


signature:
public class RpcServer { /** * One way invocation with a { @ link InvokeContext } , common api notice please see { @ link # oneway ( String , Object ) }
 * @ param addr
 * @ param request
 * @ param invokeContext
 * @ throws RemotingException
 * @ throws InterruptedException */
public void oneway ( final String addr , final Object request , final InvokeContext invokeContext ) throws RemotingException , InterruptedException { } }

implementation: check ( ) ; this . rpcRemoting . oneway ( addr , request , invokeContext ) ;


signature:
public class RpcServer { /** * One way invocation using a parsed { @ link Url } < br >
 * Notice : < br >
 * < ol >
 * < li > < b > DO NOT modify the request object concurrently when this method is called . < / b > < / li >
 * < li > When do invocation , use the parsed { @ link Url } to find a available client connection , if none then throw exception < / li >
 * < / ol >
 * @ param url
 * @ param request
 * @ throws RemotingException
 * @ throws InterruptedException */
public void oneway ( final Url url , final Object request ) throws RemotingException , InterruptedException { } }

implementation: check ( ) ; this . rpcRemoting . oneway ( url , request , null ) ;


signature:
public class RpcServer { /** * One way invocation with a { @ link InvokeContext } , common api notice please see { @ link # oneway ( Url , Object ) }
 * @ param url
 * @ param request
 * @ param invokeContext
 * @ throws RemotingException
 * @ throws InterruptedException */
public void oneway ( final Url url , final Object request , final InvokeContext invokeContext ) throws RemotingException , InterruptedException { } }

implementation: check ( ) ; this . rpcRemoting . oneway ( url , request , invokeContext ) ;


signature:
public class RpcServer { /** * Future invocation using a string address , address format example - 127.0.0.1:12200 ? key1 = value1 & key2 = value2 < br >
 * You can get result use the returned { @ link RpcResponseFuture } .
 * Notice : < br >
 * < ol >
 * < li > < b > DO NOT modify the request object concurrently when this method is called . < / b > < / li >
 * < li > When do invocation , use the string address to find a available client connection , if none then throw exception < / li >
 * < li > Unlike rpc client , address arguments takes no effect here , for rpc server will not create connection . < / li >
 * < / ol >
 * @ param addr
 * @ param request
 * @ param timeoutMillis
 * @ return RpcResponseFuture
 * @ throws RemotingException
 * @ throws InterruptedException */
public RpcResponseFuture invokeWithFuture ( final String addr , final Object request , final int timeoutMillis ) throws RemotingException , InterruptedException { } }

implementation: check ( ) ; return this . rpcRemoting . invokeWithFuture ( addr , request , null , timeoutMillis ) ;


signature:
public class RpcServer { /** * Future invocation with a { @ link InvokeContext } , common api notice please see { @ link # invokeWithFuture ( String , Object , int ) }
 * @ param addr
 * @ param request
 * @ param invokeContext
 * @ param timeoutMillis
 * @ return
 * @ throws RemotingException
 * @ throws InterruptedException */
public RpcResponseFuture invokeWithFuture ( final String addr , final Object request , final InvokeContext invokeContext , final int timeoutMillis ) throws RemotingException , InterruptedException { } }

implementation: check ( ) ; return this . rpcRemoting . invokeWithFuture ( addr , request , invokeContext , timeoutMillis ) ;


signature:
public class RpcServer { /** * Future invocation using a parsed { @ link Url } < br >
 * You can get result use the returned { @ link RpcResponseFuture } .
 * Notice : < br >
 * < ol >
 * < li > < b > DO NOT modify the request object concurrently when this method is called . < / b > < / li >
 * < li > When do invocation , use the parsed { @ link Url } to find a available client connection , if none then throw exception < / li >
 * < / ol >
 * @ param url
 * @ param request
 * @ param timeoutMillis
 * @ return RpcResponseFuture
 * @ throws RemotingException
 * @ throws InterruptedException */
public RpcResponseFuture invokeWithFuture ( final Url url , final Object request , final int timeoutMillis ) throws RemotingException , InterruptedException { } }

implementation: check ( ) ; return this . rpcRemoting . invokeWithFuture ( url , request , null , timeoutMillis ) ;


signature:
public class RpcServer { /** * Future invocation with a { @ link InvokeContext } , common api notice please see { @ link # invokeWithFuture ( Url , Object , int ) }
 * @ param url
 * @ param request
 * @ param invokeContext
 * @ param timeoutMillis
 * @ return
 * @ throws RemotingException
 * @ throws InterruptedException */
public RpcResponseFuture invokeWithFuture ( final Url url , final Object request , final InvokeContext invokeContext , final int timeoutMillis ) throws RemotingException , InterruptedException { } }

implementation: check ( ) ; return this . rpcRemoting . invokeWithFuture ( url , request , invokeContext , timeoutMillis ) ;