rsocket-kotlin
RSocket Kotlin multi-platform implementation based on
kotlinx.coroutines and ktor-io.
RSocket is a binary application protocol providing Reactive Streams semantics for use on byte stream transports such as
TCP, WebSockets, QUIC and Aeron.
It enables the following symmetric interaction models via async message passing over a single connection:
Learn more at http://rsocket.io
Supported platforms and transports :
Local (in memory) transport is supported on all targets.
Most of other transports are implemented using ktor to ensure Kotlin multiplatform.
So it depends on ktor
client/server engines for available transports and platforms.
Client transports:
| TCP | WebSocket |
---|
JVM | ✅ via ktor | ✅ via ktor |
JS | ✅ via nodeJS (not supported in browser) | ✅ via ktor |
Native (except windows) | ✅ via ktor | ✅ via ktor |
Server transports:
| TCP | WebSocket |
---|
JVM | ✅ via ktor | ✅ via ktor |
JS | ✅ via nodeJS (not supported in browser) | ❌ |
Native (except windows) | ✅ via ktor | ✅ via ktor |
Using in your projects
rsocket-kotlin is available on Maven Central
Make sure, that you use Kotlin 1.6.20+, ktor 2.0.0+ and have mavenCentral()
in the list of repositories:
repositories {
mavenCentral()
}
Ktor plugins
rsocket-kotlin provides client
and server plugins for ktor
Dependencies:
dependencies {
implementation("io.rsocket.kotlin:rsocket-ktor-client:0.15.0")
implementation("io.rsocket.kotlin:rsocket-ktor-server:0.15.0")
}
Example of client plugin usage:
val client = HttpClient {
install(WebSockets)
install(RSocketSupport) {
connector {
maxFragmentSize = 1024
connectionConfig {
keepAlive = KeepAlive(
interval = 30.seconds,
maxLifetime = 2.minutes
)
setupPayload {
buildPayload {
data("""{ "data": "setup" }""")
}
}
payloadMimeType = PayloadMimeType(
data = WellKnownMimeType.ApplicationJson,
metadata = WellKnownMimeType.MessageRSocketCompositeMetadata
)
}
acceptor {
RSocketRequestHandler {
requestResponse { it }
}
}
}
}
}
val rSocket: RSocket = client.rSocket("wss://demo.rsocket.io/rsocket")
val stream: Flow<Payload> = rSocket.requestStream(
buildPayload {
data("""{ "data": "hello world" }""")
}
)
stream.take(5).collect { payload: Payload ->
println(payload.data.readText())
}
Example of server plugin usage:
embeddedServer(CIO) {
install(WebSockets)
install(RSocketSupport) {
server {
maxFragmentSize = 1024
interceptors {
forConnection(::SomeConnectionInterceptor)
}
}
}
routing {
rSocket("rsocket") {
println(config.setupPayload.data.readText())
RSocketRequestHandler {
requestResponse { request: Payload ->
println(request.data.readText())
delay(500)
buildPayload {
data("""{ "data": "Server response" }""")
}
}
requestStream { request: Payload ->
println(request.data.readText())
flow {
repeat(10) { i ->
emit(
buildPayload {
data("""{ "data": "Server stream response: $i" }""")
}
)
}
}
}
}
}
}
}.start(true)
Standalone transports
rsocket-kotlin also provides standalone transports which can be used to establish RSocket connection:
Dependencies:
dependencies {
implementation("io.rsocket.kotlin:rsocket-core:0.15.0")
implementation("io.rsocket.kotlin:rsocket-transport-ktor-tcp:0.15.0")
implementation("io.rsocket.kotlin:rsocket-transport-ktor-websocket-client:0.15.0")
implementation("io.rsocket.kotlin:rsocket-transport-ktor-websocket-server:0.15.0")
implementation("io.rsocket.kotlin:rsocket-transport-nodejs-tcp:0.15.0")
}
Example of usage standalone client transport:
val transport = TcpClientTransport("0.0.0.0", 8080)
val connector = RSocketConnector {
}
val rsocket: RSocket = connector.connect(transport)
val response = rsocket.requestResponse(buildPayload { data("""{ "data": "hello world" }""") })
println(response.data.readText())
Example of usage standalone server transport:
val transport = TcpServerTransport("0.0.0.0", 8080)
val connector = RSocketServer {
}
val server: TcpServer = server.bind(transport) {
RSocketRequestHandler {
requestResponse { request: Payload ->
println(request.data.readText())
delay(500)
buildPayload {
data("""{ "data": "Server response" }""")
}
}
}
}
server.handlerJob.join()
More samples:
Reactive Streams Semantics
From RSocket protocol:
Reactive Streams semantics are used for flow control of Streams, Subscriptions, and Channels.
This is a credit-based model where the Requester grants the Responder credit for the number of PAYLOADs it can send.
It is sometimes referred to as "request-n" or "request(n)".
kotlinx.coroutines
doesn't truly support request(n)
semantic,
but it has flexible CoroutineContext
which can be used to achieve something similar.
rsocket-kotlin
contains RequestStrategy
coroutine context element, which defines,
strategy for sending of requestN
frames.
Example:
val client: RSocket = TODO()
val stream: Flow<Payload> = client.requestStream(Payload("data"))
stream.flowOn(PrefetchStrategy(requestSize = 10, requestOn = 5)).collect { payload: Payload ->
println(payload.data.readText())
}
Bugs and Feedback
For bugs, questions and discussions please use the Github Issues.
LICENSE
Copyright 2015-2022 the original author or authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
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.