dorkbox/Network
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Cleaned up duplicate aeron pub/sub logs

Browse Source
This commit is contained in:
Robinson 2023-02-15 23:28:45 +01:00
parent c57f0564db
commit 024ff53e20
No known key found for this signature in database
GPG Key ID: 8E7DB78588BD6F5C
1 changed files with 14 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
src/dorkbox/network/handshake/ServerHandshakePollers.kt
View File

@ -14,6 +14,7 @@ import dorkbox.network.aeron.mediaDriver.controlEndpoint
import dorkbox.network.connection.Connection
import dorkbox.network.connection.ConnectionParams
import dorkbox.network.connection.EndPoint
import io.aeron.CommonContext
import io.aeron.FragmentAssembler
import io.aeron.Image
import io.aeron.logbuffer.Header
@ -54,10 +55,9 @@ internal object ServerHandshakePollers {
} else {
// we create a NEW publication for the handshake, which connects directly to the client handshake subscription
val publicationUri = uri("ipc", message.sessionId)
logger.trace { "Server IPC connection pub ${publicationUri.build()},stream-id=${message.streamId}" }
val publication = try {
aeronDriver.addExclusivePublication(publicationUri, message.streamId)
aeronDriver.addExclusivePublication(publicationUri, "IPC", message.streamId)
} catch (e: Exception) {
logger.error(e) { "Cannot create IPC publication back to remote" }
return
@ -136,12 +136,16 @@ internal object ServerHandshakePollers {
}
val isRemoteIpv4 = clientAddress is Inet4Address
if (!isRemoteIpv4) {
val type: String
if (isRemoteIpv4) {
type = "IPv4"
} else {
// this is necessary to clean up the address when adding it to aeron, since different formats mess it up
clientAddressString = IP.toString(clientAddress)
type = "IPv6"
}
// NOTE: publications are REMOVED from Aeron clients when their linger timeout has expired!!!
// if we are listening on :: (ipv6), and a connection via ipv4 arrives, aeron MUST publish on the IPv4 version
@ -149,18 +153,19 @@ internal object ServerHandshakePollers {
// we create a NEW publication for the handshake, which connects directly to the client handshake subscription CONTROL (which then goes to the proper endpoint)
val publicationUri = uri("udp", message.sessionId, isReliable)
publicationUri.controlEndpoint(isRemoteIpv4, properPubAddress, port)
logger.trace { "Server connection pub $publicationUri,stream-id=${message.streamId}" }
.controlEndpoint(isRemoteIpv4, properPubAddress, port)
.controlMode(CommonContext.MDC_CONTROL_MODE_DYNAMIC)
val publication = try {
aeronDriver.addExclusivePublication(publicationUri, message.streamId)
aeronDriver.addExclusivePublication(publicationUri, type, message.streamId)
} catch (e: Exception) {
logger.error(e) { "Cannot create publication back to $clientAddressString" }
return
}
// we actually have to wait for it to connect before we continue
val startTime = System.nanoTime()