Connection buffer configuration

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

@@ -27,6 +27,7 @@ import kotlinx.coroutines.*
 @OptIn(TransportApi::class, RSocketLoggingApi::class)
 public class RSocketConnector internal constructor(
     private val loggerFactory: LoggerFactory,
+    private val connectionBufferCapacity: Int,
     private val maxFragmentSize: Int,
     private val interceptors: Interceptors,
     private val connectionConfigProvider: () -> ConnectionConfig,
@@ -64,6 +65,7 @@ public class RSocketConnector internal constructor(
         try {
             val requester = connect(
                 connection = connection,
+                connectionBufferCapacity = connectionBufferCapacity,
                 isServer = false,
                 maxFragmentSize = maxFragmentSize,
                 interceptors = interceptors,

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

@@ -34,6 +34,11 @@ public class RSocketConnectorBuilder internal constructor() {
             }
             field = value
         }
+    public var connectionBufferCapacity: Int = 64
+        set(value) {
+            require(value >= 0) { "connectionBufferCapacity should be positive or equal to Int.MAX_VALUE" }
+            field = value
+        }
 
     private val connectionConfig: ConnectionConfigBuilder = ConnectionConfigBuilder()
     private val interceptors: InterceptorsBuilder = InterceptorsBuilder()
@@ -103,12 +108,13 @@ public class RSocketConnectorBuilder internal constructor() {
 
     @OptIn(RSocketLoggingApi::class)
     internal fun build(): RSocketConnector = RSocketConnector(
-        loggerFactory,
-        maxFragmentSize,
-        interceptors.build(),
-        connectionConfig.producer(),
-        acceptor ?: defaultAcceptor,
-        reconnectPredicate
+        loggerFactory = loggerFactory,
+        connectionBufferCapacity = connectionBufferCapacity,
+        maxFragmentSize = maxFragmentSize,
+        interceptors = interceptors.build(),
+        connectionConfigProvider = connectionConfig.producer(),
+        acceptor = acceptor ?: defaultAcceptor,
+        reconnectPredicate = reconnectPredicate
     )
 
     private companion object {

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

@@ -27,6 +27,7 @@ import kotlinx.coroutines.*
 @OptIn(TransportApi::class, RSocketLoggingApi::class)
 public class RSocketServer internal constructor(
     private val loggerFactory: LoggerFactory,
+    private val connectionBufferCapacity: Int,
     private val maxFragmentSize: Int,
     private val interceptors: Interceptors,
 ) {
@@ -58,6 +59,7 @@ public class RSocketServer internal constructor(
                     connection = this,
                     isServer = true,
                     maxFragmentSize = maxFragmentSize,
+                    connectionBufferCapacity = connectionBufferCapacity,
                     interceptors = interceptors,
                     connectionConfig = ConnectionConfig(
                         keepAlive = setupFrame.keepAlive,

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

@@ -29,6 +29,11 @@ public class RSocketServerBuilder internal constructor() {
             }
             field = value
         }
+    public var connectionBufferCapacity: Int = 64
+        set(value) {
+            require(value >= 0) { "connectionBufferCapacity should be positive or equal to Int.MAX_VALUE" }
+            field = value
+        }
 
     private val interceptors: InterceptorsBuilder = InterceptorsBuilder()
 
@@ -37,7 +42,12 @@ public class RSocketServerBuilder internal constructor() {
     }
 
     @OptIn(RSocketLoggingApi::class)
-    internal fun build(): RSocketServer = RSocketServer(loggerFactory, maxFragmentSize, interceptors.build())
+    internal fun build(): RSocketServer = RSocketServer(
+        loggerFactory = loggerFactory,
+        connectionBufferCapacity = connectionBufferCapacity,
+        maxFragmentSize = maxFragmentSize,
+        interceptors = interceptors.build()
+    )
 }
 
 public fun RSocketServer(configure: RSocketServerBuilder.() -> Unit = {}): RSocketServer {

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/internal/Connect.kt

@@ -25,12 +25,13 @@ import kotlinx.coroutines.*
 internal suspend inline fun connect(
     connection: Connection,
     isServer: Boolean,
+    connectionBufferCapacity: Int,
     maxFragmentSize: Int,
     interceptors: Interceptors,
     connectionConfig: ConnectionConfig,
     acceptor: ConnectionAcceptor
 ): RSocket {
-    val prioritizer = Prioritizer()
+    val prioritizer = Prioritizer(connectionBufferCapacity)
     val frameSender = FrameSender(prioritizer, connection.pool, maxFragmentSize)
     val streamsStorage = StreamsStorage(isServer, connection.pool)
     val keepAliveHandler = KeepAliveHandler(connectionConfig.keepAlive, frameSender)

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/internal/Prioritizer.kt

@@ -25,9 +25,9 @@ import kotlin.native.concurrent.*
 @SharedImmutable
 private val selectFrame: suspend (Frame) -> Frame = { it }
 
-internal class Prioritizer {
-    private val priorityChannel = SafeChannel<Frame>(Channel.UNLIMITED)
-    private val commonChannel = SafeChannel<Frame>(Channel.UNLIMITED)
+internal class Prioritizer(capacity: Int) {
+    private val priorityChannel = SafeChannel<Frame>(capacity)
+    private val commonChannel = SafeChannel<Frame>(capacity)
 
     suspend fun send(frame: Frame) {
         currentCoroutineContext().ensureActive()

rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/internal/FrameSenderTest.kt

@@ -3,11 +3,12 @@ package io.rsocket.kotlin.internal
 import io.rsocket.kotlin.frame.*
 import io.rsocket.kotlin.payload.*
 import io.rsocket.kotlin.test.*
+import kotlinx.coroutines.channels.*
 import kotlin.test.*
 
 class FrameSenderTest : SuspendTest, TestWithLeakCheck {
 
-    private val prioritizer = Prioritizer()
+    private val prioritizer = Prioritizer(Channel.UNLIMITED)
     private fun sender(maxFragmentSize: Int) = FrameSender(prioritizer, InUseTrackingPool, maxFragmentSize)
 
     @Test

rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/internal/PrioritizerTest.kt

@@ -20,10 +20,11 @@ import io.ktor.utils.io.core.*
 import io.rsocket.kotlin.frame.*
 import io.rsocket.kotlin.test.*
 import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.*
 import kotlin.test.*
 
 class PrioritizerTest : SuspendTest, TestWithLeakCheck {
-    private val prioritizer = Prioritizer()
+    private val prioritizer = Prioritizer(Channel.UNLIMITED)
 
     @Test
     fun testOrdering() = test {

rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/internal/RSocketRequesterTest.kt

@@ -37,6 +37,7 @@ class RSocketRequesterTest : TestWithConnection(), TestWithLeakCheck {
         requester = connect(
             connection = connection,
             isServer = false,
+            connectionBufferCapacity = 64,
             maxFragmentSize = 0,
             interceptors = InterceptorsBuilder().build(),
             connectionConfig = ConnectionConfig(

rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/keepalive/KeepAliveTest.kt

@@ -34,6 +34,7 @@ class KeepAliveTest : TestWithConnection(), TestWithLeakCheck {
     ): RSocket = connect(
         connection = connection,
         isServer = false,
+        connectionBufferCapacity = 64,
         maxFragmentSize = 0,
         interceptors = InterceptorsBuilder().build(),
         connectionConfig = ConnectionConfig(keepAlive, DefaultPayloadMimeType, Payload.Empty)