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

Compare commits

base: SergeychWorks:master
Branches Tags
SergeychWorks:master
SergeychWorks:experimental_udp
SergeychWorks:v0.6.3
SergeychWorks:0.4.1
SergeychWorks:0.3.3
SergeychWorks:0.3.1
SergeychWorks:0.2.6
SergeychWorks:v0.2.4
SergeychWorks:v0.2.3
..
compare: SergeychWorks:0.3.3
Branches Tags
SergeychWorks:master
SergeychWorks:experimental_udp
SergeychWorks:v0.6.3
SergeychWorks:0.4.1
SergeychWorks:0.3.3
SergeychWorks:0.3.1
SergeychWorks:0.2.6
SergeychWorks:v0.2.4
SergeychWorks:v0.2.3

No commits in common. "master" and "0.3.3" have entirely different histories.
master ... 0.3.3

52 changed files with 276 additions and 1201 deletions
Show Stats Expand all files Collapse all files

3
.gitignore vendored
View File

@ -5,7 +5,6 @@ build/
!**/src/test/**/build/ !**/src/test/**/build/
### IntelliJ IDEA ### ### IntelliJ IDEA ###
.idea
.idea/modules.xml .idea/modules.xml
.idea/jarRepositories.xml .idea/jarRepositories.xml
.idea/compiler.xml .idea/compiler.xml
@ -44,5 +43,3 @@ out/
# More # More
.kotlin .kotlin
/.idea/workspace.xml /.idea/workspace.xml
/.gigaide/gigaide.properties
local.properties

8
.idea/artifacts/kiloparsec_js_0_4_1.xml generated
View File

@ -1,8 +0,0 @@
<component name="ArtifactManager">
<artifact type="jar" name="kiloparsec-js-0.4.1">
<output-path>$PROJECT_DIR$/build/libs</output-path>
<root id="archive" name="kiloparsec-js-0.4.1.jar">
<element id="module-output" name="kiloparsec.jsMain" />
</root>
</artifact>
</component>

8
.idea/artifacts/kiloparsec_js_0_4_3.xml generated
View File

@ -1,8 +0,0 @@
<component name="ArtifactManager">
<artifact type="jar" name="kiloparsec-js-0.4.3">
<output-path>$PROJECT_DIR$/build/libs</output-path>
<root id="archive" name="kiloparsec-js-0.4.3.jar">
<element id="module-output" name="kiloparsec.jsMain" />
</root>
</artifact>
</component>

8
.idea/artifacts/kiloparsec_jvm_0_4_1.xml generated
View File

@ -1,8 +0,0 @@
<component name="ArtifactManager">
<artifact type="jar" name="kiloparsec-jvm-0.4.1">
<output-path>$PROJECT_DIR$/build/libs</output-path>
<root id="archive" name="kiloparsec-jvm-0.4.1.jar">
<element id="module-output" name="kiloparsec.jvmMain" />
</root>
</artifact>
</component>

8
.idea/artifacts/kiloparsec_jvm_0_4_3.xml generated
View File

@ -1,8 +0,0 @@
<component name="ArtifactManager">
<artifact type="jar" name="kiloparsec-jvm-0.4.3">
<output-path>$PROJECT_DIR$/build/libs</output-path>
<root id="archive" name="kiloparsec-jvm-0.4.3.jar">
<element id="module-output" name="kiloparsec.jvmMain" />
</root>
</artifact>
</component>

24
.idea/codeStyles/Project.xml generated
View File

@ -1,29 +1,5 @@
<component name="ProjectCodeStyleConfiguration"> <component name="ProjectCodeStyleConfiguration">
<code_scheme name="Project" version="173"> <code_scheme name="Project" version="173">
<DBN-PSQL>
<case-options enabled="true">
<option name="KEYWORD_CASE" value="lower" />
<option name="FUNCTION_CASE" value="lower" />
<option name="PARAMETER_CASE" value="lower" />
<option name="DATATYPE_CASE" value="lower" />
<option name="OBJECT_CASE" value="preserve" />
</case-options>
<formatting-settings enabled="false" />
</DBN-PSQL>
<DBN-SQL>
<case-options enabled="true">
<option name="KEYWORD_CASE" value="lower" />
<option name="FUNCTION_CASE" value="lower" />
<option name="PARAMETER_CASE" value="lower" />
<option name="DATATYPE_CASE" value="lower" />
<option name="OBJECT_CASE" value="preserve" />
</case-options>
<formatting-settings enabled="false">
<option name="STATEMENT_SPACING" value="one_line" />
<option name="CLAUSE_CHOP_DOWN" value="chop_down_if_statement_long" />
<option name="ITERATION_ELEMENTS_WRAPPING" value="chop_down_if_not_single" />
</formatting-settings>
</DBN-SQL>
<ScalaCodeStyleSettings> <ScalaCodeStyleSettings>
<option name="MULTILINE_STRING_CLOSING_QUOTES_ON_NEW_LINE" value="true" /> <option name="MULTILINE_STRING_CLOSING_QUOTES_ON_NEW_LINE" value="true" />
</ScalaCodeStyleSettings> </ScalaCodeStyleSettings>

3
.idea/misc.xml generated
View File

@ -1,10 +1,9 @@
<?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">
<file type="web" url="file://$PROJECT_DIR$" /> <file type="web" url="file://$PROJECT_DIR$" />
</component> </component>
<component name="ProjectRootManager" version="2" languageLevel="JDK_21" project-jdk-name="corretto-17" project-jdk-type="JavaSDK"> <component name="ProjectRootManager" version="2" languageLevel="JDK_17" default="true" project-jdk-name="17 (5)" project-jdk-type="JavaSDK">
<output url="file://$PROJECT_DIR$/out" /> <output url="file://$PROJECT_DIR$/out" />
</component> </component>
</project> </project>

156
README.md
View File

@ -1,42 +1,27 @@
# Kiloparsec # Kiloparsec
> 0.7.0-SNAPSHOT is recommended for newer builds; due to extremely poor Kotlin time migration architecture, on Apple targets it is the only working version. __Recommended version is `0.3.3`: to keep the code compatible with current and further versions we
ask to upgrade to `0.3.2` at least.__ Starting from this version some pacakage names are changed for
__Recommended version is `0.6.8`: to keep the code compatible with current and further versions we
ask to upgrade to `0.4.2` at least.__ Starting from this version some package names are changed for
better clarity and fast UDP endpoints are added. better clarity and fast UDP endpoints are added.
The new generation of __PARanoid SECurity__ protocol, advanced, faster, more secure. It also allows connecting any " 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 block device" transport to the same local interface. Out if the box it
provides the following transports: provides the following transports:
| name | JVM | JS | wasmJS | native | | name | JVM | JS | native |
|-------------------|:----:|:---:|:-------:|:------:| |----------------|--------|----|--------|
| TCP/IP server | ✓ | | | ✓ | | TCP/IP server | ✓ | | 0.2.6+ |
| TCP/IP client | ✓ | | | ✓ | | TCP/IP client | ✓ | | 0.2.6+ |
| UDP server | ✓ | | | ✓ | | UDP server | 0.3.2+ | | 0.3.2+ |
| UDP client | ✓ | | | ✓ | | UDP client | 0.3.2+ | | 0.3.2+ |
| Websockets server | ✓ | | | | | Websock server | ✓ | | |
| Websockets client | ✓ | ✓ | ✓ | ✓ | | Websock client | ✓ | ✓ | ✓ |
### Note on version compatibility ### Note on version compatibility
We recommend using `0.6.12` Protocols >= 0.3.0 are not binary compatible with previous version due to more compact binary
Since version 0.6.9 websocket protocol supports both text and binary frames; old clients are backward compatible with
mew servers, but new clients only can work with older servers only in default binary frame mode. Upgrade also your
servers to get better websocket compatibility[^1].
Version 0.5.1 could be backward incompatible due to the upgrade of the crypto2.
Protocols >= 0.3.0 are not binary compatible with the previous version due to a more compact binary
format. The format from 0.3.0 onwards is supposed to keep compatible. format. The format from 0.3.0 onwards is supposed to keep compatible.
#### ID calculation algorithm is changed since 0.4.1
We recommend to upgrade to 0.4+ ASAP as public/shared key id derivation method was changed for even higher security.
### Supported native targets ### Supported native targets
- iosArm64, iosX64 - iosArm64, iosX64
@ -45,35 +30,26 @@ We recommend to upgrade to 0.4+ ASAP as public/shared key id derivation method w
### Non-native targets ### Non-native targets
- JS (browser and Node.js) - JS (browser and nodeJS)
- JVM (android, macOS, windows, linux, everywhere where JRE is installed) - JVM (android, macos, windows, linx, everywhere where JRE is installed)
## TCP/IP and UDP transports ## TCP/IP transport
These are the fastest based on async socket implementation of a ktor client. They work everywhere but JS target as It is the fastest based on async socket implementation of ktor client. It works everywhere but JS target as
there are currently no widely adopted sockets for browser JavaScript. there is currently no widely adopted sockets for browser javascript.
While UDP is faster than TCP/IP, it is less reliable, especially with commands and return values that serialize to more ## Websock server
than 240 bytes approx, and has no retransmission facilities (use TCP!). UDP, though, shines when all you need is
to [push](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec/-remote-interface/push.html) with
little or no data in it.
## Websockets server While it is much slower than pure TCP, it is still faster than any http-based transport. It uses binary frames based on
While it is much slower than TCP or UDP, it is still faster than any http-based API; 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 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. 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 websockets and tcp servers in real 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. applications to get easy access from anywhere.
## Websocket client
It is slower than TCP or UDP, but it works on literally all platforms. See the sample below.
# Usage # Usage
The library should be used as a maven dependency, not as source. The library should be used as maven dependency, not as source.
## Adding dependency ## Adding dependency
@ -94,12 +70,12 @@ 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.6.8") api("net.sergeych:kiloparsec:0.3.2")
} }
} }
``` ```
## Create a shared interface for your server and the client ## Create shared interface for your server and the client
It could be a multiplatform library that exports it or just a shared or copied source file declaring structures It could be a multiplatform library that exports it or just a shared or copied source file declaring structures
and functions available, like: and functions available, like:
@ -123,7 +99,7 @@ val cmdPushClient by command<String, Unit>()
## Call it from the client: ## Call it from the client:
Remember, we need to implement client interface `cmdPushClient` in our example, so we need to provide Remember, we need to implement client interface `cmdPushClient` in our example, so we need to provide
local interface too: local interace too:
```kotlin ```kotlin
// Unit: no session on the client: // Unit: no session on the client:
@ -150,11 +126,10 @@ assertEquals(FooArgs("bar", 117), client.call(cmdGetFoo))
``` ```
## Create a ktor-based server ## Create ktor-based server
Normally the server side needs some session. It is convenient and avoids sending repeating data on each request speeding Normally server side needs some session. It is convenient and avoid sending repeating data on each request speeding up
up the protocol. With KILOPARSEC it is rather basic operation:
the protocol. With KILOPARSEC, it is a 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:
@ -181,11 +156,11 @@ val ns: NettyApplicationEngine = embeddedServer(Netty, port = 8080, host = "0.0.
~~~ ~~~
## Create a TCP / IP client and server ## 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 Using plain TCP/IP is even simpler, and it works way faster than websocket one, and is _the same
protected as `wss://` (and `ws://`) the variant above due to same kiloparsec encryption in both cases. Still, a TCP/IP 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. client is not available in Javascript browser targets and custom TCP ports could often be blocked by firewalls.
Documentation is available in samples here: Documentation is available in samples here:
@ -193,13 +168,11 @@ Documentation is available in samples here:
- [TCP/IP client](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec/-kilo-client/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 implement asynchronous TCP/IP transport on all platforms buy JS: In short, there are two functions that implements asynchronous 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) - [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
to create a server
- [connectTcpDevice](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec.adapter/connect-tcp-device.html) - [connectTcpDevice](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec.adapter/connect-tcp-device.html) to connect to the server
to connect to the server
## UDP client and server ## UDP client and server
@ -213,49 +186,36 @@ Is very much straightforward, same as with TCP/IP:
#### Command size #### Command size
Each command invocation and result are packed in a separate UDP diagram using effective binary packing. Each command invocation and result are packed in a separate UDP diagram using effective binary packing.
Thus, for the best results commands and results should be relatively short, best to fit into 240 bytes. While bigger Thus for the best results commands and results should be relatively short, best to fit into 240 bytes. While bigger datagrams are often transmitted successfully, the probability of the effective transmission drops with the size.
datagrams are often transmitted successfully, the probability of the effective transmission drops with the size.
Kiloparsec UDP transport does not retransmit not delivered packets. Use TCP/IP or websocket if it is a concern. Kiloparsec UDP transport does not retransmits not delivered packets. Use TCP/IP or websocket if it is a concern.
For the best results, we recommend For the best results we recommend using [push](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec/-remote-interface/index.html#1558240250%2FFunctions%2F788909594) for remote interfaces with UDP.
using [push](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec/-remote-interface/index.html#1558240250%2FFunctions%2F788909594)
for remote interfaces with UDP.
#### Timeouts #### Timeouts
As Datagrams do not form protocol itself, kiloparsec issues pings when no data is circulated between parties. As Datagrams do not form protocol itself, kiloparsec issues pings when no data is circulated between parties.
When no pings are received long enough, the kiloparsec connection is closed. There are `maxInactivityTimeout` in all When no pings are received long enough, kiloparsec connection is closed. There are `maxInactivityTimeout` in all
relevant functions and constructors. relevant functions and constructors.
Client should not issue pings manually. Client shoudl not issue pings manually.
## Reusing code between servers ## Reusing code between servers
The same instance of 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.
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 practice to create a business logic in a `KiloInterface`, then create a TCP/IP and Websocket servers 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.
passing the same instance of the logic to both.
## Note on the server identification ## Note on the server identification
We do not recommend relying on TLS (HTTPS://, WSS://) host identification solely; in the modern world there is 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 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 attack] and spoofing will be undetected. Check 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) when creating a server.
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)
when creating a server.
This will effectively protect against certificate chain spoofing in the case of the application installed from the This will effectively protetcs against certificate chain spoofing in the case of the application installed from the trusted source.
trusted source.
__Important note__. The web application could not be completely secured this way unless is loaded from the IP-address, __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
as the DNS could be spoofed the same, especially when used with `Cloudflare` or other CDN that can transparently 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.
substitute the whole site. For applications, we strongly recommend not using CDN except your own, controlled ones. You
generally can't neither detect nor repel [MITM attack] performed from _any single cloudflare 'ray'_.
## See also: ## See also:
@ -264,49 +224,37 @@ generally can't neither detect nor repel [MITM attack] performed from _any singl
# Details # Details
It is not compatible with Parsec family and no more based on the 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
the modern state of threats and rate of cyber crimes, KiloParsec uses more encryption and random key exchange on each the modern state of threats and rate of cyber crimes, KiloParsec uses more encryption and random key exchange on each
and every connection (while parsec caches session keys to avoid time-consuming keys exchange). For the same reason, and every connection (while parsec caches session keys to avoid time-consuming keys exchange). For the same reason,
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 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 Kiloparsec also uses a denser binary format [bipack](https://gitea.sergeych.net/SergeychWorks/mp_bintools), no more
key-values, key-values,
which reveals much less on the inner data structure, providing advanced which reveals much less on the inner data structure, providing advanced
typed RPC interfaces with kotlinx.serialization. There is also Rust typed RPC interfaces with kotlinx.serialization. There is also Rust
implementation [bipack_ru](https://gitea.sergeych.net/DiWAN/bipack_ru). implementation [bipack_ru](https://gitea.sergeych.net/DiWAN/bipack_ru).
The architecture allows connecting the 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
Kiloparsec is a full-duplex fully async (coroutine-based) Remote Procedure Call protocol with typed parameters Kiloparsec is a dull-duplex fully async (coroutine based) Remote Procedure Call protocol with typed parameters
and support for serializing exceptions (e.g. exception thrown while executing remote command will be caught and and support for serializing exceptions (e.g. exception thrown while executing remote command will be caught and
rethrown at the caller context). rethrown at the caller context).
Kiloparsec is not REST, it _has advanced session mechanisms_ and built-in authentication based on the same curve keys. Kiloparsec is not REST, it _has advanced session mechanisms_ and built-in authentication based on the same curve keys.
Integrated tools to prevent MITM attacks include also non-transferred independently generated token that is calculated Integrated tools to prevent MITM attacks include also non-transferred independently generated token that is calculated
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 a built-in completely asynchronous (coroutine-based top-down) transport layer based on TCP (JVM only as Kiloparsec has built-in completely asynchronous (coroutine based top-down) transport layer based on TCP (JVM only as for
for
now) and the same async Websocket-based transport based on KTOR. Websocket client is multiplatform, though the server is now) and the same async Websocket-based transport based on KTOR. Websocket client is multiplatform, though the server is
JVM only insofar. JVM only insofar.
# Licensing # Licensing
This is a work in progress, not yet moved to the public domain; Currently, you need to obtain a license from https://8-rays.dev or Sergey Chernov.
you need to obtain a license from https://8-rays.dev or [Sergey Chernov]. For open source projects it will most be free
on some special terms.
It will be moved to open source; we also guarantee that it will be moved to open source immediately if the software
export restrictions are lifted. We do not support such practices here at 8-rays.dev.
[MITM]: https://en.wikipedia.org/wiki/Man-in-the-middle_attack
[Sergey Chernov]: https://t.me/real_sergeych
[^1]: On some new Xiaomi phones we found problems with websocket binary frames, probably in ktor; use text frames
otherwise.

10
bin/pubdocs
View File

@ -1,14 +1,4 @@
#!/bin/bash #!/bin/bash
#
# Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
#
# You may use, distribute and modify this code under the
# terms of the private license, which you must obtain from the author
#
# To obtain the license, contact the author: https://t.me/real_sergeych or email to
# real dot sergeych at gmail.
#
set -e set -e
./gradlew dokkaHtml ./gradlew dokkaHtml
rsync -avz ./build/dokka/* code.sergeych.net:/bigstore/sergeych_pub/code/docs/kiloparsec rsync -avz ./build/dokka/* code.sergeych.net:/bigstore/sergeych_pub/code/docs/kiloparsec

74
build.gradle.kts
View File

@ -1,80 +1,50 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
import org.jetbrains.kotlin.gradle.ExperimentalWasmDsl
plugins { plugins {
kotlin("multiplatform") version "2.2.21" kotlin("multiplatform") version "2.0.10"
id("org.jetbrains.kotlin.plugin.serialization") version "2.2.21" id("org.jetbrains.kotlin.plugin.serialization") version "2.0.0"
id("com.android.library") version "8.7.2" apply true
`maven-publish` `maven-publish`
id("org.jetbrains.dokka") version "1.9.20" id("org.jetbrains.dokka") version "1.9.20"
} }
group = "net.sergeych" group = "net.sergeych"
version = "0.7.0-SNAPSHOT" version = "0.3.3"
repositories { repositories {
mavenCentral() mavenCentral()
mavenLocal() mavenLocal()
maven("https://maven.universablockchain.com/") maven("https://maven.universablockchain.com/")
maven("https://gitea.sergeych.net/api/packages/SergeychWorks/maven") maven("https://gitea.sergeych.net/api/packages/SergeychWorks/maven")
maven("https://gitea.sergeych.net/api/packages/YoungBlood/maven")
} }
kotlin { kotlin {
jvmToolchain(21)
jvm() jvm()
js { js {
browser() browser {
}
nodejs() nodejs()
} }
linuxX64()
linuxArm64()
macosArm64() macosArm64()
macosX64()
iosX64() iosX64()
iosArm64() iosArm64()
iosSimulatorArm64() iosSimulatorArm64()
linuxX64()
linuxArm64()
macosX64()
mingwX64() val ktor_version = "2.3.12"
androidTarget()
@OptIn(ExperimentalWasmDsl::class)
wasmJs {
browser()
}
val ktor_version = "3.1.1"
sourceSets { sourceSets {
all { all {
languageSettings.optIn("kotlinx.serialization.ExperimentalSerializationApi") languageSettings.optIn("kotlinx.serialization.ExperimentalSerializationApi")
languageSettings.optIn("kotlinx.coroutines.ExperimentalCoroutinesApi") languageSettings.optIn("kotlinx.coroutines.ExperimentalCoroutinesApi")
languageSettings.optIn("kotlin.ExperimentalUnsignedTypes") languageSettings.optIn("kotlin.ExperimentalUnsignedTypes")
languageSettings.optIn("kotlin.time.ExperimentalTime")
} }
val commonMain by getting { val commonMain by getting {
dependencies { dependencies {
implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.10.2") implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.7.3")
implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.9.0") implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.7.1")
api("io.ktor:ktor-client-core:$ktor_version") api("io.ktor:ktor-client-core:$ktor_version")
api("net.sergeych:crypto2:0.9.0") api("net.sergeych:crypto2:0.4.2")
}
}
val androidMain by getting {
dependencies {
implementation("io.ktor:ktor-client-okhttp:$ktor_version")
} }
} }
val ktorSocketMain by creating { val ktorSocketMain by creating {
@ -86,8 +56,8 @@ kotlin {
val commonTest by getting { val commonTest by getting {
dependencies { dependencies {
implementation(kotlin("test")) implementation(kotlin("test"))
implementation("org.slf4j:slf4j-simple:2.0.17") implementation("org.slf4j:slf4j-simple:2.0.9")
implementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.10.2") implementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.8.1")
} }
} }
val ktorSocketTest by creating { val ktorSocketTest by creating {
@ -106,6 +76,7 @@ kotlin {
val jvmTest by getting { val jvmTest by getting {
dependsOn(ktorSocketTest) 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")
@ -146,7 +117,6 @@ kotlin {
dependsOn(ktorSocketTest) dependsOn(ktorSocketTest)
} }
} }
}
publishing { publishing {
val mavenToken by lazy { val mavenToken by lazy {
@ -165,7 +135,7 @@ publishing {
} }
} }
} }
}
tasks.dokkaHtml.configure { tasks.dokkaHtml.configure {
outputDirectory.set(buildDir.resolve("dokka")) outputDirectory.set(buildDir.resolve("dokka"))
@ -176,15 +146,3 @@ tasks.dokkaHtml.configure {
} }
} }
android {
namespace = "net.sergeych.kiloparsec"
compileSdk = 34
defaultConfig {
minSdk = 24
}
compileOptions {
sourceCompatibility = JavaVersion.VERSION_17
targetCompatibility = JavaVersion.VERSION_17
}
}

18
gradle.properties
View File

@ -1,20 +1,2 @@
#
# Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
#
# You may use, distribute and modify this code under the
# terms of the private license, which you must obtain from the author
#
# To obtain the license, contact the author: https://t.me/real_sergeych or email to
# real dot sergeych at gmail.
#
kotlin.code.style=official kotlin.code.style=official
kotlin.mpp.applyDefaultHierarchyTemplate=false kotlin.mpp.applyDefaultHierarchyTemplate=false
kotlin.daemon.jvmargs=-Xmx2048m -Dfile.encoding=UTF-8
kotlin.native.ignoreDisabledTargets=true
org.gradle.parallel=true
org.gradle.configuration-cache=true
org.gradle.caching=true

12
gradle/wrapper/gradle-wrapper.properties vendored
View File

@ -1,15 +1,5 @@
#
# Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
#
# You may use, distribute and modify this code under the
# terms of the private license, which you must obtain from the author
#
# To obtain the license, contact the author: https://t.me/real_sergeych or email to
# real dot sergeych at gmail.
#
distributionBase=GRADLE_USER_HOME distributionBase=GRADLE_USER_HOME
distributionPath=wrapper/dists distributionPath=wrapper/dists
distributionUrl=https\://services.gradle.org/distributions/gradle-8.14.3-bin.zip distributionUrl=https\://services.gradle.org/distributions/gradle-8.2-bin.zip
zipStoreBase=GRADLE_USER_HOME zipStoreBase=GRADLE_USER_HOME
zipStorePath=wrapper/dists zipStorePath=wrapper/dists

16
gradlew vendored
View File

@ -1,13 +1,19 @@
#!/bin/sh #!/bin/sh
# #
# Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved # Copyright © 2015-2021 the original authors.
# #
# You may use, distribute and modify this code under the # Licensed under the Apache License, Version 2.0 (the "License");
# terms of the private license, which you must obtain from the author # you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# #
# To obtain the license, contact the author: https://t.me/real_sergeych or email to # https://www.apache.org/licenses/LICENSE-2.0
# real dot sergeych at gmail. #
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# #
############################################################################## ##############################################################################

11
settings.gradle.kts
View File

@ -1,16 +1,5 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
pluginManagement { pluginManagement {
repositories { repositories {
google()
mavenCentral() mavenCentral()
gradlePluginPortal() gradlePluginPortal()
} }

10
src/commonMain/kotlin/net/sergeych/kiloparsec/AsyncVarint.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
import io.ktor.utils.io.* import io.ktor.utils.io.*

52
src/commonMain/kotlin/net/sergeych/kiloparsec/AtomicAsyncValue.kt
View File

@ -1,52 +0,0 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
/**
* Multiplatform atomically mutable value to be used in [kotlinx.coroutines],
* with suspending mutating operations, see [mutate].
*
* Actual value can be either changed in a block of [mutate] when
* new value _depends on the current value_ or with [reset].
*
* [value] getter is suspended because it waits until the mutation finishes
*/
open class AtomicAsyncValue<T>(initialValue: T) {
private var actualValue = initialValue
private val access = Mutex()
/**
* Change the value: get the current and set to the returned, all in the
* atomic suspend operation. All other mutating requests including assigning to [value]
* will be blocked and queued.
* @return result of the mutation. Note that immediate call to property [value]
* could already return modified bu some other thread value!
*/
suspend fun mutate(mutator: suspend (T) -> T): T = access.withLock {
actualValue = mutator(actualValue)
actualValue
}
/**
* Atomic get or set the value. Atomic get means if there is a [mutate] in progress
* it will wait until the mutation finishes and then return the correct result.
*/
suspend fun value() = access.withLock { actualValue }
/**
* Set the new value without checking it. Shortcut to
* ```mutate { value = newValue }```
*/
suspend fun reset(value: T) = mutate { value }
}

53
src/commonMain/kotlin/net/sergeych/kiloparsec/Command.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
import kotlinx.serialization.KSerializer import kotlinx.serialization.KSerializer
@ -15,28 +5,15 @@ import kotlinx.serialization.Serializable
import net.sergeych.bintools.toDataSource import net.sergeych.bintools.toDataSource
import net.sergeych.bipack.BipackDecoder import net.sergeych.bipack.BipackDecoder
import net.sergeych.bipack.BipackEncoder import net.sergeych.bipack.BipackEncoder
import net.sergeych.kiloparsec.Command.Companion.unpackCall
import net.sergeych.utools.unpack import net.sergeych.utools.unpack
/** /**
* Typesafe command definition. Command is a universal entity in kiloparsec (and also diwan): it is used * Typesafe command definition. Command is a universal entity in Divan: it is used
* in node-2-node protocols and client API, and most importantly in calling smart contract * in node-2-node protocols and client API, and most importantly in calling smart contract
* methods. This is essentially a Kotlin binding to typesafe serialize command calls and * methods. This is essentially a Kotlin binding to typesafe serialize command calls and
* deserialize results. * deserialize results.
*
* To create command instances, it is recommended to use [command] that returns [CommandDelegate].
*
* Use [packCall] to serialize the command call with some arguments.
*
* Note that `Command` instances themselves are not serialized, instead, the call is serialized,
* in the form of [Call], containing name and properly serialized arguments.
*
* [unpackCall] deserialized result of the [packCall] so the proper handler for the command could
* be used. Then the result of the execution could be packed with [exec] and then unpacked with
* [unpackResult].
*
*/ */
open class Command<A, R>( class Command<A, R>(
val name: String, val name: String,
val argsSerializer: KSerializer<A>, val argsSerializer: KSerializer<A>,
val resultSerializer: KSerializer<R> val resultSerializer: KSerializer<R>
@ -44,29 +21,13 @@ open class Command<A, R>(
@Serializable @Serializable
data class Call(val name: String,val serializedArgs: UByteArray) data class Call(val name: String,val serializedArgs: UByteArray)
/** fun packCall(args: A): UByteArray = BipackEncoder.encode(
* Pack command invocation with specified arguments. Call(name, BipackEncoder.encode(argsSerializer, args).toUByteArray())
*/ ).toUByteArray()
fun packCall(args: A): UByteArray = BipackEncoder.encode(createCall(args)).toUByteArray()
/**
* Create [Call] instance for specified args vy serializing it properly
*/
fun createCall(args: A): Call = Call(name, BipackEncoder.encode(argsSerializer, args).toUByteArray())
/**
* Unpack result, obtained by [exec].
*/
fun unpackResult(packedResult: UByteArray): R = fun unpackResult(packedResult: UByteArray): R =
unpack(resultSerializer, packedResult) unpack(resultSerializer, packedResult)
/**
* Execute a command unpacking args.
*
* @param packedArgs arguments, as provided by [packCall] in the [Call] instance
* @param handler actual code to execute the command
* @return properly serialized result to be unpacked with [unpackResult].
*/
suspend fun exec(packedArgs: UByteArray, handler: suspend (A) -> R): UByteArray = suspend fun exec(packedArgs: UByteArray, handler: suspend (A) -> R): UByteArray =
BipackEncoder.encode( BipackEncoder.encode(
resultSerializer, resultSerializer,
@ -75,10 +36,6 @@ open class Command<A, R>(
).toUByteArray() ).toUByteArray()
companion object { companion object {
/**
* Unpack command invocation instance from [packCall]. Use [exec] to deserialize arguments and
* perform command.
*/
fun unpackCall(packedCall: UByteArray): Call = BipackDecoder.decode(packedCall.toDataSource()) fun unpackCall(packedCall: UByteArray): Call = BipackDecoder.decode(packedCall.toDataSource())
} }
} }

10
src/commonMain/kotlin/net/sergeych/kiloparsec/CommandDelegate.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
import kotlinx.serialization.KSerializer import kotlinx.serialization.KSerializer

10
src/commonMain/kotlin/net/sergeych/kiloparsec/ConnectionFactory.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
/** /**

20
src/commonMain/kotlin/net/sergeych/kiloparsec/ExceptionWithCode.kt
View File

@ -1,20 +0,0 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec
/**
* External errors wich needs to be encoded with a specific code
* should implement this interface so it will be serialized properly.
*/
interface ExceptionWithCode {
val code: String
val message: String?
}

44
src/commonMain/kotlin/net/sergeych/kiloparsec/KiloClient.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CancellationException
@ -19,7 +9,6 @@ import kotlinx.coroutines.isActive
import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.sync.withLock
import net.sergeych.crypto2.SigningKey import net.sergeych.crypto2.SigningKey
import net.sergeych.crypto2.VerifyingKey
import net.sergeych.crypto2.VerifyingPublicKey import net.sergeych.crypto2.VerifyingPublicKey
import net.sergeych.mp_logger.LogTag import net.sergeych.mp_logger.LogTag
import net.sergeych.mp_logger.Loggable import net.sergeych.mp_logger.Loggable
@ -71,14 +60,6 @@ class KiloClient<S>(
@Suppress("unused") @Suppress("unused")
val connectedStateFlow = _state.asStateFlow() val connectedStateFlow = _state.asStateFlow()
/**
* The verifying, or public, key identifying client sessions. It could be used to
* restore environment on reconnection. This is what remote side, e.g. server, sees as
* [KiloScope.remoteIdentity].
*/
@Suppress("unused")
val localIdentity: VerifyingKey? = secretKey?.verifyingKey
private var deferredClient = CompletableDeferred<KiloClientConnection<S>>() private var deferredClient = CompletableDeferred<KiloClientConnection<S>>()
private val job = private val job =
@ -91,11 +72,9 @@ class KiloClient<S>(
debug { "get device and session" } debug { "get device and session" }
val client = KiloClientConnection(localInterface, kc, secretKey) val client = KiloClientConnection(localInterface, kc, secretKey)
deferredClient.complete(client) deferredClient.complete(client)
debug { "starting client run"} client.run {
val r = runCatching { client.run {
_state.value = it _state.value = it
} } }
debug { "----------- client run finished: $r" }
resetDeferredClient() resetDeferredClient()
debug { "client run finished" } debug { "client run finished" }
} catch (_: RemoteInterface.ClosedException) { } catch (_: RemoteInterface.ClosedException) {
@ -111,7 +90,7 @@ class KiloClient<S>(
_state.value = false _state.value = false
resetDeferredClient() resetDeferredClient()
// reconnection timeout // reconnection timeout
delay(700) delay(100)
} }
} }
@ -162,7 +141,7 @@ class KiloClient<S>(
* a key. Connection is established either with a properly authenticated key or no key at all. * a key. Connection is established either with a properly authenticated key or no key at all.
*/ */
@Suppress("unused") @Suppress("unused")
suspend fun remoteIdentity(): VerifyingPublicKey? = deferredClient.await().remoteId() suspend fun remoteId(): VerifyingPublicKey? = deferredClient.await().remoteId()
companion object { companion object {
class Builder<S>() { class Builder<S>() {
@ -177,19 +156,8 @@ class KiloClient<S>(
var secretIdKey: SigningKey? = null var secretIdKey: SigningKey? = null
/** /**
* Build local command implementations, those callable from the server, exception * Build local command implementations (remotely callable ones), exception
* class handlers, and anything else [KiloInterface] allows. Usage sample: * class handlers, etc.
*
* ```kotlin
* val client = KiloClient {
* connect { connectTcpDevice("localhost:$port") }
* local {
* on(cmdPing) {
* "pong! $it"
* }
* }
* }
* ```
*/ */
fun local(f: KiloInterface<S>.() -> Unit) { fun local(f: KiloInterface<S>.() -> Unit) {
interfaceBuilder = f interfaceBuilder = f

22
src/commonMain/kotlin/net/sergeych/kiloparsec/KiloClientConnection.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
import kotlinx.coroutines.* import kotlinx.coroutines.*
@ -49,6 +39,7 @@ class KiloClientConnection<S>(
try { try {
// in parallel: keys and connection // in parallel: keys and connection
val deferredKeyPair = async { SafeKeyExchange() } val deferredKeyPair = async { SafeKeyExchange() }
debug { "opening device" }
debug { "got a transport device $device" } debug { "got a transport device $device" }
@ -61,11 +52,10 @@ class KiloClientConnection<S>(
debug { "transport started" } debug { "transport started" }
val pair = deferredKeyPair.await() val pair = deferredKeyPair.await()
debug { "keypair ready (1)" } debug { "keypair ready" }
val serverHe = transport.call(L0Request, Handshake(1u, pair.publicKey)) val serverHe = transport.call(L0Request, Handshake(1u, pair.publicKey))
debug { "got server HE (2)" }
val sk = pair.clientSessionKey(serverHe.publicKey) val sk = pair.clientSessionKey(serverHe.publicKey)
var params = KiloParams(false, transport, sk, session, null, this@KiloClientConnection) var params = KiloParams(false, transport, sk, session, null, this@KiloClientConnection)
@ -90,14 +80,15 @@ class KiloClientConnection<S>(
kiloRemoteInterface.complete( kiloRemoteInterface.complete(
KiloRemoteInterface(deferredParams, clientInterface) KiloRemoteInterface(deferredParams, clientInterface)
) )
clientInterface.onConnectHandlers.invokeAll(params.scope) clientInterface.onConnectHandler?.invoke(params.scope)
onConnectedStateChanged?.invoke(true) onConnectedStateChanged?.invoke(true)
job.join() job.join()
} catch (x: CancellationException) { } catch (x: CancellationException) {
info { "client is cancelled" } info { "client is cancelled" }
} catch (x: RemoteInterface.ClosedException) { } catch (x: RemoteInterface.ClosedException) {
debug { "connection closed/refused by remote" } x.printStackTrace()
info { "connection closed by remote" }
} finally { } finally {
onConnectedStateChanged?.invoke(false) onConnectedStateChanged?.invoke(false)
job?.cancel() job?.cancel()
@ -114,6 +105,3 @@ class KiloClientConnection<S>(
kiloRemoteInterface.await().push(cmd, args) kiloRemoteInterface.await().push(cmd, args)
} }
} }
internal fun <S>Collection<KiloHandler<S>>.invokeAll(scope: KiloScope<S>) =
forEach { runCatching { scope.it() } }

26
src/commonMain/kotlin/net/sergeych/kiloparsec/KiloInterface.kt
View File

@ -1,16 +1,5 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
typealias KiloHandler<S> = KiloScope<S>.()->Unit
/** /**
* 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.
@ -25,22 +14,15 @@ typealias KiloHandler<S> = KiloScope<S>.()->Unit
*/ */
open class KiloInterface<S> : LocalInterface<KiloScope<S>>() { open class KiloInterface<S> : LocalInterface<KiloScope<S>>() {
internal val onConnectHandlers = mutableListOf<KiloHandler<S>>() internal var onConnectHandler: (KiloScope<S>.()->Unit) ? = null
/** fun onConnected(f: KiloScope<S>.()->Unit) { onConnectHandler = f }
* Registers handler [f] for [onConnected] event, to the head or the end of handler list.
*
* @param addFirst if true, [f] will be added to the beginning of the list of handlers
*/
fun onConnected(addFirst: Boolean = false, f: KiloScope<S>.()->Unit) {
if( addFirst ) onConnectHandlers.add(0, f) else onConnectHandlers += f
}
init { init {
registerError { RemoteInterface.UnknownCommand(it) } registerError { RemoteInterface.UnknownCommand() }
registerError { RemoteInterface.InternalError(it) } registerError { RemoteInterface.InternalError(it) }
registerError { RemoteInterface.ClosedException(it) } registerError { RemoteInterface.ClosedException(it) }
registerError { RemoteInterface.SecurityException(it) } // registerError { RemoteInterface.SecurityException(it) }
registerError { RemoteInterface.InvalidDataException(it) } registerError { RemoteInterface.InvalidDataException(it) }
registerError { RemoteInterface.RemoteException(it) } registerError { RemoteInterface.RemoteException(it) }
registerError { IllegalStateException() } registerError { IllegalStateException() }

10
src/commonMain/kotlin/net/sergeych/kiloparsec/KiloL0Interface.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CompletableDeferred

10
src/commonMain/kotlin/net/sergeych/kiloparsec/KiloParams.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
import kotlinx.serialization.Serializable import kotlinx.serialization.Serializable

10
src/commonMain/kotlin/net/sergeych/kiloparsec/KiloRemoteInterface.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CompletableDeferred

10
src/commonMain/kotlin/net/sergeych/kiloparsec/KiloScope.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
import net.sergeych.crypto2.SigningKey import net.sergeych.crypto2.SigningKey

10
src/commonMain/kotlin/net/sergeych/kiloparsec/KiloServer.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CancellationException

12
src/commonMain/kotlin/net/sergeych/kiloparsec/KiloServerConnection.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CompletableDeferred
@ -85,7 +75,7 @@ class KiloServerConnection<S>(
kiloRemoteInterface.complete( kiloRemoteInterface.complete(
KiloRemoteInterface(deferredParams, clientInterface) KiloRemoteInterface(deferredParams, clientInterface)
) )
clientInterface.onConnectHandlers.invokeAll(p.scope) clientInterface.onConnectHandler?.invoke(p.scope)
} }
} }

30
src/commonMain/kotlin/net/sergeych/kiloparsec/LocalInterface.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
import net.sergeych.mp_logger.LogTag import net.sergeych.mp_logger.LogTag
@ -77,18 +67,14 @@ open class LocalInterface<S> : Loggable by LogTag("LocalInterface${idCounter.inc
name: String, name: String,
packedArgs: UByteArray, packedArgs: UByteArray,
): UByteArray = ): UByteArray =
(commands[name] ?: throw RemoteInterface.UnknownCommand(name)) (commands[name] ?: throw RemoteInterface.UnknownCommand())
.invoke(scope, packedArgs) .invoke(scope, packedArgs)
private val errorByClass = mutableMapOf<KClass<*>, String>() private val errorByClass = mutableMapOf<KClass<*>, String>()
private val errorBuilder = mutableMapOf<String, (String, UByteArray?) -> Throwable>() private val errorBuilder = mutableMapOf<String, (String, UByteArray?) -> Throwable>()
/** fun <T : Throwable> registerError(
* Register exception for automatic transmission over the kiloparsec connection using its [klass]
* and possibly override code. In most cases it is simpler to use [registerError].
*/
fun <T : Throwable> registerErrorClass(
klass: KClass<T>, code: String = klass.simpleName!!, klass: KClass<T>, code: String = klass.simpleName!!,
exceptionBuilder: (String, UByteArray?) -> T, exceptionBuilder: (String, UByteArray?) -> T,
) { ) {
@ -96,17 +82,10 @@ open class LocalInterface<S> : Loggable by LogTag("LocalInterface${idCounter.inc
errorBuilder[code] = exceptionBuilder errorBuilder[code] = exceptionBuilder
} }
/**
* Register an exception to be transmitted automatically over the kiloparsec connection.
* Example:
* ```kotlin
* localInterface.registerError { IllegalArgumentException(it) }
* ```
*/
inline fun <reified T : Throwable> registerError( inline fun <reified T : Throwable> registerError(
noinline exceptionBuilder: (String) -> T, noinline exceptionBuilder: (String) -> T,
) { ) {
registerErrorClass(T::class) { msg, _ -> exceptionBuilder(msg) } registerError(T::class) { msg, _ -> exceptionBuilder(msg) }
} }
val errorProviders = mutableListOf<LocalInterface<*>>() val errorProviders = mutableListOf<LocalInterface<*>>()
@ -116,8 +95,7 @@ open class LocalInterface<S> : Loggable by LogTag("LocalInterface${idCounter.inc
} }
fun getErrorCode(t: Throwable): String? = fun getErrorCode(t: Throwable): String? =
(t as? ExceptionWithCode)?.code errorByClass[t::class] ?: errorProviders.firstNonNull { it.getErrorCode(t) }
?: errorByClass[t::class] ?: errorProviders.firstNonNull { it.getErrorCode(t) }
fun encodeError(forId: UInt, t: Throwable): Transport.Block.Error = fun encodeError(forId: UInt, t: Throwable): Transport.Block.Error =
if (t is RemoteInterface.ClosedException) { if (t is RemoteInterface.ClosedException) {

23
src/commonMain/kotlin/net/sergeych/kiloparsec/RemoteInterface.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
/** /**
@ -50,21 +40,14 @@ interface RemoteInterface {
/** /**
* Command is not supported by the remote party * Command is not supported by the remote party
*/ */
class UnknownCommand(commandName: String) : RemoteException("UnknownCommand: $commandName") class UnknownCommand : RemoteException("UnknownCommand")
open class InternalError(code: String="0"): RemoteException("Internal error: $code") open class InternalError(code: String="0"): RemoteException("Internal error: $code")
suspend fun <R> call(cmd: Command<Unit, R>): R = call(cmd, Unit) suspend fun <R> call(cmd: Command<Unit, R>): R = call(cmd, Unit)
/** /**
* Call the remote procedure with specified [args] and return its result of type [R]. The calling coroutine * Call the remote procedure with specified args and return its result
* suspends until the request is performed on the remote side and the result value (also `Unit`) will be received.
*
* When it is not necessary to wait for the return value and/or command execution, it is recommended to
* use [push] instead.
*
* @throws RemoteException if the execution caused exception on the remote size
* @throws Exception for registered exceptions, see [LocalInterface.registerError], etc.
*/ */
suspend fun <A, R> call(cmd: Command<A, R>, args: A): R suspend fun <A, R> call(cmd: Command<A, R>, args: A): R
@ -72,8 +55,6 @@ interface RemoteInterface {
* Push the notification without waiting for reception or processing. * Push the notification without waiting for reception or processing.
* It returns immediately after sending data to the transport (e.g., to the network). * 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. * Use [call] if it is necessary to wait until the command will be received and processed by the remote.
*
* Push is onlu available for commands without returned value.
*/ */
suspend fun <A> push(cmd: Command<A, Unit>, args: A) suspend fun <A> push(cmd: Command<A, Unit>, args: A)

37
src/commonMain/kotlin/net/sergeych/kiloparsec/Transport.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
import kotlinx.coroutines.* import kotlinx.coroutines.*
@ -27,12 +17,13 @@ import kotlinx.serialization.encoding.Encoder
import kotlinx.serialization.serializer import kotlinx.serialization.serializer
import net.sergeych.bipack.Unsigned import net.sergeych.bipack.Unsigned
import net.sergeych.crypto2.toDump import net.sergeych.crypto2.toDump
import net.sergeych.kiloparsec.Transport.Device
import net.sergeych.mp_logger.* import net.sergeych.mp_logger.*
import net.sergeych.utools.pack import net.sergeych.utools.pack
import net.sergeych.utools.unpack import net.sergeych.utools.unpack
/** /**
* Kiloparsec channel that operates some block [Device] exporting a given [localInterface] * Divan channel that operates some block [Device] exporting a given [localInterface]
* to remote callers. [LocalInterface] allows session managing, transmitting exceptions * to remote callers. [LocalInterface] allows session managing, transmitting exceptions
* in a scure and multiplatform way and provide local command execution (typed RPC) * in a scure and multiplatform way and provide local command execution (typed RPC)
*/ */
@ -133,13 +124,7 @@ class Transport<S>(
} }
// now we have mutex freed so we can call: // now we have mutex freed so we can call:
val r = runCatching { val r = runCatching { device.output.send(pack(b).also { debug { ">>>\n${it.toDump()}" } }) }
do {
val cr = device.output.trySend(pack(b))
if( cr.isClosed ) throw ClosedSendChannelException("can't send block: channel is closed")
delay(100)
} while(!cr.isSuccess)
}
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 }
@ -199,7 +184,10 @@ class Transport<S>(
while (isActive && !isClosed) { while (isActive && !isClosed) {
try { try {
device.input.receive().let { packed -> device.input.receive().let { packed ->
debug { "<<<\n${packed.toDump()}" }
val b = unpack<Block>(packed) val b = unpack<Block>(packed)
debug { "<<$ $b" }
debug { "access state: ${access.isLocked}" }
when (b) { when (b) {
is Block.Error -> access.withLock { is Block.Error -> access.withLock {
val error = localInterface.decodeError(b) val error = localInterface.decodeError(b)
@ -210,7 +198,10 @@ class Transport<S>(
} }
is Block.Response -> access.withLock { is Block.Response -> access.withLock {
calls.remove(b.forId)?.complete(b.packedResult) calls.remove(b.forId)?.let {
debug { "activating wait handle for ${b.forId}" }
it.complete(b.packedResult)
}
?: warning { "wait handle not found for ${b.forId}" } ?: warning { "wait handle not found for ${b.forId}" }
} }
@ -237,10 +228,12 @@ class Transport<S>(
} catch (t: Throwable) { } catch (t: Throwable) {
send(Block.Error(b.id, "UnknownError", t.message)) send(Block.Error(b.id, "UnknownError", t.message))
} }
.also { debug { "command executed: ${b.name}" } }
} }
} }
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
@ -250,7 +243,7 @@ class Transport<S>(
} 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 (_: 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
@ -276,7 +269,7 @@ class Transport<S>(
} }
debug { "no more active: $isActive / ${calls.size}" } debug { "no more active: $isActive / ${calls.size}" }
} }
debug { "exiting transport loop" } info { "exiting transport loop" }
} }
private suspend fun send(block: Block) { private suspend fun send(block: Block) {

10
src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/InetTransportDevice.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec.adapter package net.sergeych.kiloparsec.adapter
import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.Channel

10
src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/NetworkProvider.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec.adapter package net.sergeych.kiloparsec.adapter
/** /**

10
src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/ProxyDevice.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec.adapter package net.sergeych.kiloparsec.adapter
import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.Channel

91
src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/websocketClient.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec.adapter package net.sergeych.kiloparsec.adapter
import io.ktor.client.* import io.ktor.client.*
@ -20,62 +10,42 @@ import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.ClosedReceiveChannelException import kotlinx.coroutines.channels.ClosedReceiveChannelException
import kotlinx.coroutines.channels.ClosedSendChannelException import kotlinx.coroutines.channels.ClosedSendChannelException
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
import kotlinx.io.IOException
import net.sergeych.crypto2.SigningKey import net.sergeych.crypto2.SigningKey
import net.sergeych.kiloparsec.* import net.sergeych.kiloparsec.KiloClient
import net.sergeych.mp_logger.* import net.sergeych.kiloparsec.KiloConnectionData
import net.sergeych.mp_tools.decodeBase64Compact import net.sergeych.kiloparsec.KiloInterface
import net.sergeych.mp_tools.encodeToBase64Compact import net.sergeych.kiloparsec.RemoteInterface
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 net.sergeych.mp_tools.globalLaunch import net.sergeych.mp_tools.globalLaunch
import net.sergeych.tools.AtomicCounter import net.sergeych.tools.AtomicCounter
private val counter = AtomicCounter() private val counter = AtomicCounter()
/**
* Shortcut to create websocket client. Use [websocketTransportDevice] with [KiloClient]
* for fine-grained control.
*/
fun <S> websocketClient( fun <S> websocketClient(
path: String, path: String,
clientInterface: KiloInterface<S> = KiloInterface(), clientInterface: KiloInterface<S> = KiloInterface(),
client: HttpClient = HttpClient { install(WebSockets) },
secretKey: SigningKey? = null, secretKey: SigningKey? = null,
useTextFrames: Boolean = false,
sessionMaker: () -> S = { sessionMaker: () -> S = {
@Suppress("UNCHECKED_CAST") @Suppress("UNCHECKED_CAST")
Unit as S Unit as S
}, },
): KiloClient<S> { ): KiloClient<S> {
return KiloClient(clientInterface, secretKey) {
KiloConnectionData(websocketTransportDevice(path, useTextFrames), sessionMaker())
}
}
/**
* Create kilopaarsec transport over websocket (ws or wss).
* @param path websocket path (must start with ws:// or wss:// and contain a path part)
* @client use default [HttpClient], it installs [WebSockets] plugin
*/
fun websocketTransportDevice(
path: String,
useTextFrames: Boolean = false,
client: HttpClient = HttpClient {
install(WebSockets)
},
): Transport.Device {
var u = Url(path) var u = Url(path)
if (u.encodedPath.length <= 1) if (u.encodedPath.length <= 1)
u = URLBuilder(u).apply { u = URLBuilder(u).apply {
encodedPath = "/kp" encodedPath = "/kp"
}.build() }.build()
return KiloClient(clientInterface, secretKey) {
val input = Channel<UByteArray>() val input = Channel<UByteArray>()
val output = Channel<UByteArray>() val output = Channel<UByteArray>()
val closeHandle = CompletableDeferred<Boolean>() val closeHandle = CompletableDeferred<Boolean>()
val readyHandle = CompletableDeferred<Unit>()
globalLaunch { globalLaunch {
val log = LogTag("KC:${counter.incrementAndGet()}") val log = LogTag("KC:${counter.incrementAndGet()}")
try {
client.webSocket({ client.webSocket({
url.protocol = u.protocol url.protocol = u.protocol
url.host = u.host url.host = u.host
@ -87,23 +57,18 @@ fun websocketTransportDevice(
log.info { "connected to the server" } log.info { "connected to the server" }
// println("SENDING!!!") // println("SENDING!!!")
// send("Helluva") // send("Helluva")
readyHandle.complete(Unit)
launch { launch {
try { try {
for (block in output) { for (block in output) {
if (useTextFrames)
send(
Frame.Text(block.asByteArray().encodeToBase64Compact())
)
else
send(block.toByteArray()) send(block.toByteArray())
} }
log.info { "input is closed, closing the websocket" } log.info { "input is closed, closing the websocket" }
if (closeHandle.isActive) closeHandle.complete(true) if (closeHandle.isActive) closeHandle.complete(true)
} catch (_: ClosedSendChannelException) { } catch (_: ClosedSendChannelException) {
log.info { "send channel closed" } log.info { "send channel closed" }
} catch (_: CancellationException) { }
} catch (t: Throwable) { catch(_: CancellationException) {}
catch(t: Throwable) {
log.info { "unexpected exception in websock sender: ${t.stackTraceToString()}" } log.info { "unexpected exception in websock sender: ${t.stackTraceToString()}" }
closeHandle.completeExceptionally(t) closeHandle.completeExceptionally(t)
} }
@ -112,10 +77,10 @@ fun websocketTransportDevice(
launch { launch {
try { try {
for (f in incoming) { for (f in incoming) {
when (f) { if (f is Frame.Binary) {
is Frame.Binary -> input.send(f.readBytes().toUByteArray()) input.send(f.readBytes().toUByteArray())
is Frame.Text -> input.send(f.readText().decodeBase64Compact().toUByteArray()) } else {
else -> log.warning { "ignoring unexpected frame of type ${f.frameType}" } log.warning { "ignoring unexpected frame of type ${f.frameType}" }
} }
} }
if (closeHandle.isActive) closeHandle.complete(true) if (closeHandle.isActive) closeHandle.complete(true)
@ -133,27 +98,17 @@ fun websocketTransportDevice(
log.warning { "Client is closing with error" } log.warning { "Client is closing with error" }
throw RemoteInterface.ClosedException() throw RemoteInterface.ClosedException()
} }
runCatching { output.close() } output.close()
runCatching { input.close() } input.close()
runCatching { close() }
}
} catch (x: IOException) {
if ("refused" in x.toString()) log.debug { "connection refused" }
else log.warning { "unexpected IO error $x" }
runCatching { output.close() }
runCatching { input.close() }
} }
log.info { "closing connection" } log.info { "closing connection" }
} }
// Wait for connection be established or failed val device = ProxyDevice(input, output) {
val device = ProxyDevice(input, output, doClose = {
// we need to explicitly close the coroutine job, or it can hang for a long time // we need to explicitly close the coroutine job, or it can hang for a long time
// leaking resources. // leaking resources.
runCatching { output.close() }
runCatching { input.close() }
closeHandle.complete(true) closeHandle.complete(true)
// job.cancel() // job.cancel()
})
return device
} }
KiloConnectionData(device, sessionMaker())
}
}

10
src/commonMain/kotlin/net/sergeych/kiloparsec/commands.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
import kotlinx.serialization.Serializable import kotlinx.serialization.Serializable

10
src/commonMain/kotlin/net/sergeych/kiloparsec/tools.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.Mutex

10
src/commonTest/kotlin/KeysTest.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
import kotlinx.coroutines.test.runTest import kotlinx.coroutines.test.runTest
import net.sergeych.crypto2.IllegalSignatureException import net.sergeych.crypto2.IllegalSignatureException
import net.sergeych.crypto2.SealedBox import net.sergeych.crypto2.SealedBox

12
src/commonTest/kotlin/PackTest.kt
View File

@ -1,14 +1,5 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
import kotlinx.coroutines.test.runTest import kotlinx.coroutines.test.runTest
import kotlinx.datetime.Instant
import net.sergeych.bipack.BipackEncoder import net.sergeych.bipack.BipackEncoder
import net.sergeych.crypto2.initCrypto import net.sergeych.crypto2.initCrypto
import net.sergeych.kiloparsec.KiloParams import net.sergeych.kiloparsec.KiloParams
@ -20,7 +11,6 @@ import kotlin.test.Test
import kotlin.test.assertContentEquals import kotlin.test.assertContentEquals
import kotlin.test.assertEquals import kotlin.test.assertEquals
import kotlin.time.Duration.Companion.microseconds import kotlin.time.Duration.Companion.microseconds
import kotlin.time.Instant
class PackTest { class PackTest {
inline fun <reified T>check(x: T?) { inline fun <reified T>check(x: T?) {

10
src/commonTest/kotlin/ToolsTest.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
class ToolsTest { class ToolsTest {
// @Test // @Test
// fun testRemoceCmd() { // fun testRemoceCmd() {

13
src/commonTest/kotlin/TransportTest.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
import kotlinx.coroutines.* import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.ReceiveChannel import kotlinx.coroutines.channels.ReceiveChannel
@ -76,8 +66,7 @@ class TransportTest {
"p2: $it" "p2: $it"
} }
on(cmdSlow) { on(cmdSlow) {
// the problem: runTest() breaks delays so it is not enough delay(100)
for( i in 0..10000) delay(1)
"done" "done"
} }
} }

10
src/commonTest/kotlin/assertThrows.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
import kotlin.test.fail import kotlin.test.fail
inline fun <reified T: Throwable>assertThrows(f: ()->Unit): T { inline fun <reified T: Throwable>assertThrows(f: ()->Unit): T {

80
src/jvmMain/kotlin/net/sergeych/kiloparsec/adapter/WebsocketServer.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec.adapter package net.sergeych.kiloparsec.adapter
import io.ktor.server.application.* import io.ktor.server.application.*
@ -20,44 +10,25 @@ import kotlinx.coroutines.channels.ClosedReceiveChannelException
import kotlinx.coroutines.isActive import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
import net.sergeych.crypto2.SigningKey import net.sergeych.crypto2.SigningKey
import net.sergeych.crypto2.toDump
import net.sergeych.kiloparsec.KiloInterface import net.sergeych.kiloparsec.KiloInterface
import net.sergeych.kiloparsec.KiloServerConnection import net.sergeych.kiloparsec.KiloServerConnection
import net.sergeych.kiloparsec.RemoteInterface import net.sergeych.kiloparsec.RemoteInterface
import net.sergeych.mp_logger.* import net.sergeych.mp_logger.*
import net.sergeych.mp_tools.decodeBase64Compact
import net.sergeych.mp_tools.encodeToBase64Compact
import net.sergeych.tools.AtomicCounter import net.sergeych.tools.AtomicCounter
import kotlin.time.Duration.Companion.seconds import java.time.Duration
/**
* Create a ktor-based websocket server.
* This call install Routing and WebSockets with proper configuration.
*
* The course of action is:
*
* - create LocalInterface and populate it with functionality
* - call this method with localInterface
* - optionally, connect the same interface with TCP or UDP providers on supported platforms,
* in which case it might be useful to hae session creating function [createSession] separate.
*
* _Note_: [KiloInterface] as for now does not contain session creation in it as we suggest
* session could be transport specific.
*
* @param localInterface where the actual work is performed.
* @param path default http path to the websocket.
* @param serverKey optional key to authenticate the connection. If the client specify expected
* server key it should match of connection will not be established.
* @param createSession function to create a server session.
*/
fun <S> Application.setupWebsocketServer( fun <S> Application.setupWebsocketServer(
localInterface: KiloInterface<S>, localInterface: KiloInterface<S>,
path: String = "/kp", path: String = "/kp",
serverKey: SigningKey? = null, serverKey: SigningKey? = null,
createSession: () -> S, createSession: () -> S,
) { ) {
install(Routing)
install(WebSockets) { install(WebSockets) {
pingPeriod = 60.seconds //Duration.ofSeconds(15) pingPeriod = Duration.ofSeconds(15)
timeout = 45.seconds timeout = Duration.ofSeconds(15)
maxFrameSize = Long.MAX_VALUE maxFrameSize = Long.MAX_VALUE
masking = false masking = false
} }
@ -68,18 +39,13 @@ fun <S> Application.setupWebsocketServer(
log.debug { "opening the connection" } log.debug { "opening the connection" }
val input = Channel<UByteArray>(256) val input = Channel<UByteArray>(256)
val output = Channel<UByteArray>(256) val output = Channel<UByteArray>(256)
var useBinary: Boolean? = null
launch { launch {
log.debug { "starting output pump" } log.debug { "starting output pump" }
while (isActive) { while (isActive) {
try { try {
val block = output.receive() send(output.receive().toByteArray())
if (useBinary == false) }
send(block.asByteArray().encodeToBase64Compact()) catch(_: ClosedReceiveChannelException) {
else
send(block.toByteArray())
} catch (_: ClosedReceiveChannelException) {
log.debug { "closing output pump as output channel is closed" } log.debug { "closing output pump as output channel is closed" }
break break
} }
@ -98,28 +64,12 @@ fun <S> Application.setupWebsocketServer(
} }
log.debug { "KSC started, looking for incoming frames" } log.debug { "KSC started, looking for incoming frames" }
for (f in incoming) { for (f in incoming) {
log.debug { "incoming frame: ${f.frameType}" }
if (f is Frame.Binary)
try { try {
when (f) { input.send(f.readBytes().toUByteArray().also {
is Frame.Binary -> { log.debug { "in frame\n${it.toDump()}" }
if (useBinary == null) { })
log.debug { "Setting binary frame mode ------------------------------------" }
useBinary = true
}
input.send(f.readBytes().toUByteArray())
}
is Frame.Text -> {
if (useBinary == null) {
log.debug { "Setting text frame mode -----------------------------------" }
useBinary = false
}
input.send(f.readText().decodeBase64Compact().asUByteArray())
}
else -> {
log.warning { "unexpected frame type ${f.frameType}, ignoring" }
}
}
} catch (_: RemoteInterface.ClosedException) { } catch (_: RemoteInterface.ClosedException) {
log.warning { "caught local closed exception (strange!), closing" } log.warning { "caught local closed exception (strange!), closing" }
break break
@ -130,6 +80,8 @@ fun <S> Application.setupWebsocketServer(
log.exception { "unexpected exception, server connection will close" to t } log.exception { "unexpected exception, server connection will close" to t }
break break
} }
else
log.warning { "unknown frame type ${f.frameType}, ignoring" }
} }
log.debug { "closing the server" } log.debug { "closing the server" }
close() close()

127
src/jvmTest/kotlin/net/sergeych/kiloparsec/ClientTest.kt
View File

@ -1,30 +1,15 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec package net.sergeych.kiloparsec
import assertThrows import assertThrows
import io.ktor.server.engine.* import io.ktor.server.engine.*
import io.ktor.server.netty.* import io.ktor.server.netty.*
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.test.runTest import kotlinx.coroutines.test.runTest
import net.sergeych.crypto2.SigningSecretKey
import net.sergeych.crypto2.initCrypto import net.sergeych.crypto2.initCrypto
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.kiloparsec.adapter.websocketTransportDevice
import net.sergeych.mp_logger.Log import net.sergeych.mp_logger.Log
import java.net.InetAddress import java.net.InetAddress
import kotlin.random.Random
import kotlin.test.Test import kotlin.test.Test
import kotlin.test.assertEquals import kotlin.test.assertEquals
import kotlin.test.assertFalse import kotlin.test.assertFalse
@ -42,7 +27,7 @@ class ClientTest {
@Test @Test
fun webSocketTest1() = runTest { fun webSocketTest() = runTest {
initCrypto() initCrypto()
// fun Application. // fun Application.
val cmdClose by command<Unit,Unit>() val cmdClose by command<Unit,Unit>()
@ -53,7 +38,6 @@ class ClientTest {
Log.connectConsole(Log.Level.DEBUG) Log.connectConsole(Log.Level.DEBUG)
data class Session(var foo: String="not set") data class Session(var foo: String="not set")
var closeCounter = 0 var closeCounter = 0
val serverInterface = KiloInterface<Session>().apply { val serverInterface = KiloInterface<Session>().apply {
var connectedCalled = false var connectedCalled = false
@ -66,20 +50,17 @@ class ClientTest {
} }
} }
val port = Random.nextInt(8080, 9090) val ns: NettyApplicationEngine = embeddedServer(Netty, port = 8080, host = "0.0.0.0", module = {
val ns = embeddedServer(Netty, port = port, host = "0.0.0.0", module = {
setupWebsocketServer(serverInterface) { Session() } setupWebsocketServer(serverInterface) { Session() }
}).start(wait = false) }).start(wait = false)
val client = websocketClient<Unit>("ws://localhost:$port/kp") val client = websocketClient<Unit>("ws://localhost:8080/kp")
val states = mutableListOf<Boolean>() val states = mutableListOf<Boolean>()
val collector = launch { val collector = launch {
client.connectedStateFlow.collect { client.connectedStateFlow.collect {
println("got: $closeCounter/$it") println("got: $closeCounter/$it")
states += it states += it
if (!it) { if( !it) { closeCounter++ }
closeCounter++
}
} }
} }
assertEquals(true, client.call(cmdCheckConnected)) assertEquals(true, client.call(cmdCheckConnected))
@ -94,9 +75,6 @@ class ClientTest {
} }
// connection should now be closed // connection should now be closed
// the problem is: it needs some unspecified time to close
// as it is async process.
delay(100)
assertFalse { client.connectedStateFlow.value } assertFalse { client.connectedStateFlow.value }
// this should be run on automatically reopen connection // this should be run on automatically reopen connection
@ -110,101 +88,4 @@ class ClientTest {
// println("stopped server") // println("stopped server")
// println("closed client") // println("closed client")
} }
@Test
fun webSocketTest2() = runTest {
initCrypto()
// fun Application.
val cmdClose by command<Unit, Unit>()
val cmdGetFoo by command<Unit, String>()
val cmdSetFoo by command<String, Unit>()
val cmdCheckConnected by command<Unit, Boolean>()
Log.connectConsole(Log.Level.DEBUG)
data class Session(var foo: String = "not set")
var closeCounter = 0
val serverInterface = KiloInterface<Session>().apply {
var connectedCalled = false
onConnected { connectedCalled = true }
on(cmdGetFoo) { session.foo }
on(cmdSetFoo) { session.foo = it }
on(cmdCheckConnected) { connectedCalled }
on(cmdClose) {
throw LocalInterface.BreakConnectionException()
}
}
val port = Random.nextInt(8080, 9090)
val ns = embeddedServer(Netty, port = port, host = "0.0.0.0", module = {
setupWebsocketServer(serverInterface) { Session() }
}).start(wait = false)
val client = websocketClient<Unit>("ws://localhost:$port/kp", useTextFrames = true)
val states = mutableListOf<Boolean>()
val collector = launch {
client.connectedStateFlow.collect {
println("got: $closeCounter/$it")
states += it
if (!it) {
closeCounter++
}
}
}
assertEquals(true, client.call(cmdCheckConnected))
assertTrue { client.connectedStateFlow.value }
assertEquals("not set", client.call(cmdGetFoo))
client.call(cmdSetFoo, "foo")
assertEquals("foo", client.call(cmdGetFoo))
client.close()
ns.stop()
collector.cancel()
}
@Test
fun webSocketWaitForConnectTest() = runBlocking {
initCrypto()
// fun Application.
Log.connectConsole(Log.Level.DEBUG)
val clientInterface = KiloInterface<Unit>().apply {}
val port = Random.nextInt(8080, 9090)
var clientConnectCalls = 0
// It should repeatedly reconnect, and we will count:
KiloClient(clientInterface, SigningSecretKey.new()) {
clientConnectCalls++
KiloConnectionData(websocketTransportDevice("ws://localhost:$port/kp", true), Unit)
}
delay(1200)
// and check:
// println("connection attemtps: $clientConnectCalls")
assertTrue { clientConnectCalls > 1 }
}
@Test
fun webSocketWaitForConnectTest2() = runBlocking {
initCrypto()
// fun Application.
Log.connectConsole(Log.Level.DEBUG)
val clientInterface = KiloInterface<Unit>().apply {}
val port = Random.nextInt(8080, 9090)
var clientConnectCalls = 0
// It should repeatedly reconnect, and we will count:
KiloClient(clientInterface, SigningSecretKey.new()) {
clientConnectCalls++
KiloConnectionData(websocketTransportDevice("ws://localhost:$port/kp", false), Unit)
}
delay(1200)
// and check:
// println("connection attemtps: $clientConnectCalls")
assertTrue { clientConnectCalls > 1 }
}
} }

42
src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/TcpProvider.kt
View File

@ -1,19 +1,8 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec.adapter 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.utils.io.* import io.ktor.utils.io.*
import io.ktor.utils.io.writeByte
import kotlinx.coroutines.* import kotlinx.coroutines.*
import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.Channel
@ -22,11 +11,15 @@ import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.sync.withLock
import net.sergeych.kiloparsec.* import kotlinx.datetime.Clock
import net.sergeych.kiloparsec.AsyncVarint
import net.sergeych.kiloparsec.KiloClient
import net.sergeych.kiloparsec.KiloServer
import net.sergeych.kiloparsec.LocalInterface
import net.sergeych.mp_logger.* import net.sergeych.mp_logger.*
import net.sergeych.mp_tools.globalLaunch import net.sergeych.mp_tools.globalLaunch
import net.sergeych.tools.AtomicCounter import net.sergeych.tools.AtomicCounter
import kotlin.time.Clock import net.sergeych.tools.AtomicValue
import kotlin.time.Duration.Companion.seconds import kotlin.time.Duration.Companion.seconds
private val logCounter = AtomicCounter(0) private val logCounter = AtomicCounter(0)
@ -42,8 +35,8 @@ internal val PING_INACTIVITY_TIME = 30.seconds
*/ */
fun acceptTcpDevice(port: Int,localInterface: String = "0.0.0.0"): Flow<InetTransportDevice> { fun acceptTcpDevice(port: Int,localInterface: String = "0.0.0.0"): Flow<InetTransportDevice> {
val selectorManager = SelectorManager(Dispatchers.IO) val selectorManager = SelectorManager(Dispatchers.IO)
return flow {
val serverSocket = aSocket(selectorManager).tcp().bind(localInterface, port) val serverSocket = aSocket(selectorManager).tcp().bind(localInterface, port)
return flow {
while (true) { while (true) {
serverSocket.accept().let { sock -> serverSocket.accept().let { sock ->
emit(inetTransportDevice(sock, "srv")) emit(inetTransportDevice(sock, "srv"))
@ -52,13 +45,7 @@ fun acceptTcpDevice(port: Int, localInterface: String = "0.0.0.0"): Flow<InetTra
} }
} }
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]. [hostPortAddress] is a string in the form of `host:port`.
*/
suspend fun connectTcpDevice(hostPortAddress: String): InetTransportDevice =
connectTcpDevice(hostPortAddress.toNetworkAddress())
/** /**
* Connect to the TCP/IP server (see [KiloServer]) at the specified address and provide th compatible * Connect to the TCP/IP server (see [KiloServer]) at the specified address and provide th compatible
@ -87,7 +74,7 @@ private fun inetTransportDevice(
val outputBlocks = Channel<UByteArray>(4096) val outputBlocks = Channel<UByteArray>(4096)
val log = LogTag("TCPT${logCounter.incrementAndGet()}:$suffix:$networkAddress") val log = LogTag("TCPT${logCounter.incrementAndGet()}:$suffix:$networkAddress")
val job = AtomicAsyncValue<Job?>(null) val job = AtomicValue<Job?>(null)
val sockOutput = sock.openWriteChannel() val sockOutput = sock.openWriteChannel()
val sockInput = runCatching { sock.openReadChannel() }.getOrElse { val sockInput = runCatching { sock.openReadChannel() }.getOrElse {
@ -95,7 +82,7 @@ private fun inetTransportDevice(
throw IllegalStateException("failed to open read channel") throw IllegalStateException("failed to open read channel")
} }
suspend fun stop() { fun stop() {
job.mutate { job.mutate {
if ( it != null ) { if ( it != null ) {
log.debug { "stopping" } log.debug { "stopping" }
@ -104,7 +91,7 @@ private fun inetTransportDevice(
// The problem: on mac platofrms closing the socket does not close its input // The problem: on mac platofrms closing the socket does not close its input
// and output channels! // and output channels!
runCatching { sockInput.cancel() } runCatching { sockInput.cancel() }
runCatching { sockOutput.flushAndClose() } runCatching { sockOutput.close() }
if (!sock.isClosed) if (!sock.isClosed)
runCatching { runCatching {
log.debug { "closing socket by stop" } log.debug { "closing socket by stop" }
@ -121,8 +108,7 @@ private fun inetTransportDevice(
} }
var lastActiveAt = Clock.System.now() var lastActiveAt = Clock.System.now()
globalLaunch { job.value = globalLaunch {
job.reset(globalLaunch {
launch { launch {
log.debug { "opening read channel" } log.debug { "opening read channel" }
@ -140,7 +126,7 @@ private fun inetTransportDevice(
sockInput.readFully(data, 0, size) sockInput.readFully(data, 0, size)
inputBlocks.send(data.toUByteArray()) inputBlocks.send(data.toUByteArray())
} }
} catch (_: ClosedReceiveChannelException) { } catch (e: ClosedReceiveChannelException) {
log.error { "closed receive channel " } log.error { "closed receive channel " }
stop() stop()
break break
@ -154,7 +140,7 @@ private fun inetTransportDevice(
} }
} }
} }
})
launch { launch {
val outAccess = Mutex() val outAccess = Mutex()
var lastSentAt = Clock.System.now() var lastSentAt = Clock.System.now()

13
src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpBlock.kt
View File

@ -1,18 +1,7 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec.adapter package net.sergeych.kiloparsec.adapter
import io.ktor.network.sockets.* import io.ktor.network.sockets.*
import io.ktor.utils.io.core.* import io.ktor.utils.io.core.*
import kotlinx.io.readByteArray
import net.sergeych.crypto2.toDump import net.sergeych.crypto2.toDump
import net.sergeych.kiloparsec.adapter.UdpBlock.Companion.CANCEL_BLOCK import net.sergeych.kiloparsec.adapter.UdpBlock.Companion.CANCEL_BLOCK
import net.sergeych.kiloparsec.adapter.UdpBlock.Companion.ESCAPE_BYTE import net.sergeych.kiloparsec.adapter.UdpBlock.Companion.ESCAPE_BYTE
@ -108,6 +97,6 @@ sealed class UdpBlock {
} }
fun decode(datagram: Datagram) = fun decode(datagram: Datagram) =
decode(datagram.packet.readByteArray().toUByteArray()) decode(datagram.packet.readBytes().toUByteArray())
} }
} }

10
src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpConnector.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec.adapter package net.sergeych.kiloparsec.adapter
import io.ktor.network.sockets.* import io.ktor.network.sockets.*

16
src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpProvider.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec.adapter package net.sergeych.kiloparsec.adapter
import io.ktor.network.selector.* import io.ktor.network.selector.*
@ -97,7 +87,7 @@ fun acceptUdpDevice(
* the module automatically issues pings on inactivity when there is no data often enough * the module automatically issues pings on inactivity when there is no data often enough
* to maintain the connection open. * to maintain the connection open.
*/ */
suspend fun connectUdpDevice( fun connectUdpDevice(
hostPort: String, hostPort: String,
maxInactivityTimeout: Duration = 2.minutes, maxInactivityTimeout: Duration = 2.minutes,
) = connectUdpDevice(hostPort.toNetworkAddress(), maxInactivityTimeout) ) = connectUdpDevice(hostPort.toNetworkAddress(), maxInactivityTimeout)
@ -117,16 +107,16 @@ suspend fun connectUdpDevice(
* the module automatically issues pings on inactivity when there is no data often enough * the module automatically issues pings on inactivity when there is no data often enough
* to maintain the connection open. * to maintain the connection open.
*/ */
suspend fun connectUdpDevice( fun connectUdpDevice(
addr: NetworkAddress, addr: NetworkAddress,
maxInactivityTimeout: Duration = 2.minutes, maxInactivityTimeout: Duration = 2.minutes,
): InetTransportDevice { ): InetTransportDevice {
val selectorManager = SelectorManager(Dispatchers.IO) val selectorManager = SelectorManager(Dispatchers.IO)
val remoteAddress = InetSocketAddress(addr.host, addr.port) val remoteAddress = InetSocketAddress(addr.host, addr.port)
val socket = aSocket(selectorManager).udp().connect(remoteAddress)
val done = CompletableDeferred<Unit>() val done = CompletableDeferred<Unit>()
val socket = aSocket(selectorManager).udp().connect(remoteAddress)
val transport = UdpSocketTransport(object : UdpConnector { val transport = UdpSocketTransport(object : UdpConnector {
override suspend fun sendBlock(block: UdpBlock, toAddress: SocketAddress) { override suspend fun sendBlock(block: UdpBlock, toAddress: SocketAddress) {
socket.send(block.toDatagram(remoteAddress)) socket.send(block.toDatagram(remoteAddress))

25
src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpServer.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec.adapter package net.sergeych.kiloparsec.adapter
import io.ktor.network.selector.* import io.ktor.network.selector.*
@ -24,7 +14,6 @@ import net.sergeych.mp_logger.LogTag
import net.sergeych.mp_logger.Loggable import net.sergeych.mp_logger.Loggable
import net.sergeych.mp_logger.debug import net.sergeych.mp_logger.debug
import net.sergeych.mp_logger.exception import net.sergeych.mp_logger.exception
import net.sergeych.mp_tools.globalDefer
import kotlin.time.Duration import kotlin.time.Duration
import kotlin.time.Duration.Companion.minutes import kotlin.time.Duration.Companion.minutes
@ -62,9 +51,7 @@ class UdpServer(val port: Int, localInterface: String = "0.0.0.0", maxInactivity
private val access = Mutex() private val access = Mutex()
private val selectorManager = SelectorManager(Dispatchers.IO) private val selectorManager = SelectorManager(Dispatchers.IO)
private val serverSocket = globalDefer { private val serverSocket = aSocket(selectorManager).udp().bind(InetSocketAddress(localInterface, port))
aSocket(selectorManager).udp().bind(InetSocketAddress(localInterface, port))
}
override suspend fun disconnectClient(address: SocketAddress) { override suspend fun disconnectClient(address: SocketAddress) {
access.withLock { sessions.remove(address) } access.withLock { sessions.remove(address) }
@ -78,7 +65,7 @@ class UdpServer(val port: Int, localInterface: String = "0.0.0.0", maxInactivity
flow { flow {
while (true) { while (true) {
try { try {
val datagram = serverSocket.await().receive() val datagram = serverSocket.receive()
val block = UdpBlock.decode(datagram) val block = UdpBlock.decode(datagram)
val remoteAddress = datagram.address val remoteAddress = datagram.address
@ -110,10 +97,10 @@ class UdpServer(val port: Int, localInterface: String = "0.0.0.0", maxInactivity
} }
override suspend fun sendBlock(block: UdpBlock, toAddress: SocketAddress) { override suspend fun sendBlock(block: UdpBlock, toAddress: SocketAddress) {
serverSocket.await().send(block.toDatagram(toAddress)) serverSocket.send(block.toDatagram(toAddress))
} }
suspend fun isClosed(): Boolean = serverSocket.await().isClosed val isClosed: Boolean get() = serverSocket.isClosed
/** /**
* Close the UDP server. Calling it will cause: * Close the UDP server. Calling it will cause:
@ -126,8 +113,8 @@ class UdpServer(val port: Int, localInterface: String = "0.0.0.0", maxInactivity
*/ */
suspend fun close() { suspend fun close() {
access.withLock { access.withLock {
if (!isClosed()) { if (!isClosed) {
runCatching { serverSocket.await().close() } runCatching { serverSocket.close() }
} }
} }
while (sessions.isNotEmpty()) { while (sessions.isNotEmpty()) {

16
src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpSocketTransport.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec.adapter package net.sergeych.kiloparsec.adapter
import io.ktor.network.sockets.* import io.ktor.network.sockets.*
@ -19,13 +9,13 @@ import kotlinx.coroutines.channels.ClosedSendChannelException
import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.delay import kotlinx.coroutines.delay
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
import kotlinx.datetime.Clock
import net.sergeych.kiloparsec.SyncValue import net.sergeych.kiloparsec.SyncValue
import net.sergeych.mp_logger.Log import net.sergeych.mp_logger.Log
import net.sergeych.mp_logger.Loggable import net.sergeych.mp_logger.Loggable
import net.sergeych.mp_logger.debug import net.sergeych.mp_logger.debug
import net.sergeych.mp_logger.exception import net.sergeych.mp_logger.exception
import net.sergeych.mp_tools.globalLaunch import net.sergeych.mp_tools.globalLaunch
import kotlin.time.Clock
import kotlin.time.Duration import kotlin.time.Duration
/** /**
@ -171,10 +161,10 @@ internal class UdpSocketTransport(
while (!isClosed) { while (!isClosed) {
try { try {
server.sendBlock(UdpBlock.Data(outputDataBlocks.receive()), socketAddress) server.sendBlock(UdpBlock.Data(outputDataBlocks.receive()), socketAddress)
} catch (_: CancellationException) { } catch (e: CancellationException) {
// this is ok // this is ok
break break
} catch (_: ClosedReceiveChannelException) { } catch (e: ClosedReceiveChannelException) {
debug { "input channel is closed, closing" } debug { "input channel is closed, closing" }
close() close()
break break

50
src/ktorSocketTest/kotlin/net/sergeych/kiloparsec/adapter/InternetTest.kt
View File

@ -1,13 +1,3 @@
/*
* Copyright (c) 2025. Sergey S. Chernov - All Rights Reserved
*
* You may use, distribute and modify this code under the
* terms of the private license, which you must obtain from the author
*
* To obtain the license, contact the author: https://t.me/real_sergeych or email to
* real dot sergeych at gmail.
*/
package net.sergeych.kiloparsec.adapter package net.sergeych.kiloparsec.adapter
import assertThrows import assertThrows
@ -20,7 +10,7 @@ import kotlin.test.Test
import kotlin.test.assertEquals import kotlin.test.assertEquals
import kotlin.test.assertIs import kotlin.test.assertIs
class InternetTest { class InternetrTest {
class TestException : Exception("test1") class TestException : Exception("test1")
@Test @Test
@ -36,11 +26,9 @@ class InternetTest {
val cmdSave by command<String, Unit>() val cmdSave by command<String, Unit>()
val cmdLoad by command<Unit, String>() val cmdLoad by command<Unit, String>()
val cmdDrop by command<Unit, Unit>() val cmdDrop by command<Unit, Unit>()
val cmdPing by command<String, String>()
val cmdException by command<Unit, Unit>() val cmdException by command<Unit, Unit>()
val cmdCallClient by command<String, String>()
val serverInterface = KiloInterface<Session>().apply { val cli = KiloInterface<Session>().apply {
registerError { TestException() } registerError { TestException() }
onConnected { session.data = "start" } onConnected { session.data = "start" }
on(cmdSave) { session.data = it } on(cmdSave) { session.data = it }
@ -53,26 +41,15 @@ class InternetTest {
on(cmdDrop) { on(cmdDrop) {
throw LocalInterface.BreakConnectionException() throw LocalInterface.BreakConnectionException()
} }
on(cmdCallClient) {
remote.call(cmdPing, it)
} }
} val server = KiloServer(cli, acceptTcpDevice(port)) {
val server = KiloServer(serverInterface, acceptTcpDevice(port)) {
Session("unknown") Session("unknown")
} }
data class LocalSession(val localFoo: String) val client = KiloClient<Unit>() {
addErrors(cli)
val client = KiloClient {
addErrors(serverInterface)
session { LocalSession("unknown") }
// TODO: add register error variant // TODO: add register error variant
connect { connectTcpDevice("localhost:$port") } connect { connectTcpDevice("localhost:$port") }
local {
on(cmdPing) {
"pong! $it"
}
}
} }
assertEquals("start", client.call(cmdLoad)) assertEquals("start", client.call(cmdLoad))
@ -88,7 +65,7 @@ class InternetTest {
// reconnect? // reconnect?
assertEquals("start", client.call(cmdLoad)) assertEquals("start", client.call(cmdLoad))
assertEquals("pong! 42", client.call(cmdCallClient, "42"))
server.close() server.close()
} }
@ -127,7 +104,7 @@ class InternetTest {
val client = KiloClient<Unit>() { val client = KiloClient<Unit>() {
addErrors(cli) addErrors(cli)
connect { connectUdpDevice("127.0.0.1:$port") } connect { connectUdpDevice("localhost:$port") }
} }
assertEquals("start", client.call(cmdLoad)) assertEquals("start", client.call(cmdLoad))
@ -155,7 +132,7 @@ class InternetTest {
var data: String, var data: String,
) )
val port = 27170 + Random.nextInt(1, 900) val port = 27170 + Random.nextInt(1, 200)
val cmdSave by command<String, Unit>() val cmdSave by command<String, Unit>()
val cmdLoad by command<Unit, String>() val cmdLoad by command<Unit, String>()
@ -177,28 +154,19 @@ class InternetTest {
} }
} }
val uServer = UdpServer(port) val uServer = UdpServer(port)
KiloServer(cli, uServer.transportFlow) { KiloServer(cli, uServer.transportFlow) {
Session("unknown") Session("unknown")
} }
// second server
KiloServer(cli,acceptUdpDevice(uServer.port+1)) { Session("unknown2") }
val client = KiloClient<Unit>() { val client = KiloClient<Unit>() {
addErrors(cli) addErrors(cli)
connect { connectUdpDevice("127.0.0.1:$port") } connect { connectUdpDevice("localhost:$port") }
} }
val client2 = KiloClient<Unit>() { connect { connectUdpDevice("127.0.0.1:${port+1}") } }
assertEquals("start", client.call(cmdLoad)) assertEquals("start", client.call(cmdLoad))
assertEquals("start", client2.call(cmdLoad))
client.call(cmdSave, "foobar") client.call(cmdSave, "foobar")
client2.call(cmdSave, "buzz")
assertEquals("foobar", client.call(cmdLoad)) assertEquals("foobar", client.call(cmdLoad))
assertEquals("buzz", client2.call(cmdLoad))
val res = kotlin.runCatching { client.call(cmdException) } val res = kotlin.runCatching { client.call(cmdException) }
assertIs<TestException>(res.exceptionOrNull()) assertIs<TestException>(res.exceptionOrNull())