improves transport api and lifecycle (#178)

Co-authored-by: olme04 <olme04>

Author: olme04

benchmarks/src/kotlinMain/kotlin/io/rsocket/kotlin/benchmarks/RSocketKotlinBenchmark.kt

@@ -25,12 +25,12 @@ import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.*
 import kotlin.random.*
 
-@OptIn(ExperimentalStreamsApi::class)
+@OptIn(ExperimentalStreamsApi::class, DelicateCoroutinesApi::class)
 class RSocketKotlinBenchmark : RSocketBenchmark<Payload>() {
     private val requestStrategy = PrefetchStrategy(64, 0)
 
+    private val benchJob = Job()
     lateinit var client: RSocket
-    lateinit var server: Job
 
     lateinit var payload: Payload
     lateinit var payloadsFlow: Flow<Payload>
@@ -40,9 +40,7 @@ class RSocketKotlinBenchmark : RSocketBenchmark<Payload>() {
     override fun setup() {
         payload = createPayload(payloadSize)
         payloadsFlow = flow { repeat(5000) { emit(payloadCopy()) } }
-
-        val localServer = LocalServer()
-        server = RSocketServer().bind(localServer) {
+        val server = RSocketServer().bindIn(CoroutineScope(benchJob + Dispatchers.Unconfined), LocalServerTransport()) {
             RSocketRequestHandler {
                 requestResponse {
                     it.release()
@@ -59,14 +57,14 @@ class RSocketKotlinBenchmark : RSocketBenchmark<Payload>() {
             }
         }
         client = runBlocking {
-            RSocketConnector().connect(localServer)
+            RSocketConnector().connect(server)
         }
     }
 
     override fun cleanup() {
         runBlocking {
-            client.job.runCatching { cancelAndJoin() }
-            server.runCatching { cancelAndJoin() }
+            client.coroutineContext.job.cancelAndJoin()
+            benchJob.cancelAndJoin()
         }
     }
 

examples/interactions/src/jvmMain/kotlin/ReconnectExample.kt

@@ -25,8 +25,7 @@ import kotlinx.coroutines.flow.*
 
 @TransportApi
 fun main(): Unit = runBlocking {
-    val server = LocalServer()
-    RSocketServer().bind(server) {
+    val server = RSocketServer().bindIn(this, LocalServerTransport()) {
         RSocketRequestHandler {
             requestStream { requestPayload ->
                 val data = requestPayload.data.readText()

examples/interactions/src/jvmMain/kotlin/ReconnectOnConnectFailExample.kt

@@ -23,8 +23,7 @@ import kotlinx.coroutines.flow.*
 
 @TransportApi
 fun main(): Unit = runBlocking {
-    val server = LocalServer()
-    RSocketServer().bind(server) {
+    val server = RSocketServer().bindIn(this, LocalServerTransport()) {
         RSocketRequestHandler {
             requestStream { requestPayload ->
                 val data = requestPayload.data.readText()

examples/interactions/src/jvmMain/kotlin/RequestChannelExample.kt

@@ -22,8 +22,7 @@ import kotlinx.coroutines.flow.*
 
 
 fun main(): Unit = runBlocking {
-    val server = LocalServer()
-    RSocketServer().bind(server) {
+    val server = RSocketServer().bindIn(this, LocalServerTransport()) {
         RSocketRequestHandler {
             requestChannel { init, request ->
                 println("Init with: ${init.data.readText()}")

examples/interactions/src/jvmMain/kotlin/RequestResponseErrorExample.kt

@@ -20,8 +20,7 @@ import io.rsocket.kotlin.transport.local.*
 import kotlinx.coroutines.*
 
 fun main(): Unit = runBlocking {
-    val server = LocalServer()
-    RSocketServer().bind(server) {
+    val server = RSocketServer().bindIn(this, LocalServerTransport()) {
         RSocketRequestHandler {
             requestResponse {
                 val data = it.data.readText()

examples/interactions/src/jvmMain/kotlin/RequestResponseExample.kt

@@ -20,8 +20,7 @@ import io.rsocket.kotlin.transport.local.*
 import kotlinx.coroutines.*
 
 fun main(): Unit = runBlocking {
-    val server = LocalServer()
-    RSocketServer().bind(server) {
+    val server = RSocketServer().bindIn(this, LocalServerTransport()) {
         RSocketRequestHandler {
             requestResponse {
                 val data = it.data.readText()

examples/interactions/src/jvmMain/kotlin/RequestStreamExample.kt

@@ -21,8 +21,7 @@ import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.*
 
 fun main(): Unit = runBlocking {
-    val server = LocalServer()
-    RSocketServer().bind(server) {
+    val server = RSocketServer().bindIn(this, LocalServerTransport()) {
         RSocketRequestHandler {
             requestStream {
                 val data = it.data.readText()

examples/interactions/src/jvmMain/kotlin/ServerRequestExample.kt

@@ -20,8 +20,7 @@ import io.rsocket.kotlin.transport.local.*
 import kotlinx.coroutines.*
 
 fun main(): Unit = runBlocking {
-    val server = LocalServer()
-    RSocketServer().bind(server) {
+    val server = RSocketServer().bindIn(this, LocalServerTransport()) {
         RSocketRequestHandler {
             requestResponse {
                 val clientRequest = it.data.readText()

examples/interactions/src/jvmMain/kotlin/ServerSetupExample.kt

@@ -23,9 +23,7 @@ import kotlinx.coroutines.flow.*
 
 
 fun main(): Unit = runBlocking {
-
-    val server = LocalServer()
-    RSocketServer().bind(server) {
+    val server = RSocketServer().bindIn(this, LocalServerTransport()) {
         val data = config.setupPayload.metadata?.readText() ?: error("Empty metadata")
         RSocketRequestHandler {
             when (data) {
@@ -43,8 +41,8 @@ fun main(): Unit = runBlocking {
 
     suspend fun client1() {
         val rSocketClient = RSocketConnector().connect(server)
-        rSocketClient.job.join()
-        println("Client 1 canceled: ${rSocketClient.job.isCancelled}")
+        rSocketClient.coroutineContext.job.join()
+        println("Client 1 canceled: ${rSocketClient.coroutineContext.job.isCancelled}")
         try {
             rSocketClient.requestResponse(Payload.Empty)
         } catch (e: Throwable) {

examples/multiplatform-chat/src/clientMain/kotlin/Api.kt

@@ -16,13 +16,12 @@
 
 import io.ktor.client.*
 import io.ktor.client.features.websocket.*
-import io.ktor.network.selector.*
-import io.ktor.util.*
 import io.rsocket.kotlin.*
 import io.rsocket.kotlin.core.*
 import io.rsocket.kotlin.payload.*
 import io.rsocket.kotlin.transport.ktor.*
 import io.rsocket.kotlin.transport.ktor.client.*
+import kotlinx.coroutines.*
 
 class Api(rSocket: RSocket) {
     private val proto = ConfiguredProtoBuf
@@ -42,9 +41,10 @@ suspend fun connectToApiUsingWS(name: String): Api {
     return Api(client.rSocket(port = 9000))
 }
 
-@OptIn(InternalAPI::class)
 suspend fun connectToApiUsingTCP(name: String): Api {
-    val transport = TcpClientTransport(SelectorManager(), "0.0.0.0", 8000)
+    val transport = TcpClientTransport("0.0.0.0", 8000, CoroutineExceptionHandler { coroutineContext, throwable ->
+        println("FAIL: $coroutineContext, $throwable")
+    })
     return Api(connector(name).connect(transport))
 }