WIP: rework configuration API

Author: olme04

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

@@ -18,7 +18,7 @@ package io.rsocket.kotlin.benchmarks
 
 import io.ktor.utils.io.core.*
 import io.rsocket.kotlin.*
-import io.rsocket.kotlin.core.*
+import io.rsocket.kotlin.connect.*
 import io.rsocket.kotlin.payload.*
 import io.rsocket.kotlin.transport.local.*
 import kotlinx.coroutines.*
@@ -40,7 +40,7 @@ class RSocketKotlinBenchmark : RSocketBenchmark<Payload>() {
         payload = createPayload(payloadSize)
         payloadsFlow = flow { repeat(5000) { emit(payloadCopy()) } }
         val server = RSocketServer().bindIn(CoroutineScope(benchJob + Dispatchers.Unconfined), LocalServerTransport()) {
-            RSocket {
+            responder {
                 onRequestResponse {
                     it.close()
                     payloadCopy()

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

@@ -19,11 +19,6 @@ package io.rsocket.kotlin
 import io.ktor.utils.io.core.*
 import io.rsocket.kotlin.payload.*
 import kotlinx.coroutines.flow.*
-import kotlin.coroutines.*
-
-public sealed interface ConnectedRSocket : RSocket {
-    public val session: RSocketSession
-}
 
 public interface RSocket {
 
@@ -44,6 +39,8 @@ public interface RSocket {
 
 }
 
+public object EmptyRSocket : RSocket
+
 public sealed interface RSocketBuilder {
     public fun onMetadataPush(block: suspend RSocket.(metadata: ByteReadPacket) -> Unit)
     public fun onFireAndForget(block: suspend RSocket.(payload: Payload) -> Unit)
@@ -82,14 +79,6 @@ public suspend fun <C : Closeable> FlowCollector<C>.emitOrClose(value: C) {
     }
 }
 
-internal abstract class ConnectedRSocketImpl(
-    final override val coroutineContext: CoroutineContext,
-) : ConnectedRSocket, RSocketSession {
-    final override val session: RSocketSession get() = this
-}
-
-internal object EmptyRSocket : RSocket
-
 @PublishedApi
 internal class RSocketImpl : RSocketBuilder, RSocket {
     private var metadataPush: suspend RSocket.(metadata: ByteReadPacket) -> Unit =

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

@@ -0,0 +1,6 @@
+package io.rsocket.kotlin.configuration
+
+internal interface ConfigurationState {
+    fun checkConfigured()
+    fun checkNotConfigured()
+}

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/configuration/ConfigurationState.kt

@@ -0,0 +1,8 @@
+package io.rsocket.kotlin.configuration
+
+@DslMarker
+public annotation class ConnectConfigurationDsl
+
+//marker interface
+@ConnectConfigurationDsl
+public sealed interface ConnectConfiguration

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/configuration/ConnectConfigurationDsl.kt

@@ -0,0 +1,42 @@
+package io.rsocket.kotlin.configuration
+
+import kotlin.time.*
+import kotlin.time.Duration.Companion.seconds
+
+public sealed interface KeepAliveConfiguration {
+    public val interval: Duration
+    public val maxLifetime: Duration
+}
+
+public sealed interface KeepAliveConnectConfiguration : KeepAliveConfiguration, ConnectConfiguration
+
+public sealed interface KeepAliveClientConnectConfiguration : KeepAliveConnectConfiguration {
+    public fun interval(duration: Duration)
+    public fun maxLifetime(duration: Duration)
+}
+
+public sealed interface KeepAliveServerConnectConfiguration : KeepAliveConnectConfiguration
+
+internal class KeepAliveClientConnectConfigurationImpl(
+    private val configurationState: ConfigurationState,
+) : KeepAliveClientConnectConfiguration {
+    override var interval: Duration = 20.seconds
+        private set
+    override var maxLifetime: Duration = 90.seconds
+        private set
+
+    override fun interval(duration: Duration) {
+        configurationState.checkNotConfigured()
+        interval = duration
+    }
+
+    override fun maxLifetime(duration: Duration) {
+        configurationState.checkNotConfigured()
+        maxLifetime = duration
+    }
+}
+
+internal class KeepAliveServerConnectConfigurationImpl(
+    override val interval: Duration,
+    override val maxLifetime: Duration,
+) : KeepAliveServerConnectConfiguration

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/configuration/KeepAliveConfiguration.kt

@@ -0,0 +1,41 @@
+package io.rsocket.kotlin.configuration
+
+import io.rsocket.kotlin.core.*
+
+public sealed interface MimeTypeConfiguration {
+    public val metadata: MimeTypeWithName
+    public val data: MimeTypeWithName
+}
+
+public sealed interface MimeTypeConnectConfiguration : MimeTypeConfiguration, ConnectConfiguration
+
+public sealed interface MimeTypeClientConnectConfiguration : MimeTypeConnectConfiguration {
+    public fun metadata(mimeType: MimeTypeWithName)
+    public fun data(mimeType: MimeTypeWithName)
+}
+
+public sealed interface MimeTypeServerConnectConfiguration : MimeTypeConnectConfiguration
+
+internal class MimeTypeClientConnectConfigurationImpl(
+    private val configurationState: ConfigurationState,
+) : MimeTypeClientConnectConfiguration {
+    override var metadata: MimeTypeWithName = WellKnownMimeType.ApplicationOctetStream
+        private set
+    override var data: MimeTypeWithName = WellKnownMimeType.ApplicationOctetStream
+        private set
+
+    override fun metadata(mimeType: MimeTypeWithName) {
+        configurationState.checkNotConfigured()
+        metadata = mimeType
+    }
+
+    override fun data(mimeType: MimeTypeWithName) {
+        configurationState.checkNotConfigured()
+        data = mimeType
+    }
+}
+
+internal class MimeTypeServerConnectConfigurationImpl(
+    override val metadata: MimeTypeWithName,
+    override val data: MimeTypeWithName,
+) : MimeTypeServerConnectConfiguration

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/configuration/MimeTypeConfiguration.kt

@@ -0,0 +1,63 @@
+package io.rsocket.kotlin.configuration
+
+import io.rsocket.kotlin.core.*
+
+public sealed interface PayloadConfiguration {
+    //TODO: Long?
+    //TODO: add later
+    //max payload overall size
+//    public val maxSize: Long
+
+    //TODO: rename
+    //size of fragment, after fragmentation
+    // f.e. if 100, and payload = 450, will create 5 fragments when sending
+    public val maxFragmentSize: Int
+
+    public val mimeType: MimeTypeConfiguration
+}
+
+public sealed interface PayloadConnectConfiguration : PayloadConfiguration, ConnectConfiguration {
+    override val mimeType: MimeTypeConnectConfiguration
+
+    //    public fun maxSize(value: Int)
+    public fun maxFragmentSize(value: Int)
+}
+
+public sealed interface PayloadClientConnectConfiguration : PayloadConnectConfiguration {
+    override val mimeType: MimeTypeClientConnectConfiguration
+}
+
+public sealed interface PayloadServerConnectConfiguration : PayloadConnectConfiguration {
+    override val mimeType: MimeTypeServerConnectConfiguration
+}
+
+internal abstract class PayloadConnectConfigurationImpl(
+    private val configurationState: ConfigurationState,
+) : PayloadConnectConfiguration {
+    final override var maxFragmentSize: Int = 0
+        private set
+
+    final override fun maxFragmentSize(value: Int) {
+        configurationState.checkNotConfigured()
+        require(value == 0 || value >= 64) {
+            "maxFragmentSize should be zero (no fragmentation) or greater than or equal to 64, but was $value"
+        }
+        maxFragmentSize = value
+    }
+}
+
+internal class PayloadClientConnectConfigurationImpl(
+    configurationState: ConfigurationState,
+) : PayloadClientConnectConfiguration, PayloadConnectConfigurationImpl(configurationState) {
+    override val mimeType: MimeTypeClientConnectConfigurationImpl =
+        MimeTypeClientConnectConfigurationImpl(configurationState)
+}
+
+internal class PayloadServerConnectConfigurationImpl(
+    configurationState: ConfigurationState,
+    metadataMimeType: MimeTypeWithName,
+    dataMimeType: MimeTypeWithName,
+) : PayloadServerConnectConfiguration, PayloadConnectConfigurationImpl(configurationState) {
+    override val mimeType: MimeTypeServerConnectConfigurationImpl =
+        MimeTypeServerConnectConfigurationImpl(metadataMimeType, dataMimeType)
+}

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/configuration/PayloadConfiguration.kt

@@ -0,0 +1,65 @@
+package io.rsocket.kotlin.configuration
+
+import io.ktor.utils.io.core.*
+import io.rsocket.kotlin.core.*
+import io.rsocket.kotlin.payload.*
+import kotlin.time.*
+
+public sealed interface RSocketConfiguration {
+    public val setup: SetupConfiguration
+    public val payload: PayloadConfiguration
+    public val keepAlive: KeepAliveConfiguration
+}
+
+public sealed interface RSocketConnectConfiguration : RSocketConfiguration, ConnectConfiguration {
+    override val setup: SetupConnectConfiguration
+    override val payload: PayloadConnectConfiguration
+    override val keepAlive: KeepAliveConnectConfiguration
+}
+
+public sealed interface RSocketClientConnectConfiguration : RSocketConnectConfiguration {
+    override val setup: SetupClientConnectConfiguration
+    override val payload: PayloadClientConnectConfiguration
+    override val keepAlive: KeepAliveClientConnectConfiguration
+
+    public val reconnect: ReconnectConfiguration
+}
+
+public sealed interface RSocketServerConnectConfiguration : RSocketConnectConfiguration {
+    override val setup: SetupServerConnectConfiguration
+    override val payload: PayloadServerConnectConfiguration
+    override val keepAlive: KeepAliveServerConnectConfiguration
+}
+
+internal abstract class RSocketConnectConfigurationImpl : RSocketConnectConfiguration, Closeable {
+    abstract override val setup: SetupConnectConfigurationImpl
+
+    override fun close() {
+        setup.close()
+    }
+}
+
+internal class RSocketClientConnectConfigurationImpl(
+    configurationState: ConfigurationState,
+) : RSocketClientConnectConfiguration, RSocketConnectConfigurationImpl() {
+    override val setup: SetupClientConnectConfigurationImpl = SetupClientConnectConfigurationImpl(configurationState)
+    override val payload: PayloadClientConnectConfigurationImpl = PayloadClientConnectConfigurationImpl(configurationState)
+    override val keepAlive: KeepAliveClientConnectConfigurationImpl = KeepAliveClientConnectConfigurationImpl(configurationState)
+    override val reconnect: ReconnectConfigurationImpl = ReconnectConfigurationImpl(configurationState)
+}
+
+internal class RSocketServerConnectConfigurationImpl(
+    configurationState: ConfigurationState,
+    keepAliveInterval: Duration,
+    keepAliveMaxLifetime: Duration,
+    metadataMimeType: MimeTypeWithName,
+    dataMimeType: MimeTypeWithName,
+    setupPayload: Payload,
+) : RSocketServerConnectConfiguration, RSocketConnectConfigurationImpl() {
+    override val setup: SetupServerConnectConfigurationImpl =
+        SetupServerConnectConfigurationImpl(setupPayload)
+    override val payload: PayloadServerConnectConfigurationImpl =
+        PayloadServerConnectConfigurationImpl(configurationState, metadataMimeType, dataMimeType)
+    override val keepAlive: KeepAliveServerConnectConfigurationImpl =
+        KeepAliveServerConnectConfigurationImpl(keepAliveInterval, keepAliveMaxLifetime)
+}

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/configuration/RSocketConfiguration.kt

@@ -0,0 +1,31 @@
+package io.rsocket.kotlin.configuration
+
+public sealed interface ReconnectConfiguration : ConnectConfiguration {
+    //if connection failed, do we need reconnect?
+    // depending on cause, f.e. if Setup error because of unsupported setup
+    public fun reconnectOn(predicate: suspend (cause: Throwable?) -> Boolean)
+
+    //when connection establishment failed, f.e. due to network issue
+    // if specified, reconnectOn will be implicitly all time equals true
+    public fun retryWhen(predicate: suspend (cause: Throwable, attempt: Long) -> Boolean)
+}
+
+internal class ReconnectConfigurationImpl(
+    private val configurationState: ConfigurationState,
+) : ReconnectConfiguration {
+    var reconnectOn: (suspend (cause: Throwable?) -> Boolean)? = null
+        private set
+
+    var retryWhen: (suspend (cause: Throwable, attempt: Long) -> Boolean)? = null
+        private set
+
+    override fun reconnectOn(predicate: suspend (cause: Throwable?) -> Boolean) {
+        configurationState.checkNotConfigured()
+        reconnectOn = predicate
+    }
+
+    override fun retryWhen(predicate: suspend (cause: Throwable, attempt: Long) -> Boolean) {
+        configurationState.checkNotConfigured()
+        retryWhen = predicate
+    }
+}