Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 sergeych
|
515278c264 | ||
|
sergeych
|
7100fa5f76 | ||
|
sergeych
|
77e293bef2 | ||
|
sergeych
|
d6f257de14 | ||
|
sergeych
|
8a21a836e5 |
No files matched your search
+1
-2
@@ -23,8 +23,7 @@ out/
|
||||
.project
|
||||
.settings
|
||||
.springBeans
|
||||
.sts4-cache
|
||||
bin/
|
||||
.sts4-caches
|
||||
!**/src/main/**/bin/
|
||||
!**/src/test/**/bin/
|
||||
|
||||
|
||||
Generated
+8
@@ -0,0 +1,8 @@
|
||||
<component name="ArtifactManager">
|
||||
<artifact type="jar" name="kiloparsec-js-0.2.4">
|
||||
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||
<root id="archive" name="kiloparsec-js-0.2.4.jar">
|
||||
<element id="module-output" name="kiloparsec.jsMain" />
|
||||
</root>
|
||||
</artifact>
|
||||
</component>
|
||||
Generated
+8
@@ -0,0 +1,8 @@
|
||||
<component name="ArtifactManager">
|
||||
<artifact type="jar" name="kiloparsec-js-0.2.5">
|
||||
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||
<root id="archive" name="kiloparsec-js-0.2.5.jar">
|
||||
<element id="module-output" name="kiloparsec.jsMain" />
|
||||
</root>
|
||||
</artifact>
|
||||
</component>
|
||||
@@ -0,0 +1,8 @@
|
||||
<component name="ArtifactManager">
|
||||
<artifact type="jar" name="kiloparsec-js-0.2.5-SNAPSHOT">
|
||||
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||
<root id="archive" name="kiloparsec-js-0.2.5-SNAPSHOT.jar">
|
||||
<element id="module-output" name="kiloparsec.jsMain" />
|
||||
</root>
|
||||
</artifact>
|
||||
</component>
|
||||
Generated
+6
@@ -0,0 +1,6 @@
|
||||
<component name="ArtifactManager">
|
||||
<artifact type="jar" name="kiloparsec-js-0.2.6">
|
||||
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||
<root id="archive" name="kiloparsec-js-0.2.6.jar" />
|
||||
</artifact>
|
||||
</component>
|
||||
@@ -0,0 +1,8 @@
|
||||
<component name="ArtifactManager">
|
||||
<artifact type="jar" name="kiloparsec-js-0.3.1-SNAPSHOT">
|
||||
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||
<root id="archive" name="kiloparsec-js-0.3.1-SNAPSHOT.jar">
|
||||
<element id="module-output" name="kiloparsec.jsMain" />
|
||||
</root>
|
||||
</artifact>
|
||||
</component>
|
||||
Generated
+8
@@ -0,0 +1,8 @@
|
||||
<component name="ArtifactManager">
|
||||
<artifact type="jar" name="kiloparsec-jvm-0.2.4">
|
||||
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||
<root id="archive" name="kiloparsec-jvm-0.2.4.jar">
|
||||
<element id="module-output" name="kiloparsec.jvmMain" />
|
||||
</root>
|
||||
</artifact>
|
||||
</component>
|
||||
Generated
+8
@@ -0,0 +1,8 @@
|
||||
<component name="ArtifactManager">
|
||||
<artifact type="jar" name="kiloparsec-jvm-0.2.5">
|
||||
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||
<root id="archive" name="kiloparsec-jvm-0.2.5.jar">
|
||||
<element id="module-output" name="kiloparsec.jvmMain" />
|
||||
</root>
|
||||
</artifact>
|
||||
</component>
|
||||
@@ -0,0 +1,8 @@
|
||||
<component name="ArtifactManager">
|
||||
<artifact type="jar" name="kiloparsec-jvm-0.2.5-SNAPSHOT">
|
||||
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||
<root id="archive" name="kiloparsec-jvm-0.2.5-SNAPSHOT.jar">
|
||||
<element id="module-output" name="kiloparsec.jvmMain" />
|
||||
</root>
|
||||
</artifact>
|
||||
</component>
|
||||
Generated
+6
@@ -0,0 +1,6 @@
|
||||
<component name="ArtifactManager">
|
||||
<artifact type="jar" name="kiloparsec-jvm-0.2.6">
|
||||
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||
<root id="archive" name="kiloparsec-jvm-0.2.6.jar" />
|
||||
</artifact>
|
||||
</component>
|
||||
@@ -0,0 +1,8 @@
|
||||
<component name="ArtifactManager">
|
||||
<artifact type="jar" name="kiloparsec-jvm-0.3.1-SNAPSHOT">
|
||||
<output-path>$PROJECT_DIR$/build/libs</output-path>
|
||||
<root id="archive" name="kiloparsec-jvm-0.3.1-SNAPSHOT.jar">
|
||||
<element id="module-output" name="kiloparsec.jvmMain" />
|
||||
</root>
|
||||
</artifact>
|
||||
</component>
|
||||
Generated
+6
@@ -0,0 +1,6 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project version="4">
|
||||
<component name="ScalaCompilerConfiguration">
|
||||
<option name="separateProdTestSources" value="false" />
|
||||
</component>
|
||||
</project>
|
||||
@@ -1,16 +1,24 @@
|
||||
# Kiloparsec
|
||||
|
||||
__Recommended version is `0.3.1`: to keep the code compatible with current and further versions we
|
||||
ask to upgrade to `0.3.1` at least.__
|
||||
|
||||
The new generation of __PARanoid SECurity__ protocol, advanced, faster, more secure. It also allows connecting any "
|
||||
block device" transport to the same local interface. Out if the box it
|
||||
provides the following transports:
|
||||
|
||||
| name | JVM | JS | native |
|
||||
|----------------|-----|----|----------|
|
||||
| TCP/IP server | ✓ | | β @0.2.6 |
|
||||
| TCP/IP client | ✓ | | β @0.2.6 |
|
||||
| TCP/IP server | ✓ | | >= 0.2.6 |
|
||||
| TCP/IP client | ✓ | | >= 0.2.6 |
|
||||
| Websock server | ✓ | | |
|
||||
| Websock client | ✓ | ✓ | ✓ |
|
||||
|
||||
### Note on version compatibility
|
||||
|
||||
Protocols >= 0.3.0 are not binary compatible with previous version due to more compact binary
|
||||
format. The format from 0.3.0 onwards is supposed to keep compatible.
|
||||
|
||||
### Supported native targets
|
||||
|
||||
- iosArm64, iosX64
|
||||
@@ -59,7 +67,7 @@ It could be, depending on your project structure, something like:
|
||||
```kotlin
|
||||
val commonMain by getting {
|
||||
dependencies {
|
||||
api("net.sergeych:kiloparsec:0.2.6")
|
||||
api("net.sergeych:kiloparsec:0.3.1")
|
||||
}
|
||||
}
|
||||
```
|
||||
@@ -118,7 +126,7 @@ assertEquals(FooArgs("bar", 117), client.call(cmdGetFoo))
|
||||
## Create ktor-based server
|
||||
|
||||
Normally server side needs some session. It is convenient and avoid sending repeating data on each request speeding up
|
||||
the protocol. With KILOPARSEC it is rather basic operation:\
|
||||
the protocol. With KILOPARSEC it is rather basic operation:
|
||||
|
||||
~~~kotlin
|
||||
// Our session just keeps Foo for cmd{Get|Set}Foo:
|
||||
@@ -143,9 +151,48 @@ val ns: NettyApplicationEngine = embeddedServer(Netty, port = 8080, host = "0.0.
|
||||
setupWebsocketServer(serverInterface) { Session() }
|
||||
}).start(wait = false)
|
||||
|
||||
|
||||
~~~
|
||||
|
||||
## Create TCP/IP client and server
|
||||
|
||||
Using plain TCP/IP is even simpler, and it works way faster than websocket one, and is _the same
|
||||
protected as `wss://` (and `ws://`) variant above due to same kiloparsec encryption in both cases. Still, a TCP/IP
|
||||
client is not available in Javascript browser targets and custom TCP ports could often be blocked by firewalls.
|
||||
|
||||
Documentation is available in samples here:
|
||||
|
||||
- [TCP/IP server creation](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec/-kilo-server/index.html)
|
||||
|
||||
- [TCP/IP client](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec/-kilo-client/index.html)
|
||||
|
||||
In short, there are two functions that implements aysnchronous TCP/IP transport on all platforms buy JS:
|
||||
|
||||
- [acceptTcpDevice](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec.adapter/accept-tcp-device.html?query=fun%20acceptTcpDevice(port:%20Int):%20Flow%3CInetTransportDevice%3E) to create a server
|
||||
|
||||
- [connectTcpDevice](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec.adapter/connect-tcp-device.html) to connect to the server
|
||||
|
||||
## Reusing code between servers
|
||||
|
||||
The same instance of the [KiloInterface](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec/-kilo-interface/index.html?query=open%20class%20KiloInterface%3CS%3E%20:%20LocalInterface%3CKiloScope%3CS%3E%3E) could easily be reused with all instances of servers with different protocols.
|
||||
|
||||
This is a common proactive to create a business logic in a `KiloInterface`, then create a TCP/IP and Websocket servers passing the same instance of the logic to both.
|
||||
|
||||
## Note on the server identification
|
||||
|
||||
We do not recommend to rely on TLS (HTTPS://, WSS://) host identification solely, in the modern world there is
|
||||
a high probability of attacks on unfriendly (in respect to at least some of your users) states to the SSL certificates
|
||||
chain, in which case the MITM and spoofing will be undetected. Check the [remoteId](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec/-kilo-client/remote-id.html?query=suspend%20fun%20remoteId():%20VerifyingPublicKey?) in your client on each connection and provide the safe [serverSecretKey](https://code.sergeych.net/docs/kiloparsec/kiloparsec/net.sergeych.kiloparsec/-kilo-server/index.html?query=class%20KiloServer%3CS%3E(clientInterface:%20KiloInterface%3CS%3E,%20connections:%20Flow%3CInetTransportDevice%3E,%20serverSecretKey:%20SigningKey?%20=%20null,%20sessionBuilder:%20()%20-%3E%20S) when creating a server.
|
||||
|
||||
This will effectively protetcs against certificate chain spoofing in the case of the application installed from the trusted source.
|
||||
|
||||
__Important note__. The web application could not be completely secured this way unless is loaded from the IP-address, as the DNS could be spoofed the same, especially when used with `Cloudflare` or other CDN that can
|
||||
transparently substitute the whole site. In the case of we applications we strongly recommend not to use CDN except your own where you can control actual traffic rules.
|
||||
|
||||
## See also:
|
||||
|
||||
- [Source documentation](https://code.sergeych.net/docs/kiloparsec/)
|
||||
- [Project's WIKI](https://gitea.sergeych.net/sergeych/kiloparsec/wiki)
|
||||
|
||||
# Details
|
||||
|
||||
It is not compatible with parsec family and no more based on an Universa crypto library. To better fit
|
||||
|
||||
Executable
+4
@@ -0,0 +1,4 @@
|
||||
#!/bin/bash
|
||||
set -e
|
||||
./gradlew dokkaHtml
|
||||
rsync -avz ./build/dokka/* code.sergeych.net:/bigstore/sergeych_pub/code/docs/kiloparsec
|
||||
+12
-1
@@ -2,10 +2,11 @@ plugins {
|
||||
kotlin("multiplatform") version "2.0.0"
|
||||
id("org.jetbrains.kotlin.plugin.serialization") version "2.0.0"
|
||||
`maven-publish`
|
||||
id("org.jetbrains.dokka") version "1.9.20"
|
||||
}
|
||||
|
||||
group = "net.sergeych"
|
||||
version = "0.2.6"
|
||||
version = "0.3.1"
|
||||
|
||||
repositories {
|
||||
mavenCentral()
|
||||
@@ -135,3 +136,13 @@ kotlin {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
tasks.dokkaHtml.configure {
|
||||
outputDirectory.set(buildDir.resolve("dokka"))
|
||||
dokkaSourceSets {
|
||||
configureEach {
|
||||
// includes.from("docs/bipack.md")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -19,6 +19,11 @@ inline fun <reified A, reified R> command(overrideName: String? = null): Command
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Declare a Push: Unit-returning command usually used with [RemoteInterface.push]
|
||||
*/
|
||||
inline fun <reified A> push(overrideName: String? = null): CommandDelegate<A, Unit> = command(overrideName)
|
||||
|
||||
/**
|
||||
* Delegate to create [Command] via property
|
||||
*/
|
||||
|
||||
@@ -17,10 +17,33 @@ import net.sergeych.mp_logger.exception
|
||||
import net.sergeych.mp_tools.globalLaunch
|
||||
|
||||
/**
|
||||
* The auto-connecting client that reconnects to the kiloparsec server
|
||||
* The auto-connecting client that reconnects to the kiloparsec server,
|
||||
* [KiloServer],
|
||||
* and maintain connection state flow. Client factory launches a disconnected
|
||||
* set of coroutines to support automatic reconnection, so you _must_ [close]
|
||||
* it manually when it is not needed, otherwise it will continue to reconnect.
|
||||
* it manually when it is unnecessary, otherwise it will continue to reconnect.
|
||||
*
|
||||
* ## Usage
|
||||
*
|
||||
* Suppose we have TCP/IP server as in the [KiloServer] usage sample. Then we can connect
|
||||
* to it providing TCP/IP connector like:
|
||||
*
|
||||
* ```kotlin
|
||||
* val client = KiloClient<Unit>() {
|
||||
* connect { connectTcpDevice("localhost:$port") }
|
||||
* }
|
||||
*
|
||||
* // now we can invoke remote commands:
|
||||
* assertEquals("unknown", client.call(cmdLoad))
|
||||
*
|
||||
* client.call(cmdSave, "foobar")
|
||||
* assertEquals("foobar", client.call(cmdLoad))
|
||||
* ```
|
||||
*
|
||||
* ## See also
|
||||
*
|
||||
* [KiloServer]
|
||||
*
|
||||
*/
|
||||
class KiloClient<S>(
|
||||
val localInterface: KiloInterface<S>,
|
||||
@@ -28,7 +51,6 @@ class KiloClient<S>(
|
||||
connectionDataFactory: ConnectionDataFactory<S>,
|
||||
) : RemoteInterface,
|
||||
Loggable by LogTag("CLIF") {
|
||||
|
||||
val _state = MutableStateFlow(false)
|
||||
|
||||
/**
|
||||
@@ -94,6 +116,15 @@ class KiloClient<S>(
|
||||
throw t
|
||||
}
|
||||
|
||||
override suspend fun <A> push(cmd: Command<A, Unit>, args: A) {
|
||||
try {
|
||||
deferredClient.await().push(cmd, args)
|
||||
} catch (t: RemoteInterface.ClosedException) {
|
||||
resetDeferredClient()
|
||||
throw t
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Current session token. This is a per-connection unique random value same on the client and server part so
|
||||
* it could be used as a nonce to pair MITM and like attacks, be sure that the server is actually
|
||||
|
||||
@@ -100,4 +100,8 @@ class KiloClientConnection<S>(
|
||||
suspend fun token() = deferredParams.await().token
|
||||
override suspend fun <A, R> call(cmd: Command<A, R>, args: A): R =
|
||||
kiloRemoteInterface.await().call(cmd, args)
|
||||
|
||||
override suspend fun <A> push(cmd: Command<A, Unit>, args: A) {
|
||||
kiloRemoteInterface.await().push(cmd, args)
|
||||
}
|
||||
}
|
||||
@@ -4,7 +4,13 @@ package net.sergeych.kiloparsec
|
||||
* The local interface to provide functions, register errors for Kiloparsec users. Use it
|
||||
* with [KiloClient], [KiloClientConnection], [KiloServerConnection], etc.
|
||||
*
|
||||
* BAse implementation registers relevant exceptions.
|
||||
* Base class implementation does the following:
|
||||
*
|
||||
* - It registers common exceptions from [RemoteInterface] and kotlin/java `IllegalArgumentException` and
|
||||
* `IllegalStateException`
|
||||
* - It provides [onConnected] handler
|
||||
*
|
||||
* See [KiloServer] for usage sample.
|
||||
*/
|
||||
open class KiloInterface<S> : LocalInterface<KiloScope<S>>() {
|
||||
|
||||
|
||||
@@ -35,5 +35,10 @@ class KiloRemoteInterface<S>(
|
||||
else -> throw RemoteInterface.Exception("unexpected block type: $block")
|
||||
}
|
||||
}
|
||||
|
||||
override suspend fun <A> push(cmd: Command<A, Unit>, args: A) {
|
||||
val params = deferredParams.await()
|
||||
params.transport.call(L0Call, params.encrypt(cmd.packCall(args)))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -13,13 +13,100 @@ import net.sergeych.mp_tools.globalLaunch
|
||||
import net.sergeych.tools.AtomicCounter
|
||||
|
||||
private val instances = AtomicCounter()
|
||||
@Suppress("unused")
|
||||
|
||||
/**
|
||||
* The Kiloparsec server.
|
||||
* Server accepts incoming connections and serves them using the same [clientInterface].
|
||||
*
|
||||
* ## Incoming connections
|
||||
*
|
||||
* Server collecting incoming connections provided by [connections] `Flow`. For each incoming connection
|
||||
* the Kiloparsec handshake is performed, then the session object is created, see below, and connection is
|
||||
* served with [clientInterface] until closed.
|
||||
*
|
||||
* ## Session param [S]
|
||||
*
|
||||
* After the successful handshake server creates new session for each connection calling the [sessionBuilder].
|
||||
* Then it creates a [KiloScope] so [KiloScope.session] holds this connection-specific instance. Then
|
||||
* [KiloInterface.onConnected] is called with this scope. Session can be used to hold connection state. Session
|
||||
* objects are not persistent, but could be initialized in [KiloInterface.onConnected] where the remote side
|
||||
* [KiloScope.remoteIdentity] is already verified and set.
|
||||
*
|
||||
* ## Usage:
|
||||
*
|
||||
* Create a shared library between you server and clients, to specify the interface (otherwise you can
|
||||
* share sources).
|
||||
*
|
||||
* Suppose we have session with a state:
|
||||
* ```kotlin
|
||||
* data class Session(
|
||||
* var data: String,
|
||||
* )
|
||||
*```
|
||||
*
|
||||
* And some commands to access and change it, in the shared library too:
|
||||
*
|
||||
* ```kotlin
|
||||
* val cmdSave by command<String, Unit>()
|
||||
* val cmdLoad by command<Unit, String>()
|
||||
* val cmdDrop by command<Unit, Unit>()
|
||||
* val cmdException by command<Unit, Unit>()
|
||||
* ```
|
||||
* Then the server code (TCP/IP variant) could look like:
|
||||
*
|
||||
* ```kotlin
|
||||
* // The server implementation (could be shared between server instances connected
|
||||
* // to different protocol adapters):
|
||||
*
|
||||
* val cli = KiloInterface<Session>().apply {
|
||||
* // Suppose we want to throw this exception at the caller site, so we need to register it:
|
||||
* registerError { SomeException() }
|
||||
*
|
||||
* // Session initialization. If you need a sessino to depend initially on the client's identity.
|
||||
* // you can do it here:
|
||||
* onConnected {
|
||||
* // check the remoteIdentity
|
||||
* session.data = if( remoteIdentity == somePublicVerifyingKey )
|
||||
* "known"
|
||||
* else
|
||||
* "unknown"
|
||||
* }
|
||||
* on(cmdSave) { session.data = it }
|
||||
* on(cmdLoad) {
|
||||
* session.data
|
||||
* }
|
||||
* on(cmdException) {
|
||||
* throw TestException()
|
||||
* }
|
||||
* on(cmdDrop) {
|
||||
* throw LocalInterface.BreakConnectionException()
|
||||
* }
|
||||
* }
|
||||
*
|
||||
* // Create the server instance that accepts incoming TCP/IP connections on all local interfaces using the
|
||||
* // specified port:
|
||||
*
|
||||
* val server = KiloServer(cli, acceptTcpDevice(port)) {
|
||||
* // This creates a new session
|
||||
* Session("unknown")
|
||||
* }
|
||||
* ```
|
||||
*
|
||||
* See [KiloClient] to connect to the server.
|
||||
*
|
||||
* @param S the type of the server session object, returned by [sessionBuilder]. See above
|
||||
* @param clientInterface the interface available for remote calls
|
||||
* @param connections flow of incoming connections. Server stops when the flow is fully collected (normally
|
||||
* it shouldn't)
|
||||
* @param serverSecretKey the [SigningKey] used to identify this server during Kiloparsec handshake.
|
||||
* @param sessionBuilder callback that creates session objects for successful incoming connections
|
||||
*/
|
||||
class KiloServer<S>(
|
||||
private val clientInterface: KiloInterface<S>,
|
||||
private val connections: Flow<InetTransportDevice>,
|
||||
private val serverSecretKey: SigningKey? = null,
|
||||
private val sessionBuilder: ()->S,
|
||||
): LogTag("KS:${instances.incrementAndGet()}") {
|
||||
private val sessionBuilder: () -> S,
|
||||
) : LogTag("KS:${instances.incrementAndGet()}") {
|
||||
|
||||
private val job = globalLaunch {
|
||||
connections.collect { device ->
|
||||
@@ -27,22 +114,30 @@ class KiloServer<S>(
|
||||
try {
|
||||
info { "connected ${device}" }
|
||||
KiloServerConnection(clientInterface, device, sessionBuilder(), serverSecretKey)
|
||||
.apply { debug { "server connection is ready" }}
|
||||
.apply { debug { "server connection is ready" } }
|
||||
.run()
|
||||
}
|
||||
catch(_: CancellationException) {
|
||||
}
|
||||
catch(cce: LocalInterface.BreakConnectionException) {
|
||||
} catch (_: CancellationException) {
|
||||
} catch (cce: LocalInterface.BreakConnectionException) {
|
||||
info { "Closed exception caught, closing (${cce.flushSendQueue}" }
|
||||
}
|
||||
catch (t: Throwable) {
|
||||
} catch (t: Throwable) {
|
||||
exception { "unexpected while creating kiloclient" to t }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop the server and cancel all pending sessions. Unlike finishing the flow passed
|
||||
* for [KiloServer.connections], it will cancel all currently active sessions.
|
||||
*/
|
||||
fun close() {
|
||||
job.cancel()
|
||||
}
|
||||
|
||||
/**
|
||||
* Server is closed either if [close] was called or all [KiloServer.connections] were collected and flow was
|
||||
* closed.
|
||||
*/
|
||||
@Suppress("unused")
|
||||
val isClosed: Boolean get() = job.isCompleted
|
||||
}
|
||||
@@ -94,4 +94,8 @@ class KiloServerConnection<S>(
|
||||
override suspend fun <A, R> call(cmd: Command<A, R>, args: A): R {
|
||||
return kiloRemoteInterface.await().call(cmd, args)
|
||||
}
|
||||
|
||||
override suspend fun <A> push(cmd: Command<A, Unit>, args: A) {
|
||||
kiloRemoteInterface.await().push(cmd, args)
|
||||
}
|
||||
}
|
||||
@@ -50,4 +50,18 @@ interface RemoteInterface {
|
||||
* Call the remote procedure with specified args and return its result
|
||||
*/
|
||||
suspend fun <A, R> call(cmd: Command<A, R>, args: A): R
|
||||
|
||||
/**
|
||||
* Push the notification without waiting for reception or processing.
|
||||
* It returns immediately after sending data to the transport (e.g., to the network).
|
||||
* Use [call] if it is necessary to wait until the command will be received and processed by the remote.
|
||||
*/
|
||||
suspend fun <A> push(cmd: Command<A, Unit>, args: A)
|
||||
|
||||
/**
|
||||
* Push the command with no args.
|
||||
* It returns immediately after sending data to the transport (e.g., to the network).
|
||||
* Use [call] if it is necessary to wait until the command will be received and processed by the remote.
|
||||
*/
|
||||
suspend fun push(cmd: Command<Unit,Unit>) = push(cmd,Unit)
|
||||
}
|
||||
@@ -15,6 +15,7 @@ import kotlinx.serialization.descriptors.SerialDescriptor
|
||||
import kotlinx.serialization.encoding.Decoder
|
||||
import kotlinx.serialization.encoding.Encoder
|
||||
import kotlinx.serialization.serializer
|
||||
import net.sergeych.bipack.Unsigned
|
||||
import net.sergeych.crypto2.toDump
|
||||
import net.sergeych.kiloparsec.Transport.Device
|
||||
import net.sergeych.mp_logger.*
|
||||
@@ -63,7 +64,12 @@ class Transport<S>(
|
||||
@Serializable(TransportBlockSerializer::class)
|
||||
sealed class Block {
|
||||
@Serializable
|
||||
data class Call(val id: UInt, val name: String, val packedArgs: UByteArray) : Block() {
|
||||
data class Call(
|
||||
@Unsigned
|
||||
val id: UInt,
|
||||
val name: String,
|
||||
val packedArgs: UByteArray
|
||||
) : Block() {
|
||||
override fun equals(other: Any?): Boolean {
|
||||
if (this === other) return true
|
||||
if (other !is Call) return false
|
||||
@@ -83,10 +89,10 @@ class Transport<S>(
|
||||
}
|
||||
|
||||
@Serializable
|
||||
data class Response(val forId: UInt, val packedResult: UByteArray) : Block()
|
||||
data class Response(@Unsigned val forId: UInt, val packedResult: UByteArray) : Block()
|
||||
|
||||
@Serializable
|
||||
data class Error(val forId: UInt, val code: String, val text: String? = null, val extra: UByteArray? = null) :
|
||||
data class Error(@Unsigned val forId: UInt, val code: String, val text: String? = null, val extra: UByteArray? = null) :
|
||||
Block() {
|
||||
val message by lazy { text ?: "remote exception: $code" }
|
||||
}
|
||||
@@ -98,7 +104,8 @@ class Transport<S>(
|
||||
var isClosed: Boolean = false
|
||||
|
||||
/**
|
||||
* Send a call block for a command and packed args and return packed result if it is not an error
|
||||
* Send a call block for a command and packed args and return packed result if it is not an error. It suspends
|
||||
* until receiving answer from the remote side, even if returns `Unit`.
|
||||
* @throws RemoteInterface.RemoteException if the remote call caused an exception. Normally use [call] instead.
|
||||
* @throws RemoteInterface.ClosedException
|
||||
*/
|
||||
@@ -111,12 +118,13 @@ class Transport<S>(
|
||||
// We need to shield calls and lastID with mutex, but nothing more:
|
||||
access.withLock {
|
||||
if (isClosed) throw RemoteInterface.ClosedException()
|
||||
// the order is important: first id in use MUST BE >= 1, not zero:
|
||||
b = Block.Call(++lastId, name, packedArgs)
|
||||
calls[b.id] = deferred
|
||||
}
|
||||
|
||||
// now we have mutex freed so we can call:
|
||||
val r = runCatching { device.output.send(pack(b).also { debug { ">>>\n${it.toDump()}" } }) }
|
||||
val r = runCatching { device.output.send(pack(b).also { debug { ">>>\n${it.toDump()}" } }) }
|
||||
if (!r.isSuccess) {
|
||||
r.exceptionOrNull()?.let {
|
||||
exception { "failed to send output block" to it }
|
||||
@@ -131,6 +139,25 @@ class Transport<S>(
|
||||
return deferred.await()
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a call block for a command and packed args and return packed result if it is not an error. It suspends
|
||||
* until receiving answer from the remote side, even if returns `Unit`.
|
||||
* @throws RemoteInterface.RemoteException if the remote call caused an exception. Normally use [call] instead.
|
||||
* @throws RemoteInterface.ClosedException
|
||||
*/
|
||||
private suspend fun sendPushBlock(name: String, packedArgs: UByteArray) {
|
||||
if (isClosed) throw RemoteInterface.ClosedException()
|
||||
|
||||
// All push blocks have the same id == 0:
|
||||
val b = Block.Call(0u, name, packedArgs)
|
||||
val r = runCatching { device.output.send(pack(b).also { debug { ">>$\n${it.toDump()}" } }) }
|
||||
when(val e = r.exceptionOrNull()) {
|
||||
is RemoteInterface.ClosedException, is CancellationException, is RemoteInterface.RemoteException
|
||||
-> throw e
|
||||
else -> throw RemoteInterface.ClosedException()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Call the remote procedure with specified args and return its result
|
||||
*/
|
||||
@@ -139,6 +166,10 @@ class Transport<S>(
|
||||
return unpack(cmd.resultSerializer, result)
|
||||
}
|
||||
|
||||
override suspend fun <A>push(cmd: Command<A,Unit>,args: A) {
|
||||
sendPushBlock(cmd.name, pack(cmd.argsSerializer, args))
|
||||
}
|
||||
|
||||
/**
|
||||
* Start running the transport. This function suspends until the transport is closed
|
||||
* normally or by error. If you need to cancel it prematurely, cancel the coroutine
|
||||
@@ -163,7 +194,7 @@ class Transport<S>(
|
||||
warning { "decoded error: ${error::class.simpleName}: $error" }
|
||||
calls.remove(b.forId)?.completeExceptionally(localInterface.decodeError(b))
|
||||
?: warning { "error handler not found for ${b.forId}" }
|
||||
info { "error processed"}
|
||||
info { "error processed" }
|
||||
}
|
||||
|
||||
is Block.Response -> access.withLock {
|
||||
@@ -176,17 +207,21 @@ class Transport<S>(
|
||||
|
||||
is Block.Call -> launch {
|
||||
try {
|
||||
send(
|
||||
Block.Response(
|
||||
b.id,
|
||||
localInterface.execute(commandContext, b.name, b.packedArgs)
|
||||
if (b.id == 0u)
|
||||
// Command does not waits return
|
||||
localInterface.execute(commandContext, b.name, b.packedArgs)
|
||||
else
|
||||
send(
|
||||
Block.Response(
|
||||
b.id,
|
||||
localInterface.execute(commandContext, b.name, b.packedArgs)
|
||||
)
|
||||
)
|
||||
)
|
||||
} catch (x: LocalInterface.BreakConnectionException) {
|
||||
// handler forced close
|
||||
warning { "handler requested closing of the connection (${x.flushSendQueue}"}
|
||||
warning { "handler requested closing of the connection (${x.flushSendQueue}" }
|
||||
isClosed = true
|
||||
if( x.flushSendQueue ) device.flush()
|
||||
if (x.flushSendQueue) device.flush()
|
||||
device.close()
|
||||
} catch (x: RemoteInterface.RemoteException) {
|
||||
send(Block.Error(b.id, x.code, x.text, x.extra))
|
||||
@@ -196,26 +231,23 @@ class Transport<S>(
|
||||
.also { debug { "command executed: ${b.name}" } }
|
||||
}
|
||||
}
|
||||
debug { "=---------------------------------------------"}
|
||||
debug { "=---------------------------------------------" }
|
||||
}
|
||||
debug { "input step performed closed=$isClosed active=$isActive"}
|
||||
debug { "input step performed closed=$isClosed active=$isActive" }
|
||||
} catch (_: ClosedSendChannelException) {
|
||||
info { "closed send channel" }
|
||||
isClosed = true
|
||||
} catch (_: ClosedReceiveChannelException) {
|
||||
info { "closed receive channel"}
|
||||
info { "closed receive channel" }
|
||||
isClosed = true
|
||||
}
|
||||
catch(cce: LocalInterface.BreakConnectionException) {
|
||||
info { "closing connection by local request ($cce)"}
|
||||
} catch (cce: LocalInterface.BreakConnectionException) {
|
||||
info { "closing connection by local request ($cce)" }
|
||||
device.close()
|
||||
}
|
||||
catch(t: RemoteInterface.ClosedException) {
|
||||
} catch (t: RemoteInterface.ClosedException) {
|
||||
// it is ok: we just exit the coroutine normally
|
||||
// and mark we're closing
|
||||
isClosed = true
|
||||
}
|
||||
catch (_: CancellationException) {
|
||||
} catch (_: CancellationException) {
|
||||
info { "loop is cancelled with CancellationException" }
|
||||
isClosed = true
|
||||
} catch (t: Throwable) {
|
||||
@@ -226,7 +258,7 @@ class Transport<S>(
|
||||
}
|
||||
debug { "leaving transport loop" }
|
||||
access.withLock {
|
||||
debug { "access lock obtained"}
|
||||
debug { "access lock obtained" }
|
||||
isClosed = true
|
||||
debug { "closing device $device, calls in queue ${calls.size}" }
|
||||
runCatching { device.close() }
|
||||
@@ -243,8 +275,7 @@ class Transport<S>(
|
||||
private suspend fun send(block: Block) {
|
||||
try {
|
||||
device.output.send(pack(block))
|
||||
}
|
||||
catch(_: ClosedSendChannelException) {
|
||||
} catch (_: ClosedSendChannelException) {
|
||||
throw RemoteInterface.ClosedException()
|
||||
}
|
||||
}
|
||||
@@ -275,7 +306,7 @@ object TransportBlockSerializer : KSerializer<Transport.Block> {
|
||||
|
||||
|
||||
override fun deserialize(decoder: Decoder): Transport.Block =
|
||||
when( val id = decoder.decodeByte().toInt()) {
|
||||
when (val id = decoder.decodeByte().toInt()) {
|
||||
0 -> decoder.decodeSerializableValue(serializer<Transport.Block.Call>())
|
||||
1 -> decoder.decodeSerializableValue(serializer<Transport.Block.Error>())
|
||||
2 -> decoder.decodeSerializableValue(serializer<Transport.Block.Response>())
|
||||
|
||||
@@ -11,46 +11,3 @@ data class NetworkAddress(
|
||||
return "$host:$port"
|
||||
}
|
||||
}
|
||||
//
|
||||
///**
|
||||
// * Multiplatform datagram abstraction
|
||||
// */
|
||||
//interface Datagram {
|
||||
// /**
|
||||
// * Received message
|
||||
// */
|
||||
// val message: UByteArray
|
||||
//
|
||||
// /**
|
||||
// * Address from where the message was sent
|
||||
// */
|
||||
// val address: NetworkAddress
|
||||
//}
|
||||
//
|
||||
//@OptIn(ExperimentalStdlibApi::class)
|
||||
//interface DatagramConnector: AutoCloseable {
|
||||
//
|
||||
// val incoming: ReceiveChannel<Datagram>
|
||||
// suspend fun send(message: UByteArray, networkAddress: NetworkAddress)
|
||||
// @Suppress("unused")
|
||||
// suspend fun send(message: UByteArray, datagramAddress: String) {
|
||||
// send(message, datagramAddress.toNetworkAddress())
|
||||
// }
|
||||
//
|
||||
// suspend fun send(message: UByteArray,host: String,port: Int) =
|
||||
// send(message, NetworkAddress(host,port))
|
||||
// override fun close()
|
||||
//}
|
||||
//
|
||||
//expect fun NetworkAddress(host: String,port: Int): NetworkAddress
|
||||
//
|
||||
//fun String.toNetworkAddress() : NetworkAddress {
|
||||
// val (host, port) = this.split(":").map { it.trim()}
|
||||
// return NetworkAddress(host, port.toInt())
|
||||
//}
|
||||
//
|
||||
//expect fun acceptTcpDevice(port: Int): Flow<InetTransportDevice>
|
||||
//
|
||||
//expect suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice
|
||||
//
|
||||
//suspend fun connectTcpDevice(address: String) = connectTcpDevice(address.toNetworkAddress())
|
||||
@@ -184,10 +184,16 @@ class TransportTest {
|
||||
val cmdRemoteExceptionTest by command<Unit, String>()
|
||||
val cmdBreak by command<Unit, Unit>()
|
||||
|
||||
val cmdPushServer by push<String>()
|
||||
val pushedFromServer = CompletableDeferred<String>()
|
||||
|
||||
val serverInterface = KiloInterface<String>().apply {
|
||||
on(cmdPing) {
|
||||
"pong! [$it]"
|
||||
}
|
||||
on(cmdPushServer) {
|
||||
pushedFromServer.complete(it)
|
||||
}
|
||||
on(cmdGetToken) {
|
||||
sessionToken
|
||||
}
|
||||
@@ -249,11 +255,17 @@ class TransportTest {
|
||||
assertEquals("client pong: foo", kiloServerConnection.call(cmdPing, "foo"))
|
||||
assertEquals("server push: bar", kiloServerConnection.call(cmdPush, "bar"))
|
||||
|
||||
client.push(cmdPushServer, "42")
|
||||
|
||||
assertEquals("**-s1-c1-s2-c2", client.call(cmdChainCallServer1, "**"))
|
||||
|
||||
|
||||
assertThrows<TestException> { client.call(cmdException) }
|
||||
assertEquals("ok: te-local", client.call(cmdRemoteExceptionTest))
|
||||
|
||||
// wait for push to be received and check
|
||||
assertEquals("42", pushedFromServer.await())
|
||||
|
||||
assertThrows<RemoteInterface.ClosedException> {
|
||||
client.call(cmdBreak)
|
||||
}
|
||||
|
||||
@@ -13,6 +13,8 @@ import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
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_tools.globalLaunch
|
||||
@@ -27,6 +29,10 @@ class ProtocolException(text: String, cause: Throwable? = null) : RuntimeExcepti
|
||||
const val MAX_TCP_BLOCK_SIZE = 16776216
|
||||
val PING_INACTIVITY_TIME = 30.seconds
|
||||
|
||||
/**
|
||||
* Listen for incoming TCP/IP connections on all local interfaces and the specified [port]
|
||||
* anc create flow of [InetTransportDevice] suitable for [KiloClient].
|
||||
*/
|
||||
fun acceptTcpDevice(port: Int): Flow<InetTransportDevice> {
|
||||
val selectorManager = SelectorManager(Dispatchers.IO)
|
||||
val serverSocket = aSocket(selectorManager).tcp().bind("127.0.0.1", port)
|
||||
@@ -41,12 +47,19 @@ fun acceptTcpDevice(port: Int): Flow<InetTransportDevice> {
|
||||
|
||||
suspend fun connectTcpDevice(address: String) = connectTcpDevice(address.toNetworkAddress())
|
||||
|
||||
/**
|
||||
* Connect to the TCP/IP server (see [KiloServer]) at the specified address and provide th compatible
|
||||
* [InetTransportDevice] to use with [KiloClient].
|
||||
*/
|
||||
suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice {
|
||||
val selectorManager = SelectorManager(Dispatchers.IO)
|
||||
val socket = aSocket(selectorManager).tcp().connect(address.host, address.port)
|
||||
return inetTransportDevice(socket)
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse `host:port` string into the [NetworkAddress]
|
||||
*/
|
||||
fun String.toNetworkAddress(): NetworkAddress {
|
||||
val (host, port) = this.split(":").map { it.trim() }
|
||||
return NetworkAddress(host, port.toInt())
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import net.sergeych.crypto2.initCrypto
|
||||
import net.sergeych.kiloparsec.*
|
||||
@@ -15,17 +14,17 @@ class TcpTest {
|
||||
@Test
|
||||
fun tcpTest() = runTest {
|
||||
initCrypto()
|
||||
// Log.connectConsole(Log.Level.DEBUG)
|
||||
data class Session(
|
||||
var data: String
|
||||
)
|
||||
// Log.connectConsole(Log.Level.DEBUG)
|
||||
data class Session(
|
||||
var data: String,
|
||||
)
|
||||
|
||||
val port = 27170 + Random.nextInt(1, 200)
|
||||
|
||||
val cmdSave by command<String, Unit>()
|
||||
val cmdLoad by command<Unit, String>()
|
||||
val cmdDrop by command<Unit, Unit>()
|
||||
val cmdException by command<Unit, Unit>()
|
||||
val cmdSave by command<String, Unit>()
|
||||
val cmdLoad by command<Unit, String>()
|
||||
val cmdDrop by command<Unit, Unit>()
|
||||
val cmdException by command<Unit, Unit>()
|
||||
|
||||
val cli = KiloInterface<Session>().apply {
|
||||
registerError { TestException() }
|
||||
@@ -45,16 +44,16 @@ class TcpTest {
|
||||
Session("unknown")
|
||||
}
|
||||
|
||||
val client = KiloClient<Unit>() {
|
||||
addErrors(cli)
|
||||
connect { connectTcpDevice("localhost:$port") }
|
||||
}
|
||||
delay(500)
|
||||
val client = KiloClient<Unit>() {
|
||||
addErrors(cli)
|
||||
// TODO: add register error variant
|
||||
connect { connectTcpDevice("localhost:$port") }
|
||||
}
|
||||
|
||||
assertEquals("start", client.call(cmdLoad))
|
||||
assertEquals("start", client.call(cmdLoad))
|
||||
|
||||
client.call(cmdSave, "foobar")
|
||||
assertEquals("foobar", client.call(cmdLoad))
|
||||
client.call(cmdSave, "foobar")
|
||||
assertEquals("foobar", client.call(cmdLoad))
|
||||
|
||||
val res = kotlin.runCatching { client.call(cmdException) }
|
||||
assertIs<TestException>(res.exceptionOrNull())
|
||||
|
||||
Reference in new issue
Block a user