signature:
public class TermAppender { /** * Append an unfragmented message to the the term buffer .
 * @ param header for writing the default header .
 * @ param bufferOne containing the first part of the message .
 * @ param offsetOne at which the first part of the message begins .
 * @ param lengthOne of the first part of the message .
 * @ param bufferTwo containing the second part of the message .
 * @ param offsetTwo at which the second part of the message begins .
 * @ param lengthTwo of the second part of the message .
 * @ param reservedValueSupplier { @ link ReservedValueSupplier } for the frame .
 * @ param activeTermId used for flow control .
 * @ return the resulting offset of the term after the append on success otherwise { @ link # FAILED } */
public int appendUnfragmentedMessage ( final HeaderWriter header , final DirectBuffer bufferOne , final int offsetOne , final int lengthOne , final DirectBuffer bufferTwo , final int offsetTwo , final int lengthTwo , final ReservedValueSupplier reservedValueSupplier , final int activeTermId ) { } }

implementation: final int frameLength = lengthOne + lengthTwo + HEADER_LENGTH ; final int alignedLength = align ( frameLength , FRAME_ALIGNMENT ) ; final UnsafeBuffer termBuffer = this . termBuffer ; final int termLength = termBuffer . capacity ( ) ; final long rawTail = getAndAddRawTail ( alignedLength ) ; final int termId = termId ( rawTail ) ; final long termOffset = rawTail & 0xFFFF_FFFFL ; checkTerm ( activeTermId , termId ) ; long resultingOffset = termOffset + alignedLength ; if ( resultingOffset > termLength ) { resultingOffset = handleEndOfLogCondition ( termBuffer , termOffset , header , termLength , termId ) ; } else { final int frameOffset = ( int ) termOffset ; header . write ( termBuffer , frameOffset , frameLength , termId ) ; termBuffer . putBytes ( frameOffset + HEADER_LENGTH , bufferOne , offsetOne , lengthOne ) ; termBuffer . putBytes ( frameOffset + HEADER_LENGTH + lengthOne , bufferTwo , offsetTwo , lengthTwo ) ; if ( null != reservedValueSupplier ) { final long reservedValue = reservedValueSupplier . get ( termBuffer , frameOffset , frameLength ) ; termBuffer . putLong ( frameOffset + RESERVED_VALUE_OFFSET , reservedValue , LITTLE_ENDIAN ) ; } frameLengthOrdered ( termBuffer , frameOffset , frameLength ) ; } return ( int ) resultingOffset ;


signature:
public class TermAppender { /** * Append an unfragmented message to the the term buffer .
 * @ param header for writing the default header .
 * @ param buffer containing the message .
 * @ param offset at which the message begins .
 * @ param length of the message in the source buffer .
 * @ param reservedValueSupplier { @ link ReservedValueSupplier } for the frame .
 * @ param activeTermId used for flow control .
 * @ return the resulting offset of the term after the append on success otherwise { @ link # FAILED } */
public int appendUnfragmentedMessage ( final HeaderWriter header , final DirectBuffer buffer , final int offset , final int length , final ReservedValueSupplier reservedValueSupplier , final int activeTermId ) { } }

implementation: final int frameLength = length + HEADER_LENGTH ; final int alignedLength = align ( frameLength , FRAME_ALIGNMENT ) ; final UnsafeBuffer termBuffer = this . termBuffer ; final int termLength = termBuffer . capacity ( ) ; final long rawTail = getAndAddRawTail ( alignedLength ) ; final int termId = termId ( rawTail ) ; final long termOffset = rawTail & 0xFFFF_FFFFL ; checkTerm ( activeTermId , termId ) ; long resultingOffset = termOffset + alignedLength ; if ( resultingOffset > termLength ) { resultingOffset = handleEndOfLogCondition ( termBuffer , termOffset , header , termLength , termId ) ; } else { final int frameOffset = ( int ) termOffset ; header . write ( termBuffer , frameOffset , frameLength , termId ) ; termBuffer . putBytes ( frameOffset + HEADER_LENGTH , buffer , offset , length ) ; if ( null != reservedValueSupplier ) { final long reservedValue = reservedValueSupplier . get ( termBuffer , frameOffset , frameLength ) ; termBuffer . putLong ( frameOffset + RESERVED_VALUE_OFFSET , reservedValue , LITTLE_ENDIAN ) ; } frameLengthOrdered ( termBuffer , frameOffset , frameLength ) ; } return ( int ) resultingOffset ;


signature:
public class TermAppender { /** * Append a fragmented message to the the term buffer .
 * The message will be split up into fragments of MTU length minus header .
 * @ param header for writing the default header .
 * @ param bufferOne containing the first part of the message .
 * @ param offsetOne at which the first part of the message begins .
 * @ param lengthOne of the first part of the message .
 * @ param bufferTwo containing the second part of the message .
 * @ param offsetTwo at which the second part of the message begins .
 * @ param lengthTwo of the second part of the message .
 * @ param maxPayloadLength that the message will be fragmented into .
 * @ param reservedValueSupplier { @ link ReservedValueSupplier } for the frame .
 * @ param activeTermId used for flow control .
 * @ return the resulting offset of the term after the append on success otherwise { @ link # FAILED } . */
public int appendFragmentedMessage ( final HeaderWriter header , final DirectBuffer bufferOne , final int offsetOne , final int lengthOne , final DirectBuffer bufferTwo , final int offsetTwo , final int lengthTwo , final int maxPayloadLength , final ReservedValueSupplier reservedValueSupplier , final int activeTermId ) { } }

implementation: final int length = lengthOne + lengthTwo ; final int numMaxPayloads = length / maxPayloadLength ; final int remainingPayload = length % maxPayloadLength ; final int lastFrameLength = remainingPayload > 0 ? align ( remainingPayload + HEADER_LENGTH , FRAME_ALIGNMENT ) : 0 ; final int requiredLength = ( numMaxPayloads * ( maxPayloadLength + HEADER_LENGTH ) ) + lastFrameLength ; final UnsafeBuffer termBuffer = this . termBuffer ; final int termLength = termBuffer . capacity ( ) ; final long rawTail = getAndAddRawTail ( requiredLength ) ; final int termId = termId ( rawTail ) ; final long termOffset = rawTail & 0xFFFF_FFFFL ; checkTerm ( activeTermId , termId ) ; long resultingOffset = termOffset + requiredLength ; if ( resultingOffset > termLength ) { resultingOffset = handleEndOfLogCondition ( termBuffer , termOffset , header , termLength , termId ) ; } else { int frameOffset = ( int ) termOffset ; byte flags = BEGIN_FRAG_FLAG ; int remaining = length ; int positionOne = 0 ; int positionTwo = 0 ; do { final int bytesToWrite = Math . min ( remaining , maxPayloadLength ) ; final int frameLength = bytesToWrite + HEADER_LENGTH ; final int alignedLength = align ( frameLength , FRAME_ALIGNMENT ) ; header . write ( termBuffer , frameOffset , frameLength , termId ) ; int bytesWritten = 0 ; int payloadOffset = frameOffset + HEADER_LENGTH ; do { final int remainingOne = lengthOne - positionOne ; if ( remainingOne > 0 ) { final int numBytes = Math . min ( bytesToWrite - bytesWritten , remainingOne ) ; termBuffer . putBytes ( payloadOffset , bufferOne , offsetOne + positionOne , numBytes ) ; bytesWritten += numBytes ; payloadOffset += numBytes ; positionOne += numBytes ; } else { final int numBytes = Math . min ( bytesToWrite - bytesWritten , lengthTwo - positionTwo ) ; termBuffer . putBytes ( payloadOffset , bufferTwo , offsetTwo + positionTwo , numBytes ) ; bytesWritten += numBytes ; payloadOffset += numBytes ; positionTwo += numBytes ; } } while ( bytesWritten < bytesToWrite ) ; if ( remaining <= maxPayloadLength ) { flags |= END_FRAG_FLAG ; } frameFlags ( termBuffer , frameOffset , flags ) ; if ( null != reservedValueSupplier ) { final long reservedValue = reservedValueSupplier . get ( termBuffer , frameOffset , frameLength ) ; termBuffer . putLong ( frameOffset + RESERVED_VALUE_OFFSET , reservedValue , LITTLE_ENDIAN ) ; } frameLengthOrdered ( termBuffer , frameOffset , frameLength ) ; flags = 0 ; frameOffset += alignedLength ; remaining -= bytesToWrite ; } while ( remaining > 0 ) ; } return ( int ) resultingOffset ;


signature:
public class ArchiveProxy { /** * Connect to an archive on its control interface providing the response stream details .
 * @ param responseChannel for the control message responses .
 * @ param responseStreamId for the control message responses .
 * @ param correlationId for this request .
 * @ return true if successfully offered otherwise false . */
public boolean connect ( final String responseChannel , final int responseStreamId , final long correlationId ) { } }

implementation: connectRequestEncoder . wrapAndApplyHeader ( buffer , 0 , messageHeaderEncoder ) . correlationId ( correlationId ) . responseStreamId ( responseStreamId ) . version ( AeronArchive . Configuration . SEMANTIC_VERSION ) . responseChannel ( responseChannel ) ; return offerWithTimeout ( connectRequestEncoder . encodedLength ( ) , null ) ;


signature:
public class ArchiveProxy { /** * Try Connect to an archive on its control interface providing the response stream details . Only one attempt will
 * be made to offer the request .
 * @ param responseChannel for the control message responses .
 * @ param responseStreamId for the control message responses .
 * @ param correlationId for this request .
 * @ return true if successfully offered otherwise false . */
public boolean tryConnect ( final String responseChannel , final int responseStreamId , final long correlationId ) { } }

implementation: connectRequestEncoder . wrapAndApplyHeader ( buffer , 0 , messageHeaderEncoder ) . correlationId ( correlationId ) . responseStreamId ( responseStreamId ) . version ( AeronArchive . Configuration . SEMANTIC_VERSION ) . responseChannel ( responseChannel ) ; final int length = MessageHeaderEncoder . ENCODED_LENGTH + connectRequestEncoder . encodedLength ( ) ; return publication . offer ( buffer , 0 , length ) > 0 ;


signature:
public class ArchiveProxy { /** * Close this control session with the archive .
 * @ param controlSessionId with the archive .
 * @ return true if successfully offered otherwise false . */
public boolean closeSession ( final long controlSessionId ) { } }

implementation: closeSessionRequestEncoder . wrapAndApplyHeader ( buffer , 0 , messageHeaderEncoder ) . controlSessionId ( controlSessionId ) ; return offer ( closeSessionRequestEncoder . encodedLength ( ) ) ;


signature:
public class ArchiveProxy { /** * Start recording streams for a given channel and stream id pairing .
 * @ param channel to be recorded .
 * @ param streamId to be recorded .
 * @ param sourceLocation of the publication to be recorded .
 * @ param correlationId for this request .
 * @ param controlSessionId for this request .
 * @ return true if successfully offered otherwise false . */
public boolean startRecording ( final String channel , final int streamId , final SourceLocation sourceLocation , final long correlationId , final long controlSessionId ) { } }

implementation: startRecordingRequestEncoder . wrapAndApplyHeader ( buffer , 0 , messageHeaderEncoder ) . controlSessionId ( controlSessionId ) . correlationId ( correlationId ) . streamId ( streamId ) . sourceLocation ( sourceLocation ) . channel ( channel ) ; return offer ( startRecordingRequestEncoder . encodedLength ( ) ) ;


signature:
public class ArchiveProxy { /** * Stop an active recording .
 * @ param channel to be stopped .
 * @ param streamId to be stopped .
 * @ param correlationId for this request .
 * @ param controlSessionId for this request .
 * @ return true if successfully offered otherwise false . */
public boolean stopRecording ( final String channel , final int streamId , final long correlationId , final long controlSessionId ) { } }

implementation: stopRecordingRequestEncoder . wrapAndApplyHeader ( buffer , 0 , messageHeaderEncoder ) . controlSessionId ( controlSessionId ) . correlationId ( correlationId ) . streamId ( streamId ) . channel ( channel ) ; return offer ( stopRecordingRequestEncoder . encodedLength ( ) ) ;


signature:
public class ArchiveProxy { /** * Stop an active recording by the { @ link Subscription # registrationId ( ) } it was registered with .
 * @ param subscriptionId that identifies the subscription in the archive doing the recording .
 * @ param correlationId for this request .
 * @ param controlSessionId for this request .
 * @ return true if successfully offered otherwise false . */
public boolean stopRecording ( final long subscriptionId , final long correlationId , final long controlSessionId ) { } }

