902 lines
39 KiB
Kotlin
902 lines
39 KiB
Kotlin
/*
|
|
* Copyright 2020 dorkbox, llc
|
|
*
|
|
* 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
|
|
|
|
import dorkbox.bytes.toHexString
|
|
import dorkbox.dns.DnsClient
|
|
import dorkbox.netUtil.IP
|
|
import dorkbox.netUtil.IPv4
|
|
import dorkbox.netUtil.IPv6
|
|
import dorkbox.netUtil.Inet4
|
|
import dorkbox.netUtil.Inet6
|
|
import dorkbox.netUtil.dnsUtils.ResolvedAddressTypes
|
|
import dorkbox.network.aeron.AeronDriver
|
|
import dorkbox.network.aeron.IpcMediaDriverConnection
|
|
import dorkbox.network.aeron.MediaDriverConnection
|
|
import dorkbox.network.aeron.UdpMediaDriverClientConnection
|
|
import dorkbox.network.connection.Connection
|
|
import dorkbox.network.connection.ConnectionParams
|
|
import dorkbox.network.connection.EndPoint
|
|
import dorkbox.network.connection.ListenerManager
|
|
import dorkbox.network.connection.PublicKeyValidationState
|
|
import dorkbox.network.connection.eventLoop
|
|
import dorkbox.network.exceptions.ClientException
|
|
import dorkbox.network.exceptions.ClientRejectedException
|
|
import dorkbox.network.exceptions.ClientRetryException
|
|
import dorkbox.network.exceptions.ClientShutdownException
|
|
import dorkbox.network.exceptions.ClientTimedOutException
|
|
import dorkbox.network.handshake.ClientHandshake
|
|
import dorkbox.network.ping.Ping
|
|
import dorkbox.network.ping.PingManager
|
|
import kotlinx.atomicfu.atomic
|
|
import kotlinx.coroutines.launch
|
|
import kotlinx.coroutines.runBlocking
|
|
import kotlinx.coroutines.sync.Mutex
|
|
import kotlinx.coroutines.sync.withLock
|
|
import java.lang.Thread.sleep
|
|
import java.net.Inet4Address
|
|
import java.net.Inet6Address
|
|
import java.net.InetAddress
|
|
import java.util.concurrent.*
|
|
|
|
/**
|
|
* 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.
|
|
*
|
|
* @param config these are the specific connection options
|
|
* @param connectionFunc allows for custom connection implementations defined as a unit function
|
|
* @param loggerName allows for a custom logger name for this endpoint (for when there are multiple endpoints)
|
|
*/
|
|
@Suppress("unused")
|
|
open class Client<CONNECTION : Connection>(
|
|
config: Configuration = Configuration(),
|
|
connectionFunc: (connectionParameters: ConnectionParams<CONNECTION>) -> CONNECTION,
|
|
loggerName: String = Client::class.java.simpleName)
|
|
: EndPoint<CONNECTION>(config, connectionFunc, loggerName) {
|
|
|
|
/**
|
|
* 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.
|
|
*
|
|
* @param config these are the specific connection options
|
|
* @param loggerName allows for a custom logger name for this endpoint (for when there are multiple endpoints)
|
|
* @param connectionFunc allows for custom connection implementations defined as a unit function
|
|
*/
|
|
constructor(config: Configuration,
|
|
loggerName: String,
|
|
connectionFunc: (connectionParameters: ConnectionParams<CONNECTION>) -> CONNECTION)
|
|
: this(config, connectionFunc, loggerName)
|
|
|
|
|
|
/**
|
|
* 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.
|
|
*
|
|
* @param config these are the specific connection options
|
|
* @param connectionFunc allows for custom connection implementations defined as a unit function
|
|
*/
|
|
constructor(config: Configuration,
|
|
connectionFunc: (connectionParameters: ConnectionParams<CONNECTION>) -> CONNECTION)
|
|
: this(config, connectionFunc, Client::class.java.simpleName)
|
|
|
|
|
|
/**
|
|
* 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.
|
|
*
|
|
* @param config these are the specific connection options
|
|
* @param loggerName allows for a custom logger name for this endpoint (for when there are multiple endpoints)
|
|
*/
|
|
constructor(config: Configuration,
|
|
loggerName: String)
|
|
: this(config,
|
|
{
|
|
@Suppress("UNCHECKED_CAST")
|
|
Connection(it) as CONNECTION
|
|
},
|
|
loggerName)
|
|
|
|
|
|
/**
|
|
* 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.
|
|
*
|
|
* @param config these are the specific connection options
|
|
*/
|
|
constructor(config: Configuration)
|
|
: this(config,
|
|
{
|
|
@Suppress("UNCHECKED_CAST")
|
|
Connection(it) as CONNECTION
|
|
},
|
|
Client::class.java.simpleName)
|
|
|
|
|
|
|
|
companion object {
|
|
/**
|
|
* Gets the version number.
|
|
*/
|
|
const val version = "5.28"
|
|
|
|
/**
|
|
* Checks to see if a client (using the specified configuration) is running.
|
|
*
|
|
* This method should only be used to check if a client is running for a DIFFERENT configuration than the currently running client
|
|
*/
|
|
fun isRunning(configuration: Configuration): Boolean {
|
|
return AeronDriver(configuration).isRunning()
|
|
}
|
|
|
|
init {
|
|
// Add this project to the updates system, which verifies this class + UUID + version information
|
|
dorkbox.updates.Updates.add(Client::class.java, "5be42ae40cac49fb90dea86bc513141b", version)
|
|
}
|
|
}
|
|
|
|
/**
|
|
* The network or IPC address for the client to connect to.
|
|
*
|
|
* For a network address, it can be:
|
|
* - a network name ("localhost", "loopback", "lo", "bob.example.org")
|
|
* - an IP address ("127.0.0.1", "123.123.123.123", "::1")
|
|
*
|
|
* For the IPC (Inter-Process-Communication) address. it must be:
|
|
* - the IPC integer ID, "0x1337c0de", "0x12312312", etc.
|
|
*/
|
|
@Volatile
|
|
var remoteAddress: InetAddress? = IPv4.LOCALHOST
|
|
private set
|
|
|
|
/**
|
|
* the remote address, as a string.
|
|
*/
|
|
@Volatile
|
|
var remoteAddressString: String = "UNKNOWN"
|
|
private set
|
|
|
|
|
|
@Volatile
|
|
private var isConnected = false
|
|
|
|
// is valid when there is a connection to the server, otherwise it is null
|
|
private var connection0: CONNECTION? = null
|
|
|
|
|
|
// This is set by the client so if there is a "connect()" call in the the disconnect callback, we can have proper
|
|
// lock-stop ordering for how disconnect and connect work with each-other
|
|
// GUARANTEE that the callbacks for 'onDisconnect' happens-before the 'onConnect'.
|
|
private val lockStepForConnect = atomic<Mutex?>(null)
|
|
|
|
final override fun newException(message: String, cause: Throwable?): Throwable {
|
|
return ClientException(message, cause)
|
|
}
|
|
|
|
/**
|
|
* 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
|
|
*
|
|
* ### For the IPC (Inter-Process-Communication) it must be:
|
|
* - `connect()`
|
|
* - `connect("")`
|
|
* - `connectIpc()`
|
|
*
|
|
* ### Case does not matter, and "localhost" is the default.
|
|
*
|
|
* @param remoteAddress The network or if localhost, IPC address for the client to connect to
|
|
* @param connectionTimeoutSec wait for x seconds. 0 will wait indefinitely
|
|
* @param reliable true if we want to create a reliable connection (for UDP connections, is message loss acceptable?).
|
|
*
|
|
* @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
|
|
*/
|
|
fun connect(
|
|
remoteAddress: InetAddress,
|
|
connectionTimeoutSec: Int = 30,
|
|
reliable: Boolean = true)
|
|
{
|
|
val remoteAddressString = when (remoteAddress) {
|
|
is Inet4Address -> IPv4.toString(remoteAddress)
|
|
is Inet6Address -> IPv6.toString(remoteAddress, true)
|
|
else -> throw IllegalArgumentException("Cannot connect to $remoteAddress It is an invalid address!")
|
|
}
|
|
|
|
|
|
// Default IPC ports are flipped because they are in the perspective of the SERVER
|
|
connect(remoteAddress = remoteAddress,
|
|
remoteAddressString = remoteAddressString,
|
|
connectionTimeoutSec = connectionTimeoutSec,
|
|
reliable = reliable)
|
|
}
|
|
|
|
/**
|
|
* Will attempt to connect to the server via IPC, with a default 30 second connection timeout and will block until completed.
|
|
*
|
|
* @param ipcPublicationId The IPC publication address for the client to connect to
|
|
* @param ipcSubscriptionId The IPC subscription address for the client to connect to
|
|
* @param connectionTimeoutSec wait for x seconds. 0 will wait indefinitely.
|
|
*
|
|
* @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
|
|
*/
|
|
@Suppress("DuplicatedCode")
|
|
fun connectIpc(
|
|
ipcPublicationId: Int = AeronDriver.IPC_HANDSHAKE_STREAM_ID_SUB,
|
|
ipcSubscriptionId: Int = AeronDriver.IPC_HANDSHAKE_STREAM_ID_PUB,
|
|
connectionTimeoutSec: Int = 30)
|
|
{
|
|
|
|
// Default IPC ports are flipped because they are in the perspective of the SERVER
|
|
|
|
require(ipcPublicationId != ipcSubscriptionId) { "IPC publication and subscription ports cannot be the same! The must match the server's configuration." }
|
|
|
|
connect(remoteAddress = null, // required!
|
|
remoteAddressString = "IPC",
|
|
ipcPublicationId = ipcPublicationId,
|
|
ipcSubscriptionId = ipcSubscriptionId,
|
|
connectionTimeoutSec = connectionTimeoutSec)
|
|
}
|
|
|
|
/**
|
|
* 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
|
|
* - if no address is specified, and IPC is disabled in the config, then localhost will be selected
|
|
*
|
|
* ### For the IPC (Inter-Process-Communication) it must be:
|
|
* - `connect()` (only if ipc is enabled in the configuration)
|
|
* - `connect("")` (only if ipc is enabled in the configuration)
|
|
* - `connectIpc()`
|
|
*
|
|
* ### Case does not matter, and "localhost" is the default.
|
|
*
|
|
* @param remoteAddress The network host or ip address
|
|
* @param connectionTimeoutSec wait for x seconds. 0 will wait indefinitely
|
|
* @param reliable true if we want to create a reliable connection (for UDP connections, is message loss acceptable?).
|
|
*
|
|
* @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
|
|
*/
|
|
@Suppress("BlockingMethodInNonBlockingContext")
|
|
fun connect(
|
|
remoteAddress: String = "",
|
|
connectionTimeoutSec: Int = 30,
|
|
reliable: Boolean = true)
|
|
{
|
|
|
|
when {
|
|
// this is default IPC settings
|
|
remoteAddress.isEmpty() && config.enableIpc -> {
|
|
connectIpc(connectionTimeoutSec = connectionTimeoutSec)
|
|
}
|
|
|
|
IPv4.isPreferred -> {
|
|
// we have to check first if it's a valid IPv4 address. If not, maybe it's a DNS lookup
|
|
val inet4Address = if (IPv4.isValid(remoteAddress)) {
|
|
Inet4.toAddress(remoteAddress)
|
|
} else {
|
|
val client = DnsClient()
|
|
client.resolvedAddressTypes(ResolvedAddressTypes.IPV4_ONLY)
|
|
val records = client.resolve(remoteAddress)
|
|
client.stop()
|
|
records?.get(0)
|
|
}
|
|
|
|
if (inet4Address == null) {
|
|
throw IllegalArgumentException("The remote address '$remoteAddress' cannot be found.")
|
|
}
|
|
|
|
connect(remoteAddress = inet4Address,
|
|
remoteAddressString = remoteAddress,
|
|
connectionTimeoutSec = connectionTimeoutSec,
|
|
reliable = reliable)
|
|
}
|
|
|
|
IPv6.isPreferred -> {
|
|
// we have to check first if it's a valid IPv6 address. If not, maybe it's a DNS lookup
|
|
val inet6Address = if (IPv6.isValid(remoteAddress)) {
|
|
Inet6.toAddress(remoteAddress)
|
|
} else {
|
|
val client = DnsClient()
|
|
client.resolvedAddressTypes(ResolvedAddressTypes.IPV6_ONLY)
|
|
val records = client.resolve(remoteAddress)
|
|
client.stop()
|
|
records?.get(0)
|
|
}
|
|
|
|
if (inet6Address == null) {
|
|
throw IllegalArgumentException("The remote address '$remoteAddress' cannot be found.")
|
|
}
|
|
|
|
connect(remoteAddress = inet6Address,
|
|
remoteAddressString = remoteAddress,
|
|
connectionTimeoutSec = connectionTimeoutSec,
|
|
reliable = reliable)
|
|
}
|
|
|
|
// if there is no preference, then try to connect via IPv4
|
|
else -> {
|
|
// we have to check first if it's a valid IPv4 address. If not, maybe it's a DNS lookup
|
|
val inetAddress = if (IP.isValid(remoteAddress)) {
|
|
IP.toAddress(remoteAddress)
|
|
} else {
|
|
val client = DnsClient()
|
|
client.resolvedAddressTypes(ResolvedAddressTypes.IPV4_PREFERRED)
|
|
val records = client.resolve(remoteAddress)
|
|
client.stop()
|
|
records?.get(0)
|
|
}
|
|
|
|
if (inetAddress == null) {
|
|
throw IllegalArgumentException("The remote address '$remoteAddress' cannot be found.")
|
|
}
|
|
|
|
connect(remoteAddress = inetAddress,
|
|
remoteAddressString = remoteAddress,
|
|
connectionTimeoutSec = connectionTimeoutSec,
|
|
reliable = reliable
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Will attempt to connect to the server, with a default 30 second connection timeout and will block until completed.
|
|
* If unable to connect within the specified timeout an exception will be thrown
|
|
*
|
|
* 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
|
|
*
|
|
* ### For the IPC (Inter-Process-Communication) it must be:
|
|
* - `connect()`
|
|
* - `connect("")`
|
|
* - `connectIpc()`
|
|
*
|
|
* ### Case does not matter, and "localhost" is the default.
|
|
*
|
|
* @param remoteAddress The network or if localhost, IPC address for the client to connect to
|
|
* @param ipcPublicationId The IPC publication address for the client to connect to
|
|
* @param ipcSubscriptionId The IPC subscription address for the client to connect to
|
|
* @param connectionTimeoutSec wait for x seconds. 0 will wait indefinitely.
|
|
* @param reliable true if we want to create a reliable connection (for UDP connections, is message loss acceptable?).
|
|
*
|
|
* @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")
|
|
private fun connect(
|
|
remoteAddress: InetAddress? = null,
|
|
remoteAddressString: String,
|
|
// Default IPC ports are flipped because they are in the perspective of the SERVER
|
|
ipcPublicationId: Int = AeronDriver.IPC_HANDSHAKE_STREAM_ID_SUB,
|
|
ipcSubscriptionId: Int = AeronDriver.IPC_HANDSHAKE_STREAM_ID_PUB,
|
|
connectionTimeoutSec: Int = 30,
|
|
reliable: Boolean = true)
|
|
{
|
|
|
|
require(connectionTimeoutSec >= 0) { "connectionTimeoutSec '$connectionTimeoutSec' is invalid. It must be >=0" }
|
|
|
|
if (isConnected) {
|
|
logger.error { "Unable to connect when already connected!" }
|
|
return
|
|
}
|
|
|
|
connection0 = null
|
|
|
|
// localhost/loopback IP might not always be 127.0.0.1 or ::1
|
|
this.remoteAddress = remoteAddress
|
|
this.remoteAddressString = remoteAddressString
|
|
|
|
// we are done with initial configuration, now initialize aeron and the general state of this endpoint
|
|
try {
|
|
initEndpointState()
|
|
} catch (e: Exception) {
|
|
logger.error(e) { "Unable to initialize the endpoint state" }
|
|
return
|
|
}
|
|
|
|
// 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 $remoteAddressString, 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 $remoteAddressString, there are no IPv6 interfaces available!" }
|
|
}
|
|
|
|
if (remoteAddress != null && remoteAddress.isAnyLocalAddress) {
|
|
require(false) { "Cannot connect to $remoteAddressString It is an invalid address!" }
|
|
}
|
|
|
|
// IPC can be enabled TWO ways!
|
|
// - config.enableIpc
|
|
// - NULL remoteAddress
|
|
// It is entirely possible that the server does not have IPC enabled!
|
|
val autoChangeToIpc =
|
|
(config.enableIpc && (remoteAddress == null || remoteAddress.isLoopbackAddress)) || (!config.enableIpc && remoteAddress == null)
|
|
|
|
val handshake = ClientHandshake(crypto, this, logger)
|
|
|
|
val handshakeTimeout = 5
|
|
val timoutInNanos = TimeUnit.SECONDS.toNanos(connectionTimeoutSec.toLong())
|
|
val startTime = System.nanoTime()
|
|
var success = false
|
|
while (timoutInNanos == 0L || System.nanoTime() - startTime < timoutInNanos) {
|
|
if (isShutdown()) {
|
|
// If we are connecting indefinitely, we have to make sure to end the connection process
|
|
val exception = ClientShutdownException("Unable to connect while shutting down")
|
|
logger.error(exception) { "Aborting connection retry attempt to server." }
|
|
listenerManager.notifyError(exception)
|
|
throw exception
|
|
}
|
|
|
|
try {
|
|
val handshakeConnection = if (autoChangeToIpc) {
|
|
buildIpcHandshake(
|
|
ipcSubscriptionId = ipcSubscriptionId,
|
|
ipcPublicationId = ipcPublicationId,
|
|
connectionTimeoutSec = handshakeTimeout,
|
|
reliable = reliable
|
|
)
|
|
} else {
|
|
buildUdpHandshake(connectionTimeoutSec = handshakeTimeout, reliable = reliable)
|
|
}
|
|
|
|
logger.info { handshakeConnection.clientInfo }
|
|
|
|
|
|
connect0(handshake, handshakeConnection, handshakeTimeout)
|
|
success = true
|
|
|
|
// once we're done with the connection process, stop trying
|
|
break
|
|
} catch (e: ClientRetryException) {
|
|
handshake.reset()
|
|
|
|
// maybe the aeron driver isn't running? (or isn't running correctly?)
|
|
aeronDriver.closeIfSingle()
|
|
aeronDriver.start()
|
|
|
|
// 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
|
|
sleep(aeronDriver.driverTimeout()+1)
|
|
if (logger.isTraceEnabled) {
|
|
logger.trace(e) { "Unable to connect to '$remoteAddressString', retrying..." }
|
|
} else {
|
|
logger.info { "Unable to connect to '$remoteAddressString', retrying..." }
|
|
}
|
|
|
|
} catch (e: Exception) {
|
|
logger.error(e) { "[${handshake.connectKey}] : Un-recoverable error during handshake. Aborting." }
|
|
handshake.reset()
|
|
|
|
listenerManager.notifyError(e)
|
|
throw e
|
|
}
|
|
}
|
|
|
|
if (!success) {
|
|
if (System.nanoTime() - startTime < timoutInNanos) {
|
|
// we timed out. Throw the appropriate exception
|
|
val exception = ClientTimedOutException("Unable to connect to the server in $connectionTimeoutSec seconds")
|
|
logger.error(exception) { "Aborting connection attempt to server." }
|
|
listenerManager.notifyError(exception)
|
|
throw exception
|
|
}
|
|
|
|
// If we did not connect - throw an error. When `client.connect()` is called, either it connects or throws an error
|
|
val exception = ClientRejectedException("The server did not respond or permit the connection attempt")
|
|
ListenerManager.cleanStackTrace(exception)
|
|
|
|
logger.error(exception) { "Aborting connection retry attempt to server." }
|
|
listenerManager.notifyError(exception)
|
|
throw exception
|
|
}
|
|
}
|
|
|
|
private fun buildIpcHandshake(ipcSubscriptionId: Int, ipcPublicationId: Int, connectionTimeoutSec: Int, reliable: Boolean): MediaDriverConnection {
|
|
if (remoteAddress == null) {
|
|
logger.info { "IPC enabled" }
|
|
} else {
|
|
logger.info { "IPC for loopback enabled and aeron is already running. Auto-changing network connection from '$remoteAddressString' -> IPC" }
|
|
}
|
|
|
|
// MAYBE the server doesn't have IPC enabled? If no, we need to connect via network instead
|
|
val ipcConnection = IpcMediaDriverConnection(
|
|
streamIdSubscription = ipcSubscriptionId,
|
|
streamId = ipcPublicationId,
|
|
sessionId = AeronDriver.RESERVED_SESSION_ID_INVALID,
|
|
)
|
|
|
|
// throws a ConnectTimedOutException if the client cannot connect for any reason to the server handshake ports
|
|
try {
|
|
ipcConnection.buildClient(aeronDriver, logger)
|
|
return ipcConnection
|
|
} catch (e: Exception) {
|
|
if (remoteAddress == null) {
|
|
// if we specified that we MUST use IPC, then we have to throw the exception, because there is no IPC
|
|
val clientException = ClientException("Unable to connect via IPC to server. No address was specified", e)
|
|
ListenerManager.cleanStackTraceInternal(clientException)
|
|
throw clientException
|
|
}
|
|
}
|
|
|
|
if (remoteAddress == null) {
|
|
// if we specified that we MUST use IPC, then we have to throw the exception, because there is no IPC
|
|
val clientException = ClientException("Unable to connect via IPC to server. No address was specified")
|
|
ListenerManager.cleanStackTraceInternal(clientException)
|
|
throw clientException
|
|
}
|
|
|
|
logger.info { "IPC for loopback enabled, but unable to connect. Retrying with address $remoteAddressString" }
|
|
|
|
// try a UDP connection instead
|
|
val udpConnection = UdpMediaDriverClientConnection(
|
|
address = remoteAddress!!,
|
|
publicationPort = config.subscriptionPort,
|
|
subscriptionPort = config.publicationPort,
|
|
streamId = AeronDriver.UDP_HANDSHAKE_STREAM_ID,
|
|
sessionId = AeronDriver.RESERVED_SESSION_ID_INVALID,
|
|
connectionTimeoutSec = connectionTimeoutSec,
|
|
isReliable = reliable
|
|
)
|
|
|
|
// throws a ConnectTimedOutException if the client cannot connect for any reason to the server handshake ports
|
|
udpConnection.buildClient(aeronDriver, logger)
|
|
return udpConnection
|
|
}
|
|
|
|
private fun buildUdpHandshake(connectionTimeoutSec: Int, reliable: Boolean): MediaDriverConnection {
|
|
val test = UdpMediaDriverClientConnection(
|
|
address = remoteAddress!!,
|
|
publicationPort = config.subscriptionPort,
|
|
subscriptionPort = config.publicationPort,
|
|
streamId = AeronDriver.UDP_HANDSHAKE_STREAM_ID,
|
|
sessionId = AeronDriver.RESERVED_SESSION_ID_INVALID,
|
|
connectionTimeoutSec = connectionTimeoutSec,
|
|
isReliable = reliable
|
|
)
|
|
|
|
// throws a ConnectTimedOutException if the client cannot connect for any reason to the server handshake ports
|
|
test.buildClient(aeronDriver, logger)
|
|
return test
|
|
}
|
|
|
|
// the handshake process might have to restart this connection process.
|
|
private fun connect0(handshake: ClientHandshake<CONNECTION>, handshakeConnection: MediaDriverConnection, connectionTimeoutSec: Int) {
|
|
// this will block until the connection timeout, and throw an exception if we were unable to connect with the server
|
|
val isUsingIPC = handshakeConnection is IpcMediaDriverConnection
|
|
|
|
|
|
// throws(ConnectTimedOutException::class, ClientRejectedException::class, ClientException::class)
|
|
val connectionInfo = handshake.hello(handshakeConnection, connectionTimeoutSec)
|
|
|
|
// VALIDATE:: check to see if the remote connection's public key has changed!
|
|
val validateRemoteAddress = if (isUsingIPC) {
|
|
PublicKeyValidationState.VALID
|
|
} else {
|
|
crypto.validateRemoteAddress(remoteAddress!!, remoteAddressString, connectionInfo.publicKey)
|
|
}
|
|
|
|
if (validateRemoteAddress == PublicKeyValidationState.INVALID) {
|
|
handshakeConnection.close()
|
|
val exception = ClientRejectedException("Connection to $remoteAddressString not allowed! Public key mismatch.")
|
|
logger.error(exception) { "Validation error" }
|
|
throw exception
|
|
}
|
|
|
|
|
|
// VALIDATE:: If the serialization DOES NOT match between the client/server, then the server will emit a log, and the
|
|
// client will timeout. SPECIFICALLY.... we do not give class serialization/registration info to the client - in case the client
|
|
// is rogue, we do not want to carelessly provide info.
|
|
|
|
|
|
// we are now connected, so we can connect to the NEW client-specific ports
|
|
val clientConnection = if (isUsingIPC) {
|
|
IpcMediaDriverConnection(
|
|
sessionId = connectionInfo.sessionId,
|
|
// NOTE: pub/sub must be switched!
|
|
streamIdSubscription = connectionInfo.publicationPort,
|
|
streamId = connectionInfo.subscriptionPort)
|
|
}
|
|
else {
|
|
UdpMediaDriverClientConnection(
|
|
address = (handshakeConnection as UdpMediaDriverClientConnection).address,
|
|
// NOTE: pub/sub must be switched!
|
|
publicationPort = connectionInfo.subscriptionPort,
|
|
subscriptionPort = connectionInfo.publicationPort,
|
|
streamId = connectionInfo.streamId,
|
|
sessionId = connectionInfo.sessionId,
|
|
connectionTimeoutSec = connectionTimeoutSec,
|
|
isReliable = handshakeConnection.isReliable)
|
|
}
|
|
|
|
// we have to construct how the connection will communicate!
|
|
clientConnection.buildClient(aeronDriver, logger)
|
|
|
|
// only the client connects to the server, so here we have to connect. The server (when creating the new "connection" object)
|
|
// does not need to do anything
|
|
//
|
|
// throws a ConnectTimedOutException if the client cannot connect for any reason to the server-assigned client ports
|
|
logger.info { clientConnection.clientInfo }
|
|
|
|
|
|
///////////////
|
|
//// RMI
|
|
///////////////
|
|
|
|
// we set up our kryo information once we connect to a server (using the server's kryo registration details)
|
|
if (!serialization.finishInit(type, connectionInfo.kryoRegistrationDetails)) {
|
|
handshakeConnection.close()
|
|
|
|
// 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)
|
|
val exception = if (isUsingIPC) {
|
|
ClientRejectedException("[${handshake.connectKey}] Connection to IPC has incorrect class registration details!!")
|
|
} else {
|
|
ClientRejectedException("[${handshake.connectKey}] Connection to $remoteAddressString has incorrect class registration details!!")
|
|
}
|
|
ListenerManager.cleanStackTraceInternal(exception)
|
|
throw exception
|
|
}
|
|
|
|
val sessionId = clientConnection.sessionId
|
|
val streamId = clientConnection.streamId
|
|
val aeronLogInfo = "$sessionId/$streamId"
|
|
|
|
val newConnection: CONNECTION
|
|
if (isUsingIPC) {
|
|
newConnection = connectionFunc(ConnectionParams(this, clientConnection, PublicKeyValidationState.VALID))
|
|
} else {
|
|
newConnection = connectionFunc(ConnectionParams(this, clientConnection, validateRemoteAddress))
|
|
remoteAddress!!
|
|
|
|
// VALIDATE are we allowed to connect to this server (now that we have the initial server information)
|
|
val permitConnection = listenerManager.notifyFilter(newConnection)
|
|
if (!permitConnection) {
|
|
handshakeConnection.close()
|
|
val exception = ClientRejectedException("[$aeronLogInfo - ${handshake.connectKey}] Connection (${newConnection.id}) to $remoteAddressString was not permitted!")
|
|
ListenerManager.cleanStackTrace(exception)
|
|
logger.error(exception) { "Permission error" }
|
|
throw exception
|
|
}
|
|
|
|
logger.info { "[$aeronLogInfo - ${handshake.connectKey}] Connection (${newConnection.id}) adding new signature for $remoteAddressString : ${connectionInfo.publicKey.toHexString()}" }
|
|
storage.addRegisteredServerKey(remoteAddress!!, connectionInfo.publicKey)
|
|
}
|
|
|
|
|
|
//////////////
|
|
/// Extra Close action
|
|
//////////////
|
|
newConnection.closeAction = { enableNotifyDisconnect ->
|
|
// this is called whenever connection.close() is called by the framework or via client.close()
|
|
|
|
// on the client, we want to GUARANTEE that the disconnect happens-before connect.
|
|
if (!lockStepForConnect.compareAndSet(null, Mutex(locked = true))) {
|
|
logger.error { "[$aeronLogInfo - ${handshake.connectKey}] Connection ${newConnection.id} : close lockStep for disconnect was in the wrong state!" }
|
|
}
|
|
|
|
isConnected = false
|
|
// this is called whenever connection.close() is called by the framework or via client.close()
|
|
|
|
// make sure to call our client.notifyDisconnect() callbacks
|
|
|
|
// this always has to be on event dispatch, otherwise we can have weird logic loops if we reconnect within a disconnect callback
|
|
actionDispatch.launch {
|
|
if (enableNotifyDisconnect) {
|
|
listenerManager.notifyDisconnect(connection)
|
|
}
|
|
lockStepForConnect.getAndSet(null)?.unlock()
|
|
}
|
|
}
|
|
|
|
// before we finish creating the connection, we initialize it (in case there needs to be logic that happens-before `onConnect` calls occur
|
|
listenerManager.notifyInit(newConnection)
|
|
|
|
connection0 = newConnection
|
|
addConnection(newConnection)
|
|
|
|
// tell the server our connection handshake is done, and the connection can now listen for data.
|
|
// also closes the handshake (will also throw connect timeout exception)
|
|
|
|
// this value matches the server, and allows for a more robust connection attempt
|
|
val successAttemptTimeout = config.connectionCloseTimeoutInSeconds * 2
|
|
try {
|
|
handshake.done(handshakeConnection, successAttemptTimeout)
|
|
} catch (e: Exception) {
|
|
logger.error(e) { "[$aeronLogInfo - ${handshake.connectKey}] Connection (${newConnection.id}) to $remoteAddressString error during handshake" }
|
|
throw e
|
|
}
|
|
|
|
isConnected = true
|
|
|
|
logger.debug { "[$aeronLogInfo - ${handshake.connectKey}] Connection (${newConnection.id}) to $remoteAddressString done with handshake." }
|
|
|
|
// this forces the current thread to WAIT until poll system has started
|
|
val mutex = Mutex(locked = true)
|
|
|
|
// have to make a new thread to listen for incoming data!
|
|
// SUBSCRIPTIONS ARE NOT THREAD SAFE! Only one thread at a time can poll them
|
|
|
|
// these have to be in two SEPARATE actionDispatch.launch commands.... otherwise...
|
|
// if something inside-of notifyConnect is blocking or suspends, then polling will never happen!
|
|
actionDispatch.launch {
|
|
try {
|
|
mutex.unlock()
|
|
} catch (ignored: Exception) {}
|
|
|
|
val pollIdleStrategy = config.pollIdleStrategy
|
|
|
|
while (!isShutdown()) {
|
|
if (!newConnection.isClosedViaAeron()) {
|
|
// Polls the AERON media driver subscription channel for incoming messages
|
|
val pollCount = newConnection.pollSubscriptions()
|
|
|
|
// 0 means we idle. >0 means reset and don't idle (because there are likely more poll events)
|
|
pollIdleStrategy.idle(pollCount)
|
|
} else {
|
|
// If the connection has either been closed, or has expired, it needs to be cleaned-up/deleted.
|
|
logger.debug { "[$aeronLogInfo] connection expired" }
|
|
|
|
// event-loop is required, because we want to run this code AFTER the current coroutine has finished. This prevents
|
|
// odd race conditions when a client is restarted. Can only be run from inside another co-routine!
|
|
actionDispatch.eventLoop {
|
|
// NOTE: We do not shutdown the client!! The client is only closed by explicitly calling `client.close()`
|
|
newConnection.close()
|
|
}
|
|
return@launch
|
|
}
|
|
}
|
|
}
|
|
|
|
actionDispatch.launch {
|
|
mutex.withLock { }
|
|
|
|
lockStepForConnect.getAndSet(null)?.withLock { }
|
|
listenerManager.notifyConnect(newConnection)
|
|
}
|
|
}
|
|
|
|
/**
|
|
* true if the remote public key changed. This can be useful if specific actions are necessary when the key has changed.
|
|
*/
|
|
val remoteKeyHasChanged: Boolean
|
|
get() = connection.hasRemoteKeyChanged()
|
|
|
|
/**
|
|
* true if this connection is an IPC connection
|
|
*/
|
|
val isIPC: Boolean
|
|
get() = connection.isIpc
|
|
|
|
/**
|
|
* @return true if this connection is a network connection
|
|
*/
|
|
val isNetwork: Boolean
|
|
get() = connection.isNetwork
|
|
|
|
/**
|
|
* @return the connection (TCP or IPC) id of this connection.
|
|
*/
|
|
val id: Int
|
|
get() = connection.id
|
|
|
|
/**
|
|
* the connection used by the client, this is only valid after the client has connected
|
|
*/
|
|
val connection: CONNECTION
|
|
get() = connection0 as CONNECTION
|
|
|
|
|
|
/**
|
|
* 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
|
|
*/
|
|
suspend fun send(message: Any): Boolean {
|
|
val c = connection0
|
|
|
|
return if (c != null) {
|
|
c.send(message)
|
|
} else {
|
|
val exception = ClientException("Cannot send a message when there is no connection!")
|
|
logger.error(exception) { "No connection!" }
|
|
false
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 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
|
|
*/
|
|
fun sendBlocking(message: Any): Boolean {
|
|
return runBlocking {
|
|
send(message)
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Sends a "ping" packet to measure **ROUND TRIP** time to the remote connection.
|
|
*
|
|
* @param function called when the ping returns (ie: update time/latency counters/metrics/etc)
|
|
*
|
|
* @return true if the ping was successfully sent to the client
|
|
*/
|
|
suspend fun ping(pingTimeoutSeconds: Int = PingManager.DEFAULT_TIMEOUT_SECONDS, function: suspend Ping.() -> Unit): Boolean {
|
|
val c = connection0
|
|
|
|
if (c != null) {
|
|
return pingManager.ping(c, pingTimeoutSeconds, actionDispatch, responseManager, logger, function)
|
|
} else {
|
|
logger.error(ClientException("Cannot send a ping when there is no connection!")) { "No connection!" }
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
/**
|
|
* Sends a "ping" packet to measure **ROUND TRIP** time to the remote connection.
|
|
*
|
|
* @param function called when the ping returns (ie: update time/latency counters/metrics/etc)
|
|
*/
|
|
fun pingBlocking(pingTimeoutSeconds: Int = PingManager.DEFAULT_TIMEOUT_SECONDS, function: suspend Ping.() -> Unit): Boolean {
|
|
return runBlocking {
|
|
ping(pingTimeoutSeconds, function)
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Removes the specified host address from the list of registered server keys.
|
|
*/
|
|
fun removeRegisteredServerKey(address: InetAddress) {
|
|
val savedPublicKey = storage.getRegisteredServerKey(address)
|
|
if (savedPublicKey != null) {
|
|
logger.debug { "Deleting remote IP address key $address" }
|
|
storage.removeRegisteredServerKey(address)
|
|
}
|
|
}
|
|
|
|
final override fun close0() {
|
|
// no impl
|
|
}
|
|
}
|