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))
 }
 

examples/multiplatform-chat/src/serverJvmMain/kotlin/App.kt

@@ -15,11 +15,9 @@
  */
 
 import io.ktor.application.*
-import io.ktor.network.selector.*
 import io.ktor.routing.*
 import io.ktor.server.cio.*
 import io.ktor.server.engine.*
-import io.ktor.util.*
 import io.ktor.websocket.*
 import io.rsocket.kotlin.*
 import io.rsocket.kotlin.core.*
@@ -31,7 +29,7 @@ import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.*
 import kotlinx.serialization.*
 
-@OptIn(ExperimentalSerializationApi::class, ExperimentalMetadataApi::class, InternalAPI::class)
+@OptIn(ExperimentalSerializationApi::class, ExperimentalMetadataApi::class, DelicateCoroutinesApi::class)
 fun main() {
     val proto = ConfiguredProtoBuf
     val users = Users()
@@ -97,7 +95,7 @@ fun main() {
     }
 
     //start TCP server
-    rSocketServer.bind(TcpServerTransport(ActorSelectorManager(Dispatchers.IO), port = 9000), acceptor)
+    rSocketServer.bind(TcpServerTransport(port = 8000), acceptor)
 
     //start WS server
     embeddedServer(CIO, port = 9000) {

playground/src/commonMain/kotlin/TCP.kt

@@ -14,21 +14,18 @@
  * limitations under the License.
  */
 
-import io.ktor.network.selector.*
 import io.rsocket.kotlin.core.*
 import io.rsocket.kotlin.payload.*
 import io.rsocket.kotlin.transport.ktor.*
-import kotlin.coroutines.*
 
-
-suspend fun runTcpClient(dispatcher: CoroutineContext) {
-    val transport = TcpClientTransport(SelectorManager(dispatcher), "0.0.0.0", 4444)
+suspend fun runTcpClient() {
+    val transport = TcpClientTransport("0.0.0.0", 4444)
     RSocketConnector().connect(transport).doSomething()
 }
 
 //to test nodejs tcp server
-suspend fun testNodeJsServer(dispatcher: CoroutineContext) {
-    val transport = TcpClientTransport(SelectorManager(dispatcher), "127.0.0.1", 9000)
+suspend fun testNodeJsServer() {
+    val transport = TcpClientTransport("127.0.0.1", 9000)
     val client = RSocketConnector().connect(transport)
 
     val response = client.requestResponse(buildPayload { data("Hello from JVM") })

playground/src/jvmMain/kotlin/TcpClientApp.kt

@@ -14,8 +14,6 @@
  * limitations under the License.
  */
 
-import kotlinx.coroutines.*
+suspend fun main(): Unit = runTcpClient()
 
-suspend fun main(): Unit = runTcpClient(Dispatchers.IO)
-
-//suspend fun main(): Unit = testNodeJsServer(Dispatchers.IO)
+//suspend fun main(): Unit = testNodeJsServer()

playground/src/jvmMain/kotlin/TcpServerApp.kt

@@ -14,15 +14,14 @@
  * limitations under the License.
  */
 
-import io.ktor.network.selector.*
 import io.rsocket.kotlin.core.*
 import io.rsocket.kotlin.transport.ktor.*
 import kotlinx.coroutines.*
 import kotlin.coroutines.*
 
 suspend fun runTcpServer(dispatcher: CoroutineContext) {
-    val transport = TcpServerTransport(SelectorManager(dispatcher), "0.0.0.0", 4444)
-    RSocketServer().bind(transport, rSocketAcceptor).join()
+    val transport = TcpServerTransport("0.0.0.0", 4444)
+    RSocketServer().bindIn(CoroutineScope(dispatcher), transport, rSocketAcceptor).handlerJob.join()
 }
 
 suspend fun main(): Unit = runTcpServer(Dispatchers.IO)

playground/src/nativeMain/kotlin/TcpApp.kt

@@ -14,14 +14,11 @@
  * limitations under the License.
  */
 
-import io.ktor.util.*
 import kotlinx.coroutines.*
-import kotlin.coroutines.*
 
-@OptIn(InternalAPI::class)
 fun main() {
     runBlocking {
-        runTcpClient(EmptyCoroutineContext)
+        runTcpClient()
     }
 }
 

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/Connection.kt

@@ -27,17 +27,15 @@ import kotlinx.coroutines.*
  * That interface isn't stable for inheritance.
  */
 @TransportApi
-public interface Connection {
-    public val job: Job
-
+public interface Connection : CoroutineScope {
     public val pool: ObjectPool<ChunkBuffer> get() = ChunkBuffer.Pool
 
     public suspend fun send(packet: ByteReadPacket)
     public suspend fun receive(): ByteReadPacket
 }
 
 @OptIn(TransportApi::class)
-internal suspend fun Connection.receiveFrame(): Frame = receive().readFrame(pool)
+internal suspend inline fun <T> Connection.receiveFrame(block: (frame: Frame) -> T): T = receive().readFrame(pool).closeOnError(block)
 
 @OptIn(TransportApi::class)
 internal suspend fun Connection.sendFrame(frame: Frame) {

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/RSocket.kt

@@ -21,8 +21,7 @@ import io.rsocket.kotlin.payload.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.*
 
-public interface RSocket {
-    public val job: Job
+public interface RSocket : CoroutineScope {
 
     public suspend fun metadataPush(metadata: ByteReadPacket) {
         metadata.release()

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/RSocketRequestHandler.kt

@@ -20,6 +20,7 @@ import io.ktor.utils.io.core.*
 import io.rsocket.kotlin.payload.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.*
+import kotlin.coroutines.*
 
 public class RSocketRequestHandlerBuilder internal constructor() {
     private var metadataPush: (suspend RSocket.(metadata: ByteReadPacket) -> Unit)? = null
@@ -53,19 +54,29 @@ public class RSocketRequestHandlerBuilder internal constructor() {
         requestChannel = block
     }
 
-    internal fun build(job: Job): RSocket =
-        RSocketRequestHandler(job, metadataPush, fireAndForget, requestResponse, requestStream, requestChannel)
+    internal fun build(parentContext: CoroutineContext): RSocket =
+        RSocketRequestHandler(
+            parentContext + Job(parentContext[Job]),
+            metadataPush,
+            fireAndForget,
+            requestResponse,
+            requestStream,
+            requestChannel
+        )
 }
 
 @Suppress("FunctionName")
-public fun RSocketRequestHandler(parentJob: Job? = null, configure: RSocketRequestHandlerBuilder.() -> Unit): RSocket {
+public fun RSocketRequestHandler(
+    parentContext: CoroutineContext = EmptyCoroutineContext,
+    configure: RSocketRequestHandlerBuilder.() -> Unit
+): RSocket {
     val builder = RSocketRequestHandlerBuilder()
     builder.configure()
-    return builder.build(Job(parentJob))
+    return builder.build(parentContext)
 }
 
 private class RSocketRequestHandler(
-    override val job: Job,
+    override val coroutineContext: CoroutineContext,
     private val metadataPush: (suspend RSocket.(metadata: ByteReadPacket) -> Unit)? = null,
     private val fireAndForget: (suspend RSocket.(payload: Payload) -> Unit)? = null,
     private val requestResponse: (suspend RSocket.(payload: Payload) -> Payload)? = null,

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/core/RSocketConnector.kt

@@ -35,11 +35,13 @@ public class RSocketConnector internal constructor(
 ) {
 
     public suspend fun connect(transport: ClientTransport): RSocket = when (reconnectPredicate) {
-        null -> connectOnce(transport)
-        else -> ReconnectableRSocket(
-            logger = loggerFactory.logger("io.rsocket.kotlin.connection"),
-            connect = { connectOnce(transport) },
-            predicate = reconnectPredicate
+        //TODO current coroutineContext job is overriden by transport coroutineContext jov
+        null -> withContext(transport.coroutineContext) { connectOnce(transport) }
+        else -> connectWithReconnect(
+            transport.coroutineContext,
+            loggerFactory.logger("io.rsocket.kotlin.connection"),
+            { connectOnce(transport) },
+            reconnectPredicate,
         )
     }
 
@@ -48,7 +50,7 @@ public class RSocketConnector internal constructor(
         val connectionConfig = try {
             connectionConfigProvider()
         } catch (cause: Throwable) {
-            connection.job.cancel("Connection config provider failed", cause)
+            connection.cancel("Connection config provider failed", cause)
             throw cause
         }
         val setupFrame = SetupFrame(
@@ -60,7 +62,8 @@ public class RSocketConnector internal constructor(
             payload = connectionConfig.setupPayload.copy() //copy needed, as it can be used in acceptor
         )
         try {
-            val requester = connection.connect(
+            val requester = connect(
+                connection = connection,
                 isServer = false,
                 maxFragmentSize = maxFragmentSize,
                 interceptors = interceptors,
@@ -72,7 +75,7 @@ public class RSocketConnector internal constructor(
         } catch (cause: Throwable) {
             connectionConfig.setupPayload.release()
             setupFrame.release()
-            connection.job.cancel("Connection establishment failed", cause)
+            connection.cancel("Connection establishment failed", cause)
             throw cause
         }
     }