implementation: stopRecordingSubscriptionRequestEncoder . wrapAndApplyHeader ( buffer , 0 , messageHeaderEncoder ) . controlSessionId ( controlSessionId ) . correlationId ( correlationId ) . subscriptionId ( subscriptionId ) ; return offer ( stopRecordingSubscriptionRequestEncoder . encodedLength ( ) ) ;


signature:
public class ArchiveProxy { /** * Replay a recording from a given position .
 * @ param recordingId to be replayed .
 * @ param position from which the replay should be started .
 * @ param length of the stream to be replayed . Use { @ link Long # MAX _ VALUE } to follow a live stream .
 * @ param replayChannel to which the replay should be sent .
 * @ param replayStreamId to which the replay should be sent .
 * @ param correlationId for this request .
 * @ param controlSessionId for this request .
 * @ return true if successfully offered otherwise false . */
public boolean replay ( final long recordingId , final long position , final long length , final String replayChannel , final int replayStreamId , final long correlationId , final long controlSessionId ) { } }

implementation: replayRequestEncoder . wrapAndApplyHeader ( buffer , 0 , messageHeaderEncoder ) . controlSessionId ( controlSessionId ) . correlationId ( correlationId ) . recordingId ( recordingId ) . position ( position ) . length ( length ) . replayStreamId ( replayStreamId ) . replayChannel ( replayChannel ) ; return offer ( replayRequestEncoder . encodedLength ( ) ) ;


signature:
public class ArchiveProxy { /** * Stop an existing replay session .
 * @ param replaySessionId that should be stopped .
 * @ param correlationId for this request .
 * @ param controlSessionId for this request .
 * @ return true if successfully offered otherwise false . */
public boolean stopReplay ( final long replaySessionId , final long correlationId , final long controlSessionId ) { } }

implementation: stopReplayRequestEncoder . wrapAndApplyHeader ( buffer , 0 , messageHeaderEncoder ) . controlSessionId ( controlSessionId ) . correlationId ( correlationId ) . replaySessionId ( replaySessionId ) ; return offer ( replayRequestEncoder . encodedLength ( ) ) ;


signature:
public class ArchiveProxy { /** * List a range of recording descriptors .
 * @ param fromRecordingId at which to begin listing .
 * @ param recordCount for the number of descriptors to be listed .
 * @ param correlationId for this request .
 * @ param controlSessionId for this request .
 * @ return true if successfully offered otherwise false . */
public boolean listRecordings ( final long fromRecordingId , final int recordCount , final long correlationId , final long controlSessionId ) { } }

implementation: listRecordingsRequestEncoder . wrapAndApplyHeader ( buffer , 0 , messageHeaderEncoder ) . controlSessionId ( controlSessionId ) . correlationId ( correlationId ) . fromRecordingId ( fromRecordingId ) . recordCount ( recordCount ) ; return offer ( listRecordingsRequestEncoder . encodedLength ( ) ) ;


signature:
public class ArchiveProxy { /** * List a range of recording descriptors which match a channel URI fragment and stream id .
 * @ param fromRecordingId at which to begin listing .
 * @ param recordCount for the number of descriptors to be listed .
 * @ param channelFragment to match recordings on from the original channel URI in the archive descriptor .
 * @ param streamId to match recordings on .
 * @ param correlationId for this request .
 * @ param controlSessionId for this request .
 * @ return true if successfully offered otherwise false . */
public boolean listRecordingsForUri ( final long fromRecordingId , final int recordCount , final String channelFragment , final int streamId , final long correlationId , final long controlSessionId ) { } }

implementation: listRecordingsForUriRequestEncoder . wrapAndApplyHeader ( buffer , 0 , messageHeaderEncoder ) . controlSessionId ( controlSessionId ) . correlationId ( correlationId ) . fromRecordingId ( fromRecordingId ) . recordCount ( recordCount ) . streamId ( streamId ) . channel ( channelFragment ) ; return offer ( listRecordingsForUriRequestEncoder . encodedLength ( ) ) ;


signature:
public class ArchiveProxy { /** * List a recording descriptor for a given recording id .
 * @ param recordingId at which to begin listing .
 * @ param correlationId for this request .
 * @ param controlSessionId for this request .
 * @ return true if successfully offered otherwise false . */
public boolean listRecording ( final long recordingId , final long correlationId , final long controlSessionId ) { } }

implementation: listRecordingRequestEncoder . wrapAndApplyHeader ( buffer , 0 , messageHeaderEncoder ) . controlSessionId ( controlSessionId ) . correlationId ( correlationId ) . recordingId ( recordingId ) ; return offer ( listRecordingRequestEncoder . encodedLength ( ) ) ;


signature:
public class ArchiveProxy { /** * Extend an existing , non - active , recorded stream for a the same channel and stream id .
 * The channel must be configured for the initial position from which it will be extended . This can be done
 * with { @ link ChannelUriStringBuilder # initialPosition ( long , int , int ) } . The details required to initialise can
 * be found by calling { @ link # listRecording ( long , long , long ) } .
 * @ param channel to be recorded .
 * @ param streamId to be recorded .
 * @ param sourceLocation of the publication to be recorded .
 * @ param recordingId to be extended .
 * @ param correlationId for this request .
 * @ param controlSessionId for this request .
 * @ return true if successfully offered otherwise false . */
public boolean extendRecording ( final String channel , final int streamId , final SourceLocation sourceLocation , final long recordingId , final long correlationId , final long controlSessionId ) { } }

implementation: extendRecordingRequestEncoder . wrapAndApplyHeader ( buffer , 0 , messageHeaderEncoder ) . controlSessionId ( controlSessionId ) . correlationId ( correlationId ) . recordingId ( recordingId ) . streamId ( streamId ) . sourceLocation ( sourceLocation ) . channel ( channel ) ; return offer ( extendRecordingRequestEncoder . encodedLength ( ) ) ;


signature:
public class ArchiveProxy { /** * Get the recorded position of an active recording .
 * @ param recordingId of the active recording that the position is being requested for .
 * @ param correlationId for this request .
 * @ param controlSessionId for this request .
 * @ return true if successfully offered otherwise false . */
public boolean getRecordingPosition ( final long recordingId , final long correlationId , final long controlSessionId ) { } }

implementation: recordingPositionRequestEncoder . wrapAndApplyHeader ( buffer , 0 , messageHeaderEncoder ) . controlSessionId ( controlSessionId ) . correlationId ( correlationId ) . recordingId ( recordingId ) ; return offer ( recordingPositionRequestEncoder . encodedLength ( ) ) ;


signature:
public class ArchiveProxy { /** * Truncate a stopped recording to a given position that is less than the stopped position . The provided position
 * must be on a fragment boundary . Truncating a recording to the start position effectively deletes the recording .
 * @ param recordingId of the stopped recording to be truncated .
 * @ param position to which the recording will be truncated .
 * @ param correlationId for this request .
 * @ param controlSessionId for this request .
 * @ return true if successfully offered otherwise false . */
public boolean truncateRecording ( final long recordingId , final long position , final long correlationId , final long controlSessionId ) { } }

implementation: truncateRecordingRequestEncoder . wrapAndApplyHeader ( buffer , 0 , messageHeaderEncoder ) . controlSessionId ( controlSessionId ) . correlationId ( correlationId ) . recordingId ( recordingId ) . position ( position ) ; return offer ( truncateRecordingRequestEncoder . encodedLength ( ) ) ;


signature:
public class ArchiveProxy { /** * Get the stop position of a recording .
 * @ param recordingId of the recording that the stop position is being requested for .
 * @ param correlationId for this request .
 * @ param controlSessionId for this request .
 * @ return true if successfully offered otherwise false . */
public boolean getStopPosition ( final long recordingId , final long correlationId , final long controlSessionId ) { } }

implementation: stopPositionRequestEncoder . wrapAndApplyHeader ( buffer , 0 , messageHeaderEncoder ) . controlSessionId ( controlSessionId ) . correlationId ( correlationId ) . recordingId ( recordingId ) ; return offer ( stopPositionRequestEncoder . encodedLength ( ) ) ;


signature:
public class ArchiveProxy { /** * Find the last recording that matches the given criteria .
 * @ param minRecordingId to search back to .
 * @ param channelFragment for a contains match on the original channel stored with the archive descriptor .
 * @ param streamId of the recording to match .
 * @ param sessionId of the recording to match .
 * @ param correlationId for this request .
 * @ param controlSessionId for this request .
 * @ return true if successfully offered otherwise false . */
public boolean findLastMatchingRecording ( final long minRecordingId , final String channelFragment , final int streamId , final int sessionId , final long correlationId , final long controlSessionId ) { } }

implementation: findLastMatchingRecordingRequestEncoder . wrapAndApplyHeader ( buffer , 0 , messageHeaderEncoder ) . controlSessionId ( controlSessionId ) . correlationId ( correlationId ) . minRecordingId ( minRecordingId ) . sessionId ( sessionId ) . streamId ( streamId ) . channel ( channelFragment ) ; return offer ( findLastMatchingRecordingRequestEncoder . encodedLength ( ) ) ;


signature:
public class ArchiveProxy { /** * List registered subscriptions in the archive which have been used to record streams .
 * @ param pseudoIndex in the list of active recording subscriptions .
 * @ param subscriptionCount for the number of descriptors to be listed .
 * @ param channelFragment for a contains match on the stripped channel used with the registered subscription .
 * @ param streamId for the subscription .
 * @ param applyStreamId when matching .
 * @ param correlationId for this request .
 * @ param controlSessionId for this request .
 * @ return true if successfully offered otherwise false . */
public boolean listRecordingSubscriptions ( final int pseudoIndex , final int subscriptionCount , final String channelFragment , final int streamId , final boolean applyStreamId , final long correlationId , final long controlSessionId ) { } }

implementation: listRecordingSubscriptionsRequestEncoder . wrapAndApplyHeader ( buffer , 0 , messageHeaderEncoder ) . controlSessionId ( controlSessionId ) . correlationId ( correlationId ) . pseudoIndex ( pseudoIndex ) . subscriptionCount ( subscriptionCount ) . applyStreamId ( applyStreamId ? BooleanType . TRUE : BooleanType . FALSE ) . streamId ( streamId ) . channel ( channelFragment ) ; return offer ( listRecordingSubscriptionsRequestEncoder . encodedLength ( ) ) ;


signature:
public class UdpChannel { /** * Parse channel URI and create a { @ link UdpChannel } .
 * @ param channelUriString to parse
 * @ return a new { @ link UdpChannel }
 * @ throws InvalidChannelException if an error occurs . */
@ SuppressWarnings ( "MethodLength" ) public static UdpChannel parse ( final String channelUriString ) { } }

implementation: try { final ChannelUri channelUri = ChannelUri . parse ( channelUriString ) ; validateConfiguration ( channelUri ) ; InetSocketAddress endpointAddress = getEndpointAddress ( channelUri ) ; final InetSocketAddress explicitControlAddress = getExplicitControlAddress ( channelUri ) ; final String tagIdStr = channelUri . channelTag ( ) ; final String controlMode = channelUri . get ( CommonContext . MDC_CONTROL_MODE_PARAM_NAME ) ; final boolean hasNoDistinguishingCharacteristic = null == endpointAddress && null == explicitControlAddress && null == tagIdStr ; if ( hasNoDistinguishingCharacteristic && null == controlMode ) { throw new IllegalArgumentException ( "Aeron URIs for UDP must specify an endpoint address, control address, tag-id, or control-mode" ) ; } if ( null != endpointAddress && endpointAddress . isUnresolved ( ) ) { throw new UnknownHostException ( "could not resolve endpoint address: " + endpointAddress ) ; } if ( null != explicitControlAddress && explicitControlAddress . isUnresolved ( ) ) { throw new UnknownHostException ( "could not resolve control address: " + explicitControlAddress ) ; } final Context context = new Context ( ) . uriStr ( channelUriString ) . channelUri ( channelUri ) . hasNoDistinguishingCharacteristic ( hasNoDistinguishingCharacteristic ) ; if ( null != tagIdStr ) { context . hasTagId ( true ) . tagId ( Long . parseLong ( tagIdStr ) ) ; } if ( null == endpointAddress ) { endpointAddress = new InetSocketAddress ( "0.0.0.0" , 0 ) ; } if ( endpointAddress . getAddress ( ) . isMulticastAddress ( ) ) { final InetSocketAddress controlAddress = getMulticastControlAddress ( endpointAddress ) ; final InterfaceSearchAddress searchAddress = getInterfaceSearchAddress ( channelUri ) ; final NetworkInterface localInterface = findInterface ( searchAddress ) ; final InetSocketAddress resolvedAddress = resolveToAddressOfInterface ( localInterface , searchAddress ) ; context . isMulticast ( true ) . localControlAddress ( resolvedAddress ) . remoteControlAddress ( controlAddress ) . localDataAddress ( resolvedAddress ) . remoteDataAddress ( endpointAddress ) . localInterface ( localInterface ) . protocolFamily ( getProtocolFamily ( endpointAddress . getAddress ( ) ) ) . canonicalForm ( canonicalise ( resolvedAddress , endpointAddress ) ) ; final String ttlValue = channelUri . get ( CommonContext . TTL_PARAM_NAME ) ; if ( null != ttlValue ) { context . hasMulticastTtl ( true ) . multicastTtl ( Integer . parseInt ( ttlValue ) ) ; } } else if ( null != explicitControlAddress ) { context . hasExplicitControl ( true ) . remoteControlAddress ( endpointAddress ) . remoteDataAddress ( endpointAddress ) . localControlAddress ( explicitControlAddress ) . localDataAddress ( explicitControlAddress ) . protocolFamily ( getProtocolFamily ( endpointAddress . getAddress ( ) ) ) . canonicalForm ( canonicalise ( explicitControlAddress , endpointAddress ) ) ; } else { final InterfaceSearchAddress searchAddress = getInterfaceSearchAddress ( channelUri ) ; final InetSocketAddress localAddress = searchAddress . getInetAddress ( ) . isAnyLocalAddress ( ) ? searchAddress . getAddress ( ) : resolveToAddressOfInterface ( findInterface ( searchAddress ) , searchAddress ) ; final String uniqueCanonicalFormSuffix = hasNoDistinguishingCharacteristic ? ( "-" + UNIQUE_CANONICAL_FORM_VALUE . getAndAdd ( 1 ) ) : "" ; context . remoteControlAddress ( endpointAddress ) . remoteDataAddress ( endpointAddress ) . localControlAddress ( localAddress ) . localDataAddress ( localAddress ) . protocolFamily ( getProtocolFamily ( endpointAddress . getAddress ( ) ) ) . canonicalForm ( canonicalise ( localAddress , endpointAddress ) + uniqueCanonicalFormSuffix ) ; } return new UdpChannel ( context ) ; } catch ( final Exception ex ) { throw new InvalidChannelException ( ErrorCode . INVALID_CHANNEL , ex ) ; }


