Adapt local transport implementation to the latest change in transport API and simplify some parts

* drop `inheritDispatcher` and use `Unconfined` dispatcher by default
* drop possibility to configure channels buffering - use default

Author: whyoleg

rsocket-transports/local/api/rsocket-transport-local.api

@@ -8,7 +8,6 @@ public final class io/rsocket/kotlin/transport/local/LocalClientTransport$Factor
 
 public abstract interface class io/rsocket/kotlin/transport/local/LocalClientTransportBuilder : io/rsocket/kotlin/transport/RSocketTransportBuilder {
 	public abstract fun dispatcher (Lkotlin/coroutines/CoroutineContext;)V
-	public fun inheritDispatcher ()V
 }
 
 public final class io/rsocket/kotlin/transport/local/LocalServer : io/rsocket/kotlin/transport/ClientTransport {
@@ -35,10 +34,7 @@ public final class io/rsocket/kotlin/transport/local/LocalServerTransport$Factor
 
 public abstract interface class io/rsocket/kotlin/transport/local/LocalServerTransportBuilder : io/rsocket/kotlin/transport/RSocketTransportBuilder {
 	public abstract fun dispatcher (Lkotlin/coroutines/CoroutineContext;)V
-	public fun inheritDispatcher ()V
-	public abstract fun multiplexed (II)V
-	public static synthetic fun multiplexed$default (Lio/rsocket/kotlin/transport/local/LocalServerTransportBuilder;IIILjava/lang/Object;)V
-	public abstract fun sequential (I)V
-	public static synthetic fun sequential$default (Lio/rsocket/kotlin/transport/local/LocalServerTransportBuilder;IILjava/lang/Object;)V
+	public abstract fun multiplexed ()V
+	public abstract fun sequential ()V
 }
 

rsocket-transports/local/src/commonMain/kotlin/io/rsocket/kotlin/transport/local/LocalClientTransport.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2024 the original author or authors.
+ * Copyright 2015-2025 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.
@@ -32,11 +32,10 @@ public sealed interface LocalClientTransport : RSocketTransport {
 @OptIn(RSocketTransportApi::class)
 public sealed interface LocalClientTransportBuilder : RSocketTransportBuilder<LocalClientTransport> {
     public fun dispatcher(context: CoroutineContext)
-    public fun inheritDispatcher(): Unit = dispatcher(EmptyCoroutineContext)
 }
 
 private class LocalClientTransportBuilderImpl : LocalClientTransportBuilder {
-    private var dispatcher: CoroutineContext = Dispatchers.Default
+    private var dispatcher: CoroutineContext = Dispatchers.Unconfined
 
     override fun dispatcher(context: CoroutineContext) {
         check(context[Job] == null) { "Dispatcher shouldn't contain job" }
@@ -63,10 +62,11 @@ private class LocalClientTargetImpl(
     override val coroutineContext: CoroutineContext,
     private val serverName: String,
 ) : RSocketClientTarget {
-
     @RSocketTransportApi
-    override fun connectClient(handler: RSocketConnectionHandler): Job {
+    override suspend fun connectClient(): RSocketConnection {
+        currentCoroutineContext().ensureActive()
         coroutineContext.ensureActive()
-        return LocalServerRegistry.get(serverName).connect(clientScope = this, clientHandler = handler)
+
+        return LocalServerRegistry.connectClient(serverName, coroutineContext)
     }
 }

rsocket-transports/local/src/commonMain/kotlin/io/rsocket/kotlin/transport/local/LocalServerConnector.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2024 the original author or authors.
+ * Copyright 2015-2025 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.
@@ -22,174 +22,159 @@ import io.rsocket.kotlin.transport.internal.*
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.*
 import kotlinx.io.*
+import kotlin.coroutines.*
 
 internal sealed class LocalServerConnector {
     @RSocketTransportApi
-    abstract fun connect(
-        clientScope: CoroutineScope,
-        clientHandler: RSocketConnectionHandler,
-        serverScope: CoroutineScope,
-        serverHandler: RSocketConnectionHandler,
-    ): Job
-
-    internal class Sequential(
-        private val prioritizationQueueBuffersCapacity: Int,
-    ) : LocalServerConnector() {
-
-        @RSocketTransportApi
-        override fun connect(
-            clientScope: CoroutineScope,
-            clientHandler: RSocketConnectionHandler,
-            serverScope: CoroutineScope,
-            serverHandler: RSocketConnectionHandler,
-        ): Job {
-            val clientToServer = PrioritizationFrameQueue(prioritizationQueueBuffersCapacity)
-            val serverToClient = PrioritizationFrameQueue(prioritizationQueueBuffersCapacity)
-
-            launchLocalConnection(serverScope, serverToClient, clientToServer, serverHandler)
-            return launchLocalConnection(clientScope, clientToServer, serverToClient, clientHandler)
-        }
+    abstract suspend fun connect(
+        clientContext: CoroutineContext,
+        serverContext: CoroutineContext,
+        onConnection: (RSocketConnection) -> Unit,
+    ): RSocketConnection
 
+    object Sequential : LocalServerConnector() {
         @RSocketTransportApi
-        private fun launchLocalConnection(
-            scope: CoroutineScope,
-            outbound: PrioritizationFrameQueue,
-            inbound: PrioritizationFrameQueue,
-            handler: RSocketConnectionHandler,
-        ): Job = scope.launch {
-            handler.handleConnection(Connection(outbound, inbound))
-        }.onCompletion {
-            outbound.close()
-            inbound.cancel()
+        override suspend fun connect(
+            clientContext: CoroutineContext,
+            serverContext: CoroutineContext,
+            onConnection: (RSocketConnection) -> Unit,
+        ): RSocketConnection {
+            val frames = Frames()
+            onConnection(Connection(serverContext.childContext(), frames.clientToServer, frames.serverToClient))
+            return Connection(clientContext.childContext(), frames.serverToClient, frames.clientToServer)
         }
 
         @RSocketTransportApi
         private class Connection(
-            private val outbound: PrioritizationFrameQueue,
-            private val inbound: PrioritizationFrameQueue,
+            override val coroutineContext: CoroutineContext,
+            private val incomingFrames: ReceiveChannel<Buffer>,
+            private val outgoingFrames: SendChannel<Buffer>,
         ) : RSocketSequentialConnection {
-            override val isClosedForSend: Boolean get() = outbound.isClosedForSend
+            private val outboundQueue = PrioritizationFrameQueue()
+
+            init {
+                @OptIn(DelicateCoroutinesApi::class)
+                launch(start = CoroutineStart.ATOMIC) {
+                    launch {
+                        nonCancellable {
+                            while (true) outgoingFrames.send(outboundQueue.dequeueFrame() ?: break)
+                        }
+                    }.invokeOnCompletion {
+                        outboundQueue.cancel()
+                        outgoingFrames.close()
+                    }
+                    try {
+                        awaitCancellation()
+                    } finally {
+                        outboundQueue.close()
+                        incomingFrames.cancel()
+                    }
+                }
+            }
 
             override suspend fun sendFrame(streamId: Int, frame: Buffer) {
-                return outbound.enqueueFrame(streamId, frame)
+                return outboundQueue.enqueueFrame(streamId, frame)
             }
 
             override suspend fun receiveFrame(): Buffer? {
-                return inbound.dequeueFrame()
+                return incomingFrames.receiveCatching().getOrNull()
             }
         }
     }
 
-    // TODO: better parameters naming
-    class Multiplexed(
-        private val streamsQueueCapacity: Int,
-        private val streamBufferCapacity: Int,
-    ) : LocalServerConnector() {
+    object Multiplexed : LocalServerConnector() {
         @RSocketTransportApi
-        override fun connect(
-            clientScope: CoroutineScope,
-            clientHandler: RSocketConnectionHandler,
-            serverScope: CoroutineScope,
-            serverHandler: RSocketConnectionHandler,
-        ): Job {
-            val streams = Streams(streamsQueueCapacity)
-
-            launchLocalConnection(serverScope, streams.serverToClient, streams.clientToServer, serverHandler)
-            return launchLocalConnection(clientScope, streams.clientToServer, streams.serverToClient, clientHandler)
-        }
-
-        @RSocketTransportApi
-        private fun launchLocalConnection(
-            scope: CoroutineScope,
-            outbound: SendChannel<Frames>,
-            inbound: ReceiveChannel<Frames>,
-            handler: RSocketConnectionHandler,
-        ): Job = scope.launch {
-            handler.handleConnection(Connection(SupervisorJob(coroutineContext.job), outbound, inbound, streamBufferCapacity))
-        }.onCompletion {
-            outbound.close()
-            inbound.cancel()
+        override suspend fun connect(
+            clientContext: CoroutineContext,
+            serverContext: CoroutineContext,
+            onConnection: (RSocketConnection) -> Unit,
+        ): RSocketConnection {
+            val streams = Streams()
+            onConnection(Connection(serverContext.childContext(), streams.clientToServer, streams.serverToClient))
+            return Connection(clientContext.childContext(), streams.serverToClient, streams.clientToServer)
         }
 
         @RSocketTransportApi
         private class Connection(
-            private val streamsJob: Job,
-            private val outbound: SendChannel<Frames>,
-            private val inbound: ReceiveChannel<Frames>,
-            private val streamBufferCapacity: Int,
+            override val coroutineContext: CoroutineContext,
+            private val incomingStreams: ReceiveChannel<Frames>,
+            private val outgoingStreams: SendChannel<Frames>,
         ) : RSocketMultiplexedConnection {
-            override suspend fun createStream(): RSocketMultiplexedConnection.Stream {
-                val frames = Frames(streamBufferCapacity)
+            private val streamsContext = coroutineContext.supervisorContext()
 
-                outbound.send(frames)
+            init {
+                coroutineContext.job.invokeOnCompletion {
+                    outgoingStreams.close()
+                    incomingStreams.cancel()
+                }
+            }
 
+            override suspend fun createStream(): RSocketMultiplexedConnection.Stream {
+                val frames = Frames()
+                outgoingStreams.send(frames)
                 return Stream(
-                    parentJob = streamsJob,
-                    outbound = frames.clientToServer,
-                    inbound = frames.serverToClient
+                    coroutineContext = streamsContext.childContext(),
+                    incoming = frames.clientToServer,
+                    outgoing = frames.serverToClient
                 )
             }
 
             override suspend fun acceptStream(): RSocketMultiplexedConnection.Stream? {
-                val frames = inbound.receiveCatching().getOrNull() ?: return null
-
+                val frames = incomingStreams.receiveCatching().getOrNull() ?: return null
                 return Stream(
-                    parentJob = streamsJob,
-                    outbound = frames.serverToClient,
-                    inbound = frames.clientToServer
+                    coroutineContext = streamsContext.childContext(),
+                    incoming = frames.serverToClient,
+                    outgoing = frames.clientToServer
                 )
             }
         }
 
         @RSocketTransportApi
         private class Stream(
-            parentJob: Job,
-            private val outbound: SendChannel<Buffer>,
-            private val inbound: ReceiveChannel<Buffer>,
+            override val coroutineContext: CoroutineContext,
+            private val incoming: ReceiveChannel<Buffer>,
+            private val outgoing: SendChannel<Buffer>,
         ) : RSocketMultiplexedConnection.Stream {
-            private val streamJob = Job(parentJob).onCompletion {
-                outbound.close()
-                inbound.cancel()
+            init {
+                coroutineContext.job.invokeOnCompletion {
+                    outgoing.close()
+                    incoming.cancel()
+                }
             }
 
-            override fun close() {
-                streamJob.complete()
+            override fun setSendPriority(priority: Int) {
+                // no-op
             }
 
-            @OptIn(DelicateCoroutinesApi::class)
-            override val isClosedForSend: Boolean get() = outbound.isClosedForSend
-
-            override fun setSendPriority(priority: Int) {}
-
             override suspend fun sendFrame(frame: Buffer) {
-                return outbound.send(frame)
+                return outgoing.send(frame)
             }
 
             override suspend fun receiveFrame(): Buffer? {
-                return inbound.receiveCatching().getOrNull()
+                return incoming.receiveCatching().getOrNull()
             }
         }
+    }
+}
 
-        private class Streams(bufferCapacity: Int) : AutoCloseable {
-            val clientToServer = channelForCloseable<Frames>(bufferCapacity)
-            val serverToClient = channelForCloseable<Frames>(bufferCapacity)
+private class Streams : AutoCloseable {
+    val clientToServer = channelForCloseable<Frames>(Channel.BUFFERED)
+    val serverToClient = channelForCloseable<Frames>(Channel.BUFFERED)
 
-            // only for undelivered element case
-            override fun close() {
-                clientToServer.cancel()
-                serverToClient.cancel()
-            }
-        }
+    // only for undelivered element case
+    override fun close() {
+        clientToServer.cancel()
+        serverToClient.cancel()
+    }
+}
 
-        private class Frames(bufferCapacity: Int) : AutoCloseable {
-            val clientToServer = bufferChannel(bufferCapacity)
-            val serverToClient = bufferChannel(bufferCapacity)
+private class Frames : AutoCloseable {
+    val clientToServer = bufferChannel(Channel.BUFFERED)
+    val serverToClient = bufferChannel(Channel.BUFFERED)
 
-            // only for undelivered element case
-            override fun close() {
-                clientToServer.cancel()
-                serverToClient.cancel()
-            }
-        }
+    // only for undelivered element case
+    override fun close() {
+        clientToServer.cancel()
+        serverToClient.cancel()
     }
 }