network transport with remote address, universal server

This commit is contained in:
sergeych committed 2023-11-15 02:25:57 +03:00
1 parent 96edbb2040
commit f92431a281
8 files changed
+39 -33

No files matched your search

@@ -1,12 +1,10 @@
package net.sergeych.kiloparsec.adapter
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.channels.SendChannel
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.suspendCancellableCoroutine
import kotlinx.coroutines.withContext
import net.sergeych.kiloparsec.Transport
import java.net.InetAddress
import java.net.InetSocketAddress
import java.nio.channels.AsynchronousServerSocketChannel
@@ -16,7 +14,7 @@ import kotlin.coroutines.suspendCoroutine
actual fun NetworkAddress(host: String, port: Int): NetworkAddress =
JvmNetworkAddress(InetAddress.getByName(host), port)
actual fun acceptTcpDevice(port: Int): Flow<Transport.Device> {
actual fun acceptTcpDevice(port: Int): Flow<InetTransportDevice> {
return flow {
val socket = withContext(Dispatchers.IO) {
AsynchronousServerSocketChannel.open().also {
@@ -35,7 +33,7 @@ actual fun acceptTcpDevice(port: Int): Flow<Transport.Device> {
}
}
actual suspend fun connectTcpDevice(address: NetworkAddress): Transport.Device {
actual suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice {
address as JvmNetworkAddress
val socket = withContext(Dispatchers.IO) {
AsynchronousSocketChannel.open()
@@ -45,7 +43,3 @@ actual suspend fun connectTcpDevice(address: NetworkAddress): Transport.Device {
}
return asyncSocketToDevice(socket)
}
suspend fun SendChannel<UByte>.sendAll(bytes: Collection<UByte>) {
for (b in bytes) send(b)
}
@@ -31,8 +31,8 @@ private fun encode(block: UByteArray): ByteArray
* to the socket in a global scope.These are closed when transport is closed
* or the socket is closed, for example, by network failure.
*/
suspend fun asyncSocketToDevice(socket: AsynchronousSocketChannel): InetProxyDevice {
val deferredDevice = CompletableDeferred<InetProxyDevice>()
suspend fun asyncSocketToDevice(socket: AsynchronousSocketChannel): InetTransportDevice {
val deferredDevice = CompletableDeferred<InetTransportDevice>()
globalLaunch {
coroutineScope {
fun stop() {
@@ -109,7 +109,7 @@ suspend fun asyncSocketToDevice(socket: AsynchronousSocketChannel): InetProxyDev
// SocketAddress.
val addr = socket.remoteAddress as InetSocketAddress
deferredDevice.complete(
InetProxyDevice(inputBlocks, outputBlocks, JvmNetworkAddress(addr.address,addr.port)) { stop() }
InetTransportDevice(inputBlocks, outputBlocks, JvmNetworkAddress(addr.address,addr.port)) { stop() }
)
}
globalLaunch { socket.close() }