signature:
public class Configuration { /** * Validate that the timeouts for unblocking publications from a client are valid .
 * @ param publicationUnblockTimeoutNs after which an uncommitted publication will be unblocked .
 * @ param clientLivenessTimeoutNs after which a client will be considered not alive .
 * @ param timerIntervalNs interval at which the driver will check timeouts .
 * @ throws ConfigurationException if the values are not valid . */
public static void validateUnblockTimeout ( final long publicationUnblockTimeoutNs , final long clientLivenessTimeoutNs , final long timerIntervalNs ) { } }

implementation: if ( publicationUnblockTimeoutNs <= clientLivenessTimeoutNs ) { throw new ConfigurationException ( "publicationUnblockTimeoutNs=" + publicationUnblockTimeoutNs + " <= clientLivenessTimeoutNs=" + clientLivenessTimeoutNs ) ; } if ( clientLivenessTimeoutNs <= timerIntervalNs ) { throw new ConfigurationException ( "clientLivenessTimeoutNs=" + clientLivenessTimeoutNs + " <= timerIntervalNs=" + timerIntervalNs ) ; }


signature:
public class ErrorResponseFlyweight { /** * Set the error code for the command .
 * @ param code for the error
 * @ return flyweight */
public ErrorResponseFlyweight errorCode ( final ErrorCode code ) { } }

implementation: buffer . putInt ( offset + ERROR_CODE_OFFSET , code . value ( ) ) ; return this ;


signature:
public class EventConfiguration { /** * Get the { @ link Set } of { @ link ClusterEventCode } s that are enabled for the logger .
 * @ param enabledClusterEventCodes that can be " all " or a comma separated list of Event Code ids or names .
 * @ return the { @ link Set } of { @ link ClusterEventCode } s that are enabled for the logger . */
static Set < ClusterEventCode > getEnabledClusterEventCodes ( final String enabledClusterEventCodes ) { } }

implementation: if ( null == enabledClusterEventCodes || "" . equals ( enabledClusterEventCodes ) ) { return EnumSet . noneOf ( ClusterEventCode . class ) ; } final Function < Integer , ClusterEventCode > eventCodeById = ClusterEventCode :: get ; final Function < String , ClusterEventCode > eventCodeByName = ClusterEventCode :: valueOf ; final EnumSet < ClusterEventCode > allEventsSet = EnumSet . allOf ( ClusterEventCode . class ) ; return parseEventCodes ( enabledClusterEventCodes , eventCodeById , eventCodeByName , allEventsSet ) ;


signature:
public class EventConfiguration { /** * Get the { @ link Set } of { @ link ArchiveEventCode } s that are enabled for the logger .
 * @ param enabledArchiveEventCodes that can be " all " or a comma separated list of Event Code ids or names .
 * @ return the { @ link Set } of { @ link ArchiveEventCode } s that are enabled for the logger . */
static Set < ArchiveEventCode > getEnabledArchiveEventCodes ( final String enabledArchiveEventCodes ) { } }

implementation: if ( null == enabledArchiveEventCodes || "" . equals ( enabledArchiveEventCodes ) ) { return EnumSet . noneOf ( ArchiveEventCode . class ) ; } final Function < Integer , ArchiveEventCode > eventCodeById = ArchiveEventCode :: get ; final Function < String , ArchiveEventCode > eventCodeByName = ArchiveEventCode :: valueOf ; final EnumSet < ArchiveEventCode > allEventsSet = EnumSet . allOf ( ArchiveEventCode . class ) ; return parseEventCodes ( enabledArchiveEventCodes , eventCodeById , eventCodeByName , allEventsSet ) ;


signature:
public class EventConfiguration { /** * Get the { @ link Set } of { @ link DriverEventCode } s that are enabled for the logger .
 * @ param enabledLoggerEventCodes that can be " all " , " admin " , or a comma separated list of Event Code ids or names .
 * @ return the { @ link Set } of { @ link DriverEventCode } s that are enabled for the logger . */
static Set < DriverEventCode > getEnabledDriverEventCodes ( final String enabledLoggerEventCodes ) { } }

implementation: if ( null == enabledLoggerEventCodes || "" . equals ( enabledLoggerEventCodes ) ) { return EnumSet . noneOf ( DriverEventCode . class ) ; } final Set < DriverEventCode > eventCodeSet = new HashSet < > ( ) ; final String [ ] codeIds = enabledLoggerEventCodes . split ( "," ) ; for ( final String codeId : codeIds ) { switch ( codeId ) { case "all" : eventCodeSet . addAll ( ALL_LOGGER_EVENT_CODES ) ; break ; case "admin" : eventCodeSet . addAll ( ADMIN_ONLY_EVENT_CODES ) ; break ; default : { DriverEventCode code = null ; try { code = DriverEventCode . valueOf ( codeId ) ; } catch ( final IllegalArgumentException ignore ) { } if ( null == code ) { try { code = DriverEventCode . get ( Integer . parseInt ( codeId ) ) ; } catch ( final IllegalArgumentException ignore ) { } } if ( null != code ) { eventCodeSet . add ( code ) ; } else { System . err . println ( "unknown event code: " + codeId ) ; } } } } return eventCodeSet ;


signature:
public class ClusterMember { /** * Reset the state of a cluster member so it can be canvassed and reestablished . */
public void reset ( ) { } }

implementation: isBallotSent = false ; isLeader = false ; hasRequestedJoin = false ; hasSentTerminationAck = false ; vote = null ; candidateTermId = Aeron . NULL_VALUE ; leadershipTermId = Aeron . NULL_VALUE ; logPosition = NULL_POSITION ;


signature:
public class ClusterMember { /** * Parse the details for a cluster members from a string .
 * < code >
 * member - id , client - facing : port , member - facing : port , log : port , transfer : port , archive : port | 1 , . . .
 * < / code >
 * @ param value of the string to be parsed .
 * @ return An array of cluster members . */
public static ClusterMember [ ] parse ( final String value ) { } }

implementation: if ( null == value || value . length ( ) == 0 ) { return ClusterMember . EMPTY_CLUSTER_MEMBER_ARRAY ; } final String [ ] memberValues = value . split ( "\\|" ) ; final int length = memberValues . length ; final ClusterMember [ ] members = new ClusterMember [ length ] ; for ( int i = 0 ; i < length ; i ++ ) { final String endpointsDetail = memberValues [ i ] ; final String [ ] memberAttributes = endpointsDetail . split ( "," ) ; if ( memberAttributes . length != 6 ) { throw new ClusterException ( "invalid member value: " + endpointsDetail + " within: " + value ) ; } final String justEndpoints = String . join ( "," , memberAttributes [ 1 ] , memberAttributes [ 2 ] , memberAttributes [ 3 ] , memberAttributes [ 4 ] , memberAttributes [ 5 ] ) ; members [ i ] = new ClusterMember ( Integer . parseInt ( memberAttributes [ 0 ] ) , memberAttributes [ 1 ] , memberAttributes [ 2 ] , memberAttributes [ 3 ] , memberAttributes [ 4 ] , memberAttributes [ 5 ] , justEndpoints ) ; } return members ;


signature:
public class ClusterMember { /** * Encode member details from a cluster members array to a string .
 * @ param clusterMembers to fill the details from
 * @ return String representation suitable for use with { @ link ClusterMember # parse } */
public static String encodeAsString ( final ClusterMember [ ] clusterMembers ) { } }

implementation: final StringBuilder builder = new StringBuilder ( ) ; for ( int i = 0 , length = clusterMembers . length ; i < length ; i ++ ) { final ClusterMember member = clusterMembers [ i ] ; builder . append ( member . id ( ) ) . append ( ',' ) . append ( member . endpointsDetail ( ) ) ; if ( ( length - 1 ) != i ) { builder . append ( '|' ) ; } } return builder . toString ( ) ;


signature:
public class ClusterMember { /** * Add the publications for sending status messages to the other members of the cluster .
 * @ param members of the cluster .
 * @ param exclude this member when adding publications .
 * @ param channelUri for the publication .
 * @ param streamId for the publication .
 * @ param aeron to add the publications to . */
public static void addMemberStatusPublications ( final ClusterMember [ ] members , final ClusterMember exclude , final ChannelUri channelUri , final int streamId , final Aeron aeron ) { } }

implementation: for ( final ClusterMember member : members ) { if ( member != exclude ) { channelUri . put ( ENDPOINT_PARAM_NAME , member . memberFacingEndpoint ( ) ) ; member . publication = aeron . addExclusivePublication ( channelUri . toString ( ) , streamId ) ; } }


signature:
public class ClusterMember { /** * Close the publications associated with members of the cluster .
 * @ param clusterMembers to close the publications for . */
public static void closeMemberPublications ( final ClusterMember [ ] clusterMembers ) { } }

implementation: for ( final ClusterMember member : clusterMembers ) { CloseHelper . close ( member . publication ) ; }


signature:
public class ClusterMember { /** * Add an exclusive { @ link Publication } for communicating to a member on the member status channel .
 * @ param member to which the publication is addressed .
 * @ param channelUri for the target member .
 * @ param streamId for the target member .
 * @ param aeron from which the publication will be created . */
public static void addMemberStatusPublication ( final ClusterMember member , final ChannelUri channelUri , final int streamId , final Aeron aeron ) { } }

implementation: channelUri . put ( ENDPOINT_PARAM_NAME , member . memberFacingEndpoint ( ) ) ; member . publication = aeron . addExclusivePublication ( channelUri . toString ( ) , streamId ) ;


signature:
public class ClusterMember { /** * Populate map of { @ link ClusterMember } s which can be looked up by id .
 * @ param clusterMembers to populate the map .
 * @ param clusterMemberByIdMap to be populated . */
public static void addClusterMemberIds ( final ClusterMember [ ] clusterMembers , final Int2ObjectHashMap < ClusterMember > clusterMemberByIdMap ) { } }

implementation: for ( final ClusterMember member : clusterMembers ) { clusterMemberByIdMap . put ( member . id ( ) , member ) ; }


signature:
public class ClusterMember { /** * Check if the cluster leader has an active quorum of cluster followers .
 * @ param clusterMembers for the current cluster .
 * @ param nowMs for the current time .
 * @ param timeoutMs after which a follower is not considered active .
 * @ return true if quorum of cluster members are considered active . */
public static boolean hasActiveQuorum ( final ClusterMember [ ] clusterMembers , final long nowMs , final long timeoutMs ) { } }

implementation: int threshold = quorumThreshold ( clusterMembers . length ) ; for ( final ClusterMember member : clusterMembers ) { if ( member . isLeader ( ) || nowMs <= ( member . timeOfLastAppendPositionMs ( ) + timeoutMs ) ) { if ( -- threshold <= 0 ) { return true ; } } } return false ;


signature:
public class ClusterMember { /** * Calculate the position reached by a quorum of cluster members .
 * @ param members of the cluster .
 * @ param rankedPositions temp array to be used for sorting the positions to avoid allocation .
 * @ return the position reached by a quorum of cluster members . */
public static long quorumPosition ( final ClusterMember [ ] members , final long [ ] rankedPositions ) { } }

implementation: final int length = rankedPositions . length ; for ( int i = 0 ; i < length ; i ++ ) { rankedPositions [ i ] = 0 ; } for ( final ClusterMember member : members ) { long newPosition = member . logPosition ; for ( int i = 0 ; i < length ; i ++ ) { final long rankedPosition = rankedPositions [ i ] ; if ( newPosition > rankedPosition ) { rankedPositions [ i ] = newPosition ; newPosition = rankedPosition ; } } } return rankedPositions [ length - 1 ] ;


signature:
public class ClusterMember { /** * Reset the log position of all the members to the provided value .
 * @ param clusterMembers to be reset .
 * @ param logPosition to set for them all . */
public static void resetLogPositions ( final ClusterMember [ ] clusterMembers , final long logPosition ) { } }

implementation: for ( final ClusterMember member : clusterMembers ) { member . logPosition ( logPosition ) ; }


signature:
public class ClusterMember { /** * Has the members of the cluster the voted reached the provided position in their log .
 * @ param clusterMembers to check .
 * @ param position to compare the { @ link # logPosition ( ) } against .
 * @ param leadershipTermId expected of the members .
 * @ return true if all members have reached this position otherwise false . */
public static boolean haveVotersReachedPosition ( final ClusterMember [ ] clusterMembers , final long position , final long leadershipTermId ) { } }

implementation: for ( final ClusterMember member : clusterMembers ) { if ( member . vote != null && ( member . logPosition < position || member . leadershipTermId != leadershipTermId ) ) { return false ; } } return true ;


signature:
public class ClusterMember { /** * Become a candidate by voting for yourself and resetting the other votes to { @ link Aeron # NULL _ VALUE } .
 * @ param members to reset the votes for .
 * @ param candidateTermId for the candidacy .
 * @ param candidateMemberId for the election . */
public static void becomeCandidate ( final ClusterMember [ ] members , final long candidateTermId , final int candidateMemberId ) { } }

implementation: for ( final ClusterMember member : members ) { if ( member . id == candidateMemberId ) { member . vote ( Boolean . TRUE ) . candidateTermId ( candidateTermId ) . isBallotSent ( true ) ; } else { member . vote ( null ) . candidateTermId ( Aeron . NULL_VALUE ) . isBallotSent ( false ) ; } }


signature:
public class ClusterMember { /** * Has the candidate got unanimous support of the cluster ?
 * @ param members to check for votes .
 * @ param candidateTermId for the vote .
 * @ return false if any member has not voted for the candidate . */
public static boolean hasWonVoteOnFullCount ( final ClusterMember [ ] members , final long candidateTermId ) { } }

implementation: int votes = 0 ; for ( final ClusterMember member : members ) { if ( null == member . vote || member . candidateTermId != candidateTermId ) { return false ; } votes += member . vote ? 1 : 0 ; } return votes >= ClusterMember . quorumThreshold ( members . length ) ;


signature:
public class ClusterMember { /** * Has sufficient votes being counted for a majority for all members observed during { @ link Election . State # CANVASS } ?
 * @ param members to check for votes .
 * @ param candidateTermId for the vote .
 * @ return false if any member has not voted for the candidate . */
public static boolean hasMajorityVoteWithCanvassMembers ( final ClusterMember [ ] members , final long candidateTermId ) { } }

implementation: int votes = 0 ; for ( final ClusterMember member : members ) { if ( NULL_POSITION != member . logPosition && null == member . vote ) { return false ; } if ( Boolean . TRUE . equals ( member . vote ) && member . candidateTermId == candidateTermId ) { ++ votes ; } } return votes >= ClusterMember . quorumThreshold ( members . length ) ;


signature:
public class ClusterMember { /** * Has sufficient votes being counted for a majority ?
 * @ param clusterMembers to check for votes .
 * @ param candidateTermId for the vote .
 * @ return true if a majority of positive votes . */
public static boolean hasMajorityVote ( final ClusterMember [ ] clusterMembers , final long candidateTermId ) { } }

implementation: int votes = 0 ; for ( final ClusterMember member : clusterMembers ) { if ( Boolean . TRUE . equals ( member . vote ) && member . candidateTermId == candidateTermId ) { ++ votes ; } } return votes >= ClusterMember . quorumThreshold ( clusterMembers . length ) ;


signature:
public class ClusterMember { /** * Determine which member of a cluster this is and check endpoints .
 * @ param clusterMembers for the current cluster which can be null .
 * @ param memberId for this member .
 * @ param memberEndpoints for this member .
 * @ return the { @ link ClusterMember } determined . */
public static ClusterMember determineMember ( final ClusterMember [ ] clusterMembers , final int memberId , final String memberEndpoints ) { } }

implementation: ClusterMember member = NULL_VALUE != memberId ? ClusterMember . findMember ( clusterMembers , memberId ) : null ; if ( ( null == clusterMembers || 0 == clusterMembers . length ) && null == member ) { member = ClusterMember . parseEndpoints ( NULL_VALUE , memberEndpoints ) ; } else { if ( null == member ) { throw new ClusterException ( "memberId=" + memberId + " not found in clusterMembers" ) ; } if ( ! "" . equals ( memberEndpoints ) ) { ClusterMember . validateMemberEndpoints ( member , memberEndpoints ) ; } } return member ;


signature:
public class ClusterMember { /** * Check the member with the memberEndpoints
 * @ param member to check memberEndpoints against
 * @ param memberEndpoints to check member against
 * @ see ConsensusModule . Context # memberEndpoints ( )
 * @ see ConsensusModule . Context # clusterMembers ( ) */
public static void validateMemberEndpoints ( final ClusterMember member , final String memberEndpoints ) { } }

implementation: final ClusterMember endpointMember = ClusterMember . parseEndpoints ( Aeron . NULL_VALUE , memberEndpoints ) ; if ( ! areSameEndpoints ( member , endpointMember ) ) { throw new ClusterException ( "clusterMembers and memberEndpoints differ: " + member . endpointsDetail ( ) + " != " + memberEndpoints ) ; }


signature:
public class ClusterMember { /** * Are two cluster members using the same endpoints ?
 * @ param lhs to check
 * @ param rhs to check
 * @ return true if both are using the same endpoints or false if not . */
public static boolean areSameEndpoints ( final ClusterMember lhs , final ClusterMember rhs ) { } }

implementation: return lhs . clientFacingEndpoint ( ) . equals ( rhs . clientFacingEndpoint ( ) ) && lhs . memberFacingEndpoint ( ) . equals ( rhs . memberFacingEndpoint ( ) ) && lhs . logEndpoint ( ) . equals ( rhs . logEndpoint ( ) ) && lhs . transferEndpoint ( ) . equals ( rhs . transferEndpoint ( ) ) && lhs . archiveEndpoint ( ) . equals ( rhs . archiveEndpoint ( ) ) ;


signature:
public class ClusterMember { /** * Has the member achieved a unanimous view to be a suitable candidate in an election .
 * @ param clusterMembers to compare the candidate against .
 * @ param candidate for leadership .
 * @ return true if the candidate is suitable otherwise false . */
public static boolean isUnanimousCandidate ( final ClusterMember [ ] clusterMembers , final ClusterMember candidate ) { } }

implementation: for ( final ClusterMember member : clusterMembers ) { if ( NULL_POSITION == member . logPosition || compareLog ( candidate , member ) < 0 ) { return false ; } } return true ;


signature:
public class ClusterMember { /** * Has the member achieved a quorum view to be a suitable candidate in an election .
 * @ param clusterMembers to compare the candidate against .
 * @ param candidate for leadership .
 * @ return true if the candidate is suitable otherwise false . */
public static boolean isQuorumCandidate ( final ClusterMember [ ] clusterMembers , final ClusterMember candidate ) { } }

implementation: int possibleVotes = 0 ; for ( final ClusterMember member : clusterMembers ) { if ( NULL_POSITION == member . logPosition || compareLog ( candidate , member ) < 0 ) { continue ; } ++ possibleVotes ; } return possibleVotes >= ClusterMember . quorumThreshold ( clusterMembers . length ) ;


signature:
public class ClusterMember { /** * The result is positive if lhs has the more recent log , zero if logs are equal , and negative if rhs has the more
 * recent log .
 * @ param lhsLogLeadershipTermId term for which the position is most recent .
 * @ param lhsLogPosition reached in the provided term .
 * @ param rhsLogLeadershipTermId term for which the position is most recent .
 * @ param rhsLogPosition reached in the provided term .
 * @ return positive if lhs has the more recent log , zero if logs are equal , and negative if rhs has the more
 * recent log . */
public static int compareLog ( final long lhsLogLeadershipTermId , final long lhsLogPosition , final long rhsLogLeadershipTermId , final long rhsLogPosition ) { } }

implementation: if ( lhsLogLeadershipTermId > rhsLogLeadershipTermId ) { return 1 ; } else if ( lhsLogLeadershipTermId < rhsLogLeadershipTermId ) { return - 1 ; } else if ( lhsLogPosition > rhsLogPosition ) { return 1 ; } else if ( lhsLogPosition < rhsLogPosition ) { return - 1 ; } return 0 ;


signature:
public class ClusterMember { /** * The result is positive if lhs has the more recent log , zero if logs are equal , and negative if rhs has the more
 * recent log .
 * @ param lhs member to compare .
 * @ param rhs member to compare .
 * @ return positive if lhs has the more recent log , zero if logs are equal , and negative if rhs has the more
 * recent log . */
public static int compareLog ( final ClusterMember lhs , final ClusterMember rhs ) { } }

implementation: return compareLog ( lhs . leadershipTermId , lhs . logPosition , rhs . leadershipTermId , rhs . logPosition ) ;


signature:
public class ClusterMember { /** * Is the string of member endpoints not duplicated in the members .
 * @ param members to check if the provided endpoints have a duplicate .
 * @ param memberEndpoints to check for duplicates .
 * @ return true if no duplicate is found otherwise false . */
public static boolean isNotDuplicateEndpoints ( final ClusterMember [ ] members , final String memberEndpoints ) { } }

implementation: for ( final ClusterMember member : members ) { if ( member . endpointsDetail ( ) . equals ( memberEndpoints ) ) { return false ; } } return true ;


signature:
public class ClusterMember { /** * Find the index at which a member id is present .
 * @ param clusterMembers to be searched .
 * @ param memberId to search for .
 * @ return the index at which the member id is found otherwise { @ link ArrayUtil # UNKNOWN _ INDEX } . */
public static int findMemberIndex ( final ClusterMember [ ] clusterMembers , final int memberId ) { } }

implementation: final int length = clusterMembers . length ; int index = ArrayUtil . UNKNOWN_INDEX ; for ( int i = 0 ; i < length ; i ++ ) { if ( clusterMembers [ i ] . id ( ) == memberId ) { index = i ; } } return index ;


signature:
public class ClusterMember { /** * Find a { @ link ClusterMember } with a given id .
 * @ param clusterMembers to search .
 * @ param memberId to search for .
 * @ return the { @ link ClusterMember } if found otherwise null . */
public static ClusterMember findMember ( final ClusterMember [ ] clusterMembers , final int memberId ) { } }

implementation: for ( final ClusterMember member : clusterMembers ) { if ( member . id ( ) == memberId ) { return member ; } } return null ;


signature:
public class ClusterMember { /** * Add a new member to an array of { @ link ClusterMember } s .
 * @ param oldMembers to add to .
 * @ param newMember to add .
 * @ return a new array containing the old members plus the new member . */
public static ClusterMember [ ] addMember ( final ClusterMember [ ] oldMembers , final ClusterMember newMember ) { } }

implementation: return ArrayUtil . add ( oldMembers , newMember ) ;


signature:
public class ClusterMember { /** * Remove a member from an array if found , otherwise return the array unmodified .
 * @ param oldMembers to remove a member from .
 * @ param memberId of the member to remove .
 * @ return a new array with the member removed or the existing array if not found . */
public static ClusterMember [ ] removeMember ( final ClusterMember [ ] oldMembers , final int memberId ) { } }

implementation: return ArrayUtil . remove ( oldMembers , findMemberIndex ( oldMembers , memberId ) ) ;


signature:
public class ClusterMember { /** * Find the highest member id in an array of members .
 * @ param clusterMembers to search for the highest id .
 * @ return the highest id otherwise { @ link Aeron # NULL _ VALUE } if empty . */
public static int highMemberId ( final ClusterMember [ ] clusterMembers ) { } }

implementation: int highId = Aeron . NULL_VALUE ; for ( final ClusterMember member : clusterMembers ) { highId = Math . max ( highId , member . id ( ) ) ; } return highId ;


signature:
public class ClusterMember { /** * Create a string of member facing endpoints by id in format { @ code id = endpoint , id = endpoint , . . . } .
 * @ param members for which the endpoints string will be generated .
 * @ return a string of member facing endpoints by id . */
public static String clientFacingEndpoints ( final ClusterMember [ ] members ) { } }

implementation: final StringBuilder builder = new StringBuilder ( 100 ) ; for ( int i = 0 , length = members . length ; i < length ; i ++ ) { if ( 0 != i ) { builder . append ( ',' ) ; } final ClusterMember member = members [ i ] ; builder . append ( member . id ( ) ) . append ( '=' ) . append ( member . clientFacingEndpoint ( ) ) ; } return builder . toString ( ) ;


signature:
public class MinMulticastFlowControl { /** * { @ inheritDoc } */
public long onIdle ( final long timeNs , final long senderLimit , final long senderPosition , final boolean isEos ) { } }

implementation: long minPosition = Long . MAX_VALUE ; long minLimitPosition = Long . MAX_VALUE ; final ArrayList < Receiver > receiverList = this . receiverList ; for ( int lastIndex = receiverList . size ( ) - 1 , i = lastIndex ; i >= 0 ; i -- ) { final Receiver receiver = receiverList . get ( i ) ; if ( ( receiver . timeOfLastStatusMessageNs + RECEIVER_TIMEOUT ) - timeNs < 0 ) { ArrayListUtil . fastUnorderedRemove ( receiverList , i , lastIndex -- ) ; } else { minPosition = Math . min ( minPosition , receiver . lastPosition ) ; minLimitPosition = Math . min ( minLimitPosition , receiver . lastPositionPlusWindow ) ; } } if ( isEos && shouldLinger ) { if ( 0 == receiverList . size ( ) || minPosition >= senderPosition ) { shouldLinger = false ; } } return receiverList . size ( ) > 0 ? minLimitPosition : senderLimit ;


signature:
public class TermUnblocker { /** * Attempt to unblock the current term at the current offset .
 * < ol >
 * < li > Current position length is & gt ; 0 , then return < / li >
 * < li > Current position length is 0 , scan forward by frame alignment until , one of the following :
 * < ol >
 * < li > reach a non - 0 length , unblock up to indicated position ( check original frame length for non - 0 ) < / li >
 * < li > reach end of term and tail position & gt ; = end of term , unblock up to end of term ( check original
 * frame length for non - 0)
 * < / li >
 * < li > reach tail position & lt ; end of term , do NOT unblock < / li >
 * < / ol >
 * < / li >
 * < / ol >
 * @ param logMetaDataBuffer containing the default headers
 * @ param termBuffer to unblock
 * @ param blockedOffset to unblock at
 * @ param tailOffset to unblock up to
 * @ param termId for the current term .
 * @ return whether unblocking was done , not done , or applied to end of term */
public static Status unblock ( final UnsafeBuffer logMetaDataBuffer , final UnsafeBuffer termBuffer , final int blockedOffset , final int tailOffset , final int termId ) { } }

implementation: Status status = NO_ACTION ; int frameLength = frameLengthVolatile ( termBuffer , blockedOffset ) ; if ( frameLength < 0 ) { resetHeader ( logMetaDataBuffer , termBuffer , blockedOffset , termId , - frameLength ) ; status = UNBLOCKED ; } else if ( 0 == frameLength ) { int currentOffset = blockedOffset + FRAME_ALIGNMENT ; while ( currentOffset < tailOffset ) { frameLength = frameLengthVolatile ( termBuffer , currentOffset ) ; if ( frameLength != 0 ) { if ( scanBackToConfirmZeroed ( termBuffer , currentOffset , blockedOffset ) ) { final int length = currentOffset - blockedOffset ; resetHeader ( logMetaDataBuffer , termBuffer , blockedOffset , termId , length ) ; status = UNBLOCKED ; } break ; } currentOffset += FRAME_ALIGNMENT ; } if ( currentOffset == termBuffer . capacity ( ) ) { if ( 0 == frameLengthVolatile ( termBuffer , blockedOffset ) ) { final int length = currentOffset - blockedOffset ; resetHeader ( logMetaDataBuffer , termBuffer , blockedOffset , termId , length ) ; status = UNBLOCKED_TO_END ; } } } return status ;


signature:
public class CommonContext { /** * Map the CnC file if it exists .
 * @ param logger for feedback
 * @ return a new mapping for the file if it exists otherwise null ; */
public MappedByteBuffer mapExistingCncFile ( final Consumer < String > logger ) { } }

implementation: final File cncFile = new File ( aeronDirectory , CncFileDescriptor . CNC_FILE ) ; if ( cncFile . exists ( ) && cncFile . length ( ) > 0 ) { if ( null != logger ) { logger . accept ( "INFO: Aeron CnC file exists: " + cncFile ) ; } return IoUtil . mapExistingFile ( cncFile , CncFileDescriptor . CNC_FILE ) ; } return null ;


signature:
public class CommonContext { /** * Is a media driver active in the given directory ?
 * @ param directory to check
 * @ param driverTimeoutMs for the driver liveness check .
 * @ param logger for feedback as liveness checked .
 * @ return true if a driver is active or false if not . */
public static boolean isDriverActive ( final File directory , final long driverTimeoutMs , final Consumer < String > logger ) { } }

implementation: final File cncFile = new File ( directory , CncFileDescriptor . CNC_FILE ) ; if ( cncFile . exists ( ) && cncFile . length ( ) > 0 ) { logger . accept ( "INFO: Aeron CnC file exists: " + cncFile ) ; final MappedByteBuffer cncByteBuffer = IoUtil . mapExistingFile ( cncFile , "CnC file" ) ; try { return isDriverActive ( driverTimeoutMs , logger , cncByteBuffer ) ; } finally { IoUtil . unmap ( cncByteBuffer ) ; } } return false ;


signature:
public class CommonContext { /** * Is a media driver active in the current Aeron directory ?
 * @ param driverTimeoutMs for the driver liveness check .
 * @ param logger for feedback as liveness checked .
 * @ return true if a driver is active or false if not . */
public boolean isDriverActive ( final long driverTimeoutMs , final Consumer < String > logger ) { } }

implementation: final MappedByteBuffer cncByteBuffer = mapExistingCncFile ( logger ) ; try { return isDriverActive ( driverTimeoutMs , logger , cncByteBuffer ) ; } finally { IoUtil . unmap ( cncByteBuffer ) ; }


signature:
public class CommonContext { /** * Is a media driver active in the current mapped CnC buffer ? If the driver is mid start then it will wait for
 * up to the driverTimeoutMs by checking for the cncVersion being set .
 * @ param driverTimeoutMs for the driver liveness check .
 * @ param logger for feedback as liveness checked .
 * @ param cncByteBuffer for the existing CnC file .
 * @ return true if a driver is active or false if not . */
public static boolean isDriverActive ( final long driverTimeoutMs , final Consumer < String > logger , final ByteBuffer cncByteBuffer ) { } }

implementation: if ( null == cncByteBuffer ) { return false ; } final UnsafeBuffer cncMetaDataBuffer = CncFileDescriptor . createMetaDataBuffer ( cncByteBuffer ) ; final long startTimeMs = System . currentTimeMillis ( ) ; int cncVersion ; while ( 0 == ( cncVersion = cncMetaDataBuffer . getIntVolatile ( CncFileDescriptor . cncVersionOffset ( 0 ) ) ) ) { if ( System . currentTimeMillis ( ) > ( startTimeMs + driverTimeoutMs ) ) { throw new DriverTimeoutException ( "CnC file is created but not initialised." ) ; } sleep ( 1 ) ; } if ( CNC_VERSION != cncVersion ) { throw new AeronException ( "Aeron CnC version does not match: required=" + CNC_VERSION + " version=" + cncVersion ) ; } final ManyToOneRingBuffer toDriverBuffer = new ManyToOneRingBuffer ( CncFileDescriptor . createToDriverBuffer ( cncByteBuffer , cncMetaDataBuffer ) ) ; final long timestamp = toDriverBuffer . consumerHeartbeatTime ( ) ; final long now = System . currentTimeMillis ( ) ; final long timestampAge = now - timestamp ; logger . accept ( "INFO: Aeron toDriver consumer heartbeat is (ms): " + timestampAge ) ; return timestampAge <= driverTimeoutMs ;


signature:
public class CommonContext { /** * Request a driver to run its termination hook .
 * @ param directory for the driver .
 * @ param tokenBuffer containing the optional token for the request .
 * @ param tokenOffset within the tokenBuffer at which the token begins .
 * @ param tokenLength of the token in the tokenBuffer .
 * @ return true if request was sent or false if request could not be sent . */
public static boolean requestDriverTermination ( final File directory , final DirectBuffer tokenBuffer , final int tokenOffset , final int tokenLength ) { } }

implementation: final File cncFile = new File ( directory , CncFileDescriptor . CNC_FILE ) ; if ( cncFile . exists ( ) && cncFile . length ( ) > 0 ) { final MappedByteBuffer cncByteBuffer = IoUtil . mapExistingFile ( cncFile , "CnC file" ) ; try { final UnsafeBuffer cncMetaDataBuffer = CncFileDescriptor . createMetaDataBuffer ( cncByteBuffer ) ; final int cncVersion = cncMetaDataBuffer . getIntVolatile ( cncVersionOffset ( 0 ) ) ; if ( CncFileDescriptor . CNC_VERSION != cncVersion ) { throw new AeronException ( "Aeron CnC version does not match: required=" + CNC_VERSION + " version=" + cncVersion ) ; } final ManyToOneRingBuffer toDriverBuffer = new ManyToOneRingBuffer ( CncFileDescriptor . createToDriverBuffer ( cncByteBuffer , cncMetaDataBuffer ) ) ; final long clientId = toDriverBuffer . nextCorrelationId ( ) ; final DriverProxy driverProxy = new DriverProxy ( toDriverBuffer , clientId ) ; return driverProxy . terminateDriver ( tokenBuffer , tokenOffset , tokenLength ) ; } finally { IoUtil . unmap ( cncByteBuffer ) ; } } return false ;


signature:
public class CommonContext { /** * Read the error log to a given { @ link PrintStream }
 * @ param out to write the error log contents to .
 * @ return the number of observations from the error log . */
public int saveErrorLog ( final PrintStream out ) { } }

implementation: final MappedByteBuffer cncByteBuffer = mapExistingCncFile ( null ) ; try { return saveErrorLog ( out , cncByteBuffer ) ; } finally { IoUtil . unmap ( cncByteBuffer ) ; }


signature:
public class CommonContext { /** * Read the error log to a given { @ link PrintStream }
 * @ param out to write the error log contents to .
 * @ param cncByteBuffer containing the error log .
 * @ return the number of observations from the error log . */
public int saveErrorLog ( final PrintStream out , final ByteBuffer cncByteBuffer ) { } }

implementation: if ( null == cncByteBuffer ) { return 0 ; } final UnsafeBuffer cncMetaDataBuffer = CncFileDescriptor . createMetaDataBuffer ( cncByteBuffer ) ; final int cncVersion = cncMetaDataBuffer . getInt ( CncFileDescriptor . cncVersionOffset ( 0 ) ) ; if ( CNC_VERSION != cncVersion ) { throw new AeronException ( "Aeron CnC version does not match: required=" + CNC_VERSION + " version=" + cncVersion ) ; } int distinctErrorCount = 0 ; final AtomicBuffer buffer = CncFileDescriptor . createErrorLogBuffer ( cncByteBuffer , cncMetaDataBuffer ) ; if ( ErrorLogReader . hasErrors ( buffer ) ) { final SimpleDateFormat dateFormat = new SimpleDateFormat ( "yyyy-MM-dd HH:mm:ss.SSSZ" ) ; final ErrorConsumer errorConsumer = ( count , firstTimestamp , lastTimestamp , ex ) -> formatError ( out , dateFormat , count , firstTimestamp , lastTimestamp , ex ) ; distinctErrorCount = ErrorLogReader . read ( buffer , errorConsumer ) ; } out . println ( ) ; out . println ( distinctErrorCount + " distinct errors observed." ) ; return distinctErrorCount ;


signature:
public class FrameDescriptor { /** * Get the length of a frame from the header as a volatile read .
 * @ param buffer containing the frame .
 * @ param termOffset at which a frame begins .
 * @ return the value for the frame length . */
public static int frameLengthVolatile ( final UnsafeBuffer buffer , final int termOffset ) { } }

implementation: int frameLength = buffer . getIntVolatile ( termOffset ) ; if ( ByteOrder . nativeOrder ( ) != LITTLE_ENDIAN ) { frameLength = Integer . reverseBytes ( frameLength ) ; } return frameLength ;


signature:
public class FrameDescriptor { /** * Write the length header for a frame in a memory ordered fashion .
 * @ param buffer containing the frame .
 * @ param termOffset at which a frame begins .
 * @ param frameLength field to be set for the frame . */
public static void frameLengthOrdered ( final UnsafeBuffer buffer , final int termOffset , final int frameLength ) { } }

implementation: int length = frameLength ; if ( ByteOrder . nativeOrder ( ) != LITTLE_ENDIAN ) { length = Integer . reverseBytes ( frameLength ) ; } buffer . putIntOrdered ( termOffset , length ) ;


signature:
public class FrameDescriptor { /** * Write the type field for a frame .
 * @ param buffer containing the frame .
 * @ param termOffset at which a frame begins .
 * @ param type type value for the frame . */
public static void frameType ( final UnsafeBuffer buffer , final int termOffset , final int type ) { } }

implementation: buffer . putShort ( typeOffset ( termOffset ) , ( short ) type , LITTLE_ENDIAN ) ;


signature:
public class FrameDescriptor { /** * Write the flags field for a frame .
 * @ param buffer containing the frame .
 * @ param termOffset at which a frame begins .
 * @ param flags value for the frame . */
public static void frameFlags ( final UnsafeBuffer buffer , final int termOffset , final byte flags ) { } }

implementation: buffer . putByte ( flagsOffset ( termOffset ) , flags ) ;


signature:
public class FrameDescriptor { /** * Write the term offset field for a frame .
 * @ param buffer containing the frame .
 * @ param termOffset at which a frame begins . */
public static void frameTermOffset ( final UnsafeBuffer buffer , final int termOffset ) { } }

implementation: buffer . putInt ( termOffsetOffset ( termOffset ) , termOffset , LITTLE_ENDIAN ) ;


signature:
public class FrameDescriptor { /** * Write the term id field for a frame .
 * @ param buffer containing the frame .
 * @ param termOffset at which a frame begins .
 * @ param termId value for the frame . */
public static void frameTermId ( final UnsafeBuffer buffer , final int termOffset , final int termId ) { } }

implementation: buffer . putInt ( termIdOffset ( termOffset ) , termId , LITTLE_ENDIAN ) ;


signature:
public class SendChannelEndpoint { /** * Called from the { @ link Sender } to add information to the control packet dispatcher .
 * @ param publication to add to the dispatcher */
public void registerForSend ( final NetworkPublication publication ) { } }

implementation: publicationBySessionAndStreamId . put ( publication . sessionId ( ) , publication . streamId ( ) , publication ) ;


signature:
public class SendChannelEndpoint { /** * Send contents of a { @ link ByteBuffer } to connected address .
 * This is used on the sender side for performance over send ( ByteBuffer , SocketAddress ) .
 * @ param buffer to send
 * @ return number of bytes sent */
public int send ( final ByteBuffer buffer ) { } }

implementation: int bytesSent = 0 ; if ( null != sendDatagramChannel ) { final int bytesToSend = buffer . remaining ( ) ; if ( null == multiDestination ) { try { sendHook ( buffer , connectAddress ) ; if ( sendDatagramChannel . isConnected ( ) ) { bytesSent = sendDatagramChannel . write ( buffer ) ; } } catch ( final PortUnreachableException ignore ) { } catch ( final IOException ex ) { sendError ( bytesToSend , ex , connectAddress ) ; } } else { bytesSent = multiDestination . send ( sendDatagramChannel , buffer , this , bytesToSend ) ; } } return bytesSent ;


signature:
public class RecordingPos { /** * Find the active counter id for a stream based on the recording id .
 * @ param countersReader to search within .
 * @ param recordingId for the active recording .
 * @ return the counter id if found otherwise { @ link CountersReader # NULL _ COUNTER _ ID } . */
public static int findCounterIdByRecording ( final CountersReader countersReader , final long recordingId ) { } }

implementation: final DirectBuffer buffer = countersReader . metaDataBuffer ( ) ; for ( int i = 0 , size = countersReader . maxCounterId ( ) ; i < size ; i ++ ) { if ( countersReader . getCounterState ( i ) == RECORD_ALLOCATED ) { final int recordOffset = CountersReader . metaDataOffset ( i ) ; if ( buffer . getInt ( recordOffset + TYPE_ID_OFFSET ) == RECORDING_POSITION_TYPE_ID && buffer . getLong ( recordOffset + KEY_OFFSET + RECORDING_ID_OFFSET ) == recordingId ) { return i ; } } } return NULL_COUNTER_ID ;


signature:
public class RecordingPos { /** * Find the active counter id for a stream based on the session id .
 * @ param countersReader to search within .
 * @ param sessionId for the active recording .
 * @ return the counter id if found otherwise { @ link CountersReader # NULL _ COUNTER _ ID } . */
public static int findCounterIdBySession ( final CountersReader countersReader , final int sessionId ) { } }

implementation: final DirectBuffer buffer = countersReader . metaDataBuffer ( ) ; for ( int i = 0 , size = countersReader . maxCounterId ( ) ; i < size ; i ++ ) { if ( countersReader . getCounterState ( i ) == RECORD_ALLOCATED ) { final int recordOffset = CountersReader . metaDataOffset ( i ) ; if ( buffer . getInt ( recordOffset + TYPE_ID_OFFSET ) == RECORDING_POSITION_TYPE_ID && buffer . getInt ( recordOffset + KEY_OFFSET + SESSION_ID_OFFSET ) == sessionId ) { return i ; } } } return NULL_COUNTER_ID ;


signature:
public class RecordingPos { /** * Get the recording id for a given counter id .
 * @ param countersReader to search within .
 * @ param counterId for the active recording .
 * @ return the counter id if found otherwise { @ link # NULL _ RECORDING _ ID } . */
public static long getRecordingId ( final CountersReader countersReader , final int counterId ) { } }

implementation: final DirectBuffer buffer = countersReader . metaDataBuffer ( ) ; if ( countersReader . getCounterState ( counterId ) == RECORD_ALLOCATED ) { final int recordOffset = CountersReader . metaDataOffset ( counterId ) ; if ( buffer . getInt ( recordOffset + TYPE_ID_OFFSET ) == RECORDING_POSITION_TYPE_ID ) { return buffer . getLong ( recordOffset + KEY_OFFSET + RECORDING_ID_OFFSET ) ; } } return NULL_RECORDING_ID ;


signature:
public class RecordingPos { /** * Get the { @ link Image # sourceIdentity ( ) } for the recording .
 * @ param countersReader to search within .
 * @ param counterId for the active recording .
 * @ return { @ link Image # sourceIdentity ( ) } for the recording or null if not found . */
public static String getSourceIdentity ( final CountersReader countersReader , final int counterId ) { } }

implementation: final DirectBuffer buffer = countersReader . metaDataBuffer ( ) ; if ( countersReader . getCounterState ( counterId ) == RECORD_ALLOCATED ) { final int recordOffset = CountersReader . metaDataOffset ( counterId ) ; if ( buffer . getInt ( recordOffset + TYPE_ID_OFFSET ) == RECORDING_POSITION_TYPE_ID ) { return buffer . getStringAscii ( recordOffset + KEY_OFFSET + SOURCE_IDENTITY_LENGTH_OFFSET ) ; } } return null ;


signature:
public class RecordingPos { /** * Is the recording counter still active .
 * @ param countersReader to search within .
 * @ param counterId to search for .
 * @ param recordingId to confirm it is still the same value .
 * @ return true if the counter is still active otherwise false . */
public static boolean isActive ( final CountersReader countersReader , final int counterId , final long recordingId ) { } }

implementation: final DirectBuffer buffer = countersReader . metaDataBuffer ( ) ; if ( countersReader . getCounterState ( counterId ) == RECORD_ALLOCATED ) { final int recordOffset = CountersReader . metaDataOffset ( counterId ) ; return buffer . getInt ( recordOffset + TYPE_ID_OFFSET ) == RECORDING_POSITION_TYPE_ID && buffer . getLong ( recordOffset + KEY_OFFSET + RECORDING_ID_OFFSET ) == recordingId ; } return false ;


signature:
public class SocketAddressUtil { /** * Utility for parsing socket addresses from a { @ link CharSequence } . Supports
 * hostname : port , ipV4Address : port and [ ipV6Address ] : port
 * @ param cs to be parsed for the socket address .
 * @ return An { @ link InetSocketAddress } for the parsed input . */
static InetSocketAddress parse ( final CharSequence cs ) { } }

implementation: if ( null == cs || cs . length ( ) == 0 ) { throw new NullPointerException ( "Input string must not be null or empty" ) ; } InetSocketAddress address = tryParseIpV4 ( cs ) ; if ( null == address ) { address = tryParseIpV6 ( cs ) ; } if ( null == address ) { throw new IllegalArgumentException ( "Invalid format: " + cs ) ; } return address ;


signature:
public class PreferredMulticastFlowControl { /** * { @ inheritDoc } */
public long onStatusMessage ( final StatusMessageFlyweight flyweight , final InetSocketAddress receiverAddress , final long senderLimit , final int initialTermId , final int positionBitsToShift , final long timeNs ) { } }

implementation: final long position = computePosition ( flyweight . consumptionTermId ( ) , flyweight . consumptionTermOffset ( ) , positionBitsToShift , initialTermId ) ; final long windowLength = flyweight . receiverWindowLength ( ) ; final long receiverId = flyweight . receiverId ( ) ; final boolean isFromPreferred = isFromPreferred ( flyweight ) ; final long lastPositionPlusWindow = position + windowLength ; boolean isExisting = false ; long minPosition = Long . MAX_VALUE ; final ArrayList < Receiver > receiverList = this . receiverList ; for ( int i = 0 , size = receiverList . size ( ) ; i < size ; i ++ ) { final Receiver receiver = receiverList . get ( i ) ; if ( isFromPreferred && receiverId == receiver . receiverId ) { receiver . lastPosition = Math . max ( position , receiver . lastPosition ) ; receiver . lastPositionPlusWindow = lastPositionPlusWindow ; receiver . timeOfLastStatusMessageNs = timeNs ; isExisting = true ; } minPosition = Math . min ( minPosition , receiver . lastPositionPlusWindow ) ; } if ( isFromPreferred && ! isExisting ) { receiverList . add ( new Receiver ( position , lastPositionPlusWindow , timeNs , receiverId , receiverAddress ) ) ; minPosition = Math . min ( minPosition , lastPositionPlusWindow ) ; } return receiverList . size ( ) > 0 ? Math . max ( senderLimit , minPosition ) : Math . max ( senderLimit , lastPositionPlusWindow ) ;


signature:
public class TermGapScanner { /** * Scan for gaps from the scanOffset up to a limit offset . Each gap will be reported to the { @ link GapHandler } .
 * @ param termBuffer to be scanned for a gap .
 * @ param termId of the current term buffer .
 * @ param termOffset at which to start scanning .
 * @ param limitOffset at which to stop scanning .
 * @ param handler to call if a gap is found .
 * @ return offset of last contiguous frame */
public static int scanForGap ( final UnsafeBuffer termBuffer , final int termId , final int termOffset , final int limitOffset , final GapHandler handler ) { } }

implementation: int offset = termOffset ; do { final int frameLength = frameLengthVolatile ( termBuffer , offset ) ; if ( frameLength <= 0 ) { break ; } offset += align ( frameLength , FRAME_ALIGNMENT ) ; } while ( offset < limitOffset ) ; final int gapBeginOffset = offset ; if ( offset < limitOffset ) { final int limit = limitOffset - ALIGNED_HEADER_LENGTH ; while ( offset < limit ) { offset += FRAME_ALIGNMENT ; if ( 0 != termBuffer . getIntVolatile ( offset ) ) { offset -= ALIGNED_HEADER_LENGTH ; break ; } } final int gapLength = ( offset - gapBeginOffset ) + ALIGNED_HEADER_LENGTH ; handler . onGap ( termId , gapBeginOffset , gapLength ) ; } return gapBeginOffset ;


signature:
public class ClusteredMediaDriver { /** * Launch a new { @ link ClusteredMediaDriver } with provided contexts .
 * @ param driverCtx for configuring the { @ link MediaDriver } .
 * @ param archiveCtx for configuring the { @ link Archive } .
 * @ param consensusModuleCtx for the configuration of the { @ link ConsensusModule } .
 * @ return a new { @ link ClusteredMediaDriver } with the provided contexts . */
public static ClusteredMediaDriver launch ( final MediaDriver . Context driverCtx , final Archive . Context archiveCtx , final ConsensusModule . Context consensusModuleCtx ) { } }

implementation: final MediaDriver driver = MediaDriver . launch ( driverCtx . spiesSimulateConnection ( true ) ) ; final Archive archive = Archive . launch ( archiveCtx . mediaDriverAgentInvoker ( driver . sharedAgentInvoker ( ) ) . errorHandler ( driverCtx . errorHandler ( ) ) . errorCounter ( driverCtx . systemCounters ( ) . get ( SystemCounterDescriptor . ERRORS ) ) ) ; final ConsensusModule consensusModule = ConsensusModule . launch ( consensusModuleCtx ) ; return new ClusteredMediaDriver ( driver , archive , consensusModule ) ;


signature:
public class StatusMessageFlyweight { /** * Identifier for the receiver to distinguish them for FlowControl strategies .
 * @ return identifier for the receiver to distinguish them for FlowControl strategies . */
public long receiverId ( ) { } }

implementation: final long value ; if ( ByteOrder . nativeOrder ( ) == LITTLE_ENDIAN ) { value = ( ( ( ( long ) getByte ( RECEIVER_ID_FIELD_OFFSET + 7 ) ) << 56 ) | ( ( ( long ) getByte ( RECEIVER_ID_FIELD_OFFSET + 6 ) & 0xFF ) << 48 ) | ( ( ( long ) getByte ( RECEIVER_ID_FIELD_OFFSET + 5 ) & 0xFF ) << 40 ) | ( ( ( long ) getByte ( RECEIVER_ID_FIELD_OFFSET + 4 ) & 0xFF ) << 32 ) | ( ( ( long ) getByte ( RECEIVER_ID_FIELD_OFFSET + 3 ) & 0xFF ) << 24 ) | ( ( ( long ) getByte ( RECEIVER_ID_FIELD_OFFSET + 2 ) & 0xFF ) << 16 ) | ( ( ( long ) getByte ( RECEIVER_ID_FIELD_OFFSET + 1 ) & 0xFF ) << 8 ) | ( ( ( long ) getByte ( RECEIVER_ID_FIELD_OFFSET + 0 ) & 0xFF ) ) ) ; } else { value = ( ( ( ( long ) getByte ( RECEIVER_ID_FIELD_OFFSET + 0 ) ) << 56 ) | ( ( ( long ) getByte ( RECEIVER_ID_FIELD_OFFSET + 1 ) & 0xFF ) << 48 ) | ( ( ( long ) getByte ( RECEIVER_ID_FIELD_OFFSET + 2 ) & 0xFF ) << 40 ) | ( ( ( long ) getByte ( RECEIVER_ID_FIELD_OFFSET + 3 ) & 0xFF ) << 32 ) | ( ( ( long ) getByte ( RECEIVER_ID_FIELD_OFFSET + 4 ) & 0xFF ) << 24 ) | ( ( ( long ) getByte ( RECEIVER_ID_FIELD_OFFSET + 5 ) & 0xFF ) << 16 ) | ( ( ( long ) getByte ( RECEIVER_ID_FIELD_OFFSET + 6 ) & 0xFF ) << 8 ) | ( ( ( long ) getByte ( RECEIVER_ID_FIELD_OFFSET + 7 ) & 0xFF ) ) ) ; } return value ;


signature:
public class StatusMessageFlyweight { /** * Identifier for the receiver to distinguish them for FlowControl strategies .
 * @ param id for the receiver to distinguish them for FlowControl strategies .
 * @ return flyweight */
public StatusMessageFlyweight receiverId ( final long id ) { } }

implementation: if ( ByteOrder . nativeOrder ( ) == LITTLE_ENDIAN ) { putByte ( RECEIVER_ID_FIELD_OFFSET + 7 , ( byte ) ( id >> 56 ) ) ; putByte ( RECEIVER_ID_FIELD_OFFSET + 6 , ( byte ) ( id >> 48 ) ) ; putByte ( RECEIVER_ID_FIELD_OFFSET + 5 , ( byte ) ( id >> 40 ) ) ; putByte ( RECEIVER_ID_FIELD_OFFSET + 4 , ( byte ) ( id >> 32 ) ) ; putByte ( RECEIVER_ID_FIELD_OFFSET + 3 , ( byte ) ( id >> 24 ) ) ; putByte ( RECEIVER_ID_FIELD_OFFSET + 2 , ( byte ) ( id >> 16 ) ) ; putByte ( RECEIVER_ID_FIELD_OFFSET + 1 , ( byte ) ( id >> 8 ) ) ; putByte ( RECEIVER_ID_FIELD_OFFSET + 0 , ( byte ) ( id ) ) ; } else { putByte ( RECEIVER_ID_FIELD_OFFSET + 0 , ( byte ) ( id >> 56 ) ) ; putByte ( RECEIVER_ID_FIELD_OFFSET + 1 , ( byte ) ( id >> 48 ) ) ; putByte ( RECEIVER_ID_FIELD_OFFSET + 2 , ( byte ) ( id >> 40 ) ) ; putByte ( RECEIVER_ID_FIELD_OFFSET + 3 , ( byte ) ( id >> 32 ) ) ; putByte ( RECEIVER_ID_FIELD_OFFSET + 4 , ( byte ) ( id >> 24 ) ) ; putByte ( RECEIVER_ID_FIELD_OFFSET + 5 , ( byte ) ( id >> 16 ) ) ; putByte ( RECEIVER_ID_FIELD_OFFSET + 6 , ( byte ) ( id >> 8 ) ) ; putByte ( RECEIVER_ID_FIELD_OFFSET + 7 , ( byte ) ( id ) ) ; } return this ;


signature:
public class StatusMessageFlyweight { /** * Retrieve the Application Specific Feedback ( if present ) from the Status Message .
 * @ param destination to store the feedback
 * @ return the number of bytes in the feedback copied into the destination */
public int applicationSpecificFeedback ( final byte [ ] destination ) { } }

implementation: final int frameLength = frameLength ( ) ; int result = 0 ; if ( frameLength > HEADER_LENGTH ) { if ( frameLength > capacity ( ) ) { throw new AeronException ( String . format ( "SM application specific feedback (%d) is truncated (%d)" , frameLength - HEADER_LENGTH , capacity ( ) - HEADER_LENGTH ) ) ; } final int copyLength = Math . min ( destination . length , frameLength - HEADER_LENGTH ) ; getBytes ( APP_SPECIFIC_FEEDBACK_FIELD_OFFSET , destination , 0 , copyLength ) ; result = copyLength ; } return result ;


signature:
public class StatusMessageFlyweight { /** * Set the Application Specific Feedback for the Status Message .
 * @ param source of the feedback to set
 * @ param offset of the feedback in the source
 * @ param length of the feedback in bytes
 * @ return flyweight */
public StatusMessageFlyweight applicationSpecificFeedback ( final byte [ ] source , final int offset , final int length ) { } }

implementation: frameLength ( HEADER_LENGTH + length ) ; putBytes ( APP_SPECIFIC_FEEDBACK_FIELD_OFFSET , source , offset , length ) ; return this ;


signature:
public class RecordingLog { /** * Reload the log from disk . */
public void reload ( ) { } }

implementation: entries . clear ( ) ; indexByLeadershipTermIdMap . clear ( ) ; indexByLeadershipTermIdMap . compact ( ) ; nextEntryIndex = 0 ; byteBuffer . clear ( ) ; try { while ( true ) { final int bytes = fileChannel . read ( byteBuffer ) ; if ( byteBuffer . remaining ( ) == 0 ) { byteBuffer . flip ( ) ; captureEntriesFromBuffer ( byteBuffer , buffer , entries ) ; byteBuffer . clear ( ) ; } if ( - 1 == bytes ) { if ( byteBuffer . position ( ) > 0 ) { byteBuffer . flip ( ) ; captureEntriesFromBuffer ( byteBuffer , buffer , entries ) ; byteBuffer . clear ( ) ; } break ; } } } catch ( final IOException ex ) { LangUtil . rethrowUnchecked ( ex ) ; }


signature:
public class RecordingLog { /** * Find the last recording id used for a leader ship term . If not found then { @ link RecordingPos # NULL _ RECORDING _ ID } .
 * @ return the last leadership term recording id or { @ link RecordingPos # NULL _ RECORDING _ ID } if not found . */
public long findLastTermRecordingId ( ) { } }

implementation: for ( int i = entries . size ( ) - 1 ; i >= 0 ; i -- ) { final Entry entry = entries . get ( i ) ; if ( ENTRY_TYPE_TERM == entry . type ) { return entry . recordingId ; } } return RecordingPos . NULL_RECORDING_ID ;


signature:
public class RecordingLog { /** * Find the last leadership term in the recording log .
 * @ return the last leadership term in the recording log . */
public Entry findLastTerm ( ) { } }

implementation: for ( int i = entries . size ( ) - 1 ; i >= 0 ; i -- ) { final Entry entry = entries . get ( i ) ; if ( ENTRY_TYPE_TERM == entry . type ) { return entry ; } } return null ;


signature:
public class RecordingLog { /** * Get the term { @ link Entry } for a given leadership term id .
 * @ param leadershipTermId to get { @ link Entry } for .
 * @ return the { @ link Entry } if found or throw { @ link IllegalArgumentException } if no entry exists for term . */
public Entry getTermEntry ( final long leadershipTermId ) { } }

implementation: final int index = ( int ) indexByLeadershipTermIdMap . get ( leadershipTermId ) ; if ( NULL_VALUE == index ) { throw new ClusterException ( "unknown leadershipTermId=" + leadershipTermId ) ; } return entries . get ( index ) ;


signature:
public class RecordingLog { /** * Create a recovery plan for the cluster that when the steps are replayed will bring the cluster back to the
 * latest stable state .
 * @ param archive to lookup recording descriptors .
 * @ param serviceCount of services that may have snapshots .
 * @ return a new { @ link RecoveryPlan } for the cluster . */
public RecoveryPlan createRecoveryPlan ( final AeronArchive archive , final int serviceCount ) { } }

implementation: final ArrayList < Snapshot > snapshots = new ArrayList < > ( ) ; final ArrayList < Log > logs = new ArrayList < > ( ) ; planRecovery ( snapshots , logs , entries , archive , serviceCount ) ; long lastLeadershipTermId = NULL_VALUE ; long lastTermBaseLogPosition = 0 ; long committedLogPosition = - 1 ; long appendedLogPosition = 0 ; final int snapshotStepsSize = snapshots . size ( ) ; if ( snapshotStepsSize > 0 ) { final Snapshot snapshot = snapshots . get ( 0 ) ; lastLeadershipTermId = snapshot . leadershipTermId ; lastTermBaseLogPosition = snapshot . termBaseLogPosition ; appendedLogPosition = snapshot . logPosition ; committedLogPosition = snapshot . logPosition ; } if ( ! logs . isEmpty ( ) ) { final Log log = logs . get ( 0 ) ; lastLeadershipTermId = log . leadershipTermId ; lastTermBaseLogPosition = log . termBaseLogPosition ; appendedLogPosition = log . stopPosition ; committedLogPosition = log . logPosition ; } return new RecoveryPlan ( lastLeadershipTermId , lastTermBaseLogPosition , appendedLogPosition , committedLogPosition , snapshots , logs ) ;


signature:
public class RecordingLog { /** * Create a recovery plan that has only snapshots . Used for dynamicJoin snapshot load .
 * @ param snapshots to construct plan from
 * @ return a new { @ link RecoveryPlan } for the cluster */
public static RecoveryPlan createRecoveryPlan ( final ArrayList < RecordingLog . Snapshot > snapshots ) { } }

implementation: long lastLeadershipTermId = NULL_VALUE ; long lastTermBaseLogPosition = 0 ; long committedLogPosition = - 1 ; long appendedLogPosition = 0 ; final int snapshotStepsSize = snapshots . size ( ) ; if ( snapshotStepsSize > 0 ) { final Snapshot snapshot = snapshots . get ( 0 ) ; lastLeadershipTermId = snapshot . leadershipTermId ; lastTermBaseLogPosition = snapshot . termBaseLogPosition ; appendedLogPosition = snapshot . logPosition ; committedLogPosition = snapshot . logPosition ; } return new RecoveryPlan ( lastLeadershipTermId , lastTermBaseLogPosition , appendedLogPosition , committedLogPosition , snapshots , new ArrayList < > ( ) ) ;


signature:
public class RecordingLog { /** * Append a log entry for a leadership term .
 * @ param recordingId of the log .
 * @ param leadershipTermId for the current term .
 * @ param termBaseLogPosition reached at the beginning of the term .
 * @ param timestamp at the beginning of the term . */
public void appendTerm ( final long recordingId , final long leadershipTermId , final long termBaseLogPosition , final long timestamp ) { } }

implementation: final int size = entries . size ( ) ; if ( size > 0 ) { final Entry lastEntry = entries . get ( size - 1 ) ; if ( lastEntry . type != NULL_VALUE && lastEntry . leadershipTermId >= leadershipTermId ) { throw new ClusterException ( "leadershipTermId out of sequence: previous " + lastEntry . leadershipTermId + " this " + leadershipTermId ) ; } } indexByLeadershipTermIdMap . put ( leadershipTermId , nextEntryIndex ) ; append ( ENTRY_TYPE_TERM , recordingId , leadershipTermId , termBaseLogPosition , NULL_POSITION , timestamp , NULL_VALUE ) ;


signature:
public class RecordingLog { /** * Append a log entry for a snapshot .
 * @ param recordingId in the archive for the snapshot .
 * @ param leadershipTermId for the current term
 * @ param termBaseLogPosition at the beginning of the leadership term .
 * @ param logPosition for the position in the current term or length so far for that term .
 * @ param timestamp at which the snapshot was taken .
 * @ param serviceId for which the snapshot is recorded . */
public void appendSnapshot ( final long recordingId , final long leadershipTermId , final long termBaseLogPosition , final long logPosition , final long timestamp , final int serviceId ) { } }

implementation: final int size = entries . size ( ) ; if ( size > 0 ) { final Entry entry = entries . get ( size - 1 ) ; if ( entry . type == ENTRY_TYPE_TERM && entry . leadershipTermId != leadershipTermId ) { throw new ClusterException ( "leadershipTermId out of sequence: previous " + entry . leadershipTermId + " this " + leadershipTermId ) ; } } append ( ENTRY_TYPE_SNAPSHOT , recordingId , leadershipTermId , termBaseLogPosition , logPosition , timestamp , serviceId ) ;


signature:
public class RecordingLog { /** * Commit the position reached in a leadership term before a clean shutdown .
 * @ param leadershipTermId for committing the term position reached .
 * @ param logPosition reached in the leadership term . */
public void commitLogPosition ( final long leadershipTermId , final long logPosition ) { } }

implementation: final int index = getLeadershipTermEntryIndex ( leadershipTermId ) ; commitEntryValue ( index , logPosition , LOG_POSITION_OFFSET ) ; final Entry entry = entries . get ( index ) ; entries . set ( index , new Entry ( entry . recordingId , entry . leadershipTermId , entry . termBaseLogPosition , logPosition , entry . timestamp , entry . serviceId , entry . type , entry . entryIndex ) ) ;


signature:
public class RecordingLog { /** * Tombstone an entry in the log so it is no longer valid .
 * @ param leadershipTermId to match for validation .
 * @ param entryIndex reached in the leadership term . */
public void tombstoneEntry ( final long leadershipTermId , final int entryIndex ) { } }

implementation: int index = - 1 ; for ( int i = 0 , size = entries . size ( ) ; i < size ; i ++ ) { final Entry entry = entries . get ( i ) ; if ( entry . leadershipTermId == leadershipTermId && entry . entryIndex == entryIndex ) { index = entry . entryIndex ; if ( ENTRY_TYPE_TERM == entry . type ) { indexByLeadershipTermIdMap . remove ( leadershipTermId ) ; } break ; } } if ( - 1 == index ) { throw new ClusterException ( "unknown entry index: " + entryIndex ) ; } buffer . putInt ( 0 , NULL_VALUE , LITTLE_ENDIAN ) ; byteBuffer . limit ( SIZE_OF_INT ) . position ( 0 ) ; final long filePosition = ( index * ( long ) ENTRY_LENGTH ) + ENTRY_TYPE_OFFSET ; try { if ( SIZE_OF_INT != fileChannel . write ( byteBuffer , filePosition ) ) { throw new ClusterException ( "failed to write field atomically" ) ; } } catch ( final Exception ex ) { LangUtil . rethrowUnchecked ( ex ) ; }


signature:
public class AeronCluster { /** * Connect to the cluster providing { @ link Context } for configuration .
 * @ param ctx for configuration .
 * @ return allocated cluster client if the connection is successful . */
public static AeronCluster connect ( final AeronCluster . Context ctx ) { } }

implementation: Subscription subscription = null ; AsyncConnect asyncConnect = null ; try { ctx . conclude ( ) ; final Aeron aeron = ctx . aeron ( ) ; final long deadlineNs = aeron . context ( ) . nanoClock ( ) . nanoTime ( ) + ctx . messageTimeoutNs ( ) ; subscription = aeron . addSubscription ( ctx . egressChannel ( ) , ctx . egressStreamId ( ) ) ; final IdleStrategy idleStrategy = ctx . idleStrategy ( ) ; asyncConnect = new AsyncConnect ( ctx , subscription , deadlineNs ) ; final AgentInvoker aeronClientInvoker = aeron . conductorAgentInvoker ( ) ; AeronCluster aeronCluster ; while ( null == ( aeronCluster = asyncConnect . poll ( ) ) ) { if ( null != aeronClientInvoker ) { aeronClientInvoker . invoke ( ) ; } idleStrategy . idle ( ) ; } return aeronCluster ; } catch ( final Exception ex ) { if ( ! ctx . ownsAeronClient ( ) ) { CloseHelper . close ( subscription ) ; CloseHelper . close ( asyncConnect ) ; } ctx . close ( ) ; throw ex ; }


signature:
public class AeronCluster { /** * Begin an attempt at creating a connection which can be completed by calling { @ link AsyncConnect # poll ( ) } until
 * it returns the client , before complete it will return null .
 * @ param ctx for the cluster .
 * @ return the { @ link AsyncConnect } that can be polled for completion . */
public static AsyncConnect asyncConnect ( final Context ctx ) { } }

implementation: Subscription subscription = null ; try { ctx . conclude ( ) ; final long deadlineNs = ctx . aeron ( ) . context ( ) . nanoClock ( ) . nanoTime ( ) + ctx . messageTimeoutNs ( ) ; subscription = ctx . aeron ( ) . addSubscription ( ctx . egressChannel ( ) , ctx . egressStreamId ( ) ) ; return new AsyncConnect ( ctx , subscription , deadlineNs ) ; } catch ( final Exception ex ) { if ( ! ctx . ownsAeronClient ( ) ) { CloseHelper . quietClose ( subscription ) ; } ctx . close ( ) ; throw ex ; }


signature:
public class AeronCluster { /** * Close session and release associated resources . */
public void close ( ) { } }

implementation: if ( null != publication && publication . isConnected ( ) ) { closeSession ( ) ; } if ( ! ctx . ownsAeronClient ( ) ) { CloseHelper . close ( subscription ) ; CloseHelper . close ( publication ) ; } ctx . close ( ) ;


signature:
public class AeronCluster { /** * Non - blocking publish of a partial buffer containing a message plus session header to a cluster .
 * This version of the method will set the timestamp value in the header to zero .
 * @ param buffer containing message .
 * @ param offset offset in the buffer at which the encoded message begins .
 * @ param length in bytes of the encoded message .
 * @ return the same as { @ link Publication # offer ( DirectBuffer , int , int ) } . */
public long offer ( final DirectBuffer buffer , final int offset , final int length ) { } }

implementation: return publication . offer ( headerBuffer , 0 , INGRESS_HEADER_LENGTH , buffer , offset , length , null ) ;


signature:
public class AeronCluster { /** * Non - blocking publish by gathering buffer vectors into a message . The first vector will be replaced by the cluster
 * ingress header so must be left unused .
 * @ param vectors which make up the message .
 * @ return the same as { @ link Publication # offer ( DirectBufferVector [ ] ) } .
 * @ see Publication # offer ( DirectBufferVector [ ] ) */
public long offer ( final DirectBufferVector [ ] vectors ) { } }

implementation: if ( headerVector != vectors [ 0 ] ) { vectors [ 0 ] = headerVector ; } return publication . offer ( vectors , null ) ;


signature:
public class AeronCluster { /** * Send a keep alive message to the cluster to keep this session open .
 * < b > Note : < / b > keepalives can fail during a leadership transition . The consumer should continue to call
 * { @ link # pollEgress ( ) } to ensure a connection to the new leader is established .
 * @ return true if successfully sent otherwise false . */
public boolean sendKeepAlive ( ) { } }

implementation: idleStrategy . reset ( ) ; int attempts = SEND_ATTEMPTS ; while ( true ) { final long result = publication . offer ( keepaliveMsgBuffer , 0 , keepaliveMsgBuffer . capacity ( ) , null ) ; if ( result > 0 ) { return true ; } if ( result == Publication . NOT_CONNECTED || result == Publication . CLOSED ) { return false ; } if ( result == Publication . MAX_POSITION_EXCEEDED ) { throw new ClusterException ( "unexpected publication state: " + result ) ; } if ( -- attempts <= 0 ) { break ; } idleStrategy . idle ( ) ; } return false ;


signature:
public class AeronCluster { /** * To be called when a new leader event is delivered . This method needs to be called when using the
 * { @ link EgressAdapter } or { @ link EgressPoller } rather than { @ link # pollEgress ( ) } method .
 * @ param clusterSessionId which must match { @ link # clusterSessionId ( ) } .
 * @ param leadershipTermId that identifies the term for which the new leader has been elected .
 * @ param leaderMemberId which has become the new leader .
 * @ param memberEndpoints comma separated list of cluster members endpoints to connect to with the leader first . */
public void onNewLeader ( final long clusterSessionId , final long leadershipTermId , final int leaderMemberId , final String memberEndpoints ) { } }

implementation: if ( clusterSessionId != this . clusterSessionId ) { throw new ClusterException ( "invalid clusterSessionId=" + clusterSessionId + " expected " + this . clusterSessionId ) ; } this . leadershipTermId = leadershipTermId ; this . leaderMemberId = leaderMemberId ; ingressMessageHeaderEncoder . leadershipTermId ( leadershipTermId ) ; sessionKeepAliveEncoder . leadershipTermId ( leadershipTermId ) ; if ( ctx . clusterMemberEndpoints ( ) != null ) { CloseHelper . close ( publication ) ; ctx . clusterMemberEndpoints ( memberEndpoints ) ; updateMemberEndpoints ( memberEndpoints , leaderMemberId ) ; } fragmentAssembler . clear ( ) ; controlledFragmentAssembler . clear ( ) ; egressListener . newLeader ( clusterSessionId , leadershipTermId , leaderMemberId , memberEndpoints ) ; controlledEgressListener . newLeader ( clusterSessionId , leadershipTermId , leaderMemberId , memberEndpoints ) ;


signature:
public class LogBufferUnblocker { /** * Attempt to unblock a log buffer at given position
 * @ param termBuffers for current blockedOffset
 * @ param logMetaDataBuffer for log buffer
 * @ param blockedPosition to attempt to unblock
 * @ param termLength of the buffer for each term in the log
 * @ return whether unblocked or not */
public static boolean unblock ( final UnsafeBuffer [ ] termBuffers , final UnsafeBuffer logMetaDataBuffer , final long blockedPosition , final int termLength ) { } }

implementation:
final int positionBitsToShift = LogBufferDescriptor . positionBitsToShift ( termLength ) ; final int blockedTermCount = ( int ) ( blockedPosition >> positionBitsToShift ) ; final int blockedOffset = ( int ) blockedPosition & ( termLength - 1 ) ; final int activeTermCount = activeTermCount ( logMetaDataBuffer ) ; if ( activeTermCount == ( blockedTermCount - 1 ) && blockedOffset == 0 ) { final int currentTermId = termId ( rawTailVolatile ( logMetaDataBuffer , indexByTermCount ( activeTermCount ) ) ) ; return rotateLog ( logMetaDataBuffer , activeTermCount , currentTermId ) ; } final int blockedIndex = indexByTermCount ( blockedTermCount ) ; final long rawTail = rawTailVolatile ( logMetaDataBuffer , blockedIndex ) ; final int termId = termId ( rawTail ) ; final int tailOffset = termOffset ( rawTail , termLength ) ; final UnsafeBuffer termBuffer = termBuffers [ blockedIndex ] ; switch ( TermUnblocker . unblock ( logMetaDataBuffer , termBuffer , blockedOffset , tailOffset , termId ) ) { case UNBLOCKED_TO_END : rotateLog ( logMetaDataBuffer , blockedTermCount , termId ) ; // fall through
case UNBLOCKED : return true ; } return false ;


signature:
public class ImageMessageFlyweight { /** * Get the channel field as ASCII
 * @ return channel field */
public String channel ( ) { } }

implementation: final int length = buffer . getInt ( offset + CHANNEL_OFFSET ) ; lengthOfChannel = SIZE_OF_INT + length ; return buffer . getStringAscii ( offset + CHANNEL_OFFSET , length ) ;


signature:
public class ImageMessageFlyweight { /** * Set the channel field as ASCII
 * @ param channel field value
 * @ return flyweight */
public ImageMessageFlyweight channel ( final String channel ) { } }

implementation: lengthOfChannel = buffer . putStringAscii ( offset + CHANNEL_OFFSET , channel ) ; return this ;


signature:
public class LogBufferDescriptor { /** * Check that term length is valid and alignment is valid .
 * @ param termLength to be checked .
 * @ throws IllegalStateException if the length is not as expected . */
public static void checkTermLength ( final int termLength ) { } }

implementation: if ( termLength < TERM_MIN_LENGTH ) { throw new IllegalStateException ( "Term length less than min length of " + TERM_MIN_LENGTH + ": length=" + termLength ) ; } if ( termLength > TERM_MAX_LENGTH ) { throw new IllegalStateException ( "Term length more than max length of " + TERM_MAX_LENGTH + ": length=" + termLength ) ; } if ( ! BitUtil . isPowerOfTwo ( termLength ) ) { throw new IllegalStateException ( "Term length not a power of 2: length=" + termLength ) ; }


signature:
public class LogBufferDescriptor { /** * Check that page size is valid and alignment is valid .
 * @ param pageSize to be checked .
 * @ throws IllegalStateException if the size is not as expected . */
public static void checkPageSize ( final int pageSize ) { } }

implementation: if ( pageSize < PAGE_MIN_SIZE ) { throw new IllegalStateException ( "Page size less than min size of " + PAGE_MIN_SIZE + ": page size=" + pageSize ) ; } if ( pageSize > PAGE_MAX_SIZE ) { throw new IllegalStateException ( "Page size more than max size of " + PAGE_MAX_SIZE + ": page size=" + pageSize ) ; } if ( ! BitUtil . isPowerOfTwo ( pageSize ) ) { throw new IllegalStateException ( "Page size not a power of 2: page size=" + pageSize ) ; }


signature:
public class LogBufferDescriptor { /** * Compare and set the value of the current active term count .
 * @ param metadataBuffer containing the meta data .
 * @ param expectedTermCount value of the active term count expected in the log
 * @ param updateTermCount value of the active term count to be updated in the log
 * @ return true if successful otherwise false . */
public static boolean casActiveTermCount ( final UnsafeBuffer metadataBuffer , final int expectedTermCount , final int updateTermCount ) { } }

implementation: return metadataBuffer . compareAndSetInt ( LOG_ACTIVE_TERM_COUNT_OFFSET , expectedTermCount , updateTermCount ) ;


signature:
public class LogBufferDescriptor { /** * Compute the current position in absolute number of bytes .
 * @ param activeTermId active term id .
 * @ param termOffset in the term .
 * @ param positionBitsToShift number of times to left shift the term count
 * @ param initialTermId the initial term id that this stream started on
 * @ return the absolute position in bytes */
public static long computePosition ( final int activeTermId , final int termOffset , final int positionBitsToShift , final int initialTermId ) { } }

implementation:
final long termCount = activeTermId - initialTermId ; // copes with negative activeTermId on rollover
return ( termCount << positionBitsToShift ) + termOffset ;


signature:
public class LogBufferDescriptor { /** * Compute the total length of a log file given the term length .
 * Assumes { @ link # TERM _ MAX _ LENGTH } is 1GB and that filePageSize is 1GB or less and a power of 2.
 * @ param termLength on which to base the calculation .
 * @ param filePageSize to use for log .
 * @ return the total length of the log file . */
public static long computeLogLength ( final int termLength , final int filePageSize ) { } }

implementation: if ( termLength < ( 1024 * 1024 * 1024 ) ) { return align ( ( termLength * PARTITION_COUNT ) + LOG_META_DATA_LENGTH , filePageSize ) ; } return ( PARTITION_COUNT * ( long ) termLength ) + align ( LOG_META_DATA_LENGTH , filePageSize ) ;


signature:
public class LogBufferDescriptor { /** * Store the default frame header to the log meta data buffer .
 * @ param metadataBuffer into which the default headers should be stored .
 * @ param defaultHeader to be stored .
 * @ throws IllegalArgumentException if the defaultHeader larger than { @ link # LOG _ DEFAULT _ FRAME _ HEADER _ MAX _ LENGTH } */
public static void storeDefaultFrameHeader ( final UnsafeBuffer metadataBuffer , final DirectBuffer defaultHeader ) { } }

implementation: if ( defaultHeader . capacity ( ) != HEADER_LENGTH ) { throw new IllegalArgumentException ( "Default header length not equal to HEADER_LENGTH: length=" + defaultHeader . capacity ( ) ) ; } metadataBuffer . putInt ( LOG_DEFAULT_FRAME_HEADER_LENGTH_OFFSET , HEADER_LENGTH ) ; metadataBuffer . putBytes ( LOG_DEFAULT_FRAME_HEADER_OFFSET , defaultHeader , 0 , HEADER_LENGTH ) ;


signature:
public class LogBufferDescriptor { /** * Apply the default header for a message in a term .
 * @ param metadataBuffer containing the default headers .
 * @ param termBuffer to which the default header should be applied .
 * @ param termOffset at which the default should be applied . */
public static void applyDefaultHeader ( final UnsafeBuffer metadataBuffer , final UnsafeBuffer termBuffer , final int termOffset ) { } }

implementation: termBuffer . putBytes ( termOffset , metadataBuffer , LOG_DEFAULT_FRAME_HEADER_OFFSET , HEADER_LENGTH ) ;


signature:
public class LogBufferDescriptor { /** * Rotate the log and update the tail counter for the new term .
 * This method is safe for concurrent use .
 * @ param metadataBuffer for the log .
 * @ param termCount from which to rotate .
 * @ param termId to be used in the default headers .
 * @ return true if log was rotated . */
public static boolean rotateLog ( final UnsafeBuffer metadataBuffer , final int termCount , final int termId ) { } }

implementation: final int nextTermId = termId + 1 ; final int nextTermCount = termCount + 1 ; final int nextIndex = indexByTermCount ( nextTermCount ) ; final int expectedTermId = nextTermId - PARTITION_COUNT ; long rawTail ; do { rawTail = rawTail ( metadataBuffer , nextIndex ) ; if ( expectedTermId != termId ( rawTail ) ) { break ; } } while ( ! casRawTail ( metadataBuffer , nextIndex , rawTail , packTail ( nextTermId , 0 ) ) ) ; return casActiveTermCount ( metadataBuffer , termCount , nextTermCount ) ;


signature:
public class LogBufferDescriptor { /** * Set the initial value for the termId in the upper bits of the tail counter .
 * @ param metadataBuffer contain the tail counter .
 * @ param partitionIndex to be initialised .
 * @ param termId to be set . */
public static void initialiseTailWithTermId ( final UnsafeBuffer metadataBuffer , final int partitionIndex , final int termId ) { } }

implementation: metadataBuffer . putLong ( TERM_TAIL_COUNTERS_OFFSET + ( partitionIndex * SIZE_OF_LONG ) , packTail ( termId , 0 ) ) ;


signature:
public class LogBufferDescriptor { /** * Read the termOffset from a packed raw tail value .
 * @ param rawTail containing the termOffset .
 * @ param termLength that the offset cannot exceed .
 * @ return the termOffset value . */
public static int termOffset ( final long rawTail , final long termLength ) { } }

implementation: final long tail = rawTail & 0xFFFF_FFFFL ; return ( int ) Math . min ( tail , termLength ) ;


signature:
public class LogBufferDescriptor { /** * Set the raw value of the tail for the given partition .
 * @ param metadataBuffer containing the tail counters .
 * @ param partitionIndex for the tail counter .
 * @ param rawTail to be stored */
public static void rawTail ( final UnsafeBuffer metadataBuffer , final int partitionIndex , final long rawTail ) { } }

implementation: metadataBuffer . putLong ( TERM_TAIL_COUNTERS_OFFSET + ( SIZE_OF_LONG * partitionIndex ) , rawTail ) ;


signature:
public class LogBufferDescriptor { /** * Set the raw value of the tail for the given partition .
 * @ param metadataBuffer containing the tail counters .
 * @ param partitionIndex for the tail counter .
 * @ param rawTail to be stored */
public static void rawTailVolatile ( final UnsafeBuffer metadataBuffer , final int partitionIndex , final long rawTail ) { } }

implementation: metadataBuffer . putLongVolatile ( TERM_TAIL_COUNTERS_OFFSET + ( SIZE_OF_LONG * partitionIndex ) , rawTail ) ;


signature:
public class LogBufferDescriptor { /** * Get the raw value of the tail for the current active partition .
 * @ param metadataBuffer containing the tail counters .
 * @ return the raw value of the tail for the current active partition . */
public static long rawTailVolatile ( final UnsafeBuffer metadataBuffer ) { } }

implementation: final int partitionIndex = indexByTermCount ( activeTermCount ( metadataBuffer ) ) ; return metadataBuffer . getLongVolatile ( TERM_TAIL_COUNTERS_OFFSET + ( SIZE_OF_LONG * partitionIndex ) ) ;


signature:
public class LogBufferDescriptor { /** * Compare and set the raw value of the tail for the given partition .
 * @ param metadataBuffer containing the tail counters .
 * @ param partitionIndex for the tail counter .
 * @ param expectedRawTail expected current value .
 * @ param updateRawTail to be applied .
 * @ return true if the update was successful otherwise false . */
public static boolean casRawTail ( final UnsafeBuffer metadataBuffer , final int partitionIndex , final long expectedRawTail , final long updateRawTail ) { } }

implementation: final int index = TERM_TAIL_COUNTERS_OFFSET + ( SIZE_OF_LONG * partitionIndex ) ; return metadataBuffer . compareAndSetLong ( index , expectedRawTail , updateRawTail ) ;


signature:
public class Archive { /** * Launch an { @ link Archive } with that communicates with an out of process { @ link io . aeron . driver . MediaDriver }
 * and await a shutdown signal .
 * @ param args command line argument which is a list for properties files as URLs or filenames . */
public static void main ( final String [ ] args ) { } }

implementation: loadPropertiesFiles ( args ) ; try ( Archive ignore = launch ( ) ) { new ShutdownSignalBarrier ( ) . await ( ) ; System . out . println ( "Shutdown Archive..." ) ; }


signature:
public class TerminateDriverFlyweight { /** * Fill the token buffer .
 * @ param tokenBuffer containing the optional token for the request .
 * @ param tokenOffset within the tokenBuffer at which the token begins .
 * @ param tokenLength of the token in the tokenBuffer .
 * @ return flyweight */
public TerminateDriverFlyweight tokenBuffer ( final DirectBuffer tokenBuffer , final int tokenOffset , final int tokenLength ) { } }

implementation: buffer . putInt ( TOKEN_LENGTH_OFFSET , tokenLength ) ; if ( null != tokenBuffer && tokenLength > 0 ) { buffer . putBytes ( tokenBufferOffset ( ) , tokenBuffer , tokenOffset , tokenLength ) ; } return this ;


signature:
public class RecoveryState { /** * Allocate a counter to represent the snapshot services should load on start .
 * @ param aeron to allocate the counter .
 * @ param tempBuffer to use for building the key and label without allocation .
 * @ param leadershipTermId at which the snapshot was taken .
 * @ param logPosition at which the snapshot was taken .
 * @ param timestamp the snapshot was taken .
 * @ param hasReplay flag is true if all or part of the log must be replayed .
 * @ param snapshotRecordingIds for the services to use during recovery indexed by service id .
 * @ return the { @ link Counter } for the recovery state . */
public static Counter allocate ( final Aeron aeron , final MutableDirectBuffer tempBuffer , final long leadershipTermId , final long logPosition , final long timestamp , final boolean hasReplay , final long ... snapshotRecordingIds ) { } }

implementation: tempBuffer . putLong ( LEADERSHIP_TERM_ID_OFFSET , leadershipTermId ) ; tempBuffer . putLong ( LOG_POSITION_OFFSET , logPosition ) ; tempBuffer . putLong ( TIMESTAMP_OFFSET , timestamp ) ; tempBuffer . putInt ( REPLAY_FLAG_OFFSET , hasReplay ? 1 : 0 ) ; final int serviceCount = snapshotRecordingIds . length ; tempBuffer . putInt ( SERVICE_COUNT_OFFSET , serviceCount ) ; final int keyLength = SNAPSHOT_RECORDING_IDS_OFFSET + ( serviceCount * SIZE_OF_LONG ) ; if ( keyLength > MAX_KEY_LENGTH ) { throw new ClusterException ( keyLength + " exceeds max key length " + MAX_KEY_LENGTH ) ; } for ( int i = 0 ; i < serviceCount ; i ++ ) { tempBuffer . putLong ( SNAPSHOT_RECORDING_IDS_OFFSET + ( i * SIZE_OF_LONG ) , snapshotRecordingIds [ i ] ) ; } final int labelOffset = BitUtil . align ( keyLength , SIZE_OF_INT ) ; int labelLength = 0 ; labelLength += tempBuffer . putStringWithoutLengthAscii ( labelOffset + labelLength , NAME ) ; labelLength += tempBuffer . putLongAscii ( keyLength + labelLength , leadershipTermId ) ; labelLength += tempBuffer . putStringWithoutLengthAscii ( labelOffset + labelLength , " logPosition=" ) ; labelLength += tempBuffer . putLongAscii ( labelOffset + labelLength , logPosition ) ; labelLength += tempBuffer . putStringWithoutLengthAscii ( labelOffset + labelLength , " hasReplay=" + hasReplay ) ; return aeron . addCounter ( RECOVERY_STATE_TYPE_ID , tempBuffer , 0 , keyLength , tempBuffer , labelOffset , labelLength ) ;


signature:
public class RecoveryState { /** * Find the active counter id for recovery state .
 * @ param counters to search within .
 * @ return the counter id if found otherwise { @ link CountersReader # NULL _ COUNTER _ ID } . */
public static int findCounterId ( final CountersReader counters ) { } }

implementation: final DirectBuffer buffer = counters . metaDataBuffer ( ) ; for ( int i = 0 , size = counters . maxCounterId ( ) ; i < size ; i ++ ) { if ( counters . getCounterState ( i ) == RECORD_ALLOCATED ) { final int recordOffset = CountersReader . metaDataOffset ( i ) ; if ( buffer . getInt ( recordOffset + TYPE_ID_OFFSET ) == RECOVERY_STATE_TYPE_ID ) { return i ; } } } return NULL_COUNTER_ID ;


signature:
public class RecoveryState { /** * Get the position at which the snapshot was taken . { @ link Aeron # NULL _ VALUE } if no snapshot for recovery .
 * @ param counters to search within .
 * @ param counterId for the active recovery counter .
 * @ return the log position if found otherwise { @ link Aeron # NULL _ VALUE } . */
public static long getLogPosition ( final CountersReader counters , final int counterId ) { } }

implementation: final DirectBuffer buffer = counters . metaDataBuffer ( ) ; if ( counters . getCounterState ( counterId ) == RECORD_ALLOCATED ) { final int recordOffset = CountersReader . metaDataOffset ( counterId ) ; if ( buffer . getInt ( recordOffset + TYPE_ID_OFFSET ) == RECOVERY_STATE_TYPE_ID ) { return buffer . getLong ( recordOffset + KEY_OFFSET + LOG_POSITION_OFFSET ) ; } } return NULL_VALUE ;


signature:
public class RecoveryState { /** * Has the recovery process got a log to replay ?
 * @ param counters to search within .
 * @ param counterId for the active recovery counter .
 * @ return true if a replay is required . */
public static boolean hasReplay ( final CountersReader counters , final int counterId ) { } }

implementation: final DirectBuffer buffer = counters . metaDataBuffer ( ) ; if ( counters . getCounterState ( counterId ) == RECORD_ALLOCATED ) { final int recordOffset = CountersReader . metaDataOffset ( counterId ) ; if ( buffer . getInt ( recordOffset + TYPE_ID_OFFSET ) == RECOVERY_STATE_TYPE_ID ) { return buffer . getInt ( recordOffset + KEY_OFFSET + REPLAY_FLAG_OFFSET ) == 1 ; } } return false ;


signature:
public class RecoveryState { /** * Get the recording id of the snapshot for a service .
 * @ param counters to search within .
 * @ param counterId for the active recovery counter .
 * @ param serviceId for the snapshot required .
 * @ return the count of replay terms if found otherwise { @ link Aeron # NULL _ VALUE } . */
public static long getSnapshotRecordingId ( final CountersReader counters , final int counterId , final int serviceId ) { } }

implementation: final DirectBuffer buffer = counters . metaDataBuffer ( ) ; if ( counters . getCounterState ( counterId ) == RECORD_ALLOCATED ) { final int recordOffset = CountersReader . metaDataOffset ( counterId ) ; if ( buffer . getInt ( recordOffset + TYPE_ID_OFFSET ) == RECOVERY_STATE_TYPE_ID ) { final int serviceCount = buffer . getInt ( recordOffset + KEY_OFFSET + SERVICE_COUNT_OFFSET ) ; if ( serviceId < 0 || serviceId >= serviceCount ) { throw new ClusterException ( "invalid serviceId " + serviceId + " for count of " + serviceCount ) ; } return buffer . getLong ( recordOffset + KEY_OFFSET + SNAPSHOT_RECORDING_IDS_OFFSET + ( serviceId * SIZE_OF_LONG ) ) ; } } throw new ClusterException ( "Active counter not found " + counterId ) ;


signature:
public class HeaderFlyweight { /** * Convert header flags to an array of chars to be human readable .
 * @ param flags to be converted .
 * @ return header flags converted to an array of chars to be human readable . */
public static char [ ] flagsToChars ( final short flags ) { } }

implementation: final char [ ] chars = new char [ ] { '0' , '0' , '0' , '0' , '0' , '0' , '0' , '0' } ; final int length = chars . length ; short mask = ( short ) ( 1 << ( length - 1 ) ) ; for ( int i = 0 ; i < length ; i ++ ) { if ( ( flags & mask ) == mask ) { chars [ i ] = '1' ; } mask >>= 1 ; } return chars ;


signature:
public class LossReport { /** * Create a new entry for recording loss on a given stream .
 * If not space is remaining in the error report then null is returned .
 * @ param initialBytesLost on the stream .
 * @ param timestampMs at which the first loss was observed .
 * @ param sessionId for the stream .
 * @ param streamId for the stream .
 * @ param channel for the stream .
 * @ param source of the stream .
 * @ return a new record or null if the error log has insufficient space . */
public ReportEntry createEntry ( final long initialBytesLost , final long timestampMs , final int sessionId , final int streamId , final String channel , final String source ) { } }

implementation: ReportEntry reportEntry = null ; final int requiredCapacity = CHANNEL_OFFSET + ( SIZE_OF_INT * 2 ) + channel . length ( ) + source . length ( ) ; if ( requiredCapacity <= ( buffer . capacity ( ) - nextRecordOffset ) ) { final int offset = nextRecordOffset ; buffer . putLong ( offset + TOTAL_BYTES_LOST_OFFSET , initialBytesLost ) ; buffer . putLong ( offset + FIRST_OBSERVATION_OFFSET , timestampMs ) ; buffer . putLong ( offset + LAST_OBSERVATION_OFFSET , timestampMs ) ; buffer . putInt ( offset + SESSION_ID_OFFSET , sessionId ) ; buffer . putInt ( offset + STREAM_ID_OFFSET , streamId ) ; final int encodedChannelLength = buffer . putStringAscii ( offset + CHANNEL_OFFSET , channel ) ; buffer . putStringAscii ( offset + CHANNEL_OFFSET + encodedChannelLength , source ) ; buffer . putLongOrdered ( offset + OBSERVATION_COUNT_OFFSET , 1 ) ; reportEntry = new ReportEntry ( buffer , offset ) ; nextRecordOffset += BitUtil . align ( requiredCapacity , ENTRY_ALIGNMENT ) ; } return reportEntry ;


signature:
public class IngressSessionDecorator { /** * Non - blocking publish of a partial buffer containing a message plus session header to a cluster .
 * This version of the method will set the timestamp value in the header to { @ link Aeron # NULL _ VALUE } .
 * @ param publication to be offer to .
 * @ param buffer containing message .
 * @ param offset offset in the buffer at which the encoded message begins .
 * @ param length in bytes of the encoded message .
 * @ return the same as { @ link Publication # offer ( DirectBuffer , int , int ) } . */
public long offer ( final Publication publication , final DirectBuffer buffer , final int offset , final int length ) { } }

implementation: return publication . offer ( headerBuffer , 0 , HEADER_LENGTH , buffer , offset , length , null ) ;


signature:
public class PublicationMessageFlyweight { /** * Set the channel field in ASCII
 * @ param channel field value
 * @ return flyweight */
public PublicationMessageFlyweight channel ( final String channel ) { } }

implementation: lengthOfChannel = buffer . putStringAscii ( offset + CHANNEL_OFFSET , channel ) ; return this ;


signature:
public class TermGapFiller { /** * Try to gap fill the current term at a given offset if the gap contains no data .
 * Note : the gap offset plus gap length must end on a { @ link FrameDescriptor # FRAME _ ALIGNMENT } boundary .
 * @ param logMetaDataBuffer containing the default headers
 * @ param termBuffer to gap fill
 * @ param termId for the current term .
 * @ param gapOffset to fill from
 * @ param gapLength to length of the gap .
 * @ return true if the gap has been filled with a padding record or false if data found . */
public static boolean tryFillGap ( final UnsafeBuffer logMetaDataBuffer , final UnsafeBuffer termBuffer , final int termId , final int gapOffset , final int gapLength ) { } }

implementation: int offset = ( gapOffset + gapLength ) - FRAME_ALIGNMENT ; while ( offset >= gapOffset ) { if ( 0 != termBuffer . getInt ( offset ) ) { return false ; } offset -= FRAME_ALIGNMENT ; } applyDefaultHeader ( logMetaDataBuffer , termBuffer , gapOffset ) ; frameType ( termBuffer , gapOffset , HDR_TYPE_PAD ) ; frameTermOffset ( termBuffer , gapOffset ) ; frameTermId ( termBuffer , gapOffset , termId ) ; frameLengthOrdered ( termBuffer , gapOffset , gapLength ) ; return true ;


signature:
public class TermReader { /** * Reads data from a term in a log buffer and updates a passed { @ link Position } so progress is not lost in the
 * event of an exception .
 * @ param termBuffer to be read for fragments .
 * @ param termOffset within the buffer that the read should begin .
 * @ param handler the handler for data that has been read
 * @ param fragmentsLimit limit the number of fragments read .
 * @ param header to be used for mapping over the header for a given fragment .
 * @ param errorHandler to be notified if an error occurs during the callback .
 * @ param currentPosition prior to reading further fragments
 * @ param subscriberPosition to be updated after reading with new position
 * @ return the number of fragments read */
public static int read ( final UnsafeBuffer termBuffer , final int termOffset , final FragmentHandler handler , final int fragmentsLimit , final Header header , final ErrorHandler errorHandler , final long currentPosition , final Position subscriberPosition ) { } }

implementation: int fragmentsRead = 0 ; int offset = termOffset ; final int capacity = termBuffer . capacity ( ) ; header . buffer ( termBuffer ) ; try { while ( fragmentsRead < fragmentsLimit && offset < capacity ) { final int frameLength = frameLengthVolatile ( termBuffer , offset ) ; if ( frameLength <= 0 ) { break ; } final int frameOffset = offset ; offset += BitUtil . align ( frameLength , FRAME_ALIGNMENT ) ; if ( ! isPaddingFrame ( termBuffer , frameOffset ) ) { header . offset ( frameOffset ) ; handler . onFragment ( termBuffer , frameOffset + HEADER_LENGTH , frameLength - HEADER_LENGTH , header ) ; ++ fragmentsRead ; } } } catch ( final Throwable t ) { errorHandler . onError ( t ) ; } finally { final long newPosition = currentPosition + ( offset - termOffset ) ; if ( newPosition > currentPosition ) { subscriberPosition . setOrdered ( newPosition ) ; } } return fragmentsRead ;


signature:
public class HeaderWriter { /** * Write a header to the term buffer in { @ link ByteOrder # LITTLE _ ENDIAN } format using the minimum instructions .
 * @ param termBuffer to be written to .
 * @ param offset at which the header should be written .
 * @ param length of the fragment including the header .
 * @ param termId of the current term buffer . */
public void write ( final UnsafeBuffer termBuffer , final int offset , final int length , final int termId ) { } }

implementation: termBuffer . putLongOrdered ( offset + FRAME_LENGTH_FIELD_OFFSET , versionFlagsType | ( ( - length ) & 0xFFFF_FFFFL ) ) ; UnsafeAccess . UNSAFE . storeFence ( ) ; termBuffer . putLong ( offset + TERM_OFFSET_FIELD_OFFSET , sessionId | offset ) ; termBuffer . putLong ( offset + STREAM_ID_FIELD_OFFSET , streamId | ( ( ( long ) termId ) << 32 ) ) ;


signature:
public class ClientSession { /** * Non - blocking publish of a partial buffer containing a message to a cluster .
 * @ param buffer containing message .
 * @ param offset offset in the buffer at which the encoded message begins .
 * @ param length in bytes of the encoded message .
 * @ return the same as { @ link Publication # offer ( DirectBuffer , int , int ) } when in { @ link Cluster . Role # LEADER }
 * otherwise { @ link # MOCKED _ OFFER } . */
public long offer ( final DirectBuffer buffer , final int offset , final int length ) { } }

implementation: return cluster . offer ( id , responsePublication , buffer , offset , length ) ;


signature:
public class BufferClaim { /** * Wrap a region of an underlying log buffer so can can represent a claimed space for use by a publisher .
 * @ param buffer to be wrapped .
 * @ param offset at which the claimed region begins including space for the header .
 * @ param length length of the underlying claimed region including space for the header . */
public final void wrap ( final AtomicBuffer buffer , final int offset , final int length ) { } }

implementation: this . buffer . wrap ( buffer , offset , length ) ;


signature:
public class BufferClaim { /** * Put bytes into the claimed buffer space for a message . To write multiple parts then use { @ link # buffer ( ) }
 * and { @ link # offset ( ) } .
 * @ param srcBuffer to copy into the claimed space .
 * @ param srcIndex in the source buffer from which to copy .
 * @ param length of the source buffer to copy .
 * @ return this for a fluent API . */
public final BufferClaim putBytes ( final DirectBuffer srcBuffer , final int srcIndex , final int length ) { } }

implementation: buffer . putBytes ( HEADER_LENGTH , srcBuffer , srcIndex , length ) ; return this ;


signature:
public class BufferClaim { /** * Commit the message to the log buffer so that is it available to subscribers . */
public final void commit ( ) { } }

implementation: int frameLength = buffer . capacity ( ) ; if ( ByteOrder . nativeOrder ( ) != LITTLE_ENDIAN ) { frameLength = Integer . reverseBytes ( frameLength ) ; } buffer . putIntOrdered ( FRAME_LENGTH_FIELD_OFFSET , frameLength ) ;


signature:
public class BufferClaim { /** * Abort a claim of the message space to the log buffer so that the log can progress by ignoring this claim . */
public final void abort ( ) { } }

implementation: int frameLength = buffer . capacity ( ) ; if ( ByteOrder . nativeOrder ( ) != LITTLE_ENDIAN ) { frameLength = Integer . reverseBytes ( frameLength ) ; } buffer . putShort ( TYPE_FIELD_OFFSET , ( short ) HDR_TYPE_PAD , LITTLE_ENDIAN ) ; buffer . putIntOrdered ( FRAME_LENGTH_FIELD_OFFSET , frameLength ) ;


signature:
public class MediaDriver { /** * Start Media Driver as a stand - alone process .
 * @ param args command line arguments */
public static void main ( final String [ ] args ) { } }

implementation: loadPropertiesFiles ( args ) ; final ShutdownSignalBarrier barrier = new ShutdownSignalBarrier ( ) ; final MediaDriver . Context ctx = new MediaDriver . Context ( ) ; ctx . terminationHook ( barrier :: signal ) ; try ( MediaDriver ignore = MediaDriver . launch ( ctx ) ) { barrier . await ( ) ; System . out . println ( "Shutdown Driver..." ) ; }


signature:
public class MediaDriver { /** * Launch an isolated MediaDriver embedded in the current process with a provided configuration ctx and a generated
 * aeronDirectoryName ( overwrites configured { @ link Context # aeronDirectoryName ( ) } ) that can be retrieved by calling
 * aeronDirectoryName .
 * If the aeronDirectoryName is set as a system property , or via context , to something different than
 * { @ link CommonContext # AERON _ DIR _ PROP _ DEFAULT } then this set value will be used .
 * @ param ctx containing the configuration options .
 * @ return the newly started MediaDriver . */
public static MediaDriver launchEmbedded ( final Context ctx ) { } }

implementation: if ( CommonContext . AERON_DIR_PROP_DEFAULT . equals ( ctx . aeronDirectoryName ( ) ) ) { ctx . aeronDirectoryName ( CommonContext . generateRandomDirName ( ) ) ; } return launch ( ctx ) ;


signature:
public class MediaDriver { /** * Shutdown the media driver by stopping all threads and freeing resources . */
public void close ( ) { } }

implementation: CloseHelper . close ( sharedRunner ) ; CloseHelper . close ( sharedNetworkRunner ) ; CloseHelper . close ( receiverRunner ) ; CloseHelper . close ( senderRunner ) ; CloseHelper . close ( conductorRunner ) ; CloseHelper . close ( sharedInvoker ) ; if ( ctx . useWindowsHighResTimer ( ) && SystemUtil . osName ( ) . startsWith ( "win" ) ) { if ( ! wasHighResTimerEnabled ) { HighResolutionTimer . disable ( ) ; } }