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

Added guarantee of aeron publication log deletion (warning if it cannot delete the file)

Browse Source
This commit is contained in:
nathan 2020-09-01 20:36:24 +02:00
parent 3ade7f229e
commit e09fd43e37
1 changed files with 19 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
src/dorkbox/network/connection/Connection.kt
View File

@ -32,6 +32,7 @@ import kotlinx.coroutines.delay
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking import kotlinx.coroutines.runBlocking
import org.agrona.DirectBuffer import org.agrona.DirectBuffer
import java.io.IOException
import java.util.concurrent.TimeUnit import java.util.concurrent.TimeUnit
/** /**
@ -188,7 +189,7 @@ open class Connection(connectionParameters: ConnectionParams<*>) {
internal fun pollSubscriptions(): Int { internal fun pollSubscriptions(): Int {
// NOTE: regarding fragment limit size. Repeated calls to '.poll' will reassemble a fragment. // NOTE: regarding fragment limit size. Repeated calls to '.poll' will reassemble a fragment.
// `.poll(handler, 4)` == `.poll(handler, 2)` + `.poll(handler, 2)` // `.poll(handler, 4)` == `.poll(handler, 2)` + `.poll(handler, 2)`
return subscription.poll(messageHandler, 2) return subscription.poll(messageHandler, 1)
} }
/** /**
@ -303,7 +304,8 @@ open class Connection(connectionParameters: ConnectionParams<*>) {
subscription.close() subscription.close()
val closeTimeoutTime = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(endPoint.config.connectionCloseTimeoutInSeconds.toLong()) val timeOut = TimeUnit.SECONDS.toMillis(endPoint.config.connectionCloseTimeoutInSeconds.toLong())
var closeTimeoutTime = System.currentTimeMillis() + timeOut
// we do not want to close until AFTER all publications have been sent. Calling this WITHOUT waiting will instantly stop everything // we do not want to close until AFTER all publications have been sent. Calling this WITHOUT waiting will instantly stop everything
// we want a timeout-check, otherwise this will run forever // we want a timeout-check, otherwise this will run forever
@ -311,8 +313,23 @@ open class Connection(connectionParameters: ConnectionParams<*>) {
delay(100) delay(100)
} }
// on close, we want to make sure this file is DELETED!
val logFile = endPoint.mediaDriverContext!!.aeronDirectory().resolve("publications").resolve("${publication.registrationId()}.logbuffer")
publication.close() publication.close()
closeTimeoutTime = System.currentTimeMillis() + timeOut
while (logFile.exists() && System.currentTimeMillis() < closeTimeoutTime) {
if (logFile.delete()) {
break
}
}
if (logFile.exists()) {
listenerManager.value?.notifyError(this, IOException("Unable to delete aeron publication log on close: $logFile"))
}
rmiConnectionSupport.clearProxyObjects() rmiConnectionSupport.clearProxyObjects()
// this always has to be on a new dispatch, otherwise we can have weird logic loops if we reconnect within a disconnect callback // this always has to be on a new dispatch, otherwise we can have weird logic loops if we reconnect within a disconnect callback