• Stars
    star
    545
  • Rank 81,554 (Top 2 %)
  • Language
    Kotlin
  • License
    Apache License 2.0
  • Created about 7 years ago
  • Updated 4 months ago

Reviews

There are no reviews yet. Be the first to send feedback to the community and the maintainers!

Repository Details

RSocket Kotlin multi-platform implementation

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 {
    //for client
    implementation("io.rsocket.kotlin:rsocket-ktor-client:0.15.4")

    //for server
    implementation("io.rsocket.kotlin:rsocket-ktor-server:0.15.4")
}

Example of client plugin usage:

//create ktor client
val client = HttpClient {
    install(WebSockets) //rsocket requires websockets plugin installed
    install(RSocketSupport) {
        //configure rSocket connector (all values have defaults)
        connector {
            maxFragmentSize = 1024

            connectionConfig {
                keepAlive = KeepAlive(
                    interval = 30.seconds,
                    maxLifetime = 2.minutes
                )

                //payload for setup frame
                setupPayload {
                    buildPayload {
                        data("""{ "data": "setup" }""")
                    }
                }

                //mime types
                payloadMimeType = PayloadMimeType(
                    data = WellKnownMimeType.ApplicationJson,
                    metadata = WellKnownMimeType.MessageRSocketCompositeMetadata
                )
            }

            //optional acceptor for server requests
            acceptor {
                RSocketRequestHandler {
                    requestResponse { it } //echo request payload
                }
            }
        }
    }
}

//connect to some url
val rSocket: RSocket = client.rSocket("wss://demo.rsocket.io/rsocket")

//request stream
val stream: Flow<Payload> = rSocket.requestStream(
    buildPayload {
        data("""{ "data": "hello world" }""")
    }
)

//take 5 values and print response
stream.take(5).collect { payload: Payload ->
    println(payload.data.readText())
}

Example of server plugin usage:

//create ktor server
embeddedServer(CIO) {
    install(WebSockets) //rsocket requires websockets plugin installed
    install(RSocketSupport) {
        //configure rSocket server (all values have defaults)

        server {
            maxFragmentSize = 1024

            //install interceptors
            interceptors {
                forConnection(::SomeConnectionInterceptor)
            }
        }
    }
    //configure routing
    routing {
        //configure route `/rsocket`
        rSocket("rsocket") {
            println(config.setupPayload.data.readText()) //print setup payload data

            RSocketRequestHandler {
                //handler for request/response
                requestResponse { request: Payload ->
                    println(request.data.readText()) //print request payload data
                    delay(500) // work emulation
                    buildPayload {
                        data("""{ "data": "Server response" }""")
                    }
                }
                //handler for request/stream      
                requestStream { request: Payload ->
                    println(request.data.readText()) //print request payload data
                    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.4")

    // TCP ktor client/server transport
    implementation("io.rsocket.kotlin:rsocket-transport-ktor-tcp:0.15.4")

    // WS ktor client transport
    implementation("io.rsocket.kotlin:rsocket-transport-ktor-websocket-client:0.15.4")

    // WS ktor server transport
    implementation("io.rsocket.kotlin:rsocket-transport-ktor-websocket-server:0.15.4")

    // TCP nodeJS client/server transport
    implementation("io.rsocket.kotlin:rsocket-transport-nodejs-tcp:0.15.4")
}

Example of usage standalone client transport:

val transport = TcpClientTransport("0.0.0.0", 8080)
val connector = RSocketConnector {
    //configuration goes here
}
val rsocket: RSocket = connector.connect(transport)
//use rsocket to do request
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 {
    //configuration goes here
}
val server: TcpServer = server.bind(transport) {
    RSocketRequestHandler {
        //handler for request/response
        requestResponse { request: Payload ->
            println(request.data.readText()) //print request payload data
            delay(500) // work emulation
            buildPayload {
                data("""{ "data": "Server response" }""")
            }
        }
    }
}
server.handlerJob.join() //wait for server to finish

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 requestNframes.

Example:

//assume we have client
val client: RSocket = TODO()

//and stream
val stream: Flow<Payload> = client.requestStream(Payload("data"))

//now we can use `flowOn` to add request strategy to context of flow
//here we use prefetch strategy which will send requestN for 10 elements, when, there is 5 elements left to collect
//so on call `collect`, requestStream frame with requestN will be sent, and then, after 5 elements will be collected
//new requestN with 5 will be sent, so collect will be smooth 
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.

More Repositories

1

rsocket-java

Java implementation of RSocket
Java
2,354
star
2

rsocket

RSocket Protocol Definition
1,837
star
3

rsocket-js

JavaScript implementation of RSocket
TypeScript
593
star
4

rsocket-go

rsocket-go implementation
Go
504
star
5

rsocket-cpp

C++ implementation of RSocket
C++
253
star
6

rsocket-net

.NET implementation of RSocket
C#
251
star
7

rsocket-rust

RSocket Rust Implementation using Tokio
Rust
205
star
8

rsocket-rpc-java

Standard RSocket RPC Java Implementation
Java
172
star
9

rsocket-swift

Swift implementation of RSocket Protocol
Swift
76
star
10

rsocket-py

RSocket implementation in Python
Python
76
star
11

rsocket-cli

Command-line client for ReactiveSocket
Kotlin
71
star
12

rsocket-chrome-devtools

RSocket Chrome Dev Tools
TypeScript
42
star
13

rsocket-dart

Dart implementation of RSocket
Dart
31
star
14

rsocket-rpc-js

Standard RSocket RPC Implementation
JavaScript
29
star
15

rsocket-transport-akka

Akka TCP and WebSocket transports for RSocket
Java
28
star
16

rsocket-demo-android-java8

RSocket Android Demo
Kotlin
23
star
17

rsocket-wireshark

Wireshark Plugin for RSocket
C
19
star
18

rsocket-rpc-kotlin

Standard RSocket RPC Kotlin Implementation
Kotlin
19
star
19

rsocket-rpc-net

Standard RSocket RPC .NET Implementation
C#
17
star
20

rsocket-rpc-go

RSocket RPC Golang
Go
15
star
21

rsocket-php

RSocket PHP implementation using ReactPHP
PHP
10
star
22

rsocket-tck

Technology Compatibility Kit (TCK) for RSocket implementations
Kotlin
10
star
23

rsocket-transport-aeron

Aeron Transport for RSocket
Java
10
star
24

ewma

Exponentially Weighted Moving Average
JavaScript
10
star
25

rsocket-rb

Ruby
8
star
26

rsocket-website

rsocket.io website
MDX
7
star
27

bomberman-workshop

JavaScript
7
star
28

rsocket-jmeter

RSocket JMeter Plugin
Java
3
star
29

tsc

RSocket Technical Steering Committee (TSC)
3
star
30

rsocket-c

C implementation of RSocket
3
star
31

rsocket-tck-server

RSocket TCK Server
Kotlin
2
star
32

gradle-nebula-plugin-rsocket

Groovy
1
star
33

rsocket-artwork

RSocket Logos
1
star
34

.github

Template repository for rsocket projects, includes typical README + labels
1
star