2020-08-12 23:30:16 +02:00
/ *
2024-02-05 21:57:40 +01:00
* Copyright 2024 dorkbox , llc
2020-08-12 23:30:16 +02:00
*
* Licensed under the Apache License , Version 2.0 ( the " License " ) ;
* you may not use this file except in compliance with the License .
* You may obtain a copy of the License at
*
* http : //www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing , software
* distributed under the License is distributed on an " AS IS " BASIS ,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND , either express or implied .
* See the License for the specific language governing permissions and
* limitations under the License .
* /
package dorkbox.network
2022-06-04 01:32:58 +02:00
import dorkbox.dns.DnsClient
2023-08-21 02:20:41 +02:00
import dorkbox.hex.toHexString
2022-03-08 08:41:06 +01:00
import dorkbox.netUtil.IP
import dorkbox.netUtil.IPv4
import dorkbox.netUtil.IPv6
2022-06-04 01:32:58 +02:00
import dorkbox.netUtil.dnsUtils.ResolvedAddressTypes
2021-04-27 13:57:47 +02:00
import dorkbox.network.aeron.AeronDriver
2023-09-03 21:17:37 +02:00
import dorkbox.network.aeron.EventActionOperator
import dorkbox.network.aeron.EventCloseOperator
2023-05-08 09:58:24 +02:00
import dorkbox.network.aeron.EventPoller
2023-10-28 20:54:09 +02:00
import dorkbox.network.connection.Connection
import dorkbox.network.connection.ConnectionParams
import dorkbox.network.connection.EndPoint
2023-06-14 23:35:03 +02:00
import dorkbox.network.connection.IpInfo.Companion.formatCommonAddress
2023-05-08 09:58:24 +02:00
import dorkbox.network.connection.ListenerManager.Companion.cleanStackTrace
import dorkbox.network.connection.ListenerManager.Companion.cleanStackTraceInternal
2023-10-28 20:54:09 +02:00
import dorkbox.network.connection.PublicKeyValidationState
import dorkbox.network.connection.buffer.BufferManager
2023-06-26 19:28:55 +02:00
import dorkbox.network.exceptions.*
2023-06-29 01:31:31 +02:00
import dorkbox.network.handshake.ClientConnectionDriver
2020-08-12 23:30:16 +02:00
import dorkbox.network.handshake.ClientHandshake
2023-06-29 01:31:31 +02:00
import dorkbox.network.handshake.ClientHandshakeDriver
2021-04-30 16:01:25 +02:00
import dorkbox.network.ping.Ping
2023-12-12 13:23:51 +01:00
import dorkbox.util.Sys
2023-09-13 16:57:32 +02:00
import org.slf4j.LoggerFactory
2020-09-09 01:33:09 +02:00
import java.net.Inet4Address
2020-09-10 00:35:01 +02:00
import java.net.Inet6Address
2020-09-09 01:33:09 +02:00
import java.net.InetAddress
2023-11-28 20:53:37 +01:00
import java.util.*
2023-06-25 12:21:21 +02:00
import java.util.concurrent.*
2020-08-12 23:30:16 +02:00
/ * *
* The client is both SYNC and ASYNC . It starts off SYNC ( blocks thread until it ' s done ) , then once it ' s connected to the server , it ' s
* ASYNC .
2021-07-09 15:16:12 +02:00
*
* @param config these are the specific connection options
2022-05-28 12:15:45 +02:00
* @param loggerName allows for a custom logger name for this endpoint ( for when there are multiple endpoints )
2020-08-12 23:30:16 +02:00
* /
2022-03-24 00:32:58 +01:00
@Suppress ( " unused " )
2023-09-17 02:36:45 +02:00
open class Client < CONNECTION : Connection > ( config : ClientConfiguration = ClientConfiguration ( ) , loggerName : String = Client :: class . java . simpleName )
: EndPoint < CONNECTION > ( config , loggerName ) {
2021-07-09 15:16:12 +02:00
2020-08-12 23:30:16 +02:00
companion object {
/ * *
* Gets the version number .
* /
2023-09-04 00:47:46 +02:00
const val version = Configuration . version
2021-04-09 20:24:45 +02:00
2023-05-08 09:58:24 +02:00
/ * *
* Ensures that the client ( using the specified configuration ) is NO LONGER running .
*
* NOTE : This method should only be used to check if a client is running for a DIFFERENT configuration than the currently running client
*
* By default , we will wait the [ Configuration . connectionCloseTimeoutInSeconds ] * 2 amount of time before returning .
*
* @return true if the media driver is STOPPED .
* /
2023-09-04 14:23:06 +02:00
fun ensureStopped ( configuration : Configuration ) : Boolean {
2023-05-08 09:58:24 +02:00
val timeout = TimeUnit . SECONDS . toMillis ( configuration . connectionCloseTimeoutInSeconds . toLong ( ) * 2 )
2023-09-13 16:57:32 +02:00
val logger = LoggerFactory . getLogger ( Client :: class . java . simpleName )
2023-09-04 14:23:06 +02:00
return AeronDriver . ensureStopped ( configuration . copy ( ) , logger , timeout )
2023-05-08 09:58:24 +02:00
}
2021-04-30 21:18:57 +02:00
/ * *
* Checks to see if a client ( using the specified configuration ) is running .
*
2023-05-08 09:58:24 +02:00
* NOTE : This method should only be used to check if a client is running for a DIFFERENT configuration than the currently running client
*
* @return true if the media driver is active and running
2021-04-30 21:18:57 +02:00
* /
2023-09-04 14:23:06 +02:00
fun isRunning ( configuration : Configuration ) : Boolean {
2023-09-13 16:57:32 +02:00
val logger = LoggerFactory . getLogger ( Client :: class . java . simpleName )
2023-09-04 14:23:06 +02:00
return AeronDriver . isRunning ( configuration . copy ( ) , logger )
2021-04-30 21:18:57 +02:00
}
2021-04-09 20:24:45 +02:00
init {
// Add this project to the updates system, which verifies this class + UUID + version information
2021-04-27 14:00:31 +02:00
dorkbox . updates . Updates . add ( Client :: class . java , " 5be42ae40cac49fb90dea86bc513141b " , version )
2021-04-09 20:24:45 +02:00
}
2020-08-12 23:30:16 +02:00
}
/ * *
2023-06-29 01:31:31 +02:00
* The network address of the remote machine that the client connected to . This will be null for IPC connections .
2023-06-26 19:28:55 +02:00
* /
@Volatile
var address : InetAddress ? = IPv4 . LOCALHOST
private set
/ * *
2023-06-29 01:31:31 +02:00
* The network address of the remote machine that the client connected to , as a string . This will be " IPC " for IPC connections .
2023-06-26 19:28:55 +02:00
* /
@Volatile
var addressString : String = " UNKNOWN "
private set
/ * *
2023-06-29 01:31:31 +02:00
* The network address of the remote machine that the client connected to , as a pretty string . This will be " IPC " for IPC connections .
2023-06-26 19:28:55 +02:00
* /
@Volatile
var addressPrettyString : String = " UNKNOWN "
private set
2023-10-28 20:55:49 +02:00
/ * *
* The tag name assigned ( by the configuration ) to the client . The server will receive this tag during the handshake . The max length is
* 32 characters .
* /
@Volatile
var tag : String = " "
private set
2022-06-27 15:31:40 +02:00
/ * *
2023-06-26 19:28:55 +02:00
* The default connection reliability type ( ie : can the lower - level network stack throw away data that has errors , for example real - time - voice )
2022-06-27 15:31:40 +02:00
* /
@Volatile
2023-06-26 19:28:55 +02:00
var reliable : Boolean = true
private set
/ * *
* How long ( in seconds ) will connections wait to connect . 0 will wait indefinitely ,
* /
@Volatile
var connectionTimeoutSec : Int = 0
2022-06-27 15:31:40 +02:00
private set
2023-10-26 21:19:36 +02:00
/ * *
* - if the client is internally going to reconnect ( because of a network error )
* - we have specified that we will run the disconnect logic
* - there is reconnect logic in the disconnect handler
*
* Then ultimately , we want to ignore the disconnect - handler reconnect ( we do not want to have multiple reconnects happening concurrently )
* /
@Volatile
private var autoReconnect = false
2023-06-28 20:31:08 +02:00
2023-06-14 23:35:03 +02:00
private val handshake = ClientHandshake ( this , logger )
2024-02-05 21:57:40 +01:00
@Volatile
internal var clientConnectionInProgress = CountDownLatch ( 0 )
2023-05-08 09:58:24 +02:00
@Volatile
private var slowDownForException = false
2020-08-12 23:30:16 +02:00
2023-09-08 13:21:12 +02:00
@Volatile
private var stopConnectOnShutdown = false
2023-10-28 20:54:09 +02:00
/ * *
* Different connections ( to the same client ) can be " buffered " , meaning that if they " go down " because of a network glitch -- the data
* being sent is not lost ( it is buffered ) and then re - sent once the new connection is established . References to the old connection
* will also redirect to the new connection .
* /
@Volatile
internal var bufferedManager : BufferManager < CONNECTION > ? = null
2020-08-12 23:30:16 +02:00
// is valid when there is a connection to the server, otherwise it is null
2023-06-25 12:21:21 +02:00
@Volatile
2020-09-03 01:31:08 +02:00
private var connection0 : CONNECTION ? = null
2020-08-12 23:30:16 +02:00
2023-07-02 21:59:45 +02:00
private val string0 : String by lazy {
2023-08-21 02:20:41 +02:00
" EndPoint [Client: ${storage.publicKey.toHexString()} ] "
2023-07-02 21:59:45 +02:00
}
2020-09-25 14:49:17 +02:00
final override fun newException ( message : String , cause : Throwable ? ) : Throwable {
2023-06-30 00:06:53 +02:00
// +2 because we do not want to see the stack for the abstract `newException`
val clientException = ClientException ( message , cause )
clientException . cleanStackTrace ( 2 )
return clientException
2020-08-12 23:30:16 +02:00
}
2023-06-26 19:28:55 +02:00
/ * *
* Will attempt to re - connect to the server , with the settings previously used when calling connect ( )
*
* @throws IllegalArgumentException if the remote address is invalid
* @throws ClientTimedOutException if the client is unable to connect in x amount of time
* @throws ClientRejectedException if the client connection is rejected
* @throws ClientShutdownException if the client connection is shutdown while trying to connect
* @throws ClientException if there are misc errors
* /
@Suppress ( " DuplicatedCode " )
2023-09-04 00:47:46 +02:00
fun reconnect ( ) {
2023-10-26 21:19:36 +02:00
if ( autoReconnect ) {
// we must check if we should permit a MANUAL reconnect, because the auto-reconnect MIGHT ALSO re-connect!
// autoReconnect will be "reset" when the connection closes. If in a happy state, then a manual reconnect is permitted.
logger . info ( " Ignoring reconnect, auto-reconnect is in progress " )
return
}
2023-08-10 05:07:16 +02:00
if ( connectionTimeoutSec == 0 ) {
2023-09-13 16:57:32 +02:00
logger . info ( " Reconnecting... " )
2023-08-10 05:07:16 +02:00
} else {
2023-09-13 16:57:32 +02:00
logger . info ( " Reconnecting... (timeout in $connectionTimeoutSec seconds) " )
2023-08-10 05:07:16 +02:00
}
2023-08-30 12:02:04 +02:00
if ( !is Shutdown ( ) ) {
// if we aren't closed already, close now.
close ( false )
waitForClose ( )
}
2023-06-26 19:28:55 +02:00
connect (
remoteAddress = address ,
remoteAddressString = addressString ,
remoteAddressPrettyString = addressPrettyString ,
2023-07-11 00:11:58 +02:00
port1 = port1 ,
port2 = port2 ,
2023-06-26 19:28:55 +02:00
connectionTimeoutSec = connectionTimeoutSec ,
reliable = reliable ,
)
}
/ * *
* Will attempt to connect via IPC to the server , with a default 30 second connection timeout and will block until completed .
*
* # # # For the IPC ( Inter - Process - Communication ) it must be :
* - ` connectIpc ( ) `
*
* @param connectionTimeoutSec wait for x seconds . 0 will wait indefinitely
*
* @throws ClientTimedOutException if the client is unable to connect in x amount of time
* /
2023-09-04 14:23:06 +02:00
fun connectIpc ( connectionTimeoutSec : Int = 30 ) {
2023-06-26 19:28:55 +02:00
connect ( remoteAddress = null , // required!
2023-07-11 00:11:58 +02:00
port1 = 0 ,
port2 = 0 ,
2023-06-26 19:28:55 +02:00
remoteAddressString = IPC _NAME ,
remoteAddressPrettyString = IPC _NAME ,
connectionTimeoutSec = connectionTimeoutSec )
}
2022-06-27 15:31:40 +02:00
/ * *
* Will attempt to connect to the server , with a default 30 second connection timeout and will block until completed .
*
* Default connection is to localhost
*
* # # # For a network address , it can be :
* - a network name ( " localhost " , " bob.example.org " )
* - an IP address ( " 127.0.0.1 " , " 123.123.123.123 " , " ::1 " )
* - an InetAddress address
2023-06-26 19:28:55 +02:00
* - ` connect ( LOCALHOST ) `
* - ` connect ( " localhost " ) `
* - ` connect ( " bob.example.org " ) `
* - ` connect ( " 127.0.0.1 " ) `
* - ` connect ( " ::1 " ) `
2022-06-27 15:31:40 +02:00
*
* # # # For the IPC ( Inter - Process - Communication ) it must be :
2023-06-26 19:28:55 +02:00
* - ` connectIPC ( ) `
2022-06-27 15:31:40 +02:00
*
* # # # Case does not matter , and " localhost " is the default .
*
* @param remoteAddress The network or if localhost , IPC address for the client to connect to
2023-07-11 00:11:58 +02:00
* @param port1 The network host port1 to connect to
* @param port2 The network host port2 to connect to . The server uses this to work around NAT firewalls . By default , this is port1 + 1 ,
* but can also be configured independently . This is required , and must be different from port1 .
2022-06-27 15:31:40 +02:00
* @param connectionTimeoutSec wait for x seconds . 0 will wait indefinitely
2023-06-26 19:28:55 +02:00
* @param reliable true if we want to create a reliable connection , can the lower - level network stack throw away data that has errors , ( IE : real - time - voice traffic )
2022-06-27 15:31:40 +02:00
*
* @throws IllegalArgumentException if the remote address is invalid
* @throws ClientTimedOutException if the client is unable to connect in x amount of time
* @throws ClientRejectedException if the client connection is rejected
* /
2022-07-15 06:58:49 +02:00
fun connect (
remoteAddress : InetAddress ,
2023-07-11 00:11:58 +02:00
port1 : Int ,
port2 : Int = port1 + 1 ,
2022-07-15 06:58:49 +02:00
connectionTimeoutSec : Int = 30 ,
2023-09-04 14:23:06 +02:00
reliable : Boolean = true ) {
2023-05-08 09:58:24 +02:00
2022-06-27 15:31:40 +02:00
val remoteAddressString = when ( remoteAddress ) {
is Inet4Address -> IPv4 . toString ( remoteAddress )
is Inet6Address -> IPv6 . toString ( remoteAddress , true )
2022-12-17 22:24:33 +01:00
else -> throw IllegalArgumentException ( " Cannot connect to $remoteAddress It is an invalid address type! " )
2022-06-27 15:31:40 +02:00
}
connect ( remoteAddress = remoteAddress ,
remoteAddressString = remoteAddressString ,
2022-12-17 22:24:33 +01:00
remoteAddressPrettyString = remoteAddressString ,
2023-07-11 00:11:58 +02:00
port1 = port1 ,
port2 = port2 ,
2022-06-27 15:31:40 +02:00
connectionTimeoutSec = connectionTimeoutSec ,
reliable = reliable )
}
2023-06-25 17:23:44 +02:00
2020-08-12 23:30:16 +02:00
/ * *
2020-08-15 13:21:20 +02:00
* Will attempt to connect to the server , with a default 30 second connection timeout and will block until completed .
2020-08-12 23:30:16 +02:00
*
2020-08-15 13:21:20 +02:00
* Default connection is to localhost
2020-08-12 23:30:16 +02:00
*
2020-08-15 13:21:20 +02:00
* # # # For a network address , it can be :
2022-03-15 20:43:43 +01:00
* - a network name ( " localhost " , " bob.example.org " )
2020-08-15 13:21:20 +02:00
* - an IP address ( " 127.0.0.1 " , " 123.123.123.123 " , " ::1 " )
2020-09-09 01:33:09 +02:00
* - an InetAddress address
2023-06-26 19:28:55 +02:00
* - ` connect ( LOCALHOST ) `
* - ` connect ( " localhost " ) `
* - ` connect ( " bob.example.org " ) `
* - ` connect ( " 127.0.0.1 " ) `
* - ` connect ( " ::1 " ) `
2020-08-12 23:30:16 +02:00
*
2020-09-02 02:39:05 +02:00
* # # # For the IPC ( Inter - Process - Communication ) it must be :
2023-06-26 19:28:55 +02:00
* - ` connectIPC ( ) `
2020-08-12 23:30:16 +02:00
*
2021-04-30 16:01:25 +02:00
* # # # Case does not matter , and " localhost " is the default .
2020-08-12 23:30:16 +02:00
*
2023-06-26 19:28:55 +02:00
* @param remoteAddress The network host name or ip address
2023-07-11 00:11:58 +02:00
* @param port1 The network host port1 to connect to
* @param port2 The network host port2 to connect to . The server uses this to work around NAT firewalls . By default , this is port1 + 1 ,
* but can also be configured independently . This is required , and must be different from port1 .
* @param connectionTimeoutSec wait for x seconds . 0 will wait indefinitely
2022-03-08 08:41:06 +01:00
* @param connectionTimeoutSec wait for x seconds . 0 will wait indefinitely
2023-06-26 19:28:55 +02:00
* @param reliable true if we want to create a reliable connection , can the lower - level network stack throw away data that has errors , ( IE : real - time - voice traffic )
2020-08-12 23:30:16 +02:00
*
2020-08-15 13:21:20 +02:00
* @throws IllegalArgumentException if the remote address is invalid
2020-08-12 23:30:16 +02:00
* @throws ClientTimedOutException if the client is unable to connect in x amount of time
2020-08-15 13:21:20 +02:00
* @throws ClientRejectedException if the client connection is rejected
2020-08-12 23:30:16 +02:00
* /
2022-07-15 06:58:49 +02:00
fun connect (
2023-06-26 19:28:55 +02:00
remoteAddress : String ,
2023-07-11 00:11:58 +02:00
port1 : Int ,
port2 : Int = port1 + 1 ,
2022-07-15 06:58:49 +02:00
connectionTimeoutSec : Int = 30 ,
2023-05-08 09:58:24 +02:00
reliable : Boolean = true ) {
2023-09-04 14:23:06 +02:00
fun connect ( dnsResolveType : ResolvedAddressTypes ) {
2022-08-20 12:15:01 +02:00
val ipv4Requested = dnsResolveType == ResolvedAddressTypes . IPV4 _ONLY || dnsResolveType == ResolvedAddressTypes . IPV4 _PREFERRED
val inetAddress = formatCommonAddress ( remoteAddress , ipv4Requested ) {
// we already checked first if it's a valid IP address. This is called if it's not, since it might be a DNS lookup
val client = DnsClient ( )
client . resolvedAddressTypes ( dnsResolveType )
val records = client . resolve ( remoteAddress )
client . stop ( )
records ?. get ( 0 )
2022-08-19 23:32:25 +02:00
} ?: throw IllegalArgumentException ( " The remote address ' $remoteAddress ' cannot be found. " )
2022-06-04 01:32:58 +02:00
2022-12-17 22:24:33 +01:00
val remoteAddressAsIp = IP . toString ( inetAddress )
val formattedString = if ( remoteAddress == remoteAddressAsIp ) {
remoteAddress
} else {
" $remoteAddress ( $remoteAddressAsIp ) "
}
2022-08-18 12:38:44 +02:00
connect ( remoteAddress = inetAddress ,
2022-08-19 23:32:25 +02:00
// we check again, because the inetAddress that comes back from DNS, might not be what we expect
2022-12-17 22:24:33 +01:00
remoteAddressString = remoteAddressAsIp ,
remoteAddressPrettyString = formattedString ,
2023-07-11 00:11:58 +02:00
port1 = port1 ,
port2 = port2 ,
2022-03-24 00:32:58 +01:00
connectionTimeoutSec = connectionTimeoutSec ,
2022-06-27 15:31:40 +02:00
reliable = reliable )
2021-04-30 16:01:25 +02:00
}
2020-09-09 01:33:09 +02:00
2022-08-04 03:39:14 +02:00
when {
// this is default IPC settings
2023-09-04 14:23:06 +02:00
remoteAddress . isEmpty ( ) && config . enableIpc -> {
2023-06-25 12:21:21 +02:00
connect ( remoteAddress = null , // required!
2023-07-11 00:11:58 +02:00
port1 = 0 ,
port2 = 0 ,
2023-06-25 12:21:21 +02:00
remoteAddressString = IPC _NAME ,
remoteAddressPrettyString = IPC _NAME ,
connectionTimeoutSec = connectionTimeoutSec )
2022-06-04 01:32:58 +02:00
}
2022-08-20 12:15:01 +02:00
// IPv6 takes precedence ONLY if it's enabled manually
2023-06-25 12:21:21 +02:00
config . enableIPv6 -> { connect ( ResolvedAddressTypes . IPV6 _ONLY ) }
config . enableIPv4 -> { connect ( ResolvedAddressTypes . IPV4 _ONLY ) }
IPv4 . isPreferred -> { connect ( ResolvedAddressTypes . IPV4 _PREFERRED ) }
IPv6 . isPreferred -> { connect ( ResolvedAddressTypes . IPV6 _PREFERRED ) }
else -> { connect ( ResolvedAddressTypes . IPV4 _PREFERRED ) }
2021-04-30 16:01:25 +02:00
}
2020-09-09 01:33:09 +02:00
}
/ * *
* Will attempt to connect to the server , with a default 30 second connection timeout and will block until completed .
2022-06-14 22:18:08 +02:00
* If unable to connect within the specified timeout an exception will be thrown
2020-09-09 01:33:09 +02:00
*
* Default connection is to localhost
*
* # # # For a network address , it can be :
2022-03-15 20:43:43 +01:00
* - a network name ( " localhost " , " bob.example.org " )
2020-09-09 01:33:09 +02:00
* - an IP address ( " 127.0.0.1 " , " 123.123.123.123 " , " ::1 " )
2021-04-30 16:01:25 +02:00
* - an InetAddress address
2023-06-26 19:28:55 +02:00
* - ` connect ( ) ` ( same as localhost , but only if ipc is disabled in the configuration )
* - ` connect ( " localhost " ) `
* - ` connect ( " bob.example.org " ) `
* - ` connect ( " 127.0.0.1 " ) `
* - ` connect ( " ::1 " ) `
2020-09-09 01:33:09 +02:00
*
* # # # For the IPC ( Inter - Process - Communication ) it must be :
2023-06-26 19:28:55 +02:00
* - ` connect ( ) ` ( only if ipc is enabled in the configuration )
2021-04-30 16:01:25 +02:00
*
* # # # Case does not matter , and " localhost " is the default .
2020-09-09 01:33:09 +02:00
*
2023-05-28 18:43:05 +02:00
* @param remoteAddress The network or if localhost for the client to connect to
2023-07-11 00:11:58 +02:00
* @param port1 The network host port1 to connect to
* @param port2 The network host port2 to connect to . The server uses this to work around NAT firewalls . By default , this is port1 + 1 ,
* but can also be configured independently . This is required , and must be different from port1 .
* @param connectionTimeoutSec wait for x seconds . 0 will wait indefinitely
2022-03-08 08:41:06 +01:00
* @param connectionTimeoutSec wait for x seconds . 0 will wait indefinitely .
2023-06-26 19:28:55 +02:00
* @param reliable true if we want to create a reliable connection , can the lower - level network stack throw away data that has errors , ( IE : real - time - voice traffic )
2020-09-09 01:33:09 +02:00
*
* @throws IllegalArgumentException if the remote address is invalid
* @throws ClientTimedOutException if the client is unable to connect in x amount of time
* @throws ClientRejectedException if the client connection is rejected
2022-06-14 22:18:08 +02:00
* @throws ClientShutdownException if the client connection is shutdown while trying to connect
2021-07-06 15:38:53 +02:00
* @throws ClientException if there are misc errors
2020-09-09 01:33:09 +02:00
* /
@Suppress ( " DuplicatedCode " )
2023-09-04 00:47:46 +02:00
private fun connect (
2023-07-11 00:11:58 +02:00
remoteAddress : InetAddress ? ,
2022-07-15 06:58:49 +02:00
remoteAddressString : String ,
2022-12-17 22:24:33 +01:00
remoteAddressPrettyString : String ,
2023-07-11 00:11:58 +02:00
port1 : Int ,
port2 : Int ,
connectionTimeoutSec : Int ,
2022-07-15 06:58:49 +02:00
reliable : Boolean = true )
2022-08-10 14:34:09 +02:00
{
2023-06-25 12:21:21 +02:00
// NOTE: it is critical to remember that Aeron DOES NOT like running from coroutines!
2023-08-08 03:55:51 +02:00
if ( ( config . enableIPv4 || config . enableIPv6 ) && remoteAddress != null ) {
require ( port1 != port2 ) { " port1 cannot be the same as port2 " }
require ( port1 > 0 ) { " port1 must be > 0 " }
require ( port2 > 0 ) { " port2 must be > 0 " }
require ( port1 < 65535 ) { " port1 must be < 65535 " }
require ( port2 < 65535 ) { " port2 must be < 65535 " }
}
require ( connectionTimeoutSec >= 0 ) { " connectionTimeoutSec ' $connectionTimeoutSec ' is invalid. It must be >=0 " }
// only try to connect via IPv4 if we have a network interface that supports it!
if ( remoteAddress is Inet4Address && ! IPv4 . isAvailable ) {
require ( false ) { " Unable to connect to the IPv4 address $remoteAddressPrettyString , there are no IPv4 interfaces available! " }
}
// only try to connect via IPv6 if we have a network interface that supports it!
if ( remoteAddress is Inet6Address && ! IPv6 . isAvailable ) {
require ( false ) { " Unable to connect to the IPv6 address $remoteAddressPrettyString , there are no IPv6 interfaces available! " }
}
if ( remoteAddress != null && remoteAddress . isAnyLocalAddress ) {
require ( false ) { " Cannot connect to $remoteAddressPrettyString It is an invalid address! " }
}
2023-05-28 18:43:05 +02:00
2023-06-25 12:21:21 +02:00
// on the client, we must GUARANTEE that the disconnect/close completes before NEW connect begins.
// we will know this if we are running inside an INTERNAL dispatch that is NOT the connect dispatcher!
2023-10-26 08:05:08 +02:00
if ( eventDispatch . isDispatch ( ) && ! eventDispatch . CONNECT . isDispatch ( ) ) {
2023-08-10 05:31:00 +02:00
// only re-dispatch if we are on the event dispatch AND it's not the CONNECT one
2023-10-26 08:05:08 +02:00
// if we are on an "outside" thread, then we don't care.
2023-11-22 09:18:17 +01:00
// we have to guarantee that CLOSE finishes running first! If we are already on close, then this will just run after close is finished.
eventDispatch . CLOSE . launch {
eventDispatch . CONNECT . launch {
connect (
remoteAddress = remoteAddress ,
remoteAddressString = remoteAddressString ,
remoteAddressPrettyString = remoteAddressPrettyString ,
port1 = port1 ,
port2 = port2 ,
connectionTimeoutSec = connectionTimeoutSec ,
reliable = reliable )
}
2023-03-02 19:46:08 +01:00
}
2023-11-22 09:18:17 +01:00
2023-03-02 19:46:08 +01:00
return
}
2023-08-10 05:35:02 +02:00
2023-06-25 12:21:21 +02:00
// the lifecycle of a client is the ENDPOINT (measured via the network event poller) and CONNECTION (measure from connection closed)
2023-08-10 05:35:02 +02:00
// if we are reconnecting, then we do not want to wait for the ENDPOINT to close first!
if ( ! waitForEndpointShutdown ( ) ) {
2023-06-25 12:21:21 +02:00
if ( endpointIsRunning . value ) {
listenerManager . notifyError ( ServerException ( " Unable to start, the client is already running! " ) )
} else {
listenerManager . notifyError ( ClientException ( " Unable to connect the client! " ) )
}
return
}
2022-07-29 04:52:11 +02:00
config as ClientConfiguration
2022-07-15 06:58:49 +02:00
2020-09-25 18:31:23 +02:00
2020-09-03 01:31:08 +02:00
connection0 = null
2020-09-02 02:39:05 +02:00
2020-09-10 00:35:01 +02:00
2023-02-20 20:57:04 +01:00
// we are done with initial configuration, now initialize aeron and the general state of this endpoint
2023-06-14 23:35:03 +02:00
// this also makes sure that the dispatchers are still active.
// Calling `client.close()` will shutdown the dispatchers (and a new client instance must be created)
2023-02-20 20:57:04 +01:00
try {
2023-09-08 13:21:12 +02:00
stopConnectOnShutdown = false
2023-02-20 20:57:04 +01:00
startDriver ( )
2023-06-25 12:21:21 +02:00
initializeState ( )
2023-02-20 20:57:04 +01:00
} catch ( e : Exception ) {
2023-06-25 12:21:21 +02:00
listenerManager . notifyError ( ClientException ( " Unable to start the client! " , e ) )
2023-09-06 16:22:37 +02:00
resetOnError ( )
2023-02-20 20:57:04 +01:00
return
}
2023-06-26 19:28:55 +02:00
// localhost/loopback IP might not always be 127.0.0.1 or ::1
// will be null if it's IPC
this . address = remoteAddress
// will be exactly 'IPC' if it's IPC
// if it's an IP address, it will be the IP address
// if it's a DNS name, the name will be resolved, and it will be DNS (IP)
this . addressString = remoteAddressString
this . addressPrettyString = remoteAddressString
2023-07-11 00:11:58 +02:00
this . port1 = port1
this . port2 = port2
2023-10-28 20:55:49 +02:00
// DOUBLE CHECK!
require ( config . tag . length <= 32 ) { " Client tag name length must be <= 32 " }
this . tag = config . tag
2023-06-26 19:28:55 +02:00
this . reliable = reliable
this . connectionTimeoutSec = connectionTimeoutSec
2023-06-14 23:35:03 +02:00
2023-05-08 09:58:24 +02:00
val isSelfMachine = remoteAddress ?. isLoopbackAddress == true || remoteAddress == lanAddress
2021-04-30 22:39:36 +02:00
// IPC can be enabled TWO ways!
// - config.enableIpc
// - NULL remoteAddress
// It is entirely possible that the server does not have IPC enabled!
2022-03-24 00:32:58 +01:00
val autoChangeToIpc =
2023-05-08 09:58:24 +02:00
( config . enableIpc && ( remoteAddress == null || isSelfMachine ) ) || ( ! config . enableIpc && remoteAddress == null )
2020-08-12 23:30:16 +02:00
2023-06-14 23:35:03 +02:00
// how long does the initial handshake take to connect
2023-12-12 13:23:51 +01:00
val aeronTimeoutNs = aeronDriver . publicationConnectionTimeoutNs ( ) + aeronDriver . lingerNs ( )
var handshakeTimeoutNs = TimeUnit . SECONDS . toNanos ( config . connectionCloseTimeoutInSeconds . toLong ( ) )
if ( handshakeTimeoutNs < aeronTimeoutNs ) {
handshakeTimeoutNs += aeronTimeoutNs
logger . warn ( " Handshake timeout is less than aeron timeout, increasing timeout to ${Sys.getTimePrettyFull(handshakeTimeoutNs)} to prevent aeron timeout issues " )
}
2023-06-29 20:16:42 +02:00
// how long before we COMPLETELY give up retrying. A '0' means try forever.
2023-07-03 19:18:34 +02:00
var connectionTimoutInNs = TimeUnit . SECONDS . toNanos ( connectionTimeoutSec . toLong ( ) )
2023-06-25 12:21:21 +02:00
2023-12-12 13:23:51 +01:00
if ( connectionTimoutInNs in 1. . < handshakeTimeoutNs ) {
connectionTimoutInNs = 10 * handshakeTimeoutNs
logger . warn ( " Connection timeout is less than handshake timeout, increasing timeout to ${Sys.getTimePrettyFull(connectionTimoutInNs)} to prevent initial connection issues " )
}
2022-08-10 14:34:09 +02:00
if ( DEBUG _CONNECTIONS ) {
// connections are extremely difficult to diagnose when the connection timeout is short
2023-07-03 19:18:34 +02:00
connectionTimoutInNs = TimeUnit . HOURS . toNanos ( 1 )
handshakeTimeoutNs = TimeUnit . HOURS . toNanos ( 1 )
2022-08-10 14:34:09 +02:00
}
2024-02-05 21:57:40 +01:00
2022-07-15 06:58:49 +02:00
val startTime = System . nanoTime ( )
var success = false
2023-09-08 13:21:12 +02:00
while ( ! stopConnectOnShutdown && ( connectionTimoutInNs == 0L || System . nanoTime ( ) - startTime < connectionTimoutInNs ) ) {
2023-11-08 12:44:15 +01:00
logger . trace ( " Starting connect process... " )
2024-02-05 21:57:40 +01:00
clientConnectionInProgress = CountDownLatch ( 1 )
2023-10-26 21:19:36 +02:00
2022-07-15 06:58:49 +02:00
if ( isShutdown ( ) ) {
2023-06-25 12:21:21 +02:00
resetOnError ( )
2022-07-15 06:58:49 +02:00
// If we are connecting indefinitely, we have to make sure to end the connection process
2023-06-25 12:21:21 +02:00
val exception = ClientShutdownException ( " Unable to connect while shutting down, aborting connection retry attempt to server. " )
2022-07-15 06:58:49 +02:00
listenerManager . notifyError ( exception )
throw exception
}
2022-03-24 00:32:58 +01:00
2023-04-20 17:57:14 +02:00
if ( slowDownForException ) {
// short delay, since it failed we want to limit the retry rate to something slower than "as fast as the CPU can do it"
// we also want to go at SLIGHTLY slower that the aeron driver timeout frequency, this way - if there are connection or handshake issues, the server has the chance to expire the connections.
// If we go TOO FAST, then the server will EVENTUALLY have aeron errors (since it can't keep up per client). We literally
// want to have 1 in-flight handshake, per connection attempt, during the aeron connection timeout
// ALSO, we want to make sure we DO NOT approach the linger timeout!
aeronDriver . delayLingerTimeout ( 2 )
}
2023-05-28 18:43:05 +02:00
2022-07-29 04:52:11 +02:00
2023-05-08 09:58:24 +02:00
// the handshake connection is closed when the handshake has an error, or it is finished
2023-10-28 20:54:09 +02:00
var handshakeConnection : ClientHandshakeDriver ?
2023-05-08 09:58:24 +02:00
2022-07-15 06:58:49 +02:00
try {
2023-04-20 17:57:14 +02:00
// always start the aeron driver inside the restart loop.
// If we've already started the driver (on the first "start"), then this does nothing (slowDownForException also make this not "double check")
if ( slowDownForException ) {
startDriver ( )
}
2023-02-20 20:57:04 +01:00
2023-05-08 09:58:24 +02:00
// throws a ConnectTimedOutException if the client cannot connect for any reason to the server handshake ports
2023-06-14 23:35:03 +02:00
handshakeConnection = ClientHandshakeDriver . build (
2024-02-05 21:57:40 +01:00
endpoint = this ,
2023-05-08 09:58:24 +02:00
aeronDriver = aeronDriver ,
autoChangeToIpc = autoChangeToIpc ,
remoteAddress = remoteAddress ,
remoteAddressString = remoteAddressString ,
2023-07-11 00:11:58 +02:00
remotePort1 = port1 ,
clientListenPort = config . port ,
remotePort2 = port2 ,
2023-07-03 19:18:34 +02:00
handshakeTimeoutNs = handshakeTimeoutNs ,
2024-02-05 21:57:40 +01:00
connectionTimoutInNs = connectionTimoutInNs ,
2023-05-08 09:58:24 +02:00
reliable = reliable ,
2023-10-28 20:55:49 +02:00
tagName = tag ,
2023-05-08 09:58:24 +02:00
logger = logger
)
2023-06-25 12:21:21 +02:00
val pubSub = handshakeConnection . pubSub
2022-03-24 00:32:58 +01:00
2023-10-28 20:55:49 +02:00
val logInfo = pubSub . getLogInfo ( logger . isDebugEnabled )
2023-06-25 12:21:21 +02:00
if ( logger . isDebugEnabled ) {
2023-09-13 16:57:32 +02:00
logger . debug ( " Creating new handshake to $logInfo " )
2023-06-25 12:21:21 +02:00
} else {
2023-09-13 16:57:32 +02:00
logger . info ( " Creating new handshake to $logInfo " )
2023-06-25 12:21:21 +02:00
}
2023-07-03 19:18:34 +02:00
connect0 ( handshake , handshakeConnection , handshakeTimeoutNs )
2022-07-15 06:58:49 +02:00
success = true
2023-04-20 17:57:14 +02:00
slowDownForException = false
2024-02-05 21:57:40 +01:00
clientConnectionInProgress . countDown ( )
2022-07-12 16:56:47 +02:00
2022-07-15 06:58:49 +02:00
// once we're done with the connection process, stop trying
break
} catch ( e : ClientRetryException ) {
2024-02-05 21:57:40 +01:00
clientConnectionInProgress . countDown ( )
aeronDriver . closeIfSingle ( ) // if we are the ONLY instance using the media driver, restart it
2023-09-08 13:21:12 +02:00
if ( stopConnectOnShutdown ) {
break
}
2023-08-08 03:56:38 +02:00
val inSeconds = TimeUnit . NANOSECONDS . toSeconds ( handshakeTimeoutNs )
2023-07-20 20:40:17 +02:00
val message = if ( isIPC ) {
2023-08-08 03:56:38 +02:00
" Unable to connect to IPC in $inSeconds seconds, retrying... "
2023-05-24 09:28:55 +02:00
} else {
2023-08-08 03:56:38 +02:00
" Unable to connect to $remoteAddressPrettyString ( $port1 | $port2 ) in $inSeconds seconds, retrying... "
2023-05-24 09:28:55 +02:00
}
2023-02-20 19:27:53 +01:00
if ( logger . isTraceEnabled ) {
2023-09-13 16:57:32 +02:00
logger . trace ( message , e )
2023-02-20 19:27:53 +01:00
} else {
2023-09-13 16:57:32 +02:00
logger . info ( message )
2023-02-20 19:27:53 +01:00
}
2023-04-20 17:57:14 +02:00
slowDownForException = true
2022-08-20 12:15:01 +02:00
} catch ( e : ClientRejectedException ) {
2024-02-05 21:57:40 +01:00
clientConnectionInProgress . countDown ( )
2023-09-08 13:21:12 +02:00
aeronDriver . closeIfSingle ( ) // if we are the ONLY instance using the media driver, stop it
if ( stopConnectOnShutdown ) {
break
}
2023-01-04 23:30:25 +01:00
2023-04-20 17:57:14 +02:00
slowDownForException = true
2022-06-14 22:18:08 +02:00
2022-08-20 12:15:01 +02:00
if ( e . cause is ServerException ) {
2023-06-25 12:21:21 +02:00
resetOnError ( )
2022-11-12 00:38:23 +01:00
val cause = e . cause !!
val wrapped = ClientException ( cause . message !! )
2022-08-20 12:15:01 +02:00
listenerManager . notifyError ( wrapped )
throw wrapped
} else {
2023-06-25 12:21:21 +02:00
resetOnError ( )
2022-08-20 12:15:01 +02:00
listenerManager . notifyError ( e )
throw e
}
2022-07-15 06:58:49 +02:00
} catch ( e : Exception ) {
2024-02-05 21:57:40 +01:00
clientConnectionInProgress . countDown ( )
2023-01-04 23:30:25 +01:00
aeronDriver . closeIfSingle ( ) // if we are the ONLY instance using the media driver, restart it
2023-09-08 13:21:12 +02:00
if ( stopConnectOnShutdown ) {
break
}
2023-09-08 14:18:35 +02:00
val type = if ( isIPC ) {
" IPC "
} else {
" $remoteAddressPrettyString : $port1 : $port2 "
}
listenerManager . notifyError ( ClientException ( " [ ${handshake.connectKey} ] : Un-recoverable error during handshake with $type . Aborting. " , e ) )
2023-06-25 12:21:21 +02:00
resetOnError ( )
2022-07-15 06:58:49 +02:00
throw e
}
}
2022-07-14 04:54:22 +02:00
2022-07-15 06:58:49 +02:00
if ( ! success ) {
2023-06-25 12:21:21 +02:00
endpointIsRunning . lazySet ( false )
2023-09-08 13:21:12 +02:00
if ( stopConnectOnShutdown ) {
2024-02-05 21:57:40 +01:00
val exception = ClientException ( " Client closed during connection attempt to ' $remoteAddressString '. Aborting connection attempts. " ) . cleanStackTrace ( 3 )
2023-09-08 13:21:12 +02:00
listenerManager . notifyError ( exception )
2023-10-26 21:19:36 +02:00
// if we are waiting for this connection to connect (on a different thread, for example), make sure to release it.
closeLatch . countDown ( )
2023-09-08 13:21:12 +02:00
throw exception
}
2023-07-03 19:18:34 +02:00
if ( System . nanoTime ( ) - startTime < connectionTimoutInNs ) {
2023-09-08 14:18:35 +02:00
val type = if ( isIPC ) {
2023-05-28 16:24:16 +02:00
" IPC "
2023-05-24 09:28:55 +02:00
} else {
2023-07-11 00:11:58 +02:00
" $remoteAddressPrettyString : $port1 : $port2 "
2023-05-24 09:28:55 +02:00
}
2023-06-25 12:21:21 +02:00
2022-07-15 06:58:49 +02:00
// we timed out. Throw the appropriate exception
2023-06-25 12:21:21 +02:00
val exception = ClientTimedOutException ( " Unable to connect to the server at $type in $connectionTimeoutSec seconds, aborting connection attempt to server. " )
2022-06-14 22:18:08 +02:00
listenerManager . notifyError ( exception )
throw exception
}
2022-07-15 06:58:49 +02:00
// If we did not connect - throw an error. When `client.connect()` is called, either it connects or throws an error
2023-06-25 12:21:21 +02:00
val exception = ClientRejectedException ( " The server did not respond or permit the connection attempt within $connectionTimeoutSec seconds, aborting connection retry attempt to server. " )
2023-05-08 09:58:24 +02:00
exception . cleanStackTrace ( )
2022-07-15 06:58:49 +02:00
listenerManager . notifyError ( exception )
throw exception
2022-03-24 00:32:58 +01:00
}
}
2020-08-12 23:30:16 +02:00
2023-06-25 12:21:21 +02:00
2022-03-24 00:32:58 +01:00
// the handshake process might have to restart this connection process.
2023-09-04 00:47:46 +02:00
private fun connect0 ( handshake : ClientHandshake < CONNECTION > , handshakeConnection : ClientHandshakeDriver , handshakeTimeoutNs : Long ) {
2020-09-02 02:39:05 +02:00
// this will block until the connection timeout, and throw an exception if we were unable to connect with the server
2022-03-24 00:32:58 +01:00
2023-06-25 12:21:21 +02:00
2021-07-06 15:38:53 +02:00
// throws(ConnectTimedOutException::class, ClientRejectedException::class, ClientException::class)
2023-07-03 19:18:34 +02:00
val connectionInfo = handshake . hello (
2023-10-28 20:55:49 +02:00
tagName = tag ,
2023-09-26 19:53:27 +02:00
endPoint = this ,
2023-07-03 19:18:34 +02:00
handshakeConnection = handshakeConnection ,
handshakeTimeoutNs = handshakeTimeoutNs
)
2020-08-12 23:30:16 +02:00
2023-10-28 20:54:09 +02:00
bufferedManager = BufferManager ( config , listenerManager , aeronDriver , connectionInfo . sessionTimeout )
2020-09-02 02:39:05 +02:00
// VALIDATE:: check to see if the remote connection's public key has changed!
2023-06-14 23:35:03 +02:00
val validateRemoteAddress = if ( handshakeConnection . pubSub . isIpc ) {
2020-09-02 02:39:05 +02:00
PublicKeyValidationState . VALID
} else {
2023-06-26 19:28:55 +02:00
crypto . validateRemoteAddress ( address !! , addressString , connectionInfo . publicKey )
2020-09-02 02:39:05 +02:00
}
2020-08-12 23:30:16 +02:00
2020-09-02 02:39:05 +02:00
if ( validateRemoteAddress == PublicKeyValidationState . INVALID ) {
2023-09-26 19:53:27 +02:00
handshakeConnection . close ( this )
2022-07-29 04:52:11 +02:00
2023-06-26 19:28:55 +02:00
val exception = ClientRejectedException ( " Connection to [ $addressString ] not allowed! Public key mismatch. " )
2023-06-25 12:21:21 +02:00
listenerManager . notifyError ( exception )
2020-09-02 02:39:05 +02:00
throw exception
}
2020-08-12 23:30:16 +02:00
2020-08-25 17:45:08 +02:00
2022-03-24 00:32:58 +01:00
// VALIDATE:: If the serialization DOES NOT match between the client/server, then the server will emit a log, and the
2022-06-04 01:32:58 +02:00
// client will timeout. SPECIFICALLY.... we do not give class serialization/registration info to the client - in case the client
2020-09-02 02:39:05 +02:00
// is rogue, we do not want to carelessly provide info.
2023-09-21 12:53:47 +02:00
2022-08-04 03:39:14 +02:00
///////////////
//// RMI
///////////////
2023-07-12 14:08:12 +02:00
try {
// only have ot do one
2023-07-23 23:02:29 +02:00
serialization . finishClientConnect ( connectionInfo . kryoRegistrationDetails )
2023-07-12 14:08:12 +02:00
} catch ( e : Exception ) {
2023-09-26 19:53:27 +02:00
handshakeConnection . close ( this )
2022-08-04 03:39:14 +02:00
// because we are getting the class registration details from the SERVER, this should never be the case.
// It is still and edge case where the reconstruction of the registration details fails (maybe because of custom serializers)
2023-06-14 23:35:03 +02:00
val exception = if ( handshakeConnection . pubSub . isIpc ) {
2023-07-12 14:08:12 +02:00
ClientRejectedException ( " [ ${handshake.connectKey} ] Connection to IPC has incorrect class registration details!! " , e )
2022-08-04 03:39:14 +02:00
} else {
2023-07-12 14:08:12 +02:00
ClientRejectedException ( " [ ${handshake.connectKey} ] Connection to [ $addressString ] has incorrect class registration details!! " , e )
2022-08-04 03:39:14 +02:00
}
2023-07-12 14:08:12 +02:00
2023-05-08 09:58:24 +02:00
exception . cleanStackTraceInternal ( )
2023-06-25 12:21:21 +02:00
listenerManager . notifyError ( exception )
2022-08-04 03:39:14 +02:00
throw exception
}
2023-07-12 14:08:12 +02:00
// we set up our kryo information once we connect to a server (using the server's kryo registration details)
// every time we connect to a server, we have to reconfigure AND reassign kryo
2023-07-23 23:02:29 +02:00
readKryo = serialization . newReadKryo ( )
2023-02-08 22:56:23 +01:00
2022-08-04 03:39:14 +02:00
///////////////
//// CONFIG THE CLIENT
///////////////
2020-09-02 02:39:05 +02:00
// we are now connected, so we can connect to the NEW client-specific ports
2023-07-03 19:18:34 +02:00
val clientConnection = ClientConnectionDriver . build (
2024-02-05 21:57:40 +01:00
shutdown = shutdown ,
2023-07-03 19:18:34 +02:00
aeronDriver = aeronDriver ,
handshakeTimeoutNs = handshakeTimeoutNs ,
handshakeConnection = handshakeConnection ,
2023-07-11 00:11:58 +02:00
connectionInfo = connectionInfo ,
2023-10-28 20:55:49 +02:00
port2Server = port2 ,
tagName = tag
2023-07-03 19:18:34 +02:00
)
2021-07-06 15:38:53 +02:00
2023-07-01 13:26:06 +02:00
val pubSub = clientConnection . connectionInfo
2023-10-28 20:54:09 +02:00
2023-11-28 20:53:37 +01:00
val bufferedMessages = connectionInfo . bufferedMessages
val connectionType = if ( bufferedMessages ) {
2023-11-27 11:14:52 +01:00
" buffered connection "
} else {
" connection "
}
2023-11-28 20:53:37 +01:00
val connectionTypeCaps = connectionType . replaceFirstChar { if ( it . isLowerCase ( ) ) it . titlecase ( Locale . getDefault ( ) ) else it . toString ( ) }
2023-11-27 11:14:52 +01:00
2023-10-28 20:54:09 +02:00
val logInfo = pubSub . getLogInfo ( logger . isDebugEnabled )
2023-06-25 12:21:21 +02:00
if ( logger . isDebugEnabled ) {
2023-11-27 11:14:52 +01:00
logger . debug ( " Creating new $connectionType to $logInfo " )
2023-06-25 12:21:21 +02:00
} else {
2023-11-27 11:14:52 +01:00
logger . info ( " Creating new $connectionType to $logInfo " )
2023-06-25 12:21:21 +02:00
}
2022-08-04 03:39:14 +02:00
2023-09-17 02:36:45 +02:00
val newConnection = newConnection ( ConnectionParams (
2023-11-27 11:14:52 +01:00
publicKey = connectionInfo . publicKey ,
endPoint = this ,
connectionInfo = clientConnection . connectionInfo ,
publicKeyValidation = validateRemoteAddress ,
2023-11-28 20:53:37 +01:00
enableBufferedMessages = bufferedMessages ,
2023-11-27 11:14:52 +01:00
cryptoKey = connectionInfo . secretKey
2023-07-15 13:11:50 +02:00
) )
2023-10-28 20:54:09 +02:00
bufferedManager ?. onConnect ( newConnection )
2023-09-21 12:53:47 +02:00
2023-07-15 13:11:50 +02:00
if ( ! handshakeConnection . pubSub . isIpc ) {
2023-04-20 17:55:58 +02:00
// NOTE: Client can ALWAYS connect to the server. The server makes the decision if the client can connect or not.
2023-10-18 19:11:03 +02:00
if ( logger . isTraceEnabled ) {
2023-11-28 20:53:37 +01:00
logger . trace ( " [ ${handshakeConnection.details} ] ( ${handshake.connectKey} ) $connectionTypeCaps ( ${newConnection.id} ) adding new signature for [ $addressString -> ${connectionInfo.publicKey.toHexString()} ] " )
2023-10-18 19:46:34 +02:00
} else if ( logger . isDebugEnabled ) {
2023-11-28 20:53:37 +01:00
logger . debug ( " [ ${handshakeConnection.details} ] $connectionTypeCaps ( ${newConnection.id} ) adding new signature for [ $addressString -> ${connectionInfo.publicKey.toHexString()} ] " )
2023-10-18 19:11:03 +02:00
} else if ( logger . isInfoEnabled ) {
2023-11-28 20:53:37 +01:00
logger . info ( " [ ${handshakeConnection.details} ] $connectionTypeCaps adding new signature for [ $addressString -> ${connectionInfo.publicKey.toHexString()} ] " )
2023-09-13 16:57:32 +02:00
}
2023-05-08 09:58:24 +02:00
2023-06-26 19:28:55 +02:00
storage . addRegisteredServerKey ( address !! , connectionInfo . publicKey )
2020-09-02 02:39:05 +02:00
}
2020-08-12 23:30:16 +02:00
2020-09-03 01:31:08 +02:00
// tell the server our connection handshake is done, and the connection can now listen for data.
2020-09-11 11:06:53 +02:00
// also closes the handshake (will also throw connect timeout exception)
2022-06-27 15:31:40 +02:00
try {
2023-09-26 19:53:27 +02:00
handshake . done (
endPoint = this ,
handshakeConnection , clientConnection ,
handshakeTimeoutNs = handshakeTimeoutNs ,
logInfo = handshakeConnection . details
2023-07-03 19:18:34 +02:00
)
2022-06-27 15:31:40 +02:00
} catch ( e : Exception ) {
2023-06-26 19:28:55 +02:00
listenerManager . notifyError ( ClientHandshakeException ( " [ ${handshakeConnection.details} ] ( ${handshake.connectKey} ) Connection ( ${newConnection.id} ) to [ $addressString ] error during handshake " , e ) )
2022-06-27 15:31:40 +02:00
throw e
2021-07-06 15:38:53 +02:00
}
2023-05-08 09:58:24 +02:00
// finished with the handshake, so always close these!
2023-09-26 19:53:27 +02:00
handshakeConnection . close ( this )
2022-12-15 23:57:22 +01:00
2023-10-18 19:11:03 +02:00
if ( logger . isTraceEnabled ) {
2023-11-28 20:53:37 +01:00
logger . debug ( " [ ${handshakeConnection.details} ] ( ${handshake.connectKey} ) $connectionTypeCaps ( ${newConnection.id} ) done with handshake. " )
2023-10-18 19:11:03 +02:00
} else if ( logger . isDebugEnabled ) {
2023-11-28 20:53:37 +01:00
logger . debug ( " [ ${handshakeConnection.details} ] $connectionTypeCaps ( ${newConnection.id} ) done with handshake. " )
2023-09-13 16:01:14 +02:00
}
2022-06-16 18:04:52 +02:00
2023-10-26 21:19:36 +02:00
connection0 = newConnection
newConnection . setImage ( )
2023-06-25 12:21:21 +02:00
// before we finish creating the connection, we initialize it (in case there needs to be logic that happens-before `onConnect` calls
2023-10-28 20:54:09 +02:00
listenerManager . notifyInit ( newConnection )
2023-06-25 12:21:21 +02:00
2023-09-25 13:53:37 +02:00
// this enables the connection to start polling for messages
2023-09-22 15:53:46 +02:00
addConnection ( newConnection )
2023-10-26 21:19:36 +02:00
2023-09-08 13:21:12 +02:00
// if we shutdown/close before the poller starts, we don't want to block forever
pollerClosedLatch = CountDownLatch ( 1 )
2023-06-25 12:21:21 +02:00
networkEventPoller . submit (
2023-09-03 21:17:37 +02:00
action = object : EventActionOperator {
override fun invoke ( ) : Int {
2023-10-26 21:19:36 +02:00
val connection = connection0
2023-09-03 21:17:37 +02:00
// if we initiate a disconnect manually, then there is no need to wait for aeron to verify it's closed
// we only want to wait for aeron to verify it's closed if we are SUPPOSED to be connected, but there's a network blip
2023-10-26 21:19:36 +02:00
return if ( connection != null ) {
if ( ! shutdownEventPoller && connection . canPoll ( ) ) {
connection . poll ( )
} else {
// If the connection has either been closed, or has expired, it needs to be cleaned-up/deleted.
logger . error ( " [ ${connection} ] connection expired (cleanup). shutdownEventPoller= $shutdownEventPoller isClosed()= ${connection.isClosed()} isClosedWithTimeout= ${connection.isClosedWithTimeout()} " )
if ( logger . isDebugEnabled ) {
logger . debug ( " [{}] connection expired (cleanup) " , connection )
}
// remove ourselves from processing
EventPoller . REMOVE
2023-09-13 16:01:14 +02:00
}
2023-10-26 21:19:36 +02:00
} else {
2023-09-03 21:17:37 +02:00
// remove ourselves from processing
EventPoller . REMOVE
}
2021-04-30 16:01:25 +02:00
}
2023-06-25 12:21:21 +02:00
} ,
2023-09-03 21:17:37 +02:00
onClose = object : EventCloseOperator {
2023-09-04 14:23:06 +02:00
override fun invoke ( ) {
2023-10-26 21:19:36 +02:00
val connection = connection0
if ( connection == null ) {
logger . error ( " Unable to continue, as the connection has been removed before event dispatch shutdown! " )
return
}
2023-09-03 21:17:37 +02:00
val mustRestartDriverOnError = aeronDriver . internal . mustRestartDriverOnError
2023-10-28 20:54:09 +02:00
val dirtyDisconnectWithSession = ! shutdownEventPoller && connection . isDirtyClose ( )
2023-10-18 19:51:28 +02:00
2023-10-26 21:19:36 +02:00
autoReconnect = mustRestartDriverOnError || dirtyDisconnectWithSession
2023-10-18 19:51:28 +02:00
if ( mustRestartDriverOnError ) {
2023-10-23 23:27:55 +02:00
logger . error ( " [{}] Critical driver error detected, reconnecting client " , connection )
2023-10-26 21:19:36 +02:00
} else if ( dirtyDisconnectWithSession ) {
2023-10-23 23:27:55 +02:00
logger . error ( " [{}] Dirty disconnect detected, reconnecting client " , connection )
2023-10-18 19:51:28 +02:00
}
2023-09-03 21:17:37 +02:00
// this can be closed when the connection is remotely closed in ADDITION to manually closing
2023-09-13 16:01:14 +02:00
if ( logger . isDebugEnabled ) {
2023-10-23 23:27:55 +02:00
logger . debug ( " [{}] Client event dispatch closing (in progress: $shutdownInProgress ) ... " , connection )
2023-09-13 16:01:14 +02:00
}
2023-09-03 21:17:37 +02:00
// we only need to run shutdown methods if there was a network outage or D/C
if ( ! shutdownInProgress . value ) {
2023-10-18 19:51:28 +02:00
// this is because we restart automatically on driver errors/weird timeouts
2023-10-26 21:19:36 +02:00
this @Client . close ( closeEverything = false , sendDisconnectMessage = true , releaseWaitingThreads = ! autoReconnect )
2023-09-03 21:17:37 +02:00
}
2022-06-27 15:31:40 +02:00
2023-08-11 04:01:05 +02:00
2023-09-03 21:17:37 +02:00
// we can now call connect again
endpointIsRunning . lazySet ( false )
pollerClosedLatch . countDown ( )
2023-04-20 17:55:17 +02:00
2023-10-26 21:19:36 +02:00
connection0 = null
2023-10-18 19:51:28 +02:00
if ( autoReconnect ) {
2023-10-26 21:19:36 +02:00
// clients can reconnect automatically ONLY if there are driver errors, otherwise it's explicit!
eventDispatch . CLOSE . launch {
2023-09-03 21:17:37 +02:00
waitForEndpointShutdown ( )
2023-08-11 04:01:05 +02:00
2023-09-03 21:17:37 +02:00
// also wait for everyone else to shutdown!!
aeronDriver . internal . endPointUsages . forEach {
2023-10-26 21:19:36 +02:00
if ( it !== this @Client ) {
it . waitForEndpointShutdown ( )
}
2023-09-03 21:17:37 +02:00
}
2023-08-11 04:01:05 +02:00
2023-09-03 21:17:37 +02:00
// if we restart/reconnect too fast, errors from the previous run will still be present!
aeronDriver . delayLingerTimeout ( )
2023-08-11 04:01:05 +02:00
2023-10-26 21:19:36 +02:00
if ( connectionTimeoutSec == 0 ) {
2023-11-27 13:11:47 +01:00
logger . info ( " Reconnecting... " )
2023-10-26 21:19:36 +02:00
} else {
2023-11-27 13:11:47 +01:00
logger . info ( " Reconnecting... (timeout in $connectionTimeoutSec seconds) " )
2023-10-26 21:19:36 +02:00
}
connect (
remoteAddress = address ,
remoteAddressString = addressString ,
remoteAddressPrettyString = addressPrettyString ,
port1 = port1 ,
port2 = port2 ,
connectionTimeoutSec = connectionTimeoutSec ,
reliable = reliable ,
)
2023-09-03 21:17:37 +02:00
}
2023-08-11 04:01:05 +02:00
}
2023-10-26 21:19:36 +02:00
logger . debug ( " [{}] Closed the Network Event Poller task. " , connection )
2023-08-11 04:01:05 +02:00
}
2023-06-25 12:21:21 +02:00
} )
2023-10-26 21:19:36 +02:00
listenerManager . notifyConnect ( newConnection )
2023-10-28 20:54:09 +02:00
newConnection . sendBufferedMessages ( )
2020-08-12 23:30:16 +02:00
}
2020-09-02 02:39:05 +02:00
/ * *
2020-09-03 01:31:08 +02:00
* true if the remote public key changed . This can be useful if specific actions are necessary when the key has changed .
2020-09-02 02:39:05 +02:00
* /
2020-09-03 01:31:08 +02:00
val remoteKeyHasChanged : Boolean
2023-07-15 13:11:50 +02:00
get ( ) = connection . remoteKeyChanged
2020-08-12 23:30:43 +02:00
2020-09-02 02:39:05 +02:00
/ * *
2020-09-03 01:31:08 +02:00
* true if this connection is an IPC connection
2020-09-02 02:39:05 +02:00
* /
2020-09-03 01:31:08 +02:00
val isIPC : Boolean
2023-07-20 20:40:17 +02:00
get ( ) = address == null
2020-08-12 23:30:43 +02:00
2020-09-02 02:39:05 +02:00
/ * *
* @return true if this connection is a network connection
* /
2020-09-03 01:31:08 +02:00
val isNetwork : Boolean
get ( ) = connection . isNetwork
2020-08-12 23:30:43 +02:00
2020-09-02 02:39:05 +02:00
/ * *
2023-06-25 12:21:21 +02:00
* @return the connection id of this connection .
2020-09-02 02:39:05 +02:00
* /
2020-09-03 01:31:08 +02:00
val id : Int
get ( ) = connection . id
2020-08-12 23:30:16 +02:00
/ * *
2020-09-03 01:31:08 +02:00
* the connection used by the client , this is only valid after the client has connected
2020-08-12 23:30:16 +02:00
* /
2020-09-03 01:31:08 +02:00
val connection : CONNECTION
get ( ) = connection0 as CONNECTION
2020-08-12 23:30:16 +02:00
/ * *
2020-09-22 19:42:04 +02:00
* Sends a message to the server , if the connection is closed for any reason , this returns false .
*
* @return true if the message was sent successfully , false if the connection has been closed
* /
2023-09-04 00:47:46 +02:00
fun send ( message : Any ) : Boolean {
2020-09-22 19:42:04 +02:00
val c = connection0
return if ( c != null ) {
c . send ( message )
2023-11-28 20:53:37 +01:00
} else if ( isShutdown ( ) ) {
logger . error ( " Cannot send a message ' ${message::class.java} ' when there is no connection! (We are shutdown) " )
false
2020-09-22 19:42:04 +02:00
} else {
2023-11-28 20:53:37 +01:00
val exception = TransmitException ( " Cannot send message ' ${message::class.java} ' when there is no connection! " )
2023-06-25 12:21:21 +02:00
listenerManager . notifyError ( exception )
2022-08-18 22:01:42 +02:00
false
}
}
2023-11-02 22:36:50 +01:00
/ * *
* Safely sends objects to a destination , where the callback is notified once the remote endpoint has received the message .
* This is to guarantee happens - before , and using this will depend upon APP + NETWORK latency , and is ( by design ) not as performant as
* sending a regular message !
*
* @return true if the message was sent successfully , false if the connection has been closed
* /
fun send ( message : Any , onSuccessCallback : CONNECTION . ( ) -> Unit ) : Boolean {
val c = connection0
return if ( c != null ) {
@Suppress ( " UNCHECKED_CAST " )
c . send ( message , onSuccessCallback as Connection . ( ) -> Unit )
2023-11-28 20:53:37 +01:00
} else if ( isShutdown ( ) ) {
logger . error ( " Cannot send-sync message ' ${message::class.java} ' when there is no connection! (We are shutdown) " )
false
2023-11-02 22:36:50 +01:00
} else {
2023-11-28 20:53:37 +01:00
val exception = TransmitException ( " Cannot send-sync message ' ${message::class.java} ' when there is no connection! " )
2023-11-02 22:36:50 +01:00
listenerManager . notifyError ( exception )
false
}
}
2021-04-29 10:25:25 +02:00
/ * *
* Sends a " ping " packet to measure * * ROUND TRIP * * time to the remote connection .
2020-09-22 19:42:04 +02:00
*
2021-04-29 10:25:25 +02:00
* @param function called when the ping returns ( ie : update time / latency counters / metrics / etc )
2021-07-06 15:38:53 +02:00
*
* @return true if the ping was successfully sent to the client
2020-08-12 23:30:16 +02:00
* /
2023-09-05 12:58:38 +02:00
fun ping ( function : Ping . ( ) -> Unit ) : Boolean {
2020-09-03 01:31:08 +02:00
val c = connection0
2021-04-29 10:25:25 +02:00
2023-11-28 20:53:37 +01:00
return if ( c != null ) {
c . ping ( function )
} else if ( isShutdown ( ) ) {
logger . error ( " Cannot send a ping when there is no connection! (We are shutdown) " )
false
2020-08-12 23:30:16 +02:00
} else {
2023-06-25 12:21:21 +02:00
val exception = TransmitException ( " Cannot send a ping when there is no connection! " )
listenerManager . notifyError ( exception )
2023-11-28 20:53:37 +01:00
false
2020-08-12 23:30:16 +02:00
}
}
2020-09-03 01:31:08 +02:00
/ * *
* Removes the specified host address from the list of registered server keys .
* /
2020-09-09 01:33:09 +02:00
fun removeRegisteredServerKey ( address : InetAddress ) {
2021-04-29 10:02:59 +02:00
val savedPublicKey = storage . getRegisteredServerKey ( address )
2020-08-12 23:30:16 +02:00
if ( savedPublicKey != null ) {
2023-09-13 16:01:14 +02:00
if ( logger . isDebugEnabled ) {
2023-09-13 16:57:32 +02:00
logger . debug ( " Deleting remote IP address key $address " )
2023-09-13 16:01:14 +02:00
}
2021-04-29 10:02:59 +02:00
storage . removeRegisteredServerKey ( address )
2020-08-12 23:30:16 +02:00
}
}
2023-06-25 12:21:21 +02:00
/ * *
* Will throw an exception if there are resources that are still in use
* /
fun checkForMemoryLeaks ( ) {
AeronDriver . checkForMemoryLeaks ( )
}
/ * *
2023-06-25 17:23:44 +02:00
* By default , if you call close ( ) on the client , it will shut down all parts of the endpoint ( listeners , driver , event polling , etc ) .
*
* @param closeEverything if true , all parts of the client will be closed ( listeners , driver , event polling , etc )
2023-06-25 12:21:21 +02:00
* /
2023-06-25 17:23:44 +02:00
fun close ( closeEverything : Boolean = true ) {
2023-09-08 13:21:12 +02:00
stopConnectOnShutdown = true
2023-10-28 20:54:09 +02:00
bufferedManager ?. close ( )
2023-10-26 21:19:36 +02:00
close ( closeEverything = closeEverything , sendDisconnectMessage = true , releaseWaitingThreads = true )
2023-06-25 12:21:21 +02:00
}
override fun toString ( ) : String {
2023-07-02 21:59:45 +02:00
return string0
2020-08-12 23:30:43 +02:00
}
2023-05-08 09:58:24 +02:00
fun < R > use ( block : ( Client < CONNECTION > ) -> R ) : R {
return try {
block ( this )
} finally {
close ( )
}
}
2020-08-12 23:30:16 +02:00
}