SergeychWorks/kiloparsec
1
0 Fork 0
Code Issues 2 Pull Requests Packages Projects Releases 7 Wiki Activity

0.3.2: UDP timeout support, fix #4

Browse Source
This commit is contained in:
sergeych committed 2024-08-11 15:11:32 +02:00
1 parent 4d178d951f
commit 40b8723132
12 files changed
+130 -37
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/TcpProvider.kt
+1 -1
View File
@@ -27,7 +27,7 @@ private val logCounter = AtomicCounter(0)
class ProtocolException(text: String, cause: Throwable? = null) : RuntimeException(text, cause)
const val MAX_TCP_BLOCK_SIZE = 16776216
val PING_INACTIVITY_TIME = 30.seconds
internal val PING_INACTIVITY_TIME = 30.seconds
/**
* Listen for incoming TCP/IP connections on all local interfaces and the specified [port]
src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpProvider.kt
+44 -17
View File
@@ -10,6 +10,8 @@ import net.sergeych.kiloparsec.KiloServer
import net.sergeych.kiloparsec.RemoteInterface
import net.sergeych.mp_tools.globalLaunch
import net.sergeych.tools.AtomicCounter
import kotlin.time.Duration
import kotlin.time.Duration.Companion.minutes
internal val udpCounter = AtomicCounter(0)
@@ -55,9 +57,40 @@ class UdpTransportException(override val message: String) : RemoteInterface.Inva
* See [connectUdpDevice] for the client sample.
*
* When it is necessary to stop listening to some port, use [UdpServer] instead.
*
* @param port port to listen
* @param localInterface string form local interface to listen
* @param maxInactivityTimeout maximum silence time after which the connection is supposed to be lost.
* the module automatically issues pings on inactivity when there is no data often enough
* to maintain the connection open.
*/
fun acceptUdpDevice(port: Int, localInterface: String = "0.0.0.0"): Flow<InetTransportDevice> =
UdpServer(port, localInterface).transportFlow
fun acceptUdpDevice(
port: Int,
localInterface: String = "0.0.0.0",
maxInactivityTimeout: Duration = 2.minutes,
): Flow<InetTransportDevice> =
UdpServer(port, localInterface,maxInactivityTimeout).transportFlow
/**
* Connect to UDP server (see [acceptUdpDevice] or [UdpServer]) and return a [InetTransportDevice] for it. It
* should be used with [KiloClient] as connection provider:
* ```kotlin
* val client = KiloClient<Unit>() {
* connect { connectUdpDevice("localhost:$port") }
* }
* // now we can execute remote commands:
* assertEquals("start", client.call(cmdLoad))
* ```
*
* @param hostPort "host:port" string address of the remote UDP port to connect to
* @param maxInactivityTimeout maximum silence time after which the connection is supposed to be lost.
* the module automatically issues pings on inactivity when there is no data often enough
* to maintain the connection open.
*/
fun connectUdpDevice(
hostPort: String,
maxInactivityTimeout: Duration = 2.minutes,
) = connectUdpDevice(hostPort.toNetworkAddress(), maxInactivityTimeout)
/**
* Connect to UDP server (see [acceptUdpDevice]) and return a [InetTransportDevice] for it. It
@@ -69,21 +102,15 @@ fun acceptUdpDevice(port: Int, localInterface: String = "0.0.0.0"): Flow<InetTra
* // now we can execute remote commands:
* assertEquals("start", client.call(cmdLoad))
* ```
* @param addr the network address where to connect to
* @param maxInactivityTimeout maximum silence time after which the connection is supposed to be lost.
* the module automatically issues pings on inactivity when there is no data often enough
* to maintain the connection open.
*/
fun connectUdpDevice(hostPort: String) = connectUdpDevice(hostPort.toNetworkAddress())
/**
* Connect to UDP server (see [acceptUdpDevice]) and return a [InetTransportDevice] for it. It
* should be used with [KiloClient] as connection provider:
* ```kotlin
* val client = KiloClient<Unit>() {
* connect { connectUdpDevice("localhost:$port") }
* }
* // now we can execute remote commands:
* assertEquals("start", client.call(cmdLoad))
* ```
*/
fun connectUdpDevice(addr: NetworkAddress): InetTransportDevice {
fun connectUdpDevice(
addr: NetworkAddress,
maxInactivityTimeout: Duration = 2.minutes,
): InetTransportDevice {
val selectorManager = SelectorManager(Dispatchers.IO)
val remoteAddress = InetSocketAddress(addr.host, addr.port)
val socket = aSocket(selectorManager).udp().connect(remoteAddress)
@@ -99,7 +126,7 @@ fun connectUdpDevice(addr: NetworkAddress): InetTransportDevice {
done.complete(Unit)
}
}, remoteAddress, false)
}, remoteAddress, false, maxInactivityTimeout)
globalLaunch {
launch {
src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpServer.kt
+17 -9
View File
@@ -14,6 +14,8 @@ import net.sergeych.mp_logger.LogTag
import net.sergeych.mp_logger.Loggable
import net.sergeych.mp_logger.debug
import net.sergeych.mp_logger.exception
import kotlin.time.Duration
import kotlin.time.Duration.Companion.minutes
/**
* UDP server for kiloparsec. Unlike [acceptUdpDevice], it allow stopping listening
@@ -34,8 +36,15 @@ import net.sergeych.mp_logger.exception
* ```
*
* See [acceptUdpDevice] for more information.
*
* @param port port to listen to
* @param localInterface string form of local interface to listen to
* @param maxInactivityTimeout maximum silence time after which the connection is supposed to be lost.
* the module automatically issues pings on inactivity when there is no data often enough
* to maintain the connection open.
*/
class UdpServer(val port: Int,localInterface: String = "0.0.0.0") :
class UdpServer(val port: Int, localInterface: String = "0.0.0.0", maxInactivityTimeout: Duration = 2.minutes) :
Loggable by LogTag("UDPS${udpCounter.incrementAndGet()}"), UdpConnector {
private val sessions = mutableMapOf<SocketAddress, UdpSocketTransport>()
@@ -54,7 +63,7 @@ class UdpServer(val port: Int,localInterface: String = "0.0.0.0") :
*/
val transportFlow by lazy {
flow {
while(true) {
while (true) {
try {
val datagram = serverSocket.receive()
val block = UdpBlock.decode(datagram)
@@ -68,18 +77,17 @@ class UdpServer(val port: Int,localInterface: String = "0.0.0.0") :
sessions.getOrPut(remoteAddress) {
// new connection: create transport
debug { "Creating new connection to $remoteAddress" }
UdpSocketTransport(this@UdpServer, remoteAddress, true)
UdpSocketTransport(this@UdpServer, remoteAddress, true, maxInactivityTimeout)
// and emit it:
.also { emit(it.transportDevice) }
}.processIncoming(block)
}
}
}
catch(_: CancellationException) { break }
catch(_: ClosedReceiveChannelException) {
} catch (_: CancellationException) {
break
}
catch(e: Exception) {
} catch (_: ClosedReceiveChannelException) {
break
} catch (e: Exception) {
exception { "unexpected exception in incoming datagram processing" to e }
close()
break
@@ -109,7 +117,7 @@ class UdpServer(val port: Int,localInterface: String = "0.0.0.0") :
runCatching { serverSocket.close() }
}
}
while(sessions.isNotEmpty()) {
while (sessions.isNotEmpty()) {
runCatching {
access.withLock { sessions.values.firstOrNull() }
?.close()
src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpSocketTransport.kt
+15 -7
View File
@@ -16,14 +16,19 @@ import net.sergeych.mp_logger.Loggable
import net.sergeych.mp_logger.debug
import net.sergeych.mp_logger.exception
import net.sergeych.mp_tools.globalLaunch
import kotlin.time.Duration.Companion.seconds
import kotlin.time.Duration
/**
* This is a common part of UDP transport shared between client and server connections.
* It should not be used directly but bu the [UdpServer], [acceptUdpDevice] and [connectUdpDevice]
* respectively.
*/
internal class UdpSocketTransport(private val server: UdpConnector, val socketAddress: SocketAddress, val isServer: Boolean) :
internal class UdpSocketTransport(
private val server: UdpConnector,
val socketAddress: SocketAddress,
val isServer: Boolean,
val maxInactivityTimeout: Duration
) :
Loggable {
// IMPORTANT! Log stuff must be the first (or you shot your leg):
@@ -34,12 +39,10 @@ internal class UdpSocketTransport(private val server: UdpConnector, val socketAd
// Pinger params: keep them first!
private var lastSendAt = Clock.System.now()
private var lastReceived = Clock.System.now()
private val pingTimeout = 30.seconds
private val pingTimeout = maxInactivityTimeout / 3
private val pingSleep = pingTimeout / 3
private val pingMinTimeout = pingTimeout / 2
private val pingMinTimeout = pingTimeout * 2 / 3
// TODO: break on inactivity
private val inactivityBreakTimeout = 30.seconds
val inputDataBlocks = Channel<UByteArray>(256, onBufferOverflow = BufferOverflow.DROP_OLDEST)
val outputDataBlocks = Channel<UByteArray>(256, onBufferOverflow = BufferOverflow.DROP_OLDEST)
@@ -142,7 +145,12 @@ internal class UdpSocketTransport(private val server: UdpConnector, val socketAd
private suspend fun pinger() {
while (!isClosed) {
delay(pingSleep)
if (Clock.System.now() - lastSendAt >= pingTimeout) {
val inactivity = Clock.System.now() - lastSendAt
if( inactivity > maxInactivityTimeout) {
debug { "inactivity timout: closing the connection" }
close()
}
if (inactivity >= pingTimeout) {
debug { "pinger sends a ping on timeout" }
send(UdpBlock.Ping)
}
src/ktorSocketTest/kotlin/InternetTest.kt → src/ktorSocketTest/kotlin/net/sergeych/kiloparsec/adapter/InternetTest.kt
+3 -1
View File
@@ -1,7 +1,9 @@
package net.sergeych.kiloparsec.adapter
import assertThrows
import kotlinx.coroutines.test.runTest
import net.sergeych.crypto2.initCrypto
import net.sergeych.kiloparsec.*
import net.sergeych.kiloparsec.adapter.*
import net.sergeych.mp_logger.Log
import kotlin.random.Random
import kotlin.test.Test