Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 sergeych
|
515278c264 | ||
|
sergeych
|
7100fa5f76 | ||
|
sergeych
|
77e293bef2 | ||
|
sergeych
|
d6f257de14 | ||
|
sergeych
|
8a21a836e5 | ||
|
sergeych
|
439e229294 | ||
|
sergeych
|
ffcdcf7350 | ||
|
sergeych
|
26d1f3522f | ||
|
sergeych
|
326b92142d |
No files matched your search
+1
-2
@@ -23,8 +23,7 @@ out/
|
|||||||
.project
|
.project
|
||||||
.settings
|
.settings
|
||||||
.springBeans
|
.springBeans
|
||||||
.sts4-cache
|
.sts4-caches
|
||||||
bin/
|
|
||||||
!**/src/main/**/bin/
|
!**/src/main/**/bin/
|
||||||
!**/src/test/**/bin/
|
!**/src/test/**/bin/
|
||||||
|
|
||||||
|
|||||||
Generated
+8
@@ -0,0 +1,8 @@
|
|||||||
|
<component name="ArtifactManager">
|
||||||
|
<artifact type="jar" name="kiloparsec-js-0.2.4">
|
||||||
|
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||||
|
<root id="archive" name="kiloparsec-js-0.2.4.jar">
|
||||||
|
<element id="module-output" name="kiloparsec.jsMain" />
|
||||||
|
</root>
|
||||||
|
</artifact>
|
||||||
|
</component>
|
||||||
Generated
+8
@@ -0,0 +1,8 @@
|
|||||||
|
<component name="ArtifactManager">
|
||||||
|
<artifact type="jar" name="kiloparsec-js-0.2.5">
|
||||||
|
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||||
|
<root id="archive" name="kiloparsec-js-0.2.5.jar">
|
||||||
|
<element id="module-output" name="kiloparsec.jsMain" />
|
||||||
|
</root>
|
||||||
|
</artifact>
|
||||||
|
</component>
|
||||||
@@ -0,0 +1,8 @@
|
|||||||
|
<component name="ArtifactManager">
|
||||||
|
<artifact type="jar" name="kiloparsec-js-0.2.5-SNAPSHOT">
|
||||||
|
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||||
|
<root id="archive" name="kiloparsec-js-0.2.5-SNAPSHOT.jar">
|
||||||
|
<element id="module-output" name="kiloparsec.jsMain" />
|
||||||
|
</root>
|
||||||
|
</artifact>
|
||||||
|
</component>
|
||||||
Generated
+6
@@ -0,0 +1,6 @@
|
|||||||
|
<component name="ArtifactManager">
|
||||||
|
<artifact type="jar" name="kiloparsec-js-0.2.6">
|
||||||
|
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||||
|
<root id="archive" name="kiloparsec-js-0.2.6.jar" />
|
||||||
|
</artifact>
|
||||||
|
</component>
|
||||||
@@ -0,0 +1,8 @@
|
|||||||
|
<component name="ArtifactManager">
|
||||||
|
<artifact type="jar" name="kiloparsec-js-0.3.1-SNAPSHOT">
|
||||||
|
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||||
|
<root id="archive" name="kiloparsec-js-0.3.1-SNAPSHOT.jar">
|
||||||
|
<element id="module-output" name="kiloparsec.jsMain" />
|
||||||
|
</root>
|
||||||
|
</artifact>
|
||||||
|
</component>
|
||||||
Generated
+8
@@ -0,0 +1,8 @@
|
|||||||
|
<component name="ArtifactManager">
|
||||||
|
<artifact type="jar" name="kiloparsec-jvm-0.2.4">
|
||||||
|
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||||
|
<root id="archive" name="kiloparsec-jvm-0.2.4.jar">
|
||||||
|
<element id="module-output" name="kiloparsec.jvmMain" />
|
||||||
|
</root>
|
||||||
|
</artifact>
|
||||||
|
</component>
|
||||||
Generated
+8
@@ -0,0 +1,8 @@
|
|||||||
|
<component name="ArtifactManager">
|
||||||
|
<artifact type="jar" name="kiloparsec-jvm-0.2.5">
|
||||||
|
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||||
|
<root id="archive" name="kiloparsec-jvm-0.2.5.jar">
|
||||||
|
<element id="module-output" name="kiloparsec.jvmMain" />
|
||||||
|
</root>
|
||||||
|
</artifact>
|
||||||
|
</component>
|
||||||
@@ -0,0 +1,8 @@
|
|||||||
|
<component name="ArtifactManager">
|
||||||
|
<artifact type="jar" name="kiloparsec-jvm-0.2.5-SNAPSHOT">
|
||||||
|
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||||
|
<root id="archive" name="kiloparsec-jvm-0.2.5-SNAPSHOT.jar">
|
||||||
|
<element id="module-output" name="kiloparsec.jvmMain" />
|
||||||
|
</root>
|
||||||
|
</artifact>
|
||||||
|
</component>
|
||||||
Generated
+6
@@ -0,0 +1,6 @@
|
|||||||
|
<component name="ArtifactManager">
|
||||||
|
<artifact type="jar" name="kiloparsec-jvm-0.2.6">
|
||||||
|
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||||
|
<root id="archive" name="kiloparsec-jvm-0.2.6.jar" />
|
||||||
|
</artifact>
|
||||||
|
</component>
|
||||||
@@ -0,0 +1,8 @@
|
|||||||
|
<component name="ArtifactManager">
|
||||||
|
<artifact type="jar" name="kiloparsec-jvm-0.3.1-SNAPSHOT">
|
||||||
|
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||||
|
<root id="archive" name="kiloparsec-jvm-0.3.1-SNAPSHOT.jar">
|
||||||
|
<element id="module-output" name="kiloparsec.jvmMain" />
|
||||||
|
</root>
|
||||||
|
</artifact>
|
||||||
|
</component>
|
||||||
Generated
-1
@@ -1,4 +1,3 @@
|
|||||||
<?xml version="1.0" encoding="UTF-8"?>
|
|
||||||
<project version="4">
|
<project version="4">
|
||||||
<component name="ExternalStorageConfigurationManager" enabled="true" />
|
<component name="ExternalStorageConfigurationManager" enabled="true" />
|
||||||
<component name="FrameworkDetectionExcludesConfiguration">
|
<component name="FrameworkDetectionExcludesConfiguration">
|
||||||
|
|||||||
Generated
+6
@@ -0,0 +1,6 @@
|
|||||||
|
<?xml version="1.0" encoding="UTF-8"?>
|
||||||
|
<project version="4">
|
||||||
|
<component name="ScalaCompilerConfiguration">
|
||||||
|
<option name="separateProdTestSources" value="false" />
|
||||||
|
</component>
|
||||||
|
</project>
|
||||||
@@ -1,35 +1,52 @@
|
|||||||
# Kiloparsec
|
# Kiloparsec
|
||||||
|
|
||||||
The new generation of __PARanoid SECurity__ protocol, advanced, faster, more secure. It also allows connecting any "block device" transport to the same local interface. Out if the box it
|
__Recommended version is `0.3.1`: to keep the code compatible with current and further versions we
|
||||||
|
ask to upgrade to `0.3.1` at least.__
|
||||||
|
|
||||||
|
The new generation of __PARanoid SECurity__ protocol, advanced, faster, more secure. It also allows connecting any "
|
||||||
|
block device" transport to the same local interface. Out if the box it
|
||||||
provides the following transports:
|
provides the following transports:
|
||||||
|
|
||||||
| name | JVM | JS | native |
|
| name | JVM | JS | native |
|
||||||
|----------------|-----|----|--------|
|
|----------------|-----|----|----------|
|
||||||
| TCP/IP server | * | | |
|
| TCP/IP server | ✓ | | >= 0.2.6 |
|
||||||
| TCP/IP client | * | | |
|
| TCP/IP client | ✓ | | >= 0.2.6 |
|
||||||
| Websock server | * | | |
|
| Websock server | ✓ | | |
|
||||||
| Websock client | * | * | * |
|
| Websock client | ✓ | ✓ | ✓ |
|
||||||
|
|
||||||
At the moment we're working on supporting TCP/IP on most native targets. This feature is planned to rach public beta in August and production in early september 2024.
|
### Note on version compatibility
|
||||||
|
|
||||||
|
Protocols >= 0.3.0 are not binary compatible with previous version due to more compact binary
|
||||||
|
format. The format from 0.3.0 onwards is supposed to keep compatible.
|
||||||
|
|
||||||
|
### Supported native targets
|
||||||
|
|
||||||
|
- iosArm64, iosX64
|
||||||
|
- macosArm64, macosArm64
|
||||||
|
- linxArm64, linuxX64
|
||||||
|
|
||||||
|
### Non-native targets
|
||||||
|
|
||||||
|
- JS (browser and nodeJS)
|
||||||
|
- JVM (android, macos, windows, linx, everywhere where JRE is installed)
|
||||||
|
|
||||||
## TCP/IP transport
|
## TCP/IP transport
|
||||||
|
|
||||||
It is the fastest. JVM implementation uses nio2 async sockets and optimizes TCP socket to play
|
It is the fastest based on async socket implementation of ktor client. It works everywhere but JS target as
|
||||||
well with blocks (smart NO_DELAY mode). It is multiplatform, nut lacks of async TCP/IP support
|
there is currently no widely adopted sockets for browser javascript.
|
||||||
on natvic targetm this is where I need help having little time. I'd prefer to use something asyn like UV on native targets.
|
|
||||||
|
|
||||||
I know no existing way to implement it in KotlinJS for the modern browsers.
|
|
||||||
|
|
||||||
## Websock server
|
## Websock server
|
||||||
|
|
||||||
While it is much slower than pure TCP, it is still faster than any http-based transport. It uses binary frames based on the Ktor server framework to easily integrate with web services. We recommend using it instead of a classic HTTP API as it beats it in terms of speed and server load even with HTTP/2.
|
While it is much slower than pure TCP, it is still faster than any http-based transport. It uses binary frames based on
|
||||||
|
the Ktor server framework to easily integrate with web services. We recommend using it instead of a classic HTTP API as
|
||||||
|
it beats it in terms of speed and server load even with HTTP/2.
|
||||||
|
|
||||||
We recommend to create the `KiloInterface<S>` instance and connect it to the websock and tcp servers in real applications to get easy access from anywhere.
|
We recommend to create the `KiloInterface<S>` instance and connect it to the websock and tcp servers in real
|
||||||
|
applications to get easy access from anywhere.
|
||||||
|
|
||||||
# Usage
|
# Usage
|
||||||
|
|
||||||
Th elibrary should be used as maven dependency, not as source.
|
The library should be used as maven dependency, not as source.
|
||||||
|
|
||||||
## Adding dependency
|
## Adding dependency
|
||||||
|
|
||||||
@@ -50,7 +67,7 @@ It could be, depending on your project structure, something like:
|
|||||||
```kotlin
|
```kotlin
|
||||||
val commonMain by getting {
|
val commonMain by getting {
|
||||||
dependencies {
|
dependencies {
|
||||||
api("net.sergeych:kiloparsec:0.2.4")
|
api("net.sergeych:kiloparsec:0.3.1")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
```
|
```
|
||||||
@@ -64,16 +81,16 @@ and functions available, like:
|
|||||||
// Api.kt
|
// Api.kt
|
||||||
|
|
||||||
@Serializable
|
@Serializable
|
||||||
class FooArgs(val text: String,val number: Int = 42)
|
class FooArgs(val text: String, val number: Int = 42)
|
||||||
|
|
||||||
// Server-side interface
|
// Server-side interface
|
||||||
val cmdSetFoo by command<FooArgs,Unit>()
|
val cmdSetFoo by command<FooArgs, Unit>()
|
||||||
val cmdGetFoo by command<Unit,FooArgs>()
|
val cmdGetFoo by command<Unit, FooArgs>()
|
||||||
val cmdPing by command<String,String>()
|
val cmdPing by command<String, String>()
|
||||||
val cmdCheckConnected by command<Unit,Boolean>()
|
val cmdCheckConnected by command<Unit, Boolean>()
|
||||||
|
|
||||||
// client-side interface (called from the server)
|
// client-side interface (called from the server)
|
||||||
val cmdPushClient by command<String,Unit>()
|
val cmdPushClient by command<String, Unit>()
|
||||||
```
|
```
|
||||||
|
|
||||||
## Call it from the client:
|
## Call it from the client:
|
||||||
@@ -93,7 +110,7 @@ val client = websocketClient<Unit>("wss://your.host.com/kp") {
|
|||||||
// If we want to collect connected state changes (this is optional)
|
// If we want to collect connected state changes (this is optional)
|
||||||
launch {
|
launch {
|
||||||
client.connectedStateFlow.collect {
|
client.connectedStateFlow.collect {
|
||||||
if( it )
|
if (it)
|
||||||
println("I am connected")
|
println("I am connected")
|
||||||
else
|
else
|
||||||
println("trying to connect...")
|
println("trying to connect...")
|
||||||
@@ -109,17 +126,17 @@ assertEquals(FooArgs("bar", 117), client.call(cmdGetFoo))
|
|||||||
## Create ktor-based server
|
## Create ktor-based server
|
||||||
|
|
||||||
Normally server side needs some session. It is convenient and avoid sending repeating data on each request speeding up
|
Normally server side needs some session. It is convenient and avoid sending repeating data on each request speeding up
|
||||||
the protocol. With KILOPARSEC it is rather basic operation:\
|
the protocol. With KILOPARSEC it is rather basic operation:
|
||||||
|
|
||||||
~~~kotlin
|
~~~kotlin
|
||||||
// Our session just keeps Foo for cmd{Get|Set}Foo:
|
// Our session just keeps Foo for cmd{Get|Set}Foo:
|
||||||
data class Session(var fooState: FooArgs?=null)
|
data class Session(var fooState: FooArgs? = null)
|
||||||
|
|
||||||
// Let's now provide interface we export, it will be used on each connection automatically:
|
// Let's now provide interface we export, it will be used on each connection automatically:
|
||||||
|
|
||||||
// Note server interface uses Session:
|
// Note server interface uses Session:
|
||||||
val serverInterface = KiloInterface<Session>().apply {
|
val serverInterface = KiloInterface<Session>().apply {
|
||||||
onConnected {
|
onConnected {
|
||||||
// Do some initialization
|
// Do some initialization
|
||||||
session.fooState = null
|
session.fooState = null
|
||||||
}
|
}
|
||||||
@@ -134,8 +151,48 @@ val ns: NettyApplicationEngine = embeddedServer(Netty, port = 8080, host = "0.0.
|
|||||||
setupWebsocketServer(serverInterface) { Session() }
|
setupWebsocketServer(serverInterface) { Session() }
|
||||||
}).start(wait = false)
|
}).start(wait = false)
|
||||||
|
|
||||||
|
|
||||||
~~~
|
~~~
|
||||||
|
|
||||||
|
## Create TCP/IP client and server
|
||||||
|
|
||||||
|
Using plain TCP/IP is even simpler, and it works way faster than websocket one, and is _the same
|
||||||
|
protected as `wss://` (and `ws://`) variant above due to same kiloparsec encryption in both cases. Still, a TCP/IP
|
||||||
|
client is not available in Javascript browser targets and custom TCP ports could often be blocked by firewalls.
|
||||||
|
|
||||||
|
Documentation is available in samples here:
|
||||||
|
|
||||||
|
- [TCP/IP server creation](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec/-kilo-server/index.html)
|
||||||
|
|
||||||
|
- [TCP/IP client](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec/-kilo-client/index.html)
|
||||||
|
|
||||||
|
In short, there are two functions that implements aysnchronous TCP/IP transport on all platforms buy JS:
|
||||||
|
|
||||||
|
- [acceptTcpDevice](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec.adapter/accept-tcp-device.html?query=fun%20acceptTcpDevice(port:%20Int):%20Flow%3CInetTransportDevice%3E) to create a server
|
||||||
|
|
||||||
|
- [connectTcpDevice](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec.adapter/connect-tcp-device.html) to connect to the server
|
||||||
|
|
||||||
|
## Reusing code between servers
|
||||||
|
|
||||||
|
The same instance of the [KiloInterface](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec/-kilo-interface/index.html?query=open%20class%20KiloInterface%3CS%3E%20:%20LocalInterface%3CKiloScope%3CS%3E%3E) could easily be reused with all instances of servers with different protocols.
|
||||||
|
|
||||||
|
This is a common proactive to create a business logic in a `KiloInterface`, then create a TCP/IP and Websocket servers passing the same instance of the logic to both.
|
||||||
|
|
||||||
|
## Note on the server identification
|
||||||
|
|
||||||
|
We do not recommend to rely on TLS (HTTPS://, WSS://) host identification solely, in the modern world there is
|
||||||
|
a high probability of attacks on unfriendly (in respect to at least some of your users) states to the SSL certificates
|
||||||
|
chain, in which case the MITM and spoofing will be undetected. Check the [remoteId](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec/-kilo-client/remote-id.html?query=suspend%20fun%20remoteId():%20VerifyingPublicKey?) in your client on each connection and provide the safe [serverSecretKey](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec/-kilo-server/index.html?query=class%20KiloServer%3CS%3E(clientInterface:%20KiloInterface%3CS%3E,%20connections:%20Flow%3CInetTransportDevice%3E,%20serverSecretKey:%20SigningKey?%20=%20null,%20sessionBuilder:%20()%20-%3E%20S) when creating a server.
|
||||||
|
|
||||||
|
This will effectively protetcs against certificate chain spoofing in the case of the application installed from the trusted source.
|
||||||
|
|
||||||
|
__Important note__. The web application could not be completely secured this way unless is loaded from the IP-address, as the DNS could be spoofed the same, especially when used with `Cloudflare` or other CDN that can
|
||||||
|
transparently substitute the whole site. In the case of we applications we strongly recommend not to use CDN except your own where you can control actual traffic rules.
|
||||||
|
|
||||||
|
## See also:
|
||||||
|
|
||||||
|
- [Source documentation](https://code.sergeych.net/docs/kiloparsec/)
|
||||||
|
- [Project's WIKI](https://gitea.sergeych.net/sergeych/kiloparsec/wiki)
|
||||||
|
|
||||||
# Details
|
# Details
|
||||||
|
|
||||||
It is not compatible with parsec family and no more based on an Universa crypto library. To better fit
|
It is not compatible with parsec family and no more based on an Universa crypto library. To better fit
|
||||||
@@ -144,12 +201,14 @@ and every connection (while parsec caches session keys to avoid time-consuming k
|
|||||||
keys cryptography for session is shifted to use ed25519 curves which are supposed to provide agreeable strength with
|
keys cryptography for session is shifted to use ed25519 curves which are supposed to provide agreeable strength with
|
||||||
enough speed to protect every connection with a unique new keys. Also, we completely get rid of SHA2.
|
enough speed to protect every connection with a unique new keys. Also, we completely get rid of SHA2.
|
||||||
|
|
||||||
Kiloparsec also uses a denser binary format [bipack](https://gitea.sergeych.net/SergeychWorks/mp_bintools), no more key-values,
|
Kiloparsec also uses a denser binary format [bipack](https://gitea.sergeych.net/SergeychWorks/mp_bintools), no more
|
||||||
which reveals much less on the inner data structure, providing advanced
|
key-values,
|
||||||
typed RPC interfaces with kotlinx.serialization. There is also Rust implementation [bipack_ru](https://gitea.sergeych.net/DiWAN/bipack_ru).
|
which reveals much less on the inner data structure, providing advanced
|
||||||
|
typed RPC interfaces with kotlinx.serialization. There is also Rust
|
||||||
|
implementation [bipack_ru](https://gitea.sergeych.net/DiWAN/bipack_ru).
|
||||||
The architecture allows connecting same functional interfaces to several various type channels at once.
|
The architecture allows connecting same functional interfaces to several various type channels at once.
|
||||||
|
|
||||||
Also, the difference from parsecs is that there are no more unencrypted layer commands available to users.
|
Also, the difference from parsecs is that there are no more unencrypted layer commands available to users.
|
||||||
All RPC is performed over the encrypted connection.
|
All RPC is performed over the encrypted connection.
|
||||||
|
|
||||||
# Technical description
|
# Technical description
|
||||||
@@ -163,9 +222,10 @@ Integrated tools to prevent MITM attacks include also non-transferred independen
|
|||||||
independently on the ends and is never transferred with the network. Comparing it somehow (visually, with QR code, etc)
|
independently on the ends and is never transferred with the network. Comparing it somehow (visually, with QR code, etc)
|
||||||
could add a very robust guarantee of the connection safety and ingenuity.
|
could add a very robust guarantee of the connection safety and ingenuity.
|
||||||
|
|
||||||
Kiloparsec has built-in completely asynchronous (coroutine based top-down) transport layer based on TCP (JVM only as for now) and the same async Websocket-based transport based on KTOR. Websocket client is multiplatform, though the server is JVM only insofar.
|
Kiloparsec has built-in completely asynchronous (coroutine based top-down) transport layer based on TCP (JVM only as for
|
||||||
|
now) and the same async Websocket-based transport based on KTOR. Websocket client is multiplatform, though the server is
|
||||||
|
JVM only insofar.
|
||||||
|
|
||||||
# Licensing
|
# Licensing
|
||||||
|
|
||||||
When the project reaches some stable state, it will be available under Apache or MIT license. I promise ;) I do
|
Currently, you need to obtain a license from https://8-rays.dev or Sergey Chernov.
|
||||||
write opensource ;)
|
|
||||||
Executable
+4
@@ -0,0 +1,4 @@
|
|||||||
|
#!/bin/bash
|
||||||
|
set -e
|
||||||
|
./gradlew dokkaHtml
|
||||||
|
rsync -avz ./build/dokka/* code.sergeych.net:/bigstore/sergeych_pub/code/docs/kiloparsec
|
||||||
+60
-29
@@ -1,12 +1,12 @@
|
|||||||
|
|
||||||
plugins {
|
plugins {
|
||||||
kotlin("multiplatform") version "2.0.0"
|
kotlin("multiplatform") version "2.0.0"
|
||||||
id("org.jetbrains.kotlin.plugin.serialization") version "2.0.0"
|
id("org.jetbrains.kotlin.plugin.serialization") version "2.0.0"
|
||||||
`maven-publish`
|
`maven-publish`
|
||||||
|
id("org.jetbrains.dokka") version "1.9.20"
|
||||||
}
|
}
|
||||||
|
|
||||||
group = "net.sergeych"
|
group = "net.sergeych"
|
||||||
version = "0.2.4"
|
version = "0.3.1"
|
||||||
|
|
||||||
repositories {
|
repositories {
|
||||||
mavenCentral()
|
mavenCentral()
|
||||||
@@ -20,20 +20,8 @@ kotlin {
|
|||||||
js {
|
js {
|
||||||
browser {
|
browser {
|
||||||
}
|
}
|
||||||
|
nodejs()
|
||||||
}
|
}
|
||||||
// val hostOs = System.getProperty("os.name")
|
|
||||||
// val isArm64 = System.getProperty("os.arch") == "aarch64"
|
|
||||||
// val isMingwX64 = hostOs.startsWith("Windows")
|
|
||||||
// @Suppress("UNUSED_VARIABLE")
|
|
||||||
// val nativeTarget = when {
|
|
||||||
// hostOs == "Mac OS X" && isArm64 -> macosArm64("native")
|
|
||||||
// hostOs == "Mac OS X" && !isArm64 -> macosX64("native")
|
|
||||||
// hostOs == "Linux" && isArm64 -> linuxArm64("native")
|
|
||||||
// hostOs == "Linux" && !isArm64 -> linuxX64("native")
|
|
||||||
// isMingwX64 -> mingwX64("native")
|
|
||||||
// else -> throw GradleException("Host OS is not supported in Kotlin/Native.")
|
|
||||||
// }
|
|
||||||
|
|
||||||
macosArm64()
|
macosArm64()
|
||||||
iosX64()
|
iosX64()
|
||||||
iosArm64()
|
iosArm64()
|
||||||
@@ -55,18 +43,16 @@ kotlin {
|
|||||||
dependencies {
|
dependencies {
|
||||||
implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.7.3")
|
implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.7.3")
|
||||||
implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.7.1")
|
implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.7.1")
|
||||||
|
|
||||||
// api("com.ionspin.kotlin:bignum:0.3.9")
|
|
||||||
api("io.ktor:ktor-client-core:$ktor_version")
|
api("io.ktor:ktor-client-core:$ktor_version")
|
||||||
api("net.sergeych:crypto2:0.4.2")
|
api("net.sergeych:crypto2:0.4.2")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// val ktorSocketMain by creating {
|
val ktorSocketMain by creating {
|
||||||
// dependsOn(commonMain)
|
dependsOn(commonMain)
|
||||||
// dependencies {
|
dependencies {
|
||||||
// implementation("io.ktor:ktor-network:$ktor_version")
|
implementation("io.ktor:ktor-network:$ktor_version")
|
||||||
// }
|
}
|
||||||
// }
|
}
|
||||||
val commonTest by getting {
|
val commonTest by getting {
|
||||||
dependencies {
|
dependencies {
|
||||||
implementation(kotlin("test"))
|
implementation(kotlin("test"))
|
||||||
@@ -74,6 +60,9 @@ kotlin {
|
|||||||
implementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.8.1")
|
implementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.8.1")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
val ktorSocketTest by creating {
|
||||||
|
dependsOn(commonTest)
|
||||||
|
}
|
||||||
val jvmMain by getting {
|
val jvmMain by getting {
|
||||||
dependencies {
|
dependencies {
|
||||||
implementation("io.ktor:ktor-server-core:$ktor_version")
|
implementation("io.ktor:ktor-server-core:$ktor_version")
|
||||||
@@ -82,19 +71,51 @@ kotlin {
|
|||||||
implementation("io.ktor:ktor-server-netty:$ktor_version")
|
implementation("io.ktor:ktor-server-netty:$ktor_version")
|
||||||
api("io.ktor:ktor-client-cio:$ktor_version")
|
api("io.ktor:ktor-client-cio:$ktor_version")
|
||||||
}
|
}
|
||||||
// dependsOn(ktorSocketMain)
|
dependsOn(ktorSocketMain)
|
||||||
}
|
}
|
||||||
val jvmTest by getting
|
val jvmTest by getting {
|
||||||
|
dependsOn(ktorSocketTest)
|
||||||
|
}
|
||||||
|
|
||||||
val jsMain by getting {
|
val jsMain by getting {
|
||||||
dependencies {
|
dependencies {
|
||||||
implementation("io.ktor:ktor-client-js:$ktor_version")
|
implementation("io.ktor:ktor-client-js:$ktor_version")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
val jsTest by getting
|
val jsTest by getting
|
||||||
|
val macosArm64Main by getting {
|
||||||
// for (pm in listOf(linuxMain, macosMain, iosMain, mingwMain))
|
dependsOn(ktorSocketMain)
|
||||||
// pm { dependsOn(ktorSocketMain) }
|
}
|
||||||
|
val macosArm64Test by getting {
|
||||||
|
dependsOn(ktorSocketTest)
|
||||||
|
}
|
||||||
|
val macosX64Main by getting {
|
||||||
|
dependsOn(ktorSocketMain)
|
||||||
|
}
|
||||||
|
val iosX64Main by getting {
|
||||||
|
dependsOn(ktorSocketMain)
|
||||||
|
}
|
||||||
|
val iosX64Test by getting {
|
||||||
|
dependsOn(ktorSocketTest)
|
||||||
|
}
|
||||||
|
val iosArm64Main by getting {
|
||||||
|
dependsOn(ktorSocketMain)
|
||||||
|
}
|
||||||
|
val iosArm64Test by getting {
|
||||||
|
dependsOn(ktorSocketTest)
|
||||||
|
}
|
||||||
|
val linuxArm64Main by getting {
|
||||||
|
dependsOn(ktorSocketMain)
|
||||||
|
}
|
||||||
|
val linuxArm64Test by getting {
|
||||||
|
dependsOn(ktorSocketTest)
|
||||||
|
}
|
||||||
|
val linuxX64Main by getting {
|
||||||
|
dependsOn(ktorSocketMain)
|
||||||
|
}
|
||||||
|
val linuxX64Test by getting {
|
||||||
|
dependsOn(ktorSocketTest)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
publishing {
|
publishing {
|
||||||
@@ -115,3 +136,13 @@ kotlin {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
tasks.dokkaHtml.configure {
|
||||||
|
outputDirectory.set(buildDir.resolve("dokka"))
|
||||||
|
dokkaSourceSets {
|
||||||
|
configureEach {
|
||||||
|
// includes.from("docs/bipack.md")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
@@ -1 +1,2 @@
|
|||||||
kotlin.code.style=official
|
kotlin.code.style=official
|
||||||
|
kotlin.mpp.applyDefaultHierarchyTemplate=false
|
||||||
@@ -19,6 +19,11 @@ inline fun <reified A, reified R> command(overrideName: String? = null): Command
|
|||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Declare a Push: Unit-returning command usually used with [RemoteInterface.push]
|
||||||
|
*/
|
||||||
|
inline fun <reified A> push(overrideName: String? = null): CommandDelegate<A, Unit> = command(overrideName)
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Delegate to create [Command] via property
|
* Delegate to create [Command] via property
|
||||||
*/
|
*/
|
||||||
|
|||||||
@@ -17,10 +17,33 @@ import net.sergeych.mp_logger.exception
|
|||||||
import net.sergeych.mp_tools.globalLaunch
|
import net.sergeych.mp_tools.globalLaunch
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* The auto-connecting client that reconnects to the kiloparsec server
|
* The auto-connecting client that reconnects to the kiloparsec server,
|
||||||
|
* [KiloServer],
|
||||||
* and maintain connection state flow. Client factory launches a disconnected
|
* and maintain connection state flow. Client factory launches a disconnected
|
||||||
* set of coroutines to support automatic reconnection, so you _must_ [close]
|
* set of coroutines to support automatic reconnection, so you _must_ [close]
|
||||||
* it manually when it is not needed, otherwise it will continue to reconnect.
|
* it manually when it is unnecessary, otherwise it will continue to reconnect.
|
||||||
|
*
|
||||||
|
* ## Usage
|
||||||
|
*
|
||||||
|
* Suppose we have TCP/IP server as in the [KiloServer] usage sample. Then we can connect
|
||||||
|
* to it providing TCP/IP connector like:
|
||||||
|
*
|
||||||
|
* ```kotlin
|
||||||
|
* val client = KiloClient<Unit>() {
|
||||||
|
* connect { connectTcpDevice("localhost:$port") }
|
||||||
|
* }
|
||||||
|
*
|
||||||
|
* // now we can invoke remote commands:
|
||||||
|
* assertEquals("unknown", client.call(cmdLoad))
|
||||||
|
*
|
||||||
|
* client.call(cmdSave, "foobar")
|
||||||
|
* assertEquals("foobar", client.call(cmdLoad))
|
||||||
|
* ```
|
||||||
|
*
|
||||||
|
* ## See also
|
||||||
|
*
|
||||||
|
* [KiloServer]
|
||||||
|
*
|
||||||
*/
|
*/
|
||||||
class KiloClient<S>(
|
class KiloClient<S>(
|
||||||
val localInterface: KiloInterface<S>,
|
val localInterface: KiloInterface<S>,
|
||||||
@@ -28,7 +51,6 @@ class KiloClient<S>(
|
|||||||
connectionDataFactory: ConnectionDataFactory<S>,
|
connectionDataFactory: ConnectionDataFactory<S>,
|
||||||
) : RemoteInterface,
|
) : RemoteInterface,
|
||||||
Loggable by LogTag("CLIF") {
|
Loggable by LogTag("CLIF") {
|
||||||
|
|
||||||
val _state = MutableStateFlow(false)
|
val _state = MutableStateFlow(false)
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -94,6 +116,15 @@ class KiloClient<S>(
|
|||||||
throw t
|
throw t
|
||||||
}
|
}
|
||||||
|
|
||||||
|
override suspend fun <A> push(cmd: Command<A, Unit>, args: A) {
|
||||||
|
try {
|
||||||
|
deferredClient.await().push(cmd, args)
|
||||||
|
} catch (t: RemoteInterface.ClosedException) {
|
||||||
|
resetDeferredClient()
|
||||||
|
throw t
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Current session token. This is a per-connection unique random value same on the client and server part so
|
* Current session token. This is a per-connection unique random value same on the client and server part so
|
||||||
* it could be used as a nonce to pair MITM and like attacks, be sure that the server is actually
|
* it could be used as a nonce to pair MITM and like attacks, be sure that the server is actually
|
||||||
|
|||||||
@@ -100,4 +100,8 @@ class KiloClientConnection<S>(
|
|||||||
suspend fun token() = deferredParams.await().token
|
suspend fun token() = deferredParams.await().token
|
||||||
override suspend fun <A, R> call(cmd: Command<A, R>, args: A): R =
|
override suspend fun <A, R> call(cmd: Command<A, R>, args: A): R =
|
||||||
kiloRemoteInterface.await().call(cmd, args)
|
kiloRemoteInterface.await().call(cmd, args)
|
||||||
|
|
||||||
|
override suspend fun <A> push(cmd: Command<A, Unit>, args: A) {
|
||||||
|
kiloRemoteInterface.await().push(cmd, args)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
@@ -4,7 +4,13 @@ package net.sergeych.kiloparsec
|
|||||||
* The local interface to provide functions, register errors for Kiloparsec users. Use it
|
* The local interface to provide functions, register errors for Kiloparsec users. Use it
|
||||||
* with [KiloClient], [KiloClientConnection], [KiloServerConnection], etc.
|
* with [KiloClient], [KiloClientConnection], [KiloServerConnection], etc.
|
||||||
*
|
*
|
||||||
* BAse implementation registers relevant exceptions.
|
* Base class implementation does the following:
|
||||||
|
*
|
||||||
|
* - It registers common exceptions from [RemoteInterface] and kotlin/java `IllegalArgumentException` and
|
||||||
|
* `IllegalStateException`
|
||||||
|
* - It provides [onConnected] handler
|
||||||
|
*
|
||||||
|
* See [KiloServer] for usage sample.
|
||||||
*/
|
*/
|
||||||
open class KiloInterface<S> : LocalInterface<KiloScope<S>>() {
|
open class KiloInterface<S> : LocalInterface<KiloScope<S>>() {
|
||||||
|
|
||||||
|
|||||||
@@ -35,5 +35,10 @@ class KiloRemoteInterface<S>(
|
|||||||
else -> throw RemoteInterface.Exception("unexpected block type: $block")
|
else -> throw RemoteInterface.Exception("unexpected block type: $block")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
override suspend fun <A> push(cmd: Command<A, Unit>, args: A) {
|
||||||
|
val params = deferredParams.await()
|
||||||
|
params.transport.call(L0Call, params.encrypt(cmd.packCall(args)))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -13,13 +13,100 @@ import net.sergeych.mp_tools.globalLaunch
|
|||||||
import net.sergeych.tools.AtomicCounter
|
import net.sergeych.tools.AtomicCounter
|
||||||
|
|
||||||
private val instances = AtomicCounter()
|
private val instances = AtomicCounter()
|
||||||
@Suppress("unused")
|
|
||||||
|
/**
|
||||||
|
* The Kiloparsec server.
|
||||||
|
* Server accepts incoming connections and serves them using the same [clientInterface].
|
||||||
|
*
|
||||||
|
* ## Incoming connections
|
||||||
|
*
|
||||||
|
* Server collecting incoming connections provided by [connections] `Flow`. For each incoming connection
|
||||||
|
* the Kiloparsec handshake is performed, then the session object is created, see below, and connection is
|
||||||
|
* served with [clientInterface] until closed.
|
||||||
|
*
|
||||||
|
* ## Session param [S]
|
||||||
|
*
|
||||||
|
* After the successful handshake server creates new session for each connection calling the [sessionBuilder].
|
||||||
|
* Then it creates a [KiloScope] so [KiloScope.session] holds this connection-specific instance. Then
|
||||||
|
* [KiloInterface.onConnected] is called with this scope. Session can be used to hold connection state. Session
|
||||||
|
* objects are not persistent, but could be initialized in [KiloInterface.onConnected] where the remote side
|
||||||
|
* [KiloScope.remoteIdentity] is already verified and set.
|
||||||
|
*
|
||||||
|
* ## Usage:
|
||||||
|
*
|
||||||
|
* Create a shared library between you server and clients, to specify the interface (otherwise you can
|
||||||
|
* share sources).
|
||||||
|
*
|
||||||
|
* Suppose we have session with a state:
|
||||||
|
* ```kotlin
|
||||||
|
* data class Session(
|
||||||
|
* var data: String,
|
||||||
|
* )
|
||||||
|
*```
|
||||||
|
*
|
||||||
|
* And some commands to access and change it, in the shared library too:
|
||||||
|
*
|
||||||
|
* ```kotlin
|
||||||
|
* val cmdSave by command<String, Unit>()
|
||||||
|
* val cmdLoad by command<Unit, String>()
|
||||||
|
* val cmdDrop by command<Unit, Unit>()
|
||||||
|
* val cmdException by command<Unit, Unit>()
|
||||||
|
* ```
|
||||||
|
* Then the server code (TCP/IP variant) could look like:
|
||||||
|
*
|
||||||
|
* ```kotlin
|
||||||
|
* // The server implementation (could be shared between server instances connected
|
||||||
|
* // to different protocol adapters):
|
||||||
|
*
|
||||||
|
* val cli = KiloInterface<Session>().apply {
|
||||||
|
* // Suppose we want to throw this exception at the caller site, so we need to register it:
|
||||||
|
* registerError { SomeException() }
|
||||||
|
*
|
||||||
|
* // Session initialization. If you need a sessino to depend initially on the client's identity.
|
||||||
|
* // you can do it here:
|
||||||
|
* onConnected {
|
||||||
|
* // check the remoteIdentity
|
||||||
|
* session.data = if( remoteIdentity == somePublicVerifyingKey )
|
||||||
|
* "known"
|
||||||
|
* else
|
||||||
|
* "unknown"
|
||||||
|
* }
|
||||||
|
* on(cmdSave) { session.data = it }
|
||||||
|
* on(cmdLoad) {
|
||||||
|
* session.data
|
||||||
|
* }
|
||||||
|
* on(cmdException) {
|
||||||
|
* throw TestException()
|
||||||
|
* }
|
||||||
|
* on(cmdDrop) {
|
||||||
|
* throw LocalInterface.BreakConnectionException()
|
||||||
|
* }
|
||||||
|
* }
|
||||||
|
*
|
||||||
|
* // Create the server instance that accepts incoming TCP/IP connections on all local interfaces using the
|
||||||
|
* // specified port:
|
||||||
|
*
|
||||||
|
* val server = KiloServer(cli, acceptTcpDevice(port)) {
|
||||||
|
* // This creates a new session
|
||||||
|
* Session("unknown")
|
||||||
|
* }
|
||||||
|
* ```
|
||||||
|
*
|
||||||
|
* See [KiloClient] to connect to the server.
|
||||||
|
*
|
||||||
|
* @param S the type of the server session object, returned by [sessionBuilder]. See above
|
||||||
|
* @param clientInterface the interface available for remote calls
|
||||||
|
* @param connections flow of incoming connections. Server stops when the flow is fully collected (normally
|
||||||
|
* it shouldn't)
|
||||||
|
* @param serverSecretKey the [SigningKey] used to identify this server during Kiloparsec handshake.
|
||||||
|
* @param sessionBuilder callback that creates session objects for successful incoming connections
|
||||||
|
*/
|
||||||
class KiloServer<S>(
|
class KiloServer<S>(
|
||||||
private val clientInterface: KiloInterface<S>,
|
private val clientInterface: KiloInterface<S>,
|
||||||
private val connections: Flow<InetTransportDevice>,
|
private val connections: Flow<InetTransportDevice>,
|
||||||
private val serverSecretKey: SigningKey? = null,
|
private val serverSecretKey: SigningKey? = null,
|
||||||
private val sessionBuilder: ()->S,
|
private val sessionBuilder: () -> S,
|
||||||
): LogTag("KS:${instances.incrementAndGet()}") {
|
) : LogTag("KS:${instances.incrementAndGet()}") {
|
||||||
|
|
||||||
private val job = globalLaunch {
|
private val job = globalLaunch {
|
||||||
connections.collect { device ->
|
connections.collect { device ->
|
||||||
@@ -27,22 +114,30 @@ class KiloServer<S>(
|
|||||||
try {
|
try {
|
||||||
info { "connected ${device}" }
|
info { "connected ${device}" }
|
||||||
KiloServerConnection(clientInterface, device, sessionBuilder(), serverSecretKey)
|
KiloServerConnection(clientInterface, device, sessionBuilder(), serverSecretKey)
|
||||||
.apply { debug { "server connection is ready" }}
|
.apply { debug { "server connection is ready" } }
|
||||||
.run()
|
.run()
|
||||||
}
|
} catch (_: CancellationException) {
|
||||||
catch(_: CancellationException) {
|
} catch (cce: LocalInterface.BreakConnectionException) {
|
||||||
}
|
|
||||||
catch(cce: LocalInterface.BreakConnectionException) {
|
|
||||||
info { "Closed exception caught, closing (${cce.flushSendQueue}" }
|
info { "Closed exception caught, closing (${cce.flushSendQueue}" }
|
||||||
}
|
} catch (t: Throwable) {
|
||||||
catch (t: Throwable) {
|
|
||||||
exception { "unexpected while creating kiloclient" to t }
|
exception { "unexpected while creating kiloclient" to t }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Stop the server and cancel all pending sessions. Unlike finishing the flow passed
|
||||||
|
* for [KiloServer.connections], it will cancel all currently active sessions.
|
||||||
|
*/
|
||||||
fun close() {
|
fun close() {
|
||||||
job.cancel()
|
job.cancel()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Server is closed either if [close] was called or all [KiloServer.connections] were collected and flow was
|
||||||
|
* closed.
|
||||||
|
*/
|
||||||
|
@Suppress("unused")
|
||||||
|
val isClosed: Boolean get() = job.isCompleted
|
||||||
}
|
}
|
||||||
@@ -94,4 +94,8 @@ class KiloServerConnection<S>(
|
|||||||
override suspend fun <A, R> call(cmd: Command<A, R>, args: A): R {
|
override suspend fun <A, R> call(cmd: Command<A, R>, args: A): R {
|
||||||
return kiloRemoteInterface.await().call(cmd, args)
|
return kiloRemoteInterface.await().call(cmd, args)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
override suspend fun <A> push(cmd: Command<A, Unit>, args: A) {
|
||||||
|
kiloRemoteInterface.await().push(cmd, args)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
@@ -50,4 +50,18 @@ interface RemoteInterface {
|
|||||||
* Call the remote procedure with specified args and return its result
|
* Call the remote procedure with specified args and return its result
|
||||||
*/
|
*/
|
||||||
suspend fun <A, R> call(cmd: Command<A, R>, args: A): R
|
suspend fun <A, R> call(cmd: Command<A, R>, args: A): R
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Push the notification without waiting for reception or processing.
|
||||||
|
* It returns immediately after sending data to the transport (e.g., to the network).
|
||||||
|
* Use [call] if it is necessary to wait until the command will be received and processed by the remote.
|
||||||
|
*/
|
||||||
|
suspend fun <A> push(cmd: Command<A, Unit>, args: A)
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Push the command with no args.
|
||||||
|
* It returns immediately after sending data to the transport (e.g., to the network).
|
||||||
|
* Use [call] if it is necessary to wait until the command will be received and processed by the remote.
|
||||||
|
*/
|
||||||
|
suspend fun push(cmd: Command<Unit,Unit>) = push(cmd,Unit)
|
||||||
}
|
}
|
||||||
@@ -15,6 +15,7 @@ import kotlinx.serialization.descriptors.SerialDescriptor
|
|||||||
import kotlinx.serialization.encoding.Decoder
|
import kotlinx.serialization.encoding.Decoder
|
||||||
import kotlinx.serialization.encoding.Encoder
|
import kotlinx.serialization.encoding.Encoder
|
||||||
import kotlinx.serialization.serializer
|
import kotlinx.serialization.serializer
|
||||||
|
import net.sergeych.bipack.Unsigned
|
||||||
import net.sergeych.crypto2.toDump
|
import net.sergeych.crypto2.toDump
|
||||||
import net.sergeych.kiloparsec.Transport.Device
|
import net.sergeych.kiloparsec.Transport.Device
|
||||||
import net.sergeych.mp_logger.*
|
import net.sergeych.mp_logger.*
|
||||||
@@ -63,7 +64,12 @@ class Transport<S>(
|
|||||||
@Serializable(TransportBlockSerializer::class)
|
@Serializable(TransportBlockSerializer::class)
|
||||||
sealed class Block {
|
sealed class Block {
|
||||||
@Serializable
|
@Serializable
|
||||||
data class Call(val id: UInt, val name: String, val packedArgs: UByteArray) : Block() {
|
data class Call(
|
||||||
|
@Unsigned
|
||||||
|
val id: UInt,
|
||||||
|
val name: String,
|
||||||
|
val packedArgs: UByteArray
|
||||||
|
) : Block() {
|
||||||
override fun equals(other: Any?): Boolean {
|
override fun equals(other: Any?): Boolean {
|
||||||
if (this === other) return true
|
if (this === other) return true
|
||||||
if (other !is Call) return false
|
if (other !is Call) return false
|
||||||
@@ -83,10 +89,10 @@ class Transport<S>(
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Serializable
|
@Serializable
|
||||||
data class Response(val forId: UInt, val packedResult: UByteArray) : Block()
|
data class Response(@Unsigned val forId: UInt, val packedResult: UByteArray) : Block()
|
||||||
|
|
||||||
@Serializable
|
@Serializable
|
||||||
data class Error(val forId: UInt, val code: String, val text: String? = null, val extra: UByteArray? = null) :
|
data class Error(@Unsigned val forId: UInt, val code: String, val text: String? = null, val extra: UByteArray? = null) :
|
||||||
Block() {
|
Block() {
|
||||||
val message by lazy { text ?: "remote exception: $code" }
|
val message by lazy { text ?: "remote exception: $code" }
|
||||||
}
|
}
|
||||||
@@ -98,7 +104,8 @@ class Transport<S>(
|
|||||||
var isClosed: Boolean = false
|
var isClosed: Boolean = false
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Send a call block for a command and packed args and return packed result if it is not an error
|
* Send a call block for a command and packed args and return packed result if it is not an error. It suspends
|
||||||
|
* until receiving answer from the remote side, even if returns `Unit`.
|
||||||
* @throws RemoteInterface.RemoteException if the remote call caused an exception. Normally use [call] instead.
|
* @throws RemoteInterface.RemoteException if the remote call caused an exception. Normally use [call] instead.
|
||||||
* @throws RemoteInterface.ClosedException
|
* @throws RemoteInterface.ClosedException
|
||||||
*/
|
*/
|
||||||
@@ -111,12 +118,13 @@ class Transport<S>(
|
|||||||
// We need to shield calls and lastID with mutex, but nothing more:
|
// We need to shield calls and lastID with mutex, but nothing more:
|
||||||
access.withLock {
|
access.withLock {
|
||||||
if (isClosed) throw RemoteInterface.ClosedException()
|
if (isClosed) throw RemoteInterface.ClosedException()
|
||||||
|
// the order is important: first id in use MUST BE >= 1, not zero:
|
||||||
b = Block.Call(++lastId, name, packedArgs)
|
b = Block.Call(++lastId, name, packedArgs)
|
||||||
calls[b.id] = deferred
|
calls[b.id] = deferred
|
||||||
}
|
}
|
||||||
|
|
||||||
// now we have mutex freed so we can call:
|
// now we have mutex freed so we can call:
|
||||||
val r = runCatching { device.output.send(pack(b).also { debug { ">>>\n${it.toDump()}" } }) }
|
val r = runCatching { device.output.send(pack(b).also { debug { ">>>\n${it.toDump()}" } }) }
|
||||||
if (!r.isSuccess) {
|
if (!r.isSuccess) {
|
||||||
r.exceptionOrNull()?.let {
|
r.exceptionOrNull()?.let {
|
||||||
exception { "failed to send output block" to it }
|
exception { "failed to send output block" to it }
|
||||||
@@ -131,6 +139,25 @@ class Transport<S>(
|
|||||||
return deferred.await()
|
return deferred.await()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Send a call block for a command and packed args and return packed result if it is not an error. It suspends
|
||||||
|
* until receiving answer from the remote side, even if returns `Unit`.
|
||||||
|
* @throws RemoteInterface.RemoteException if the remote call caused an exception. Normally use [call] instead.
|
||||||
|
* @throws RemoteInterface.ClosedException
|
||||||
|
*/
|
||||||
|
private suspend fun sendPushBlock(name: String, packedArgs: UByteArray) {
|
||||||
|
if (isClosed) throw RemoteInterface.ClosedException()
|
||||||
|
|
||||||
|
// All push blocks have the same id == 0:
|
||||||
|
val b = Block.Call(0u, name, packedArgs)
|
||||||
|
val r = runCatching { device.output.send(pack(b).also { debug { ">>$\n${it.toDump()}" } }) }
|
||||||
|
when(val e = r.exceptionOrNull()) {
|
||||||
|
is RemoteInterface.ClosedException, is CancellationException, is RemoteInterface.RemoteException
|
||||||
|
-> throw e
|
||||||
|
else -> throw RemoteInterface.ClosedException()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Call the remote procedure with specified args and return its result
|
* Call the remote procedure with specified args and return its result
|
||||||
*/
|
*/
|
||||||
@@ -139,6 +166,10 @@ class Transport<S>(
|
|||||||
return unpack(cmd.resultSerializer, result)
|
return unpack(cmd.resultSerializer, result)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
override suspend fun <A>push(cmd: Command<A,Unit>,args: A) {
|
||||||
|
sendPushBlock(cmd.name, pack(cmd.argsSerializer, args))
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Start running the transport. This function suspends until the transport is closed
|
* Start running the transport. This function suspends until the transport is closed
|
||||||
* normally or by error. If you need to cancel it prematurely, cancel the coroutine
|
* normally or by error. If you need to cancel it prematurely, cancel the coroutine
|
||||||
@@ -163,7 +194,7 @@ class Transport<S>(
|
|||||||
warning { "decoded error: ${error::class.simpleName}: $error" }
|
warning { "decoded error: ${error::class.simpleName}: $error" }
|
||||||
calls.remove(b.forId)?.completeExceptionally(localInterface.decodeError(b))
|
calls.remove(b.forId)?.completeExceptionally(localInterface.decodeError(b))
|
||||||
?: warning { "error handler not found for ${b.forId}" }
|
?: warning { "error handler not found for ${b.forId}" }
|
||||||
info { "error processed"}
|
info { "error processed" }
|
||||||
}
|
}
|
||||||
|
|
||||||
is Block.Response -> access.withLock {
|
is Block.Response -> access.withLock {
|
||||||
@@ -176,17 +207,21 @@ class Transport<S>(
|
|||||||
|
|
||||||
is Block.Call -> launch {
|
is Block.Call -> launch {
|
||||||
try {
|
try {
|
||||||
send(
|
if (b.id == 0u)
|
||||||
Block.Response(
|
// Command does not waits return
|
||||||
b.id,
|
localInterface.execute(commandContext, b.name, b.packedArgs)
|
||||||
localInterface.execute(commandContext, b.name, b.packedArgs)
|
else
|
||||||
|
send(
|
||||||
|
Block.Response(
|
||||||
|
b.id,
|
||||||
|
localInterface.execute(commandContext, b.name, b.packedArgs)
|
||||||
|
)
|
||||||
)
|
)
|
||||||
)
|
|
||||||
} catch (x: LocalInterface.BreakConnectionException) {
|
} catch (x: LocalInterface.BreakConnectionException) {
|
||||||
// handler forced close
|
// handler forced close
|
||||||
warning { "handler requested closing of the connection (${x.flushSendQueue}"}
|
warning { "handler requested closing of the connection (${x.flushSendQueue}" }
|
||||||
isClosed = true
|
isClosed = true
|
||||||
if( x.flushSendQueue ) device.flush()
|
if (x.flushSendQueue) device.flush()
|
||||||
device.close()
|
device.close()
|
||||||
} catch (x: RemoteInterface.RemoteException) {
|
} catch (x: RemoteInterface.RemoteException) {
|
||||||
send(Block.Error(b.id, x.code, x.text, x.extra))
|
send(Block.Error(b.id, x.code, x.text, x.extra))
|
||||||
@@ -196,26 +231,23 @@ class Transport<S>(
|
|||||||
.also { debug { "command executed: ${b.name}" } }
|
.also { debug { "command executed: ${b.name}" } }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
debug { "=---------------------------------------------"}
|
debug { "=---------------------------------------------" }
|
||||||
}
|
}
|
||||||
debug { "input step performed closed=$isClosed active=$isActive"}
|
debug { "input step performed closed=$isClosed active=$isActive" }
|
||||||
} catch (_: ClosedSendChannelException) {
|
} catch (_: ClosedSendChannelException) {
|
||||||
info { "closed send channel" }
|
info { "closed send channel" }
|
||||||
isClosed = true
|
isClosed = true
|
||||||
} catch (_: ClosedReceiveChannelException) {
|
} catch (_: ClosedReceiveChannelException) {
|
||||||
info { "closed receive channel"}
|
info { "closed receive channel" }
|
||||||
isClosed = true
|
isClosed = true
|
||||||
}
|
} catch (cce: LocalInterface.BreakConnectionException) {
|
||||||
catch(cce: LocalInterface.BreakConnectionException) {
|
info { "closing connection by local request ($cce)" }
|
||||||
info { "closing connection by local request ($cce)"}
|
|
||||||
device.close()
|
device.close()
|
||||||
}
|
} catch (t: RemoteInterface.ClosedException) {
|
||||||
catch(t: RemoteInterface.ClosedException) {
|
|
||||||
// it is ok: we just exit the coroutine normally
|
// it is ok: we just exit the coroutine normally
|
||||||
// and mark we're closing
|
// and mark we're closing
|
||||||
isClosed = true
|
isClosed = true
|
||||||
}
|
} catch (_: CancellationException) {
|
||||||
catch (_: CancellationException) {
|
|
||||||
info { "loop is cancelled with CancellationException" }
|
info { "loop is cancelled with CancellationException" }
|
||||||
isClosed = true
|
isClosed = true
|
||||||
} catch (t: Throwable) {
|
} catch (t: Throwable) {
|
||||||
@@ -226,7 +258,7 @@ class Transport<S>(
|
|||||||
}
|
}
|
||||||
debug { "leaving transport loop" }
|
debug { "leaving transport loop" }
|
||||||
access.withLock {
|
access.withLock {
|
||||||
debug { "access lock obtained"}
|
debug { "access lock obtained" }
|
||||||
isClosed = true
|
isClosed = true
|
||||||
debug { "closing device $device, calls in queue ${calls.size}" }
|
debug { "closing device $device, calls in queue ${calls.size}" }
|
||||||
runCatching { device.close() }
|
runCatching { device.close() }
|
||||||
@@ -243,8 +275,7 @@ class Transport<S>(
|
|||||||
private suspend fun send(block: Block) {
|
private suspend fun send(block: Block) {
|
||||||
try {
|
try {
|
||||||
device.output.send(pack(block))
|
device.output.send(pack(block))
|
||||||
}
|
} catch (_: ClosedSendChannelException) {
|
||||||
catch(_: ClosedSendChannelException) {
|
|
||||||
throw RemoteInterface.ClosedException()
|
throw RemoteInterface.ClosedException()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -275,7 +306,7 @@ object TransportBlockSerializer : KSerializer<Transport.Block> {
|
|||||||
|
|
||||||
|
|
||||||
override fun deserialize(decoder: Decoder): Transport.Block =
|
override fun deserialize(decoder: Decoder): Transport.Block =
|
||||||
when( val id = decoder.decodeByte().toInt()) {
|
when (val id = decoder.decodeByte().toInt()) {
|
||||||
0 -> decoder.decodeSerializableValue(serializer<Transport.Block.Call>())
|
0 -> decoder.decodeSerializableValue(serializer<Transport.Block.Call>())
|
||||||
1 -> decoder.decodeSerializableValue(serializer<Transport.Block.Error>())
|
1 -> decoder.decodeSerializableValue(serializer<Transport.Block.Error>())
|
||||||
2 -> decoder.decodeSerializableValue(serializer<Transport.Block.Response>())
|
2 -> decoder.decodeSerializableValue(serializer<Transport.Block.Response>())
|
||||||
|
|||||||
@@ -1,56 +1,13 @@
|
|||||||
package net.sergeych.kiloparsec.adapter
|
package net.sergeych.kiloparsec.adapter
|
||||||
|
|
||||||
import kotlinx.coroutines.channels.ReceiveChannel
|
|
||||||
import kotlinx.coroutines.flow.Flow
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Multiplatform implementation of an internet address.
|
* Multiplatform internet address.
|
||||||
* Notice to implementors. It must provide correct and effective [equals] and [hashCode].
|
|
||||||
*/
|
*/
|
||||||
interface NetworkAddress {
|
data class NetworkAddress(
|
||||||
val host: String
|
val host: String,
|
||||||
val port: Int
|
val port: Int
|
||||||
}
|
) {
|
||||||
|
override fun toString(): String {
|
||||||
/**
|
return "$host:$port"
|
||||||
* Multiplatform datagram abstraction
|
|
||||||
*/
|
|
||||||
interface Datagram {
|
|
||||||
/**
|
|
||||||
* Received message
|
|
||||||
*/
|
|
||||||
val message: UByteArray
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Address from where the message was sent
|
|
||||||
*/
|
|
||||||
val address: NetworkAddress
|
|
||||||
}
|
|
||||||
|
|
||||||
@OptIn(ExperimentalStdlibApi::class)
|
|
||||||
interface DatagramConnector: AutoCloseable {
|
|
||||||
|
|
||||||
val incoming: ReceiveChannel<Datagram>
|
|
||||||
suspend fun send(message: UByteArray, networkAddress: NetworkAddress)
|
|
||||||
@Suppress("unused")
|
|
||||||
suspend fun send(message: UByteArray, datagramAddress: String) {
|
|
||||||
send(message, datagramAddress.toNetworkAddress())
|
|
||||||
}
|
}
|
||||||
|
|
||||||
suspend fun send(message: UByteArray,host: String,port: Int) =
|
|
||||||
send(message, NetworkAddress(host,port))
|
|
||||||
override fun close()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
expect fun NetworkAddress(host: String,port: Int): NetworkAddress
|
|
||||||
|
|
||||||
fun String.toNetworkAddress() : NetworkAddress {
|
|
||||||
val (host, port) = this.split(":").map { it.trim()}
|
|
||||||
return NetworkAddress(host, port.toInt())
|
|
||||||
}
|
|
||||||
|
|
||||||
expect fun acceptTcpDevice(port: Int): Flow<InetTransportDevice>
|
|
||||||
|
|
||||||
expect suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice
|
|
||||||
|
|
||||||
suspend fun connectTcpDevice(address: String) = connectTcpDevice(address.toNetworkAddress())
|
|
||||||
@@ -184,10 +184,16 @@ class TransportTest {
|
|||||||
val cmdRemoteExceptionTest by command<Unit, String>()
|
val cmdRemoteExceptionTest by command<Unit, String>()
|
||||||
val cmdBreak by command<Unit, Unit>()
|
val cmdBreak by command<Unit, Unit>()
|
||||||
|
|
||||||
|
val cmdPushServer by push<String>()
|
||||||
|
val pushedFromServer = CompletableDeferred<String>()
|
||||||
|
|
||||||
val serverInterface = KiloInterface<String>().apply {
|
val serverInterface = KiloInterface<String>().apply {
|
||||||
on(cmdPing) {
|
on(cmdPing) {
|
||||||
"pong! [$it]"
|
"pong! [$it]"
|
||||||
}
|
}
|
||||||
|
on(cmdPushServer) {
|
||||||
|
pushedFromServer.complete(it)
|
||||||
|
}
|
||||||
on(cmdGetToken) {
|
on(cmdGetToken) {
|
||||||
sessionToken
|
sessionToken
|
||||||
}
|
}
|
||||||
@@ -249,11 +255,17 @@ class TransportTest {
|
|||||||
assertEquals("client pong: foo", kiloServerConnection.call(cmdPing, "foo"))
|
assertEquals("client pong: foo", kiloServerConnection.call(cmdPing, "foo"))
|
||||||
assertEquals("server push: bar", kiloServerConnection.call(cmdPush, "bar"))
|
assertEquals("server push: bar", kiloServerConnection.call(cmdPush, "bar"))
|
||||||
|
|
||||||
|
client.push(cmdPushServer, "42")
|
||||||
|
|
||||||
assertEquals("**-s1-c1-s2-c2", client.call(cmdChainCallServer1, "**"))
|
assertEquals("**-s1-c1-s2-c2", client.call(cmdChainCallServer1, "**"))
|
||||||
|
|
||||||
|
|
||||||
assertThrows<TestException> { client.call(cmdException) }
|
assertThrows<TestException> { client.call(cmdException) }
|
||||||
assertEquals("ok: te-local", client.call(cmdRemoteExceptionTest))
|
assertEquals("ok: te-local", client.call(cmdRemoteExceptionTest))
|
||||||
|
|
||||||
|
// wait for push to be received and check
|
||||||
|
assertEquals("42", pushedFromServer.await())
|
||||||
|
|
||||||
assertThrows<RemoteInterface.ClosedException> {
|
assertThrows<RemoteInterface.ClosedException> {
|
||||||
client.call(cmdBreak)
|
client.call(cmdBreak)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,2 +0,0 @@
|
|||||||
package net.sergeych.kiloparsec.adapter
|
|
||||||
|
|
||||||
@@ -1,15 +0,0 @@
|
|||||||
package net.sergeych.kiloparsec.adapter
|
|
||||||
|
|
||||||
import kotlinx.coroutines.flow.Flow
|
|
||||||
|
|
||||||
actual fun NetworkAddress(host: String, port: Int): NetworkAddress {
|
|
||||||
TODO("Not yet implemented")
|
|
||||||
}
|
|
||||||
|
|
||||||
actual fun acceptTcpDevice(port: Int): Flow<InetTransportDevice> {
|
|
||||||
TODO("Not yet implemented")
|
|
||||||
}
|
|
||||||
|
|
||||||
actual suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice {
|
|
||||||
TODO("Not yet implemented")
|
|
||||||
}
|
|
||||||
@@ -1,20 +0,0 @@
|
|||||||
package net.sergeych.kiloparsec.adapter
|
|
||||||
|
|
||||||
import java.nio.channels.CompletionHandler
|
|
||||||
import kotlin.coroutines.Continuation
|
|
||||||
import kotlin.coroutines.resume
|
|
||||||
import kotlin.coroutines.resumeWithException
|
|
||||||
|
|
||||||
open class ContinuationHandler<T> : CompletionHandler<T, Continuation<T>> {
|
|
||||||
override fun completed(result: T, attachment: Continuation<T>) {
|
|
||||||
attachment.resume(result)
|
|
||||||
}
|
|
||||||
|
|
||||||
override fun failed(exc: Throwable, attachment: Continuation<T>) {
|
|
||||||
attachment.resumeWithException(exc)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
object VoidCompletionHandler: ContinuationHandler<Void>()
|
|
||||||
|
|
||||||
object IntCompletionHandler: ContinuationHandler<Int>()
|
|
||||||
@@ -1,49 +0,0 @@
|
|||||||
package net.sergeych.kiloparsec.adapter
|
|
||||||
|
|
||||||
import kotlinx.coroutines.Dispatchers
|
|
||||||
import kotlinx.coroutines.flow.Flow
|
|
||||||
import kotlinx.coroutines.flow.flow
|
|
||||||
import kotlinx.coroutines.suspendCancellableCoroutine
|
|
||||||
import kotlinx.coroutines.withContext
|
|
||||||
import java.net.InetAddress
|
|
||||||
import java.net.InetSocketAddress
|
|
||||||
import java.nio.channels.AsynchronousServerSocketChannel
|
|
||||||
import java.nio.channels.AsynchronousSocketChannel
|
|
||||||
import kotlin.coroutines.suspendCoroutine
|
|
||||||
|
|
||||||
actual fun NetworkAddress(host: String, port: Int): NetworkAddress =
|
|
||||||
JvmNetworkAddress(InetAddress.getByName(host), port)
|
|
||||||
|
|
||||||
actual fun acceptTcpDevice(port: Int): Flow<InetTransportDevice> {
|
|
||||||
return flow {
|
|
||||||
val socket = withContext(Dispatchers.IO) {
|
|
||||||
AsynchronousServerSocketChannel.open().also {
|
|
||||||
it.bind(InetSocketAddress(port))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
while (true) {
|
|
||||||
println("0 --- $port")
|
|
||||||
val connectedSocket = suspendCancellableCoroutine { continuation ->
|
|
||||||
continuation.invokeOnCancellation {
|
|
||||||
socket.close()
|
|
||||||
}
|
|
||||||
socket.accept(continuation, ContinuationHandler())
|
|
||||||
}
|
|
||||||
println("1 ---")
|
|
||||||
emit(asyncSocketToDevice(connectedSocket))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
@Suppress("unused")
|
|
||||||
suspend fun connectTcpDevice(host: String, port: Int) = connectTcpDevice(NetworkAddress(host,port))
|
|
||||||
actual suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice {
|
|
||||||
address as JvmNetworkAddress
|
|
||||||
val socket = withContext(Dispatchers.IO) {
|
|
||||||
AsynchronousSocketChannel.open()
|
|
||||||
}
|
|
||||||
suspendCoroutine { cont ->
|
|
||||||
socket.connect(address.socketAddress, cont, VoidCompletionHandler)
|
|
||||||
}
|
|
||||||
return asyncSocketToDevice(socket)
|
|
||||||
}
|
|
||||||
@@ -1,118 +0,0 @@
|
|||||||
package net.sergeych.kiloparsec.adapter
|
|
||||||
|
|
||||||
import kotlinx.coroutines.*
|
|
||||||
import kotlinx.coroutines.channels.BufferOverflow
|
|
||||||
import kotlinx.coroutines.channels.Channel
|
|
||||||
import net.sergeych.mp_logger.LogTag
|
|
||||||
import net.sergeych.mp_logger.exception
|
|
||||||
import net.sergeych.mp_logger.info
|
|
||||||
import net.sergeych.mp_logger.warning
|
|
||||||
import java.net.*
|
|
||||||
import java.util.concurrent.atomic.AtomicInteger
|
|
||||||
|
|
||||||
private val counter = AtomicInteger(0)
|
|
||||||
|
|
||||||
class JvmNetworkAddress(val inetAddress: InetAddress, override val port: Int) : NetworkAddress {
|
|
||||||
override val host: String by lazy { inetAddress.canonicalHostName }
|
|
||||||
override fun equals(other: Any?): Boolean {
|
|
||||||
if (this === other) return true
|
|
||||||
if (other !is JvmNetworkAddress) return false
|
|
||||||
|
|
||||||
if (inetAddress != other.inetAddress) return false
|
|
||||||
if (port != other.port) return false
|
|
||||||
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
|
|
||||||
val socketAddress: SocketAddress by lazy { InetSocketAddress(inetAddress,port) }
|
|
||||||
|
|
||||||
override fun hashCode(): Int {
|
|
||||||
var result = inetAddress.hashCode()
|
|
||||||
result = 31 * result + port
|
|
||||||
return result
|
|
||||||
}
|
|
||||||
|
|
||||||
override fun toString(): String = "$host:$port"
|
|
||||||
}
|
|
||||||
|
|
||||||
class UdpDatagram(override val message: UByteArray, val inetAddress: InetAddress, val port: Int) : Datagram {
|
|
||||||
|
|
||||||
override val address: NetworkAddress by lazy {
|
|
||||||
JvmNetworkAddress(inetAddress, port)
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
@OptIn(DelicateCoroutinesApi::class)
|
|
||||||
class UdpServer(val port: Int) :
|
|
||||||
DatagramConnector, LogTag("UDPS:${counter.incrementAndGet()}") {
|
|
||||||
private var isClosed = false
|
|
||||||
|
|
||||||
|
|
||||||
private val deferredSocket = CompletableDeferred<DatagramSocket>()
|
|
||||||
private var job: Job? = null
|
|
||||||
|
|
||||||
private suspend fun start() = try {
|
|
||||||
coroutineScope {
|
|
||||||
val socket = DatagramSocket(port)
|
|
||||||
val buffer = ByteArray(16384)
|
|
||||||
val packet = DatagramPacket(buffer, buffer.size)
|
|
||||||
deferredSocket.complete(socket)
|
|
||||||
while (isActive && !isClosed) {
|
|
||||||
try {
|
|
||||||
socket.receive(packet)
|
|
||||||
val data = packet.data.sliceArray(0..<packet.length)
|
|
||||||
val datagram = UdpDatagram(data.toUByteArray(), packet.address, packet.port)
|
|
||||||
if (!channel.trySend(datagram).isSuccess) {
|
|
||||||
warning { "packet lost!" }
|
|
||||||
// and we cause overflow that overwrites the oldest
|
|
||||||
channel.send(datagram)
|
|
||||||
}
|
|
||||||
} catch (e: Exception) {
|
|
||||||
if (!isClosed)
|
|
||||||
e.printStackTrace()
|
|
||||||
throw e
|
|
||||||
}
|
|
||||||
}
|
|
||||||
info { "closing socket and reception loop" }
|
|
||||||
|
|
||||||
}
|
|
||||||
} catch (_: CancellationException) {
|
|
||||||
info { "server is closed" }
|
|
||||||
} catch (t: Throwable) {
|
|
||||||
exception { "unexpected end of server" to t }
|
|
||||||
}
|
|
||||||
|
|
||||||
init {
|
|
||||||
job = GlobalScope.launch(Dispatchers.IO) {
|
|
||||||
start()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
override fun close() {
|
|
||||||
if (!isClosed) {
|
|
||||||
if (deferredSocket.isCompleted) {
|
|
||||||
runCatching {
|
|
||||||
deferredSocket.getCompleted().close()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
isClosed = true
|
|
||||||
job?.cancel(); job = null
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private val channel = Channel<Datagram>(2048, BufferOverflow.DROP_OLDEST)
|
|
||||||
override val incoming = channel
|
|
||||||
|
|
||||||
override suspend fun send(message: UByteArray, networkAddress: NetworkAddress) {
|
|
||||||
networkAddress as JvmNetworkAddress
|
|
||||||
withContext(Dispatchers.IO) {
|
|
||||||
val packet = DatagramPacket(
|
|
||||||
message.toByteArray(), message.size,
|
|
||||||
networkAddress.inetAddress, networkAddress.port
|
|
||||||
)
|
|
||||||
deferredSocket.await().send(packet)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -1,155 +0,0 @@
|
|||||||
package net.sergeych.kiloparsec.adapter
|
|
||||||
|
|
||||||
import kotlinx.coroutines.*
|
|
||||||
import kotlinx.coroutines.channels.Channel
|
|
||||||
import kotlinx.coroutines.channels.ClosedReceiveChannelException
|
|
||||||
import kotlinx.coroutines.flow.MutableStateFlow
|
|
||||||
import net.sergeych.crypto2.Contrail
|
|
||||||
import net.sergeych.crypto2.encodeVarUnsigned
|
|
||||||
import net.sergeych.crypto2.readVarUnsigned
|
|
||||||
import net.sergeych.kiloparsec.RemoteInterface
|
|
||||||
import net.sergeych.kiloparsec.Transport
|
|
||||||
import net.sergeych.mp_logger.LogTag
|
|
||||||
import net.sergeych.mp_logger.warning
|
|
||||||
import net.sergeych.mp_tools.globalLaunch
|
|
||||||
import net.sergeych.tools.waitFor
|
|
||||||
import java.net.InetSocketAddress
|
|
||||||
import java.net.StandardSocketOptions.TCP_NODELAY
|
|
||||||
import java.nio.ByteBuffer
|
|
||||||
import java.nio.channels.AsynchronousSocketChannel
|
|
||||||
import kotlin.coroutines.cancellation.CancellationException
|
|
||||||
import kotlin.coroutines.suspendCoroutine
|
|
||||||
|
|
||||||
private val log = LogTag("ASTD")
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Prepend block with its size, varint-encoded
|
|
||||||
*/
|
|
||||||
private fun encode(block: UByteArray): ByteArray {
|
|
||||||
val c = Contrail.create(block)
|
|
||||||
return (encodeVarUnsigned(c.size.toUInt()) + c).toByteArray()
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Convert asynchronous socket to a [Transport.Device] using non-blocking nio,
|
|
||||||
* in a coroutine-effective manner. Note that it runs coroutines to read/write
|
|
||||||
* 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): InetTransportDevice {
|
|
||||||
val deferredDevice = CompletableDeferred<InetTransportDevice>()
|
|
||||||
globalLaunch {
|
|
||||||
coroutineScope {
|
|
||||||
val sendQueueEmpty = MutableStateFlow(true)
|
|
||||||
val receiving = MutableStateFlow(false)
|
|
||||||
// We're in block mode, every block we send worth immediate sending, we do not
|
|
||||||
// send partial blocks, so:
|
|
||||||
socket.setOption(TCP_NODELAY, true)
|
|
||||||
|
|
||||||
// socket input is to be parsed for blocks, so we receive bytes
|
|
||||||
// and decode them to blocks
|
|
||||||
val input = Channel<UByte>(1024)
|
|
||||||
val inputBlocks = Channel<UByteArray>()
|
|
||||||
// output is blocks, so we sent transformed, framed blocks:
|
|
||||||
val outputBlocks = Channel<UByteArray>()
|
|
||||||
|
|
||||||
fun stop() {
|
|
||||||
kotlin.runCatching { inputBlocks.close(RemoteInterface.ClosedException()) }
|
|
||||||
kotlin.runCatching { outputBlocks.close() }
|
|
||||||
socket.close()
|
|
||||||
cancel()
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
// copy incoming data from the socket to input channel:
|
|
||||||
launch {
|
|
||||||
val data = ByteArray(1024)
|
|
||||||
val inb = ByteBuffer.wrap(data)
|
|
||||||
kotlin.runCatching {
|
|
||||||
while (isActive) {
|
|
||||||
inb.position(0)
|
|
||||||
val size: Int = suspendCoroutine { continuation ->
|
|
||||||
socket.read(inb, continuation, IntCompletionHandler)
|
|
||||||
}
|
|
||||||
if (size < 0) stop()
|
|
||||||
else {
|
|
||||||
// println("recvd:\n${data.sliceArray(0..<size).toDump()}\n------------------")
|
|
||||||
for (i in 0..<size) input.send(data[i].toUByte())
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// copy from output to socket:
|
|
||||||
launch {
|
|
||||||
try {
|
|
||||||
while (isActive) {
|
|
||||||
// wait for the first block to send
|
|
||||||
sendQueueEmpty.value = outputBlocks.isEmpty
|
|
||||||
var data = encode(outputBlocks.receive())
|
|
||||||
|
|
||||||
// now we're sending, so queue state is sending:
|
|
||||||
sendQueueEmpty.value = false
|
|
||||||
|
|
||||||
// if there are more, take them all (NO_DELAY optimization)
|
|
||||||
while (!outputBlocks.isEmpty)
|
|
||||||
data += encode(outputBlocks.receive())
|
|
||||||
|
|
||||||
// now send it all together:
|
|
||||||
val outBuff = ByteBuffer.wrap(data)
|
|
||||||
val cnt = suspendCoroutine { continuation ->
|
|
||||||
socket.write(outBuff, continuation, IntCompletionHandler)
|
|
||||||
}
|
|
||||||
// be sure it was all sent
|
|
||||||
if (outBuff.position() != data.size || cnt != data.size) {
|
|
||||||
throw RuntimeException("unexpected partial write")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// in the case of just breaking out of the loop:
|
|
||||||
sendQueueEmpty.value = true
|
|
||||||
} catch (_: ClosedReceiveChannelException) {
|
|
||||||
stop()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// transport device copes with blocks:
|
|
||||||
// decode blocks from a byte channel read from the socket:
|
|
||||||
launch {
|
|
||||||
try {
|
|
||||||
while (isActive) {
|
|
||||||
receiving.value = !input.isEmpty
|
|
||||||
val size = readVarUnsigned(input)
|
|
||||||
receiving.value = true
|
|
||||||
if (size == 0u) log.warning { "zero size block is ignored!" }
|
|
||||||
else {
|
|
||||||
val block = UByteArray(size.toInt())
|
|
||||||
for (i in 0..<size.toInt()) {
|
|
||||||
block[i] = input.receive()
|
|
||||||
}
|
|
||||||
Contrail.unpack(block)?.let { inputBlocks.send(it) }
|
|
||||||
?: log.warning { "skipping bad block ${block.size} bytes" }
|
|
||||||
}
|
|
||||||
}
|
|
||||||
} catch (_: CancellationException) {
|
|
||||||
} catch (_: ClosedReceiveChannelException) {
|
|
||||||
stop()
|
|
||||||
}
|
|
||||||
receiving.value = false
|
|
||||||
}
|
|
||||||
|
|
||||||
val addr = socket.remoteAddress as InetSocketAddress
|
|
||||||
deferredDevice.complete(
|
|
||||||
InetTransportDevice(inputBlocks, outputBlocks, JvmNetworkAddress(addr.address, addr.port), {
|
|
||||||
yield()
|
|
||||||
// wait until all received data are parsed, but not too long
|
|
||||||
withTimeoutOrNull(500) {
|
|
||||||
receiving.waitFor { !it }
|
|
||||||
}
|
|
||||||
// then stop it
|
|
||||||
stop()
|
|
||||||
})
|
|
||||||
)
|
|
||||||
}
|
|
||||||
globalLaunch { socket.close() }
|
|
||||||
}
|
|
||||||
return deferredDevice.await()
|
|
||||||
}
|
|
||||||
@@ -6,18 +6,17 @@ import io.ktor.server.netty.*
|
|||||||
import kotlinx.coroutines.launch
|
import kotlinx.coroutines.launch
|
||||||
import kotlinx.coroutines.test.runTest
|
import kotlinx.coroutines.test.runTest
|
||||||
import net.sergeych.crypto2.initCrypto
|
import net.sergeych.crypto2.initCrypto
|
||||||
import net.sergeych.kiloparsec.adapter.acceptTcpDevice
|
|
||||||
import net.sergeych.kiloparsec.adapter.connectTcpDevice
|
|
||||||
import net.sergeych.kiloparsec.adapter.setupWebsocketServer
|
import net.sergeych.kiloparsec.adapter.setupWebsocketServer
|
||||||
import net.sergeych.kiloparsec.adapter.websocketClient
|
import net.sergeych.kiloparsec.adapter.websocketClient
|
||||||
import net.sergeych.mp_logger.Log
|
import net.sergeych.mp_logger.Log
|
||||||
import java.net.InetAddress
|
import java.net.InetAddress
|
||||||
import kotlin.test.*
|
import kotlin.test.Test
|
||||||
|
import kotlin.test.assertEquals
|
||||||
|
import kotlin.test.assertFalse
|
||||||
|
import kotlin.test.assertTrue
|
||||||
|
|
||||||
class ClientTest {
|
class ClientTest {
|
||||||
|
|
||||||
class TestException : Exception("test1")
|
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
fun testAddresses() {
|
fun testAddresses() {
|
||||||
println(InetAddress.getLocalHost())
|
println(InetAddress.getLocalHost())
|
||||||
@@ -25,59 +24,7 @@ class ClientTest {
|
|||||||
println(InetAddress.getByName("127.0.0.1"))
|
println(InetAddress.getByName("127.0.0.1"))
|
||||||
println(InetAddress.getByName("mail.ru"))
|
println(InetAddress.getByName("mail.ru"))
|
||||||
}
|
}
|
||||||
@Test
|
|
||||||
fun testClient() = runTest {
|
|
||||||
initCrypto()
|
|
||||||
Log.connectConsole(Log.Level.DEBUG)
|
|
||||||
data class Session(
|
|
||||||
var data: String
|
|
||||||
)
|
|
||||||
|
|
||||||
val cmdSave by command<String,Unit>()
|
|
||||||
val cmdLoad by command<Unit,String>()
|
|
||||||
val cmdDrop by command<Unit,Unit>()
|
|
||||||
val cmdException by command<Unit,Unit>()
|
|
||||||
|
|
||||||
val cli = KiloInterface<Session>().apply {
|
|
||||||
registerError { TestException() }
|
|
||||||
onConnected { session.data = "start" }
|
|
||||||
on(cmdSave) { session.data = it }
|
|
||||||
on(cmdLoad) {
|
|
||||||
session.data
|
|
||||||
}
|
|
||||||
on(cmdException) {
|
|
||||||
throw TestException()
|
|
||||||
}
|
|
||||||
on(cmdDrop) {
|
|
||||||
throw LocalInterface.BreakConnectionException()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
val server = KiloServer(cli, acceptTcpDevice(17101)) {
|
|
||||||
Session("unknown")
|
|
||||||
}
|
|
||||||
|
|
||||||
val client = KiloClient<Unit>() {
|
|
||||||
addErrors(cli)
|
|
||||||
connect { connectTcpDevice("localhost:17101") }
|
|
||||||
}
|
|
||||||
println(client.call(cmdLoad))
|
|
||||||
|
|
||||||
assertEquals("start", client.call(cmdLoad))
|
|
||||||
client.call(cmdSave, "foobar")
|
|
||||||
assertEquals("foobar", client.call(cmdLoad))
|
|
||||||
|
|
||||||
val res = kotlin.runCatching { client.call(cmdException) }
|
|
||||||
println(res.exceptionOrNull())
|
|
||||||
assertIs<TestException>(res.exceptionOrNull())
|
|
||||||
assertEquals("foobar", client.call(cmdLoad))
|
|
||||||
|
|
||||||
assertThrows<RemoteInterface.ClosedException> { client.call(cmdDrop) }
|
|
||||||
|
|
||||||
// reconnect?
|
|
||||||
assertEquals("start", client.call(cmdLoad))
|
|
||||||
|
|
||||||
server.close()
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
fun webSocketTest() = runTest {
|
fun webSocketTest() = runTest {
|
||||||
|
|||||||
@@ -1,100 +0,0 @@
|
|||||||
package net.sergeych.kiloparsec.adapters
|
|
||||||
|
|
||||||
import kotlinx.coroutines.*
|
|
||||||
import kotlinx.coroutines.test.runTest
|
|
||||||
import net.sergeych.crypto2.initCrypto
|
|
||||||
import net.sergeych.kiloparsec.adapter.UdpServer
|
|
||||||
import net.sergeych.kiloparsec.adapter.acceptTcpDevice
|
|
||||||
import net.sergeych.kiloparsec.adapter.connectTcpDevice
|
|
||||||
import net.sergeych.kiloparsec.adapter.toNetworkAddress
|
|
||||||
import net.sergeych.kiloparsec.decodeFromUByteArray
|
|
||||||
import net.sergeych.kiloparsec.encodeToUByteArray
|
|
||||||
import net.sergeych.mp_logger.Log
|
|
||||||
import net.sergeych.mp_logger.LogTag
|
|
||||||
import net.sergeych.synctools.ProtectedOp
|
|
||||||
import net.sergeych.synctools.invoke
|
|
||||||
import kotlin.test.Test
|
|
||||||
import kotlin.test.assertContains
|
|
||||||
import kotlin.test.assertEquals
|
|
||||||
|
|
||||||
class NetworkTest {
|
|
||||||
|
|
||||||
@Test
|
|
||||||
fun udpProviderTest() = runTest {
|
|
||||||
Log.connectConsole(Log.Level.DEBUG)
|
|
||||||
val s1 = UdpServer(17120)
|
|
||||||
val s2 = UdpServer(17121)
|
|
||||||
s1.send("Hello".encodeToUByteArray(), "localhost", 17121)
|
|
||||||
val d1 = s2.incoming.receive()
|
|
||||||
assertEquals(d1.address.port, 17120)
|
|
||||||
assertEquals("Hello", d1.message.toByteArray().decodeToString())
|
|
||||||
s1.send("world".encodeToUByteArray(), d1.address)
|
|
||||||
assertEquals("world", s1.incoming.receive().message.toByteArray().decodeToString())
|
|
||||||
// println("s1: ${s1.bindAddress()}")
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
||||||
@Test
|
|
||||||
fun tcpAsyncConnectionTest() = runTest {
|
|
||||||
initCrypto()
|
|
||||||
Log.connectConsole(Log.Level.DEBUG)
|
|
||||||
|
|
||||||
coroutineScope {
|
|
||||||
val serverFlow = acceptTcpDevice(17171)
|
|
||||||
val op = ProtectedOp()
|
|
||||||
var pills = setOf<String>()
|
|
||||||
val j = launch {
|
|
||||||
println("serf")
|
|
||||||
serverFlow.collect { device ->
|
|
||||||
println("serf 0")
|
|
||||||
launch {
|
|
||||||
println("serf 1")
|
|
||||||
device.output.send("Hello, world!".encodeToUByteArray())
|
|
||||||
device.output.send("Great".encodeToUByteArray())
|
|
||||||
while (true) {
|
|
||||||
val x = device.input.receive().decodeFromUByteArray()
|
|
||||||
if (x.startsWith("die")) {
|
|
||||||
op.invoke {
|
|
||||||
pills += x
|
|
||||||
}
|
|
||||||
cancel()
|
|
||||||
}
|
|
||||||
else
|
|
||||||
println("ignoring unexpected input: $x")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
yield()
|
|
||||||
run {
|
|
||||||
try {
|
|
||||||
println("pre-con")
|
|
||||||
val s = connectTcpDevice("127.0.1.1:17171".toNetworkAddress())
|
|
||||||
println("2")
|
|
||||||
assertEquals("Hello, world!", s.input.receive().decodeFromUByteArray())
|
|
||||||
assertEquals("Great", s.input.receive().decodeFromUByteArray())
|
|
||||||
s.output.send("Goodbye".encodeToUByteArray())
|
|
||||||
s.output.send("die1".encodeToUByteArray())
|
|
||||||
s.close()
|
|
||||||
}
|
|
||||||
catch(t: Throwable) {
|
|
||||||
t.printStackTrace()
|
|
||||||
throw t
|
|
||||||
}
|
|
||||||
}
|
|
||||||
println("pre-con2")
|
|
||||||
val s1 = connectTcpDevice("127.0.1.1:17171".toNetworkAddress())
|
|
||||||
assertEquals("Hello, world!", s1.input.receive().decodeFromUByteArray())
|
|
||||||
assertEquals("Great", s1.input.receive().decodeFromUByteArray())
|
|
||||||
s1.output.send("die2".encodeToUByteArray())
|
|
||||||
s1.close()
|
|
||||||
|
|
||||||
// check that channels were flushed prior to closed:
|
|
||||||
assertContains(pills, "die1")
|
|
||||||
assertContains(pills, "die2")
|
|
||||||
|
|
||||||
// Check that server jobs are closed
|
|
||||||
j.cancelAndJoin()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -2,42 +2,209 @@ package net.sergeych.kiloparsec.adapter
|
|||||||
|
|
||||||
import io.ktor.network.selector.*
|
import io.ktor.network.selector.*
|
||||||
import io.ktor.network.sockets.*
|
import io.ktor.network.sockets.*
|
||||||
import io.ktor.util.network.*
|
|
||||||
import io.ktor.utils.io.*
|
import io.ktor.utils.io.*
|
||||||
import io.ktor.utils.io.core.*
|
|
||||||
import kotlinx.coroutines.*
|
import kotlinx.coroutines.*
|
||||||
|
import kotlinx.coroutines.CancellationException
|
||||||
import kotlinx.coroutines.channels.Channel
|
import kotlinx.coroutines.channels.Channel
|
||||||
|
import kotlinx.coroutines.channels.ClosedReceiveChannelException
|
||||||
import kotlinx.coroutines.flow.Flow
|
import kotlinx.coroutines.flow.Flow
|
||||||
import kotlinx.coroutines.flow.flow
|
import kotlinx.coroutines.flow.flow
|
||||||
|
import kotlinx.coroutines.sync.Mutex
|
||||||
|
import kotlinx.coroutines.sync.withLock
|
||||||
|
import kotlinx.datetime.Clock
|
||||||
import net.sergeych.kiloparsec.AsyncVarint
|
import net.sergeych.kiloparsec.AsyncVarint
|
||||||
import net.sergeych.mp_logger.LogTag
|
import net.sergeych.kiloparsec.KiloClient
|
||||||
import kotlin.io.use
|
import net.sergeych.kiloparsec.KiloServer
|
||||||
|
import net.sergeych.kiloparsec.LocalInterface
|
||||||
|
import net.sergeych.mp_logger.*
|
||||||
|
import net.sergeych.mp_tools.globalLaunch
|
||||||
|
import net.sergeych.tools.AtomicCounter
|
||||||
|
import net.sergeych.tools.AtomicValue
|
||||||
|
import kotlin.time.Duration.Companion.seconds
|
||||||
|
|
||||||
//class SocketNetworkAddress(override val host: String, override val port: Int) : NetworkAddress
|
private val logCounter = AtomicCounter(0)
|
||||||
//
|
|
||||||
//actual fun NetworkAddress(host: String, port: Int): NetworkAddress = SocketNetworkAddress(host, port)
|
class ProtocolException(text: String, cause: Throwable? = null) : RuntimeException(text, cause)
|
||||||
//
|
|
||||||
//fun acceptTcpSocketDevice(port: Int): Flow<InetTransportDevice> {
|
const val MAX_TCP_BLOCK_SIZE = 16776216
|
||||||
// val selectorManager = SelectorManager(Dispatchers.IO)
|
val PING_INACTIVITY_TIME = 30.seconds
|
||||||
// val serverSocket = aSocket(selectorManager).tcp().bind("127.0.0.1", port)
|
|
||||||
// val log = LogTag("TCPS${}")
|
/**
|
||||||
// return flow {
|
* Listen for incoming TCP/IP connections on all local interfaces and the specified [port]
|
||||||
// serverSocket.accept().use { sock ->
|
* anc create flow of [InetTransportDevice] suitable for [KiloClient].
|
||||||
// val closed = CompletableDeferred<Boolean>()
|
*/
|
||||||
// val scope = coroutineScope {
|
fun acceptTcpDevice(port: Int): Flow<InetTransportDevice> {
|
||||||
// val networkAddress = sock.remoteAddress.toJavaAddress().let { NetworkAddress(it.hostname, it.port) }
|
val selectorManager = SelectorManager(Dispatchers.IO)
|
||||||
// val inputBlocks = Channel<UByteArray>(4096)
|
val serverSocket = aSocket(selectorManager).tcp().bind("127.0.0.1", port)
|
||||||
// sock.launch {
|
return flow {
|
||||||
// val sockInput = sock.openReadChannel()
|
while (true) {
|
||||||
// while (isActive && sock.isActive) {
|
serverSocket.accept().let { sock ->
|
||||||
// try {
|
emit(inetTransportDevice(sock, "srv"))
|
||||||
// val size = AsyncVarint.decodeUnsigned(sockInput)
|
}
|
||||||
// } catch (e: Exception) {
|
}
|
||||||
//
|
}
|
||||||
// }
|
}
|
||||||
// }
|
|
||||||
// }
|
suspend fun connectTcpDevice(address: String) = connectTcpDevice(address.toNetworkAddress())
|
||||||
// }
|
|
||||||
// }
|
/**
|
||||||
// }
|
* Connect to the TCP/IP server (see [KiloServer]) at the specified address and provide th compatible
|
||||||
//}
|
* [InetTransportDevice] to use with [KiloClient].
|
||||||
|
*/
|
||||||
|
suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice {
|
||||||
|
val selectorManager = SelectorManager(Dispatchers.IO)
|
||||||
|
val socket = aSocket(selectorManager).tcp().connect(address.host, address.port)
|
||||||
|
return inetTransportDevice(socket)
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Parse `host:port` string into the [NetworkAddress]
|
||||||
|
*/
|
||||||
|
fun String.toNetworkAddress(): NetworkAddress {
|
||||||
|
val (host, port) = this.split(":").map { it.trim() }
|
||||||
|
return NetworkAddress(host, port.toInt())
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun inetTransportDevice(
|
||||||
|
sock: Socket,
|
||||||
|
suffix: String = "cli",
|
||||||
|
): InetTransportDevice {
|
||||||
|
val networkAddress = (sock.remoteAddress as InetSocketAddress).let { NetworkAddress(it.hostname, it.port) }
|
||||||
|
val inputBlocks = Channel<UByteArray>(4096)
|
||||||
|
val outputBlocks = Channel<UByteArray>(4096)
|
||||||
|
|
||||||
|
val log = LogTag("TCPT${logCounter.incrementAndGet()}:$suffix:$networkAddress")
|
||||||
|
val job = AtomicValue<Job?>(null)
|
||||||
|
|
||||||
|
val sockOutput = sock.openWriteChannel()
|
||||||
|
val sockInput = runCatching { sock.openReadChannel() }.getOrElse {
|
||||||
|
log.warning { "failed to open read channel $it" }
|
||||||
|
throw IllegalStateException("failed to open read channel")
|
||||||
|
}
|
||||||
|
|
||||||
|
fun stop() {
|
||||||
|
job.mutate {
|
||||||
|
if ( it != null ) {
|
||||||
|
log.debug { "stopping" }
|
||||||
|
runCatching { inputBlocks.close() }
|
||||||
|
runCatching { outputBlocks.close() }
|
||||||
|
// The problem: on mac platofrms closing the socket does not close its input
|
||||||
|
// and output channels!
|
||||||
|
runCatching { sockInput.cancel() }
|
||||||
|
runCatching { sockOutput.close() }
|
||||||
|
if (!sock.isClosed)
|
||||||
|
runCatching {
|
||||||
|
log.debug { "closing socket by stop" }
|
||||||
|
sock.close()
|
||||||
|
}
|
||||||
|
else
|
||||||
|
log.debug { "socket is already closed when stop is called" }
|
||||||
|
it.cancel()
|
||||||
|
log.debug { "implementation job cancel called" }
|
||||||
|
} else
|
||||||
|
log.debug { "already stopped" }
|
||||||
|
null
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
var lastActiveAt = Clock.System.now()
|
||||||
|
job.value = globalLaunch {
|
||||||
|
launch {
|
||||||
|
|
||||||
|
log.debug { "opening read channel" }
|
||||||
|
|
||||||
|
while (isActive && sock.isActive) {
|
||||||
|
try {
|
||||||
|
val size = AsyncVarint.decodeUnsigned(sockInput).toInt()
|
||||||
|
if (size > MAX_TCP_BLOCK_SIZE) // 16M is a max command block
|
||||||
|
throw ProtocolException("Illegal block size: $size should be < $MAX_TCP_BLOCK_SIZE")
|
||||||
|
val data = ByteArray(size)
|
||||||
|
if (size == 0) {
|
||||||
|
log.debug { "ping received" }
|
||||||
|
lastActiveAt = Clock.System.now()
|
||||||
|
} else {
|
||||||
|
sockInput.readFully(data, 0, size)
|
||||||
|
inputBlocks.send(data.toUByteArray())
|
||||||
|
}
|
||||||
|
} catch (e: ClosedReceiveChannelException) {
|
||||||
|
log.error { "closed receive channel " }
|
||||||
|
stop()
|
||||||
|
break
|
||||||
|
} catch (_: CancellationException) {
|
||||||
|
log.error { "cancellation exception " }
|
||||||
|
break
|
||||||
|
} catch (e: Exception) {
|
||||||
|
log.exception { "unexpected exception in TCP socket read" to e }
|
||||||
|
stop()
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
launch {
|
||||||
|
val outAccess = Mutex()
|
||||||
|
var lastSentAt = Clock.System.now()
|
||||||
|
launch {
|
||||||
|
while (isActive && sock.isActive) {
|
||||||
|
delay(500)
|
||||||
|
val activityTime = if(lastSentAt > lastActiveAt) lastSentAt else lastActiveAt
|
||||||
|
if (Clock.System.now() - activityTime > PING_INACTIVITY_TIME) {
|
||||||
|
log.debug { "pinging for inactivity" }
|
||||||
|
val repeat = outAccess.withLock {
|
||||||
|
try {
|
||||||
|
sockOutput.writeByte(0)
|
||||||
|
sockOutput.flush()
|
||||||
|
lastSentAt = Clock.System.now()
|
||||||
|
true
|
||||||
|
} catch (e: ClosedReceiveChannelException) {
|
||||||
|
e.printStackTrace()
|
||||||
|
stop()
|
||||||
|
false
|
||||||
|
} catch (_: CancellationException) {
|
||||||
|
false
|
||||||
|
} catch (e: Throwable) {
|
||||||
|
e.printStackTrace()
|
||||||
|
stop()
|
||||||
|
false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (!repeat) break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
while (isActive && sock.isActive) {
|
||||||
|
try {
|
||||||
|
val block = outputBlocks.receive()
|
||||||
|
outAccess.withLock {
|
||||||
|
AsyncVarint.encodeUnsigned(block.size.toULong(), sockOutput)
|
||||||
|
sockOutput.writeFully(block.toByteArray(), 0, block.size)
|
||||||
|
sockOutput.flush()
|
||||||
|
lastSentAt = Clock.System.now()
|
||||||
|
}
|
||||||
|
} catch (_: CancellationException) {
|
||||||
|
log.debug { "cancellation exception on output" }
|
||||||
|
stop()
|
||||||
|
break
|
||||||
|
} catch (_: LocalInterface.BreakConnectionException) {
|
||||||
|
log.debug { "requested connection break" }
|
||||||
|
stop()
|
||||||
|
break
|
||||||
|
} catch (_: ClosedReceiveChannelException) {
|
||||||
|
log.debug { "receive block channel closed, closing the socket" }
|
||||||
|
stop()
|
||||||
|
break
|
||||||
|
} catch (e: Exception) {
|
||||||
|
log.exception { "unexpected exception. closing." to e }
|
||||||
|
stop()
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
val device = InetTransportDevice(inputBlocks, outputBlocks, networkAddress, {
|
||||||
|
stop()
|
||||||
|
})
|
||||||
|
log.debug { "Transport ready" }
|
||||||
|
return device
|
||||||
|
}
|
||||||
@@ -0,0 +1,69 @@
|
|||||||
|
import kotlinx.coroutines.test.runTest
|
||||||
|
import net.sergeych.crypto2.initCrypto
|
||||||
|
import net.sergeych.kiloparsec.*
|
||||||
|
import net.sergeych.kiloparsec.adapter.acceptTcpDevice
|
||||||
|
import net.sergeych.kiloparsec.adapter.connectTcpDevice
|
||||||
|
import kotlin.random.Random
|
||||||
|
import kotlin.test.Test
|
||||||
|
import kotlin.test.assertEquals
|
||||||
|
import kotlin.test.assertIs
|
||||||
|
|
||||||
|
class TcpTest {
|
||||||
|
class TestException : Exception("test1")
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun tcpTest() = runTest {
|
||||||
|
initCrypto()
|
||||||
|
// Log.connectConsole(Log.Level.DEBUG)
|
||||||
|
data class Session(
|
||||||
|
var data: String,
|
||||||
|
)
|
||||||
|
|
||||||
|
val port = 27170 + Random.nextInt(1, 200)
|
||||||
|
|
||||||
|
val cmdSave by command<String, Unit>()
|
||||||
|
val cmdLoad by command<Unit, String>()
|
||||||
|
val cmdDrop by command<Unit, Unit>()
|
||||||
|
val cmdException by command<Unit, Unit>()
|
||||||
|
|
||||||
|
val cli = KiloInterface<Session>().apply {
|
||||||
|
registerError { TestException() }
|
||||||
|
onConnected { session.data = "start" }
|
||||||
|
on(cmdSave) { session.data = it }
|
||||||
|
on(cmdLoad) {
|
||||||
|
session.data
|
||||||
|
}
|
||||||
|
on(cmdException) {
|
||||||
|
throw TestException()
|
||||||
|
}
|
||||||
|
on(cmdDrop) {
|
||||||
|
throw LocalInterface.BreakConnectionException()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
val server = KiloServer(cli, acceptTcpDevice(port)) {
|
||||||
|
Session("unknown")
|
||||||
|
}
|
||||||
|
|
||||||
|
val client = KiloClient<Unit>() {
|
||||||
|
addErrors(cli)
|
||||||
|
// TODO: add register error variant
|
||||||
|
connect { connectTcpDevice("localhost:$port") }
|
||||||
|
}
|
||||||
|
|
||||||
|
assertEquals("start", client.call(cmdLoad))
|
||||||
|
|
||||||
|
client.call(cmdSave, "foobar")
|
||||||
|
assertEquals("foobar", client.call(cmdLoad))
|
||||||
|
|
||||||
|
val res = kotlin.runCatching { client.call(cmdException) }
|
||||||
|
assertIs<TestException>(res.exceptionOrNull())
|
||||||
|
assertEquals("foobar", client.call(cmdLoad))
|
||||||
|
|
||||||
|
assertThrows<RemoteInterface.ClosedException> { client.call(cmdDrop) }
|
||||||
|
|
||||||
|
// reconnect?
|
||||||
|
assertEquals("start", client.call(cmdLoad))
|
||||||
|
|
||||||
|
server.close()
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,2 +0,0 @@
|
|||||||
package net.sergeych.kiloparsec.adapter
|
|
||||||
|
|
||||||
@@ -1,15 +0,0 @@
|
|||||||
package net.sergeych.kiloparsec.adapter
|
|
||||||
|
|
||||||
import kotlinx.coroutines.flow.Flow
|
|
||||||
|
|
||||||
actual fun NetworkAddress(host: String, port: Int): NetworkAddress {
|
|
||||||
TODO("Not yet implemented")
|
|
||||||
}
|
|
||||||
|
|
||||||
actual fun acceptTcpDevice(port: Int): Flow<InetTransportDevice> {
|
|
||||||
TODO("Not yet implemented")
|
|
||||||
}
|
|
||||||
|
|
||||||
actual suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice {
|
|
||||||
TODO("Not yet implemented")
|
|
||||||
}
|
|
||||||
Reference in new issue
Block a user