signature:
public class UdpChannel { /** * Return a string which is a canonical form of the channel suitable for use as a file or directory
 * name and also as a method of hashing , etc .
 * A canonical form :
 * < ul >
 * < li > begins with the string " UDP - " < / li >
 * < li > has all addresses converted to hexadecimal < / li >
 * < li > uses " - " as all field separators < / li >
 * < / ul >
 * The general format is :
 * UDP - interface - localPort - remoteAddress - remotePort
 * @ param localData address / interface for the channel
 * @ param remoteData address for the channel
 * @ return canonical representation as a string */
public static String canonicalise ( final InetSocketAddress localData , final InetSocketAddress remoteData ) { } }

implementation: final StringBuilder builder = new StringBuilder ( 48 ) ; builder . append ( "UDP-" ) ; toHex ( builder , localData . getAddress ( ) . getAddress ( ) ) . append ( '-' ) . append ( localData . getPort ( ) ) ; builder . append ( '-' ) ; toHex ( builder , remoteData . getAddress ( ) . getAddress ( ) ) . append ( '-' ) . append ( remoteData . getPort ( ) ) ; return builder . toString ( ) ;


signature:
public class UdpChannel { /** * Does this channel have a tag match to another channel including endpoints .
 * @ param udpChannel to match against .
 * @ return true if there is a match otherwise false . */
public boolean matchesTag ( final UdpChannel udpChannel ) { } }

implementation: if ( ! hasTag || ! udpChannel . hasTag ( ) || tag != udpChannel . tag ( ) ) { return false ; } if ( udpChannel . remoteData ( ) . getAddress ( ) . isAnyLocalAddress ( ) && udpChannel . remoteData ( ) . getPort ( ) == 0 && udpChannel . localData ( ) . getAddress ( ) . isAnyLocalAddress ( ) && udpChannel . localData ( ) . getPort ( ) == 0 ) { return true ; } throw new IllegalArgumentException ( "matching tag has set endpoint or control address" ) ;


signature:
public class UdpChannel { /** * Get the endpoint address from the URI .
 * @ param uri to check
 * @ return endpoint address for URI */
public static InetSocketAddress destinationAddress ( final ChannelUri uri ) { } }

implementation: try { validateConfiguration ( uri ) ; return getEndpointAddress ( uri ) ; } catch ( final Exception ex ) { throw new InvalidChannelException ( ErrorCode . INVALID_CHANNEL , ex ) ; }


signature:
public class UdpChannel { /** * Used for debugging to get a human readable description of the channel .
 * @ return a human readable description of the channel . */
public String description ( ) { } }

implementation: final StringBuilder builder = new StringBuilder ( "UdpChannel - " ) ; if ( null != localInterface ) { builder . append ( "interface: " ) . append ( localInterface . getDisplayName ( ) ) . append ( ", " ) ; } builder . append ( "localData: " ) . append ( localData ) . append ( ", remoteData: " ) . append ( remoteData ) . append ( ", ttl: " ) . append ( multicastTtl ) ; return builder . toString ( ) ;


signature:
public class PublicationImage { /** * { @ inheritDoc } */
public void close ( ) { } }

implementation: hwmPosition . close ( ) ; rebuildPosition . close ( ) ; for ( final ReadablePosition position : subscriberPositions ) { position . close ( ) ; } for ( int i = 0 , size = untetheredSubscriptions . size ( ) ; i < size ; i ++ ) { final UntetheredSubscription untetheredSubscription = untetheredSubscriptions . get ( i ) ; if ( UntetheredSubscription . RESTING == untetheredSubscription . state ) { untetheredSubscription . position . close ( ) ; } } congestionControl . close ( ) ; rawLog . close ( ) ;


signature:
public class PublicationImage { /** * { @ inheritDoc } */
public void addSubscriber ( final SubscriptionLink subscriptionLink , final ReadablePosition subscriberPosition ) { } }

implementation: subscriberPositions = ArrayUtil . add ( subscriberPositions , subscriberPosition ) ; if ( ! subscriptionLink . isTether ( ) ) { untetheredSubscriptions . add ( new UntetheredSubscription ( subscriptionLink , subscriberPosition , timeOfLastStatusMessageScheduleNs ) ) ; }


signature:
public class PublicationImage { /** * { @ inheritDoc } */
public void removeSubscriber ( final SubscriptionLink subscriptionLink , final ReadablePosition subscriberPosition ) { } }

implementation: subscriberPositions = ArrayUtil . remove ( subscriberPositions , subscriberPosition ) ; subscriberPosition . close ( ) ; if ( ! subscriptionLink . isTether ( ) ) { for ( int lastIndex = untetheredSubscriptions . size ( ) - 1 , i = lastIndex ; i >= 0 ; i -- ) { if ( untetheredSubscriptions . get ( i ) . subscriptionLink == subscriptionLink ) { ArrayListUtil . fastUnorderedRemove ( untetheredSubscriptions , i , lastIndex ) ; break ; } } }


signature:
public class PublicationImage { /** * Called from the { @ link LossDetector } when gap is detected by the { @ link DriverConductor } thread .
 * { @ inheritDoc } */
public void onGapDetected ( final int termId , final int termOffset , final int length ) { } }

implementation: final long changeNumber = beginLossChange + 1 ; beginLossChange = changeNumber ; lossTermId = termId ; lossTermOffset = termOffset ; lossLength = length ; endLossChange = changeNumber ; if ( null != reportEntry ) { reportEntry . recordObservation ( length , cachedEpochClock . time ( ) ) ; } else if ( null != lossReport ) { reportEntry = lossReport . createEntry ( length , cachedEpochClock . time ( ) , sessionId , streamId , channel ( ) , sourceAddress . toString ( ) ) ; if ( null == reportEntry ) { lossReport = null ; } }


signature:
public class PublicationImage { /** * Add a destination to this image so it can merge streams .
 * @ param transportIndex from which packets will arrive .
 * @ param transport from which packets will arrive . */
void addDestination ( final int transportIndex , final ReceiveDestinationUdpTransport transport ) { } }

implementation: imageConnections = ArrayUtil . ensureCapacity ( imageConnections , transportIndex + 1 ) ; if ( transport . isMulticast ( ) ) { imageConnections [ transportIndex ] = new ImageConnection ( cachedNanoClock . nanoTime ( ) , transport . udpChannel ( ) . remoteControl ( ) ) ; } else if ( transport . hasExplicitControl ( ) ) { imageConnections [ transportIndex ] = new ImageConnection ( cachedNanoClock . nanoTime ( ) , transport . explicitControlAddress ( ) ) ; }


signature:
public class PublicationImage { /** * Called from the { @ link DriverConductor } .
 * @ param nowNs in nanoseconds
 * @ param statusMessageTimeoutNs for sending of Status Messages . */
final void trackRebuild ( final long nowNs , final long statusMessageTimeoutNs ) { } }

implementation: long minSubscriberPosition = Long . MAX_VALUE ; long maxSubscriberPosition = Long . MIN_VALUE ; for ( final ReadablePosition subscriberPosition : subscriberPositions ) { final long position = subscriberPosition . getVolatile ( ) ; minSubscriberPosition = Math . min ( minSubscriberPosition , position ) ; maxSubscriberPosition = Math . max ( maxSubscriberPosition , position ) ; } final long rebuildPosition = Math . max ( this . rebuildPosition . get ( ) , maxSubscriberPosition ) ; final long hwmPosition = this . hwmPosition . getVolatile ( ) ; final long scanOutcome = lossDetector . scan ( termBuffers [ indexByPosition ( rebuildPosition , positionBitsToShift ) ] , rebuildPosition , hwmPosition , nowNs , termLengthMask , positionBitsToShift , initialTermId ) ; final int rebuildTermOffset = ( int ) rebuildPosition & termLengthMask ; final long newRebuildPosition = ( rebuildPosition - rebuildTermOffset ) + rebuildOffset ( scanOutcome ) ; this . rebuildPosition . proposeMaxOrdered ( newRebuildPosition ) ; final long ccOutcome = congestionControl . onTrackRebuild ( nowNs , minSubscriberPosition , nextSmPosition , hwmPosition , rebuildPosition , newRebuildPosition , lossFound ( scanOutcome ) ) ; final int windowLength = CongestionControl . receiverWindowLength ( ccOutcome ) ; final int threshold = CongestionControl . threshold ( windowLength ) ; if ( CongestionControl . shouldForceStatusMessage ( ccOutcome ) || ( ( timeOfLastStatusMessageScheduleNs + statusMessageTimeoutNs ) - nowNs < 0 ) || ( minSubscriberPosition > ( nextSmPosition + threshold ) ) ) { scheduleStatusMessage ( nowNs , minSubscriberPosition , windowLength ) ; cleanBufferTo ( minSubscriberPosition - ( termLengthMask + 1 ) ) ; }


signature:
public class PublicationImage { /** * Insert frame into term buffer .
 * @ param termId for the data packet to insert into the appropriate term .
 * @ param termOffset for the start of the packet in the term .
 * @ param buffer for the data packet to insert into the appropriate term .
 * @ param length of the data packet
 * @ param transportIndex which the packet came from .
 * @ param srcAddress which the packet came from .
 * @ return number of bytes applied as a result of this insertion . */
int insertPacket ( final int termId , final int termOffset , final UnsafeBuffer buffer , final int length , final int transportIndex , final InetSocketAddress srcAddress ) { } }

implementation: final boolean isHeartbeat = DataHeaderFlyweight . isHeartbeat ( buffer , length ) ; final long packetPosition = computePosition ( termId , termOffset , positionBitsToShift , initialTermId ) ; final long proposedPosition = isHeartbeat ? packetPosition : packetPosition + length ; if ( ! isFlowControlUnderRun ( packetPosition ) && ! isFlowControlOverRun ( proposedPosition ) ) { trackConnection ( transportIndex , srcAddress , lastPacketTimestampNs ) ; if ( isHeartbeat ) { if ( DataHeaderFlyweight . isEndOfStream ( buffer ) && ! isEndOfStream && allEos ( transportIndex ) ) { LogBufferDescriptor . endOfStreamPosition ( rawLog . metaData ( ) , proposedPosition ) ; isEndOfStream = true ; } heartbeatsReceived . incrementOrdered ( ) ; } else { final UnsafeBuffer termBuffer = termBuffers [ indexByPosition ( packetPosition , positionBitsToShift ) ] ; TermRebuilder . insert ( termBuffer , termOffset , buffer , length ) ; } lastPacketTimestampNs = cachedNanoClock . nanoTime ( ) ; hwmPosition . proposeMaxOrdered ( proposedPosition ) ; } return length ;


signature:
public class PublicationImage { /** * To be called from the { @ link Receiver } to see if a image should be retained .
 * @ param nowNs current time to check against for activity .
 * @ return true if the image should be retained otherwise false . */
boolean hasActivityAndNotEndOfStream ( final long nowNs ) { } }

implementation: boolean isActive = true ; if ( ( ( lastPacketTimestampNs + imageLivenessTimeoutNs ) - nowNs < 0 ) || ( isEndOfStream && rebuildPosition . getVolatile ( ) >= hwmPosition . get ( ) ) ) { isActive = false ; } return isActive ;


signature:
public class PublicationImage { /** * Called from the { @ link Receiver } to send any pending Status Messages .
 * @ return number of work items processed . */
int sendPendingStatusMessage ( ) { } }

implementation: int workCount = 0 ; if ( ACTIVE == state ) { final long changeNumber = endSmChange ; if ( changeNumber != lastSmChangeNumber ) { final long smPosition = nextSmPosition ; final int receiverWindowLength = nextSmReceiverWindowLength ; UNSAFE . loadFence ( ) ; if ( changeNumber == beginSmChange ) { final int termId = computeTermIdFromPosition ( smPosition , positionBitsToShift , initialTermId ) ; final int termOffset = ( int ) smPosition & termLengthMask ; channelEndpoint . sendStatusMessage ( imageConnections , sessionId , streamId , termId , termOffset , receiverWindowLength , ( byte ) 0 ) ; statusMessagesSent . incrementOrdered ( ) ; lastSmPosition = smPosition ; lastSmWindowLimit = smPosition + receiverWindowLength ; lastSmChangeNumber = changeNumber ; } workCount = 1 ; } } return workCount ;


signature:
public class PublicationImage { /** * Called from the { @ link Receiver } thread to processing any pending loss of packets .
 * @ return number of work items processed . */
int processPendingLoss ( ) { } }

implementation: int workCount = 0 ; final long changeNumber = endLossChange ; if ( changeNumber != lastLossChangeNumber ) { final int termId = lossTermId ; final int termOffset = lossTermOffset ; final int length = lossLength ; UNSAFE . loadFence ( ) ; if ( changeNumber == beginLossChange ) { if ( isReliable ) { channelEndpoint . sendNakMessage ( imageConnections , sessionId , streamId , termId , termOffset , length ) ; nakMessagesSent . incrementOrdered ( ) ; } else { final UnsafeBuffer termBuffer = termBuffers [ indexByTerm ( initialTermId , termId ) ] ; if ( tryFillGap ( rawLog . metaData ( ) , termBuffer , termId , termOffset , length ) ) { lossGapFills . incrementOrdered ( ) ; } } lastLossChangeNumber = changeNumber ; } workCount = 1 ; } return workCount ;


signature:
public class PublicationImage { /** * Called from the { @ link Receiver } thread to check for initiating an RTT measurement .
 * @ param nowNs in nanoseconds
 * @ return number of work items processed . */
int initiateAnyRttMeasurements ( final long nowNs ) { } }

implementation: int workCount = 0 ; if ( congestionControl . shouldMeasureRtt ( nowNs ) ) { final long preciseTimeNs = nanoClock . nanoTime ( ) ; channelEndpoint . sendRttMeasurement ( imageConnections , sessionId , streamId , preciseTimeNs , 0 , true ) ; congestionControl . onRttMeasurementSent ( preciseTimeNs ) ; workCount = 1 ; } return workCount ;


signature:
public class PublicationImage { /** * Called from the { @ link Receiver } upon receiving an RTT Measurement that is a reply .
 * @ param header of the measurement message .
 * @ param transportIndex that the RTT Measurement came in on .
 * @ param srcAddress from the sender requesting the measurement */
void onRttMeasurement ( final RttMeasurementFlyweight header , @ SuppressWarnings ( "unused" ) final int transportIndex , final InetSocketAddress srcAddress ) { } }

implementation: final long nowNs = nanoClock . nanoTime ( ) ; final long rttInNs = nowNs - header . echoTimestampNs ( ) - header . receptionDelta ( ) ; congestionControl . onRttMeasurement ( nowNs , rttInNs , srcAddress ) ;


signature:
public class PublicationImage { /** * { @ inheritDoc } */
public void onTimeEvent ( final long timeNs , final long timesMs , final DriverConductor conductor ) { } }

implementation: switch ( state ) { case ACTIVE : checkUntetheredSubscriptions ( timeNs , conductor ) ; break ; case INACTIVE : if ( isDrained ( ) ) { state = State . LINGER ; timeOfLastStateChangeNs = timeNs ; conductor . transitionToLinger ( this ) ; } isTrackingRebuild = false ; break ; case LINGER : if ( ( timeOfLastStateChangeNs + imageLivenessTimeoutNs ) - timeNs < 0 ) { state = State . DONE ; conductor . cleanupImage ( this ) ; } break ; }


signature:
public class ClusterControl { /** * Map a { @ link CountersReader } over the provided { @ link File } for the CnC file .
 * @ param cncFile for the counters .
 * @ return a { @ link CountersReader } over the provided CnC file . */
public static CountersReader mapCounters ( final File cncFile ) { } }

implementation: final MappedByteBuffer cncByteBuffer = IoUtil . mapExistingFile ( cncFile , "cnc" ) ; final DirectBuffer cncMetaData = createMetaDataBuffer ( cncByteBuffer ) ; final int cncVersion = cncMetaData . getInt ( cncVersionOffset ( 0 ) ) ; if ( CncFileDescriptor . CNC_VERSION != cncVersion ) { throw new AeronException ( "Aeron CnC version does not match: version=" + cncVersion + " required=" + CNC_VERSION ) ; } return new CountersReader ( createCountersMetaDataBuffer ( cncByteBuffer , cncMetaData ) , createCountersValuesBuffer ( cncByteBuffer , cncMetaData ) , StandardCharsets . US_ASCII ) ;


signature:
public class ClusterControl { /** * Find the control toggle counter or return null if not found .
 * @ param counters to search for the control toggle .
 * @ return the control toggle counter or return null if not found . */
public static AtomicCounter findControlToggle ( final CountersReader counters ) { } }

implementation: final AtomicBuffer buffer = counters . metaDataBuffer ( ) ; for ( int i = 0 , size = counters . maxCounterId ( ) ; i < size ; i ++ ) { final int recordOffset = CountersReader . metaDataOffset ( i ) ; if ( counters . getCounterState ( i ) == RECORD_ALLOCATED && buffer . getInt ( recordOffset + TYPE_ID_OFFSET ) == CONTROL_TOGGLE_TYPE_ID ) { return new AtomicCounter ( counters . valuesBuffer ( ) , i , null ) ; } } return null ;


signature:
public class ChannelUri { /** * Put a key and value pair in the map of params .
 * @ param key of the param to be put .
 * @ param value of the param to be put .
 * @ return the existing value otherwise null . */
public String put ( final String key , final String value ) { } }

implementation: return params . put ( key , value ) ;


signature:
public class ChannelUri { /** * Get the channel tag , if it exists , that refers to an another channel .
 * @ return channel tag if it exists or null if not in this URI .
 * @ see CommonContext # TAGS _ PARAM _ NAME
 * @ see CommonContext # TAG _ PREFIX */
public String channelTag ( ) { } }

implementation: return ( null != tags && tags . length > CHANNEL_TAG_INDEX ) ? tags [ CHANNEL_TAG_INDEX ] : null ;


signature:
public class ChannelUri { /** * Get the entity tag , if it exists , that refers to an entity such as subscription or publication .
 * @ return entity tag if it exists or null if not in this URI .
 * @ see CommonContext # TAGS _ PARAM _ NAME
 * @ see CommonContext # TAG _ PREFIX */
public String entityTag ( ) { } }

implementation: return ( null != tags && tags . length > ENTITY_TAG_INDEX ) ? tags [ ENTITY_TAG_INDEX ] : null ;


signature:
public class ChannelUri { /** * Initialise a channel for restarting a publication at a given position .
 * @ param position at which the publication should be started .
 * @ param initialTermId what which the stream would start .
 * @ param termLength for the stream . */
public void initialPosition ( final long position , final int initialTermId , final int termLength ) { } }

implementation: if ( position < 0 || 0 != ( position & ( FRAME_ALIGNMENT - 1 ) ) ) { throw new IllegalArgumentException ( "invalid position: " + position ) ; } final int bitsToShift = LogBufferDescriptor . positionBitsToShift ( termLength ) ; final int termId = LogBufferDescriptor . computeTermIdFromPosition ( position , bitsToShift , initialTermId ) ; final int termOffset = ( int ) ( position & ( termLength - 1 ) ) ; put ( INITIAL_TERM_ID_PARAM_NAME , Integer . toString ( initialTermId ) ) ; put ( TERM_ID_PARAM_NAME , Integer . toString ( termId ) ) ; put ( TERM_OFFSET_PARAM_NAME , Integer . toString ( termOffset ) ) ; put ( TERM_LENGTH_PARAM_NAME , Integer . toString ( termLength ) ) ;


signature:
public class ChannelUri { /** * Parse a { @ link CharSequence } which contains an Aeron URI .
 * @ param cs to be parsed .
 * @ return a new { @ link ChannelUri } representing the URI string . */
public static ChannelUri parse ( final CharSequence cs ) { } }

implementation: int position = 0 ; final String prefix ; if ( startsWith ( cs , 0 , SPY_PREFIX ) ) { prefix = SPY_QUALIFIER ; position = SPY_PREFIX . length ( ) ; } else { prefix = "" ; } if ( ! startsWith ( cs , position , AERON_PREFIX ) ) { throw new IllegalArgumentException ( "Aeron URIs must start with 'aeron:', found: '" + cs + "'" ) ; } else { position += AERON_PREFIX . length ( ) ; } final StringBuilder builder = new StringBuilder ( ) ; final Map < String , String > params = new Object2ObjectHashMap < > ( ) ; String media = null ; String key = null ; State state = State . MEDIA ; for ( int i = position ; i < cs . length ( ) ; i ++ ) { final char c = cs . charAt ( i ) ; switch ( state ) { case MEDIA : switch ( c ) { case '?' : media = builder . toString ( ) ; builder . setLength ( 0 ) ; state = State . PARAMS_KEY ; break ; case ':' : throw new IllegalArgumentException ( "encountered ':' within media definition" ) ; default : builder . append ( c ) ; } break ; case PARAMS_KEY : if ( c == '=' ) { key = builder . toString ( ) ; builder . setLength ( 0 ) ; state = State . PARAMS_VALUE ; } else { builder . append ( c ) ; } break ; case PARAMS_VALUE : if ( c == '|' ) { params . put ( key , builder . toString ( ) ) ; builder . setLength ( 0 ) ; state = State . PARAMS_KEY ; } else { builder . append ( c ) ; } break ; default : throw new IllegalStateException ( "unexpected state=" + state ) ; } } switch ( state ) { case MEDIA : media = builder . toString ( ) ; break ; case PARAMS_VALUE : params . put ( key , builder . toString ( ) ) ; break ; default : throw new IllegalArgumentException ( "no more input found, state=" + state ) ; } return new ChannelUri ( prefix , media , params ) ;


signature:
public class ChannelUri { /** * Add a sessionId to a given channel .
 * @ param channel to add sessionId to .
 * @ param sessionId to add to channel .
 * @ return new string that represents channel with sessionId added . */
public static String addSessionId ( final String channel , final int sessionId ) { } }

implementation: final ChannelUri channelUri = ChannelUri . parse ( channel ) ; channelUri . put ( CommonContext . SESSION_ID_PARAM_NAME , Integer . toString ( sessionId ) ) ; return channelUri . toString ( ) ;


signature:
public class ChannelUri { /** * Get the value of the tag from a given parameter value .
 * @ param paramValue to extract the tag value from .
 * @ return the value of the tag or { @ link # INVALID _ TAG } if not tagged .
 * @ see CommonContext # TAGS _ PARAM _ NAME
 * @ see CommonContext # TAG _ PREFIX */
public static long getTag ( final String paramValue ) { } }

implementation: return isTagged ( paramValue ) ? AsciiEncoding . parseLongAscii ( paramValue , 4 , paramValue . length ( ) - 4 ) : INVALID_TAG ;


signature:
public class ExclusiveTermAppender { /** * Claim length of a the term buffer for writing in the message with zero copy semantics .
 * @ param termId for the current term .
 * @ param termOffset in the term at which to append .
 * @ param header for writing the default header .
 * @ param length of the message to be written .
 * @ param bufferClaim to be updated with the claimed region .
 * @ return the resulting offset of the term after the append on success otherwise { @ link # FAILED } . */
public int claim ( final int termId , final int termOffset , final HeaderWriter header , final int length , final BufferClaim bufferClaim ) { } }

implementation: final int frameLength = length + HEADER_LENGTH ; final int alignedLength = align ( frameLength , FRAME_ALIGNMENT ) ; final UnsafeBuffer termBuffer = this . termBuffer ; final int termLength = termBuffer . capacity ( ) ; int resultingOffset = termOffset + alignedLength ; putRawTailOrdered ( termId , resultingOffset ) ; if ( resultingOffset > termLength ) { resultingOffset = handleEndOfLogCondition ( termBuffer , termOffset , header , termLength , termId ) ; } else { header . write ( termBuffer , termOffset , frameLength , termId ) ; bufferClaim . wrap ( termBuffer , termOffset , frameLength ) ; } return resultingOffset ;


signature:
public class ExclusiveTermAppender { /** * Pad a length of the term buffer with a padding record .
 * @ param termId for the current term .
 * @ param termOffset in the term at which to append .
 * @ param header for writing the default header .
 * @ param length of the padding to be written .
 * @ return the resulting offset of the term after success otherwise { @ link # FAILED } . */
public int appendPadding ( final int termId , final int termOffset , final HeaderWriter header , final int length ) { } }

implementation: final int frameLength = length + HEADER_LENGTH ; final int alignedLength = align ( frameLength , FRAME_ALIGNMENT ) ; final UnsafeBuffer termBuffer = this . termBuffer ; final int termLength = termBuffer . capacity ( ) ; int resultingOffset = termOffset + alignedLength ; putRawTailOrdered ( termId , resultingOffset ) ; if ( resultingOffset > termLength ) { resultingOffset = handleEndOfLogCondition ( termBuffer , termOffset , header , termLength , termId ) ; } else { header . write ( termBuffer , termOffset , frameLength , termId ) ; frameType ( termBuffer , termOffset , PADDING_FRAME_TYPE ) ; frameLengthOrdered ( termBuffer , termOffset , frameLength ) ; } return resultingOffset ;


signature:
public class ExclusiveTermAppender { /** * Append an unfragmented message to the the term buffer .
 * @ param termId for the current term .
 * @ param termOffset in the term at which to append .
 * @ param header for writing the default header .
 * @ param srcBuffer containing the message .
 * @ param srcOffset at which the message begins .
 * @ param length of the message in the source buffer .
 * @ param reservedValueSupplier { @ link ReservedValueSupplier } for the frame .
 * @ return the resulting offset of the term after the append on success otherwise { @ link # FAILED } . */
public int appendUnfragmentedMessage ( final int termId , final int termOffset , final HeaderWriter header , final DirectBuffer srcBuffer , final int srcOffset , final int length , final ReservedValueSupplier reservedValueSupplier ) { } }

implementation: final int frameLength = length + HEADER_LENGTH ; final int alignedLength = align ( frameLength , FRAME_ALIGNMENT ) ; final UnsafeBuffer termBuffer = this . termBuffer ; final int termLength = termBuffer . capacity ( ) ; int resultingOffset = termOffset + alignedLength ; putRawTailOrdered ( termId , resultingOffset ) ; if ( resultingOffset > termLength ) { resultingOffset = handleEndOfLogCondition ( termBuffer , termOffset , header , termLength , termId ) ; } else { header . write ( termBuffer , termOffset , frameLength , termId ) ; termBuffer . putBytes ( termOffset + HEADER_LENGTH , srcBuffer , srcOffset , length ) ; if ( null != reservedValueSupplier ) { final long reservedValue = reservedValueSupplier . get ( termBuffer , termOffset , frameLength ) ; termBuffer . putLong ( termOffset + RESERVED_VALUE_OFFSET , reservedValue , LITTLE_ENDIAN ) ; } frameLengthOrdered ( termBuffer , termOffset , frameLength ) ; } return resultingOffset ;


signature:
public class ExclusiveTermAppender { /** * Append a fragmented message to the the term buffer .
 * The message will be split up into fragments of MTU length minus header .
 * @ param termId for the current term .
 * @ param termOffset in the term at which to append .
 * @ param header for writing the default header .
 * @ param vectors to the buffers .
 * @ param length of the message in the source buffer .
 * @ param maxPayloadLength that the message will be fragmented into .
 * @ param reservedValueSupplier { @ link ReservedValueSupplier } for the frame .
 * @ return the resulting offset of the term after the append on success otherwise { @ link # FAILED } . */
public int appendFragmentedMessage ( final int termId , final int termOffset , final HeaderWriter header , final DirectBufferVector [ ] vectors , final int length , final int maxPayloadLength , final ReservedValueSupplier reservedValueSupplier ) { } }

implementation: final int numMaxPayloads = length / maxPayloadLength ; final int remainingPayload = length % maxPayloadLength ; final int lastFrameLength = remainingPayload > 0 ? align ( remainingPayload + HEADER_LENGTH , FRAME_ALIGNMENT ) : 0 ; final int requiredLength = ( numMaxPayloads * ( maxPayloadLength + HEADER_LENGTH ) ) + lastFrameLength ; final UnsafeBuffer termBuffer = this . termBuffer ; final int termLength = termBuffer . capacity ( ) ; int resultingOffset = termOffset + requiredLength ; putRawTailOrdered ( termId , resultingOffset ) ; if ( resultingOffset > termLength ) { resultingOffset = handleEndOfLogCondition ( termBuffer , termOffset , header , termLength , termId ) ; } else { int frameOffset = termOffset ; byte flags = BEGIN_FRAG_FLAG ; int remaining = length ; int vectorIndex = 0 ; int vectorOffset = 0 ; do { final int bytesToWrite = Math . min ( remaining , maxPayloadLength ) ; final int frameLength = bytesToWrite + HEADER_LENGTH ; final int alignedLength = align ( frameLength , FRAME_ALIGNMENT ) ; header . write ( termBuffer , frameOffset , frameLength , termId ) ; int bytesWritten = 0 ; int payloadOffset = frameOffset + HEADER_LENGTH ; do { final DirectBufferVector vector = vectors [ vectorIndex ] ; final int vectorRemaining = vector . length - vectorOffset ; final int numBytes = Math . min ( bytesToWrite - bytesWritten , vectorRemaining ) ; termBuffer . putBytes ( payloadOffset , vector . buffer , vector . offset + vectorOffset , numBytes ) ; bytesWritten += numBytes ; payloadOffset += numBytes ; vectorOffset += numBytes ; if ( vectorRemaining <= numBytes ) { vectorIndex ++ ; vectorOffset = 0 ; } } while ( bytesWritten < bytesToWrite ) ; if ( remaining <= maxPayloadLength ) { flags |= END_FRAG_FLAG ; } frameFlags ( termBuffer , frameOffset , flags ) ; if ( null != reservedValueSupplier ) { final long reservedValue = reservedValueSupplier . get ( termBuffer , frameOffset , frameLength ) ; termBuffer . putLong ( frameOffset + RESERVED_VALUE_OFFSET , reservedValue , LITTLE_ENDIAN ) ; } frameLengthOrdered ( termBuffer , frameOffset , frameLength ) ; flags = 0 ; frameOffset += alignedLength ; remaining -= bytesToWrite ; } while ( remaining > 0 ) ; } return resultingOffset ;


signature:
public class NetworkPublication { /** * Update the publishers limit for flow control as part of the conductor duty cycle .
 * @ return 1 if the limit has been updated otherwise 0. */
final int updatePublisherLimit ( ) { } }

implementation: int workCount = 0 ; final long senderPosition = this . senderPosition . getVolatile ( ) ; if ( hasReceivers || ( spiesSimulateConnection && spyPositions . length > 0 ) ) { long minConsumerPosition = senderPosition ; for ( final ReadablePosition spyPosition : spyPositions ) { minConsumerPosition = Math . min ( minConsumerPosition , spyPosition . getVolatile ( ) ) ; } final long proposedPublisherLimit = minConsumerPosition + termWindowLength ; if ( publisherLimit . proposeMaxOrdered ( proposedPublisherLimit ) ) { cleanBuffer ( proposedPublisherLimit ) ; workCount = 1 ; } } else if ( publisherLimit . get ( ) > senderPosition ) { publisherLimit . setOrdered ( senderPosition ) ; } return workCount ;


signature:
public class NetworkUtil { /** * Search for a list of network interfaces that match the specified address and subnet prefix .
 * The results will be ordered by the length of the subnet prefix
 * ( { @ link InterfaceAddress # getNetworkPrefixLength ( ) } ) . If no results match , then the collection
 * will be empty .
 * @ param address to search for on the { @ link NetworkInterface } s .
 * @ param subnetPrefix to limit the search .
 * @ return { @ link NetworkInterface } s that match the supplied criteria , ordered by the length
 * of the subnet prefix . Empty if none match .
 * @ throws SocketException if an error occurs */
public static NetworkInterface [ ] filterBySubnet ( final InetAddress address , final int subnetPrefix ) throws SocketException { } }

implementation: return filterBySubnet ( NetworkInterfaceShim . DEFAULT , address , subnetPrefix ) ;


signature:
public class NetworkUtil { /** * Allocate a direct { @ link ByteBuffer } that is padded at the end with at least alignment bytes .
 * @ param capacity for the buffer .
 * @ param alignment for the buffer .
 * @ return the direct { @ link ByteBuffer } . */
public static ByteBuffer allocateDirectAlignedAndPadded ( final int capacity , final int alignment ) { } }

implementation: final ByteBuffer buffer = BufferUtil . allocateDirectAligned ( capacity + alignment , alignment ) ; buffer . limit ( buffer . limit ( ) - alignment ) ; return buffer . slice ( ) ;


signature:
public class StreamCounter { /** * Allocate a counter for tracking a position on a stream of messages .
 * @ param tempBuffer to be used for labels and key .
 * @ param name of the counter for the label .
 * @ param typeId of the counter for classification .
 * @ param countersManager from which to allocated the underlying storage .
 * @ param registrationId to be associated with the counter .
 * @ param sessionId for the stream of messages .
 * @ param streamId for the stream of messages .
 * @ param channel for the stream of messages .
 * @ return a new { @ link UnsafeBufferPosition } for tracking the stream . */
public static UnsafeBufferPosition allocate ( final MutableDirectBuffer tempBuffer , final String name , final int typeId , final CountersManager countersManager , final long registrationId , final int sessionId , final int streamId , final String channel ) { } }

implementation: return new UnsafeBufferPosition ( ( UnsafeBuffer ) countersManager . valuesBuffer ( ) , allocateCounterId ( tempBuffer , name , typeId , countersManager , registrationId , sessionId , streamId , channel ) , countersManager ) ;


signature:
public class StreamCounter { /** * Return the label name for a counter type identifier .
 * @ param typeId of the counter .
 * @ return the label name as a String . */
public static String labelName ( final int typeId ) { } }

implementation: switch ( typeId ) { case PublisherLimit . PUBLISHER_LIMIT_TYPE_ID : return PublisherLimit . NAME ; case SenderPos . SENDER_POSITION_TYPE_ID : return SenderPos . NAME ; case ReceiverHwm . RECEIVER_HWM_TYPE_ID : return ReceiverHwm . NAME ; case SubscriberPos . SUBSCRIBER_POSITION_TYPE_ID : return SubscriberPos . NAME ; case ReceiverPos . RECEIVER_POS_TYPE_ID : return ReceiverPos . NAME ; case SenderLimit . SENDER_LIMIT_TYPE_ID : return SenderLimit . NAME ; case PublisherPos . PUBLISHER_POS_TYPE_ID : return PublisherPos . NAME ; case SenderBpe . SENDER_BPE_TYPE_ID : return SenderBpe . NAME ; default : return "<unknown>" ; }


signature:
public class TermScanner { /** * Scan the term buffer for availability of new message fragments from a given offset up to a maxLength of bytes .
 * @ param termBuffer to be scanned for new message fragments .
 * @ param offset at which the scan should begin .
 * @ param maxLength in bytes of how much should be scanned .
 * @ return resulting status of the scan which packs the available bytes and padding into a long . */
public static long scanForAvailability ( final UnsafeBuffer termBuffer , final int offset , final int maxLength ) { } }

implementation: final int limit = Math . min ( maxLength , termBuffer . capacity ( ) - offset ) ; int available = 0 ; int padding = 0 ; do { final int termOffset = offset + available ; final int frameLength = frameLengthVolatile ( termBuffer , termOffset ) ; if ( frameLength <= 0 ) { break ; } int alignedFrameLength = align ( frameLength , FRAME_ALIGNMENT ) ; if ( isPaddingFrame ( termBuffer , termOffset ) ) { padding = alignedFrameLength - HEADER_LENGTH ; alignedFrameLength = HEADER_LENGTH ; } available += alignedFrameLength ; if ( available > limit ) { available -= alignedFrameLength ; padding = 0 ; break ; } } while ( 0 == padding && available < limit ) ; return pack ( padding , available ) ;


signature:
public class ChannelUriStringBuilder { /** * Clear out all the values thus setting back to the initial state .
 * @ return this for a fluent API . */
public ChannelUriStringBuilder clear ( ) { } }

implementation: prefix = null ; media = null ; endpoint = null ; networkInterface = null ; controlEndpoint = null ; controlMode = null ; tags = null ; alias = null ; reliable = null ; ttl = null ; mtu = null ; termLength = null ; initialTermId = null ; termId = null ; termOffset = null ; sessionId = null ; linger = null ; sparse = null ; eos = null ; tether = null ; isSessionIdTagged = false ; return this ;


signature:
public class ChannelUriStringBuilder { /** * Validates that the collection of set parameters are valid together .
 * @ return this for a fluent API .
 * @ throws IllegalStateException if the combination of params is invalid . */
public ChannelUriStringBuilder validate ( ) { } }

implementation:
if ( null == media ) { throw new IllegalStateException ( "media type is mandatory" ) ; } if ( CommonContext . UDP_MEDIA . equals ( media ) && ( null == endpoint && null == controlEndpoint ) ) { throw new IllegalStateException ( "either 'endpoint' or 'control' must be specified for UDP." ) ; } int count = 0 ; count += null == initialTermId ? 0 : 1 ; count += null == termId ? 0 : 1 ; count += null == termOffset ? 0 : 1 ; if ( count > 0 ) { if ( count < 3 ) { throw new IllegalStateException ( "if any of then a complete set of 'initialTermId', 'termId', and 'termOffset' must be provided" ) ; } if ( termId - initialTermId < 0 ) // lgtm [ java / dereferenced - value - may - be - null ]
{ throw new IllegalStateException ( "difference greater than 2^31 - 1: termId=" + termId + " - initialTermId=" + initialTermId ) ; } if ( null != termLength && termOffset > termLength ) // lgtm [ java / dereferenced - value - may - be - null ]
{ throw new IllegalStateException ( "termOffset=" + termOffset + " > termLength=" + termLength ) ; } } return this ;


signature:
public class ChannelUriStringBuilder { /** * Set the prefix for taking an addition action such as spying on an outgoing publication with " aeron - spy " .
 * @ param prefix to be applied to the URI before the the scheme .
 * @ return this for a fluent API .
 * @ see ChannelUri # SPY _ QUALIFIER */
public ChannelUriStringBuilder prefix ( final String prefix ) { } }

implementation: if ( null != prefix && ! prefix . equals ( "" ) && ! prefix . equals ( SPY_QUALIFIER ) ) { throw new IllegalArgumentException ( "invalid prefix: " + prefix ) ; } this . prefix = prefix ; return this ;


signature:
public class ChannelUriStringBuilder { /** * Set the media for this channel . Valid values are " udp " and " ipc " .
 * @ param media for this channel .
 * @ return this for a fluent API . */
public ChannelUriStringBuilder media ( final String media ) { } }

implementation: switch ( media ) { case CommonContext . UDP_MEDIA : case CommonContext . IPC_MEDIA : break ; default : throw new IllegalArgumentException ( "invalid media: " + media ) ; } this . media = media ; return this ;


signature:
public class ChannelUriStringBuilder { /** * Set the control mode for multi - destination - cast . Set to " manual " for allowing control from the publication API .
 * @ param controlMode for taking control of MDC .
 * @ return this for a fluent API .
 * @ see Publication # addDestination ( String )
 * @ see Publication # removeDestination ( String )
 * @ see CommonContext # MDC _ CONTROL _ MODE _ PARAM _ NAME
 * @ see CommonContext # MDC _ CONTROL _ MODE _ MANUAL
 * @ see CommonContext # MDC _ CONTROL _ MODE _ DYNAMIC */
public ChannelUriStringBuilder controlMode ( final String controlMode ) { } }

implementation: if ( null != controlMode && ! controlMode . equals ( CommonContext . MDC_CONTROL_MODE_MANUAL ) && ! controlMode . equals ( CommonContext . MDC_CONTROL_MODE_DYNAMIC ) ) { throw new IllegalArgumentException ( "invalid control mode: " + controlMode ) ; } this . controlMode = controlMode ; return this ;


signature:
public class ChannelUriStringBuilder { /** * Set the Time To Live ( TTL ) for a multicast datagram . Valid values are 0-255 for the number of hops the datagram
 * can progress along .
 * @ param ttl value for a multicast datagram .
 * @ return this for a fluent API .
 * @ see CommonContext # TTL _ PARAM _ NAME */
public ChannelUriStringBuilder ttl ( final Integer ttl ) { } }

implementation: if ( null != ttl && ( ttl < 0 || ttl > 255 ) ) { throw new IllegalArgumentException ( "TTL not in range 0-255: " + ttl ) ; } this . ttl = ttl ; return this ;


signature:
public class ChannelUriStringBuilder { /** * Set the maximum transmission unit ( MTU ) including Aeron header for a datagram payload . If this is greater
 * than the network MTU for UDP then the packet will be fragmented and can amplify the impact of loss .
 * @ param mtu the maximum transmission unit including Aeron header for a datagram payload .
 * @ return this for a fluent API .
 * @ see CommonContext # MTU _ LENGTH _ PARAM _ NAME */
public ChannelUriStringBuilder mtu ( final Integer mtu ) { } }

implementation: if ( null != mtu ) { if ( mtu < 32 || mtu > 65504 ) { throw new IllegalArgumentException ( "MTU not in range 32-65504: " + mtu ) ; } if ( ( mtu & ( FRAME_ALIGNMENT - 1 ) ) != 0 ) { throw new IllegalArgumentException ( "MTU not a multiple of FRAME_ALIGNMENT: mtu=" + mtu ) ; } } this . mtu = mtu ; return this ;


signature:
public class ChannelUriStringBuilder { /** * Set the length of buffer used for each term of the log . Valid values are powers of 2 in the 64K - 1G range .
 * @ param termLength of the buffer used for each term of the log .
 * @ return this for a fluent API .
 * @ see CommonContext # TERM _ LENGTH _ PARAM _ NAME */
public ChannelUriStringBuilder termLength ( final Integer termLength ) { } }

implementation: if ( null != termLength ) { LogBufferDescriptor . checkTermLength ( termLength ) ; } this . termLength = termLength ; return this ;


signature:
public class ChannelUriStringBuilder { /** * Set the offset within a term at which a publication will start . This when combined with the term id can establish
 * a starting position .
 * @ param termOffset within a term at which a publication will start .
 * @ return this for a fluent API .
 * @ see CommonContext # TERM _ OFFSET _ PARAM _ NAME */
public ChannelUriStringBuilder termOffset ( final Integer termOffset ) { } }

implementation: if ( null != termOffset ) { if ( ( termOffset < 0 || termOffset > LogBufferDescriptor . TERM_MAX_LENGTH ) ) { throw new IllegalArgumentException ( "term offset not in range 0-1g: " + termOffset ) ; } if ( 0 != ( termOffset & ( FRAME_ALIGNMENT - 1 ) ) ) { throw new IllegalArgumentException ( "term offset not multiple of FRAME_ALIGNMENT: " + termOffset ) ; } } this . termOffset = termOffset ; return this ;


signature:
public class ChannelUriStringBuilder { /** * Set the time a network publication will linger in nanoseconds after being drained . This time is so that tail loss
 * can be recovered .
 * @ param lingerNs time for the publication after it is drained .
 * @ return this for a fluent API .
 * @ see CommonContext # LINGER _ PARAM _ NAME */
public ChannelUriStringBuilder linger ( final Long lingerNs ) { } }

implementation: if ( null != lingerNs && lingerNs < 0 ) { throw new IllegalArgumentException ( "linger value cannot be negative: " + lingerNs ) ; } this . linger = lingerNs ; return this ;


signature:
public class ChannelUriStringBuilder { /** * Initialise a channel for restarting a publication at a given position .
 * @ param position at which the publication should be started .
 * @ param initialTermId what which the stream would start .
 * @ param termLength for the stream .
 * @ return this for a fluent API . */
public ChannelUriStringBuilder initialPosition ( final long position , final int initialTermId , final int termLength ) { } }

implementation: if ( position < 0 || 0 != ( position & ( FRAME_ALIGNMENT - 1 ) ) ) { throw new IllegalArgumentException ( "invalid position: " + position ) ; } final int bitsToShift = LogBufferDescriptor . positionBitsToShift ( termLength ) ; this . initialTermId = initialTermId ; this . termId = LogBufferDescriptor . computeTermIdFromPosition ( position , bitsToShift , initialTermId ) ; this . termOffset = ( int ) ( position & ( termLength - 1 ) ) ; this . termLength = termLength ; return this ;


signature:
public class ChannelUriStringBuilder { /** * Build a channel URI String for the given parameters .
 * @ return a channel URI String for the given parameters . */
@ SuppressWarnings ( "MethodLength" ) public String build ( ) { } }

implementation: sb . setLength ( 0 ) ; if ( null != prefix && ! "" . equals ( prefix ) ) { sb . append ( prefix ) . append ( ':' ) ; } sb . append ( ChannelUri . AERON_SCHEME ) . append ( ':' ) . append ( media ) . append ( '?' ) ; if ( null != tags ) { sb . append ( TAGS_PARAM_NAME ) . append ( '=' ) . append ( tags ) . append ( '|' ) ; } if ( null != endpoint ) { sb . append ( ENDPOINT_PARAM_NAME ) . append ( '=' ) . append ( endpoint ) . append ( '|' ) ; } if ( null != networkInterface ) { sb . append ( INTERFACE_PARAM_NAME ) . append ( '=' ) . append ( networkInterface ) . append ( '|' ) ; } if ( null != controlEndpoint ) { sb . append ( MDC_CONTROL_PARAM_NAME ) . append ( '=' ) . append ( controlEndpoint ) . append ( '|' ) ; } if ( null != controlMode ) { sb . append ( MDC_CONTROL_MODE_PARAM_NAME ) . append ( '=' ) . append ( controlMode ) . append ( '|' ) ; } if ( null != mtu ) { sb . append ( MTU_LENGTH_PARAM_NAME ) . append ( '=' ) . append ( mtu . intValue ( ) ) . append ( '|' ) ; } if ( null != termLength ) { sb . append ( TERM_LENGTH_PARAM_NAME ) . append ( '=' ) . append ( termLength . intValue ( ) ) . append ( '|' ) ; } if ( null != initialTermId ) { sb . append ( INITIAL_TERM_ID_PARAM_NAME ) . append ( '=' ) . append ( initialTermId . intValue ( ) ) . append ( '|' ) ; } if ( null != termId ) { sb . append ( TERM_ID_PARAM_NAME ) . append ( '=' ) . append ( termId . intValue ( ) ) . append ( '|' ) ; } if ( null != termOffset ) { sb . append ( TERM_OFFSET_PARAM_NAME ) . append ( '=' ) . append ( termOffset . intValue ( ) ) . append ( '|' ) ; } if ( null != sessionId ) { sb . append ( SESSION_ID_PARAM_NAME ) . append ( '=' ) . append ( prefixTag ( isSessionIdTagged , sessionId ) ) . append ( '|' ) ; } if ( null != ttl ) { sb . append ( TTL_PARAM_NAME ) . append ( '=' ) . append ( ttl . intValue ( ) ) . append ( '|' ) ; } if ( null != reliable ) { sb . append ( RELIABLE_STREAM_PARAM_NAME ) . append ( '=' ) . append ( reliable ) . append ( '|' ) ; } if ( null != linger ) { sb . append ( LINGER_PARAM_NAME ) . append ( '=' ) . append ( linger . intValue ( ) ) . append ( '|' ) ; } if ( null != alias ) { sb . append ( ALIAS_PARAM_NAME ) . append ( '=' ) . append ( alias ) . append ( '|' ) ; } if ( null != sparse ) { sb . append ( SPARSE_PARAM_NAME ) . append ( '=' ) . append ( sparse ) . append ( '|' ) ; } if ( null != eos ) { sb . append ( EOS_PARAM_NAME ) . append ( '=' ) . append ( eos ) . append ( '|' ) ; } if ( null != tether ) { sb . append ( TETHER_PARAM_NAME ) . append ( '=' ) . append ( tether ) . append ( '|' ) ; } final char lastChar = sb . charAt ( sb . length ( ) - 1 ) ; if ( lastChar == '|' || lastChar == '?' ) { sb . setLength ( sb . length ( ) - 1 ) ; } return sb . toString ( ) ;


signature:
public class UdpChannelTransport { /** * Throw a { @ link AeronException } with a message for a send error .
 * @ param bytesToSend expected to be sent to the network .
 * @ param ex experienced .
 * @ param destination to which the send was addressed . */
public static void sendError ( final int bytesToSend , final IOException ex , final InetSocketAddress destination ) { } }

implementation: throw new AeronException ( "failed to send packet of " + bytesToSend + " bytes to " + destination , ex ) ;


signature:
public class UdpChannelTransport { /** * Create the underlying channel for reading and writing .
 * @ param statusIndicator to set for error status */
public void openDatagramChannel ( final AtomicCounter statusIndicator ) { } }

implementation: try { sendDatagramChannel = DatagramChannel . open ( udpChannel . protocolFamily ( ) ) ; receiveDatagramChannel = sendDatagramChannel ; if ( udpChannel . isMulticast ( ) ) { if ( null != connectAddress ) { receiveDatagramChannel = DatagramChannel . open ( udpChannel . protocolFamily ( ) ) ; } receiveDatagramChannel . setOption ( StandardSocketOptions . SO_REUSEADDR , true ) ; receiveDatagramChannel . bind ( new InetSocketAddress ( endPointAddress . getPort ( ) ) ) ; receiveDatagramChannel . join ( endPointAddress . getAddress ( ) , udpChannel . localInterface ( ) ) ; sendDatagramChannel . setOption ( StandardSocketOptions . IP_MULTICAST_IF , udpChannel . localInterface ( ) ) ; if ( udpChannel . isHasMulticastTtl ( ) ) { sendDatagramChannel . setOption ( StandardSocketOptions . IP_MULTICAST_TTL , udpChannel . multicastTtl ( ) ) ; multicastTtl = sendDatagramChannel . getOption ( StandardSocketOptions . IP_MULTICAST_TTL ) ; } else if ( context . socketMulticastTtl ( ) != 0 ) { sendDatagramChannel . setOption ( StandardSocketOptions . IP_MULTICAST_TTL , context . socketMulticastTtl ( ) ) ; multicastTtl = sendDatagramChannel . getOption ( StandardSocketOptions . IP_MULTICAST_TTL ) ; } } else { sendDatagramChannel . bind ( bindAddress ) ; } if ( null != connectAddress ) { sendDatagramChannel . connect ( connectAddress ) ; } if ( 0 != context . socketSndbufLength ( ) ) { sendDatagramChannel . setOption ( SO_SNDBUF , context . socketSndbufLength ( ) ) ; } if ( 0 != context . socketRcvbufLength ( ) ) { receiveDatagramChannel . setOption ( SO_RCVBUF , context . socketRcvbufLength ( ) ) ; } sendDatagramChannel . configureBlocking ( false ) ; receiveDatagramChannel . configureBlocking ( false ) ; } catch ( final IOException ex ) { if ( null != statusIndicator ) { statusIndicator . setOrdered ( ChannelEndpointStatus . ERRORED ) ; } CloseHelper . quietClose ( sendDatagramChannel ) ; if ( receiveDatagramChannel != sendDatagramChannel ) { CloseHelper . quietClose ( receiveDatagramChannel ) ; } sendDatagramChannel = null ; receiveDatagramChannel = null ; throw new AeronException ( "channel error - " + ex . getMessage ( ) + " (at " + ex . getStackTrace ( ) [ 0 ] . toString ( ) + "): " + udpChannel . originalUriString ( ) , ex ) ; }


signature:
public class UdpChannelTransport { /** * Close transport , canceling any pending read operations and closing channel */
public void close ( ) { } }

implementation: if ( ! isClosed ) { isClosed = true ; try { if ( null != selectionKey ) { selectionKey . cancel ( ) ; } if ( null != transportPoller ) { transportPoller . cancelRead ( this ) ; transportPoller . selectNowWithoutProcessing ( ) ; } if ( null != sendDatagramChannel ) { sendDatagramChannel . close ( ) ; } if ( receiveDatagramChannel != sendDatagramChannel && null != receiveDatagramChannel ) { receiveDatagramChannel . close ( ) ; } if ( null != transportPoller ) { transportPoller . selectNowWithoutProcessing ( ) ; } } catch ( final IOException ex ) { errorLog . record ( ex ) ; } }


signature:
public class UdpChannelTransport { /** * Is the received frame valid . This method will do some basic checks on the header and can be
 * overridden in a subclass for further validation .
 * @ param buffer containing the frame .
 * @ param length of the frame .
 * @ return true if the frame is believed valid otherwise false . */
public boolean isValidFrame ( final UnsafeBuffer buffer , final int length ) { } }

implementation: boolean isFrameValid = true ; if ( frameVersion ( buffer , 0 ) != HeaderFlyweight . CURRENT_VERSION ) { isFrameValid = false ; invalidPackets . increment ( ) ; } else if ( length < HeaderFlyweight . MIN_HEADER_LENGTH ) { isFrameValid = false ; invalidPackets . increment ( ) ; } return isFrameValid ;


signature:
public class UdpChannelTransport { /** * Receive a datagram from the media layer .
 * @ param buffer into which the datagram will be received .
 * @ return the source address of the datagram if one is available otherwise false . */
public InetSocketAddress receive ( final ByteBuffer buffer ) { } }

implementation: buffer . clear ( ) ; InetSocketAddress address = null ; try { if ( receiveDatagramChannel . isOpen ( ) ) { address = ( InetSocketAddress ) receiveDatagramChannel . receive ( buffer ) ; } } catch ( final PortUnreachableException ignored ) { } catch ( final Exception ex ) { LangUtil . rethrowUnchecked ( ex ) ; } return address ;


signature:
public class RetransmitHandler { /** * Called on reception of a NAK to start retransmits handling .
 * @ param termId from the NAK and the term id of the buffer to retransmit from
 * @ param termOffset from the NAK and the offset of the data to retransmit
 * @ param length of the missing data
 * @ param termLength of the term buffer .
 * @ param retransmitSender to call if an immediate retransmit is required */
public void onNak ( final int termId , final int termOffset , final int length , final int termLength , final RetransmitSender retransmitSender ) { } }

implementation: if ( ! isInvalid ( termOffset , termLength ) ) { if ( null == activeRetransmitsMap . get ( termId , termOffset ) && activeRetransmitsMap . size ( ) < MAX_RETRANSMITS_DEFAULT ) { final RetransmitAction action = assignRetransmitAction ( ) ; action . termId = termId ; action . termOffset = termOffset ; action . length = Math . min ( length , termLength - termOffset ) ; final long delay = delayGenerator . generateDelay ( ) ; if ( 0 == delay ) { retransmitSender . resend ( termId , termOffset , action . length ) ; action . linger ( lingerTimeoutGenerator . generateDelay ( ) , nanoClock . nanoTime ( ) ) ; } else { action . delay ( delay , nanoClock . nanoTime ( ) ) ; } activeRetransmitsMap . put ( termId , termOffset , action ) ; } }


signature:
public class RetransmitHandler { /** * Called to indicate a retransmission is received that may obviate the need to send one ourselves .
 * NOTE : Currently only called from unit tests . Would be used for retransmitting from receivers for NAK suppression .
 * @ param termId of the data
 * @ param termOffset of the data */
public void onRetransmitReceived ( final int termId , final int termOffset ) { } }

implementation:
final RetransmitAction action = activeRetransmitsMap . get ( termId , termOffset ) ; if ( null != action && DELAYED == action . state ) { activeRetransmitsMap . remove ( termId , termOffset ) ; action . cancel ( ) ; // do not go into linger
}


signature:
public class RetransmitHandler { /** * Called to process any outstanding timeouts .
 * @ param nowNs time in nanoseconds
 * @ param retransmitSender to call on retransmissions */
public void processTimeouts ( final long nowNs , final RetransmitSender retransmitSender ) { } }

implementation: if ( activeRetransmitsMap . size ( ) > 0 ) { for ( final RetransmitAction action : retransmitActionPool ) { if ( DELAYED == action . state && ( action . expireNs - nowNs < 0 ) ) { retransmitSender . resend ( action . termId , action . termOffset , action . length ) ; action . linger ( lingerTimeoutGenerator . generateDelay ( ) , nanoClock . nanoTime ( ) ) ; } else if ( LINGERING == action . state && ( action . expireNs - nowNs < 0 ) ) { action . cancel ( ) ; activeRetransmitsMap . remove ( action . termId , action . termOffset ) ; } } }


signature:
public class RecordingSubscriptionDescriptorPoller { /** * Reset the poller to dispatch the descriptors returned from a query .
 * @ param correlationId for the response .
 * @ param subscriptionCount of descriptors to expect .
 * @ param consumer to which the recording subscription descriptors are to be dispatched . */
public void reset ( final long correlationId , final int subscriptionCount , final RecordingSubscriptionDescriptorConsumer consumer ) { } }

implementation: this . correlationId = correlationId ; this . consumer = consumer ; this . remainingSubscriptionCount = subscriptionCount ; isDispatchComplete = false ;


signature:
public class ControlledFragmentAssembler { /** * The implementation of { @ link ControlledFragmentHandler } that reassembles and forwards whole messages .
 * @ param buffer containing the data .
 * @ param offset at which the data begins .
 * @ param length of the data in bytes .
 * @ param header representing the meta data for the data .
 * @ return { @ link io . aeron . logbuffer . ControlledFragmentHandler . Action } to be taken after processing fragment . */
public Action onFragment ( final DirectBuffer buffer , final int offset , final int length , final Header header ) { } }

implementation: final byte flags = header . flags ( ) ; Action action = Action . CONTINUE ; if ( ( flags & UNFRAGMENTED ) == UNFRAGMENTED ) { action = delegate . onFragment ( buffer , offset , length , header ) ; } else { if ( ( flags & BEGIN_FRAG_FLAG ) == BEGIN_FRAG_FLAG ) { final BufferBuilder builder = getBufferBuilder ( header . sessionId ( ) ) ; builder . reset ( ) . append ( buffer , offset , length ) ; } else { final BufferBuilder builder = builderBySessionIdMap . get ( header . sessionId ( ) ) ; if ( null != builder && builder . limit ( ) != 0 ) { final int limit = builder . limit ( ) ; builder . append ( buffer , offset , length ) ; if ( ( flags & END_FRAG_FLAG ) == END_FRAG_FLAG ) { final int msgLength = builder . limit ( ) ; action = delegate . onFragment ( builder . buffer ( ) , 0 , msgLength , header ) ; if ( Action . ABORT == action ) { builder . limit ( limit ) ; } else { builder . reset ( ) ; } } } } } return action ;


signature:
public class ChannelEndpointStatus { /** * String representation of the channel status .
 * @ param status to be converted .
 * @ return representation of the channel status . */
public static String status ( final long status ) { } }

implementation: if ( INITIALIZING == status ) { return "INITIALIZING" ; } if ( ERRORED == status ) { return "ERRORED" ; } if ( ACTIVE == status ) { return "ACTIVE" ; } if ( CLOSING == status ) { return "CLOSING" ; } return "unknown id=" + status ;


signature:
public class ChannelEndpointStatus { /** * Allocate an indicator for tracking the status of a channel endpoint .
 * @ param tempBuffer to be used for labels and metadata .
 * @ param name of the counter for the label .
 * @ param typeId of the counter for classification .
 * @ param countersManager from which to allocated the underlying storage .
 * @ param channel for the stream of messages .
 * @ return a new { @ link AtomicCounter } for tracking the status . */
public static AtomicCounter allocate ( final MutableDirectBuffer tempBuffer , final String name , final int typeId , final CountersManager countersManager , final String channel ) { } }

implementation: final int keyLength = tempBuffer . putStringWithoutLengthAscii ( CHANNEL_OFFSET + SIZE_OF_INT , channel , 0 , MAX_CHANNEL_LENGTH ) ; tempBuffer . putInt ( CHANNEL_OFFSET , keyLength ) ; int labelLength = 0 ; labelLength += tempBuffer . putStringWithoutLengthAscii ( keyLength + labelLength , name ) ; labelLength += tempBuffer . putStringWithoutLengthAscii ( keyLength + labelLength , ": " ) ; labelLength += tempBuffer . putStringWithoutLengthAscii ( keyLength + labelLength , channel , 0 , MAX_LABEL_LENGTH - labelLength ) ; return countersManager . newCounter ( typeId , tempBuffer , 0 , keyLength , tempBuffer , keyLength , labelLength ) ;


signature:
public class ImageFragmentAssembler { /** * The implementation of { @ link FragmentHandler } that reassembles and forwards whole messages .
 * @ param buffer containing the data .
 * @ param offset at which the data begins .
 * @ param length of the data in bytes .
 * @ param header representing the meta data for the data . */
public void onFragment ( final DirectBuffer buffer , final int offset , final int length , final Header header ) { } }

implementation: final byte flags = header . flags ( ) ; if ( ( flags & UNFRAGMENTED ) == UNFRAGMENTED ) { delegate . onFragment ( buffer , offset , length , header ) ; } else { handleFragment ( buffer , offset , length , header , flags ) ; }


signature:
public class DirectBufferVector { /** * Reset the values .
 * @ param buffer which is the super set .
 * @ param offset at which the vector begins .
 * @ param length of the vector .
 * @ return this for a fluent API . */
public DirectBufferVector reset ( final DirectBuffer buffer , final int offset , final int length ) { } }

implementation: this . buffer = buffer ; this . offset = offset ; this . length = length ; return this ;


signature:
public class DirectBufferVector { /** * Ensure the vector is valid for the buffer .
 * @ throws NullPointerException if the buffer is null .
 * @ throws IllegalArgumentException if the offset is out of range for the buffer .
 * @ throws IllegalArgumentException if the length is out of range for the buffer .
 * @ return this for a fluent API . */
public DirectBufferVector validate ( ) { } }

implementation: final int capacity = buffer . capacity ( ) ; if ( offset < 0 || offset >= capacity ) { throw new IllegalArgumentException ( "offset=" + offset + " capacity=" + capacity ) ; } if ( length < 0 || length > ( capacity - offset ) ) { throw new IllegalArgumentException ( "offset=" + offset + " capacity=" + capacity + " length=" + length ) ; } return this ;


signature:
public class DirectBufferVector { /** * Validate an array of vectors to make up a message and compute the total length .
 * @ param vectors to be validated summed .
 * @ return the sum of the vector lengths . */
public static int validateAndComputeLength ( final DirectBufferVector [ ] vectors ) { } }

implementation: int messageLength = 0 ; for ( final DirectBufferVector vector : vectors ) { vector . validate ( ) ; messageLength += vector . length ; if ( messageLength < 0 ) { throw new IllegalStateException ( "length overflow: " + Arrays . toString ( vectors ) ) ; } } return messageLength ;


signature:
public class Configuration { /** * How far ahead a producer can get from a consumer position .
 * @ param termBufferLength for when default is not set and considering an appropriate minimum .
 * @ param defaultTermWindowLength to take priority .
 * @ return the length to be used for the producer window . */
public static int producerWindowLength ( final int termBufferLength , final int defaultTermWindowLength ) { } }

implementation: int termWindowLength = termBufferLength / 2 ; if ( 0 != defaultTermWindowLength ) { termWindowLength = Math . min ( defaultTermWindowLength , termWindowLength ) ; } return termWindowLength ;


signature:
public class Configuration { /** * Get the { @ link IdleStrategy } that should be applied to { @ link org . agrona . concurrent . Agent } s .
 * @ param strategyName of the class to be created .
 * @ param controllableStatus status indicator for what the strategy should do .
 * @ return the newly created IdleStrategy . */
public static IdleStrategy agentIdleStrategy ( final String strategyName , final StatusIndicator controllableStatus ) { } }

implementation: IdleStrategy idleStrategy = null ; switch ( strategyName ) { case DEFAULT_IDLE_STRATEGY : idleStrategy = new BackoffIdleStrategy ( IDLE_MAX_SPINS , IDLE_MAX_YIELDS , IDLE_MIN_PARK_NS , IDLE_MAX_PARK_NS ) ; break ; case CONTROLLABLE_IDLE_STRATEGY : idleStrategy = new ControllableIdleStrategy ( controllableStatus ) ; controllableStatus . setOrdered ( ControllableIdleStrategy . PARK ) ; break ; default : try { idleStrategy = ( IdleStrategy ) Class . forName ( strategyName ) . getConstructor ( ) . newInstance ( ) ; } catch ( final Exception ex ) { LangUtil . rethrowUnchecked ( ex ) ; } break ; } return idleStrategy ;


signature:
public class Configuration { /** * Get the supplier of { @ link SendChannelEndpoint } s which can be used for
 * debugging , monitoring , or modifying the behaviour when sending to the channel .
 * @ return the { @ link SendChannelEndpointSupplier } . */
public static SendChannelEndpointSupplier sendChannelEndpointSupplier ( ) { } }

implementation: SendChannelEndpointSupplier supplier = null ; try { final String className = getProperty ( SEND_CHANNEL_ENDPOINT_SUPPLIER_PROP_NAME ) ; if ( null == className ) { return new DefaultSendChannelEndpointSupplier ( ) ; } supplier = ( SendChannelEndpointSupplier ) Class . forName ( className ) . getConstructor ( ) . newInstance ( ) ; } catch ( final Exception ex ) { LangUtil . rethrowUnchecked ( ex ) ; } return supplier ;


signature:
public class Configuration { /** * Get the supplier of { @ link ReceiveChannelEndpoint } s which can be used for
 * debugging , monitoring , or modifying the behaviour when receiving from the channel .
 * @ return the { @ link SendChannelEndpointSupplier } . */
public static ReceiveChannelEndpointSupplier receiveChannelEndpointSupplier ( ) { } }

implementation: ReceiveChannelEndpointSupplier supplier = null ; try { final String className = getProperty ( RECEIVE_CHANNEL_ENDPOINT_SUPPLIER_PROP_NAME ) ; if ( null == className ) { return new DefaultReceiveChannelEndpointSupplier ( ) ; } supplier = ( ReceiveChannelEndpointSupplier ) Class . forName ( className ) . getConstructor ( ) . newInstance ( ) ; } catch ( final Exception ex ) { LangUtil . rethrowUnchecked ( ex ) ; } return supplier ;


signature:
public class Configuration { /** * Get the supplier of { @ link FlowControl } s which can be used for changing behavior of flow control for unicast
 * publications .
 * @ return the { @ link FlowControlSupplier } . */
public static FlowControlSupplier unicastFlowControlSupplier ( ) { } }

implementation: FlowControlSupplier supplier = null ; try { final String className = getProperty ( UNICAST_FLOW_CONTROL_STRATEGY_SUPPLIER_PROP_NAME ) ; if ( null == className ) { return new DefaultUnicastFlowControlSupplier ( ) ; } supplier = ( FlowControlSupplier ) Class . forName ( className ) . getConstructor ( ) . newInstance ( ) ; } catch ( final Exception ex ) { LangUtil . rethrowUnchecked ( ex ) ; } return supplier ;


signature:
public class Configuration { /** * Get the supplier of { @ link FlowControl } s which can be used for changing behavior of flow control for multicast
 * publications .
 * @ return the { @ link FlowControlSupplier } . */
public static FlowControlSupplier multicastFlowControlSupplier ( ) { } }

implementation: FlowControlSupplier supplier = null ; try { final String className = getProperty ( MULTICAST_FLOW_CONTROL_STRATEGY_SUPPLIER_PROP_NAME ) ; if ( null == className ) { return new DefaultMulticastFlowControlSupplier ( ) ; } supplier = ( FlowControlSupplier ) Class . forName ( className ) . getConstructor ( ) . newInstance ( ) ; } catch ( final Exception ex ) { LangUtil . rethrowUnchecked ( ex ) ; } return supplier ;


signature:
public class Configuration { /** * Get the supplier of { @ link CongestionControl } implementations which can be used for receivers .
 * @ return the { @ link CongestionControlSupplier } */
public static CongestionControlSupplier congestionControlSupplier ( ) { } }

implementation: CongestionControlSupplier supplier = null ; try { final String className = getProperty ( CONGESTION_CONTROL_STRATEGY_SUPPLIER_PROP_NAME ) ; if ( null == className ) { return new DefaultCongestionControlSupplier ( ) ; } supplier = ( CongestionControlSupplier ) Class . forName ( className ) . getConstructor ( ) . newInstance ( ) ; } catch ( final Exception ex ) { LangUtil . rethrowUnchecked ( ex ) ; } return supplier ;


signature:
public class Configuration { /** * Validate that the MTU is an appropriate length . MTU lengths must be a multiple of
 * { @ link FrameDescriptor # FRAME _ ALIGNMENT } .
 * @ param mtuLength to be validated .
 * @ throws ConfigurationException if the MTU length is not valid . */
public static void validateMtuLength ( final int mtuLength ) { } }

implementation: if ( mtuLength < DataHeaderFlyweight . HEADER_LENGTH || mtuLength > MAX_UDP_PAYLOAD_LENGTH ) { throw new ConfigurationException ( "mtuLength must be a >= HEADER_LENGTH and <= MAX_UDP_PAYLOAD_LENGTH: " + mtuLength ) ; } if ( ( mtuLength & ( FrameDescriptor . FRAME_ALIGNMENT - 1 ) ) != 0 ) { throw new ConfigurationException ( "mtuLength must be a multiple of FRAME_ALIGNMENT: " + mtuLength ) ; }


signature:
public class Configuration { /** * Get the { @ link TerminationValidator } implementations which can be used for validating a termination request
 * sent to the driver to ensure the client has the right to terminate a driver .
 * @ return the { @ link TerminationValidator } */
public static TerminationValidator terminationValidator ( ) { } }

implementation: TerminationValidator validator = null ; try { final String className = getProperty ( TERMINATION_VALIDATOR_PROP_NAME ) ; if ( null == className ) { return new DefaultDenyTerminationValidator ( ) ; } validator = ( TerminationValidator ) Class . forName ( className ) . getConstructor ( ) . newInstance ( ) ; } catch ( final Exception ex ) { LangUtil . rethrowUnchecked ( ex ) ; } return validator ;


signature:
public class Configuration { /** * Validate that the socket buffer lengths are sufficient for the media driver configuration .
 * @ param ctx to be validated . */
public static void validateSocketBufferLengths ( final MediaDriver . Context ctx ) { } }

implementation: try ( DatagramChannel probe = DatagramChannel . open ( ) ) { final int defaultSoSndBuf = probe . getOption ( StandardSocketOptions . SO_SNDBUF ) ; probe . setOption ( StandardSocketOptions . SO_SNDBUF , Integer . MAX_VALUE ) ; final int maxSoSndBuf = probe . getOption ( StandardSocketOptions . SO_SNDBUF ) ; if ( maxSoSndBuf < ctx . socketSndbufLength ( ) ) { System . err . format ( "WARNING: Could not get desired SO_SNDBUF, adjust OS to allow %s: attempted=%d, actual=%d%n" , SOCKET_SNDBUF_LENGTH_PROP_NAME , ctx . socketSndbufLength ( ) , maxSoSndBuf ) ; } probe . setOption ( StandardSocketOptions . SO_RCVBUF , Integer . MAX_VALUE ) ; final int maxSoRcvBuf = probe . getOption ( StandardSocketOptions . SO_RCVBUF ) ; if ( maxSoRcvBuf < ctx . socketRcvbufLength ( ) ) { System . err . format ( "WARNING: Could not get desired SO_RCVBUF, adjust OS to allow %s: attempted=%d, actual=%d%n" , SOCKET_RCVBUF_LENGTH_PROP_NAME , ctx . socketRcvbufLength ( ) , maxSoRcvBuf ) ; } final int soSndBuf = 0 == ctx . socketSndbufLength ( ) ? defaultSoSndBuf : ctx . socketSndbufLength ( ) ; if ( ctx . mtuLength ( ) > soSndBuf ) { throw new ConfigurationException ( String . format ( "MTU greater than socket SO_SNDBUF, adjust %s to match MTU: mtuLength=%d, SO_SNDBUF=%d" , SOCKET_SNDBUF_LENGTH_PROP_NAME , ctx . mtuLength ( ) , soSndBuf ) ) ; } if ( ctx . initialWindowLength ( ) > maxSoRcvBuf ) { throw new ConfigurationException ( "window length greater than socket SO_RCVBUF, increase '" + Configuration . INITIAL_WINDOW_LENGTH_PROP_NAME + "' to match window: windowLength=" + ctx . initialWindowLength ( ) + ", SO_RCVBUF=" + maxSoRcvBuf ) ; } } catch ( final IOException ex ) { throw new AeronException ( "probe socket: " + ex . toString ( ) , ex ) ; }


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

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


signature:
public class Configuration { /** * Validate the range of session ids based on a high and low value provided which accounts for the values wrapping .
 * @ param low value in the range .
 * @ param high value in the range .
 * @ throws ConfigurationException if the values are not valid . */
public static void validateSessionIdRange ( final int low , final int high ) { } }

implementation: if ( low > high ) { throw new ConfigurationException ( "low session id value " + low + " must be <= high value " + high ) ; } if ( Math . abs ( ( long ) high - low ) > Integer . MAX_VALUE ) { throw new ConfigurationException ( "reserved range to too large" ) ; }