cleanups general code and public API

build.gradle.kts

@@ -14,9 +14,8 @@
  * limitations under the License.
  */
 
-import groovy.util.Node
-import groovy.util.NodeList
-import org.gradle.api.publish.maven.internal.artifact.FileBasedMavenArtifact
+import groovy.util.*
+import org.gradle.api.publish.maven.internal.artifact.*
 import org.jetbrains.kotlin.gradle.dsl.*
 import org.jetbrains.kotlin.gradle.plugin.mpp.*
 import org.jetbrains.kotlin.konan.target.*
@@ -162,20 +161,21 @@ subprojects {
                         useExperimentalAnnotation("kotlinx.coroutines.InternalCoroutinesApi")
                         useExperimentalAnnotation("kotlinx.coroutines.ObsoleteCoroutinesApi")
                         useExperimentalAnnotation("kotlinx.coroutines.FlowPreview")
+                        useExperimentalAnnotation("kotlinx.coroutines.DelicateCoroutinesApi")
 
-                        useExperimentalAnnotation("io.ktor.util.KtorExperimentalAPI")
                         useExperimentalAnnotation("io.ktor.util.InternalAPI")
                         useExperimentalAnnotation("io.ktor.utils.io.core.internal.DangerousInternalIoApi")
 
                         useExperimentalAnnotation("io.rsocket.kotlin.TransportApi")
                         useExperimentalAnnotation("io.rsocket.kotlin.ExperimentalMetadataApi")
                         useExperimentalAnnotation("io.rsocket.kotlin.ExperimentalStreamsApi")
+                        useExperimentalAnnotation("io.rsocket.kotlin.RSocketLoggingApi")
                     }
                 }
             }
 
             if (isLibProject && !isTestProject) {
-                explicitApiWarning() //TODO change to strict before release
+                explicitApi()
                 sourceSets["commonTest"].dependencies {
                     implementation(project(":rsocket-test"))
                 }
@@ -204,8 +204,10 @@ subprojects {
     }
 }
 
-fun publishPlatformArtifactsInRootModule(platformPublication:MavenPublication,
-                                         kotlinMultiplatformPublication: MavenPublication) {
+fun publishPlatformArtifactsInRootModule(
+    platformPublication: MavenPublication,
+    kotlinMultiplatformPublication: MavenPublication
+) {
     lateinit var platformXml: XmlProvider
 
     platformPublication.pom.withXml { platformXml = this }
@@ -312,7 +314,7 @@ subprojects {
             dependsOn(tasks.withType<Sign>())
         }
 
-        tasks.matching { it.name == "generatePomFileForKotlinMultiplatformPublication"}.configureEach {
+        tasks.matching { it.name == "generatePomFileForKotlinMultiplatformPublication" }.configureEach {
             dependsOn(tasks["generatePomFileForJvmPublication"])
         }
     }
@@ -421,7 +423,7 @@ if (sonatypeUsername != null && sonatypePassword != null) {
                                         it
                                     )
                                 }
-                                else -> it.artifactId = "${project.name}-$type"
+                                else                  -> it.artifactId = "${project.name}-$type"
                             }
                         }
                     }

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

@@ -16,7 +16,6 @@
 
 import io.rsocket.kotlin.*
 import io.rsocket.kotlin.core.*
-import io.rsocket.kotlin.payload.*
 import io.rsocket.kotlin.transport.local.*
 import kotlinx.coroutines.*
 

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

@@ -16,7 +16,6 @@
 
 import io.rsocket.kotlin.*
 import io.rsocket.kotlin.core.*
-import io.rsocket.kotlin.payload.*
 import io.rsocket.kotlin.transport.local.*
 import kotlinx.coroutines.*
 

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

@@ -16,7 +16,6 @@
 
 import io.rsocket.kotlin.*
 import io.rsocket.kotlin.core.*
-import io.rsocket.kotlin.payload.*
 import io.rsocket.kotlin.transport.local.*
 import kotlinx.coroutines.*
 

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

@@ -17,7 +17,6 @@
 import io.ktor.client.*
 import io.ktor.client.features.websocket.*
 import io.ktor.network.selector.*
-import io.ktor.network.sockets.*
 import io.ktor.util.*
 import io.rsocket.kotlin.*
 import io.rsocket.kotlin.core.*

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

@@ -16,7 +16,6 @@
 
 import io.ktor.utils.io.core.*
 import io.rsocket.kotlin.*
-import io.rsocket.kotlin.payload.*
 import kotlinx.serialization.*
 import kotlinx.serialization.protobuf.*
 

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

@@ -16,7 +16,6 @@
 
 import io.ktor.application.*
 import io.ktor.network.selector.*
-import io.ktor.network.sockets.*
 import io.ktor.routing.*
 import io.ktor.server.cio.*
 import io.ktor.server.engine.*

examples/nodejs-tcp-transport/src/jsMain/kotlin/Server.kt

@@ -18,7 +18,6 @@ import io.ktor.utils.io.core.*
 import io.ktor.utils.io.js.*
 import io.rsocket.kotlin.*
 import io.rsocket.kotlin.core.*
-import io.rsocket.kotlin.frame.io.*
 import io.rsocket.kotlin.payload.*
 import io.rsocket.kotlin.transport.*
 import kotlinx.coroutines.*
@@ -145,3 +144,17 @@ class NodeJsTcpConnection(private val socket: Socket) : Connection {
         return receiveChannel.receive()
     }
 }
+
+private fun ByteReadPacket.readLength(): Int {
+    val b = readByte().toInt() and 0xFF shl 16
+    val b1 = readByte().toInt() and 0xFF shl 8
+    val b2 = readByte().toInt() and 0xFF
+    return b or b1 or b2
+}
+
+private fun BytePacketBuilder.writeLength(length: Int) {
+    require(length and 0xFFFFFF.inv() == 0) { "Length is larger than 24 bits" }
+    writeByte((length shr 16).toByte())
+    writeByte((length shr 8).toByte())
+    writeByte(length.toByte())
+}

playground/src/commonMain/kotlin/TCP.kt

@@ -15,7 +15,6 @@
  */
 
 import io.ktor.network.selector.*
-import io.ktor.network.sockets.*
 import io.rsocket.kotlin.core.*
 import io.rsocket.kotlin.payload.*
 import io.rsocket.kotlin.transport.ktor.*

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

@@ -37,3 +37,12 @@ public annotation class ExperimentalMetadataApi
     message = "This is an API to customize request strategy of streams. This API can change in future in non backwards-compatible manner."
 )
 public annotation class ExperimentalStreamsApi
+
+@Retention(value = AnnotationRetention.BINARY)
+@RequiresOptIn(
+    level = RequiresOptIn.Level.WARNING,
+    message = "This is mostly internal API used for logging. This API can change in future in non backwards-compatible manner."
+)
+public annotation class RSocketLoggingApi
+
+

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

@@ -30,18 +30,16 @@ import kotlinx.coroutines.*
 public interface Connection {
     public val job: Job
 
-    @DangerousInternalIoApi
-    public val pool: ObjectPool<ChunkBuffer>
-        get() = ChunkBuffer.Pool
+    public val pool: ObjectPool<ChunkBuffer> get() = ChunkBuffer.Pool
 
     public suspend fun send(packet: ByteReadPacket)
     public suspend fun receive(): ByteReadPacket
 }
 
-@OptIn(DangerousInternalIoApi::class, TransportApi::class)
+@OptIn(TransportApi::class)
 internal suspend fun Connection.receiveFrame(): Frame = receive().readFrame(pool)
 
-@OptIn(DangerousInternalIoApi::class, TransportApi::class)
+@OptIn(TransportApi::class)
 internal suspend fun Connection.sendFrame(frame: Frame) {
     val packet = frame.toPacket(pool)
     packet.closeOnError { send(packet) }

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

@@ -19,18 +19,10 @@ package io.rsocket.kotlin
 import io.rsocket.kotlin.keepalive.*
 import io.rsocket.kotlin.payload.*
 
-
-//TODO fun interfaces don't support `suspend` functions for now... (seems will work in kotlin 1.5)
-
-public interface ConnectionAcceptor {
+public fun interface ConnectionAcceptor {
     public suspend fun ConnectionAcceptorContext.accept(): RSocket
 }
 
-public inline fun ConnectionAcceptor(crossinline block: suspend ConnectionAcceptorContext.() -> RSocket): ConnectionAcceptor =
-    object : ConnectionAcceptor {
-        override suspend fun ConnectionAcceptorContext.accept(): RSocket = block()
-    }
-
 public class ConnectionAcceptorContext internal constructor(
     public val config: ConnectionConfig,
     public val requester: RSocket,

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

@@ -46,16 +46,16 @@ public sealed class RSocketError(public val errorCode: Int, message: String) : T
             public const val MinAllowedCode: Int = ErrorCode.CustomMin
             public const val MaxAllowedCode: Int = ErrorCode.CustomMax
 
-            public inline fun checkCodeInAllowedRange(errorCode: Int): Boolean =
-                    MinAllowedCode <= errorCode || errorCode <= MaxAllowedCode
+            public fun checkCodeInAllowedRange(errorCode: Int): Boolean =
+                MinAllowedCode <= errorCode || errorCode <= MaxAllowedCode
         }
     }
 }
 
 @Suppress("FunctionName") // function name intentionally starts with an uppercase letter
 internal fun RSocketError(streamId: Int, errorCode: Int, message: String): Throwable =
     when (streamId) {
-        0 -> when (errorCode) {
+        0    -> when (errorCode) {
             ErrorCode.InvalidSetup     -> RSocketError.Setup.Invalid(message)
             ErrorCode.UnsupportedSetup -> RSocketError.Setup.Unsupported(message)
             ErrorCode.RejectedSetup    -> RSocketError.Setup.Rejected(message)
@@ -66,11 +66,11 @@ internal fun RSocketError(streamId: Int, errorCode: Int, message: String): Throw
         }
         else -> when (errorCode) {
             ErrorCode.ApplicationError -> RSocketError.ApplicationError(message)
-            ErrorCode.Rejected -> RSocketError.Rejected(message)
-            ErrorCode.Canceled -> RSocketError.Canceled(message)
-            ErrorCode.Invalid -> RSocketError.Invalid(message)
-            else -> when (RSocketError.Custom.checkCodeInAllowedRange(errorCode)) {
-                true -> RSocketError.Custom(errorCode, message)
+            ErrorCode.Rejected         -> RSocketError.Rejected(message)
+            ErrorCode.Canceled         -> RSocketError.Canceled(message)
+            ErrorCode.Invalid          -> RSocketError.Invalid(message)
+            else                       -> when (RSocketError.Custom.checkCodeInAllowedRange(errorCode)) {
+                true  -> RSocketError.Custom(errorCode, message)
                 false -> IllegalArgumentException("Invalid Error frame in Stream ID $streamId: $errorCode '$message'")
             }
         }

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

@@ -18,13 +18,13 @@ package io.rsocket.kotlin.core
 
 import io.rsocket.kotlin.frame.io.*
 
-public interface MimeType
+public sealed interface MimeType
 
-public interface MimeTypeWithName : MimeType {
+public sealed interface MimeTypeWithName : MimeType {
     public val text: String
 }
 
-public interface MimeTypeWithId : MimeType {
+public sealed interface MimeTypeWithId : MimeType {
     public val identifier: Byte
 }
 

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

@@ -19,19 +19,20 @@ package io.rsocket.kotlin.core
 import io.rsocket.kotlin.*
 import io.rsocket.kotlin.frame.*
 import io.rsocket.kotlin.frame.io.*
+import io.rsocket.kotlin.internal.*
 import io.rsocket.kotlin.logging.*
 import io.rsocket.kotlin.transport.*
 
-@OptIn(TransportApi::class)
-class RSocketConnector internal constructor(
+@OptIn(TransportApi::class, RSocketLoggingApi::class)
+public class RSocketConnector internal constructor(
     private val loggerFactory: LoggerFactory,
     private val interceptors: Interceptors,
     private val connectionConfigProvider: () -> ConnectionConfig,
     private val acceptor: ConnectionAcceptor,
     private val reconnectPredicate: ReconnectPredicate?,
 ) {
 
-    suspend fun connect(transport: ClientTransport): RSocket = when (reconnectPredicate) {
+    public suspend fun connect(transport: ClientTransport): RSocket = when (reconnectPredicate) {
         null -> connectOnce(transport)
         else -> ReconnectableRSocket(
             logger = loggerFactory.logger("io.rsocket.kotlin.connection"),

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

@@ -17,12 +17,14 @@
 package io.rsocket.kotlin.core
 
 import io.rsocket.kotlin.*
+import io.rsocket.kotlin.internal.*
 import io.rsocket.kotlin.keepalive.*
 import io.rsocket.kotlin.logging.*
 import io.rsocket.kotlin.payload.*
 import kotlinx.coroutines.*
 
 public class RSocketConnectorBuilder internal constructor() {
+    @RSocketLoggingApi
     public var loggerFactory: LoggerFactory = DefaultLoggerFactory
 
     private val connectionConfig: ConnectionConfigBuilder = ConnectionConfigBuilder()
@@ -42,10 +44,6 @@ public class RSocketConnectorBuilder internal constructor() {
         acceptor = block
     }
 
-    public fun acceptor(block: suspend ConnectionAcceptorContext.() -> RSocket) {
-        acceptor(ConnectionAcceptor(block))
-    }
-
     /**
      * When configured, [RSocketConnector.connect] will return custom [RSocket] implementation,
      * which will try to reconnect if connection lost and [retries] are not exhausted with [predicate] returning `true`.
@@ -95,6 +93,7 @@ public class RSocketConnectorBuilder internal constructor() {
         }
     }
 
+    @OptIn(RSocketLoggingApi::class)
     internal fun build(): RSocketConnector = RSocketConnector(
         loggerFactory,
         interceptors.build(),

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

@@ -19,21 +19,17 @@ package io.rsocket.kotlin.core
 import io.rsocket.kotlin.*
 import io.rsocket.kotlin.frame.*
 import io.rsocket.kotlin.frame.io.*
+import io.rsocket.kotlin.internal.*
 import io.rsocket.kotlin.logging.*
 import io.rsocket.kotlin.transport.*
 import kotlinx.coroutines.*
 
-@OptIn(TransportApi::class)
+@OptIn(TransportApi::class, RSocketLoggingApi::class)
 public class RSocketServer internal constructor(
     private val loggerFactory: LoggerFactory,
     private val interceptors: Interceptors,
 ) {
 
-    public fun <T> bind(
-        transport: ServerTransport<T>,
-        block: suspend ConnectionAcceptorContext.() -> RSocket
-    ): T = bind(transport, ConnectionAcceptor(block))
-
     public fun <T> bind(
         transport: ServerTransport<T>,
         acceptor: ConnectionAcceptor,

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

@@ -16,9 +16,11 @@
 
 package io.rsocket.kotlin.core
 
+import io.rsocket.kotlin.*
 import io.rsocket.kotlin.logging.*
 
 public class RSocketServerBuilder internal constructor() {
+    @RSocketLoggingApi
     public var loggerFactory: LoggerFactory = DefaultLoggerFactory
 
     private val interceptors: InterceptorsBuilder = InterceptorsBuilder()
@@ -27,6 +29,7 @@ public class RSocketServerBuilder internal constructor() {
         interceptors.configure()
     }
 
+    @OptIn(RSocketLoggingApi::class)
     internal fun build(): RSocketServer = RSocketServer(loggerFactory, interceptors.build())
 }
 

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/frame/CancelFrame.kt

@@ -19,8 +19,9 @@ package io.rsocket.kotlin.frame
 import io.ktor.utils.io.core.*
 
 internal class CancelFrame(
-    override val streamId: Int,
-) : Frame(FrameType.Cancel) {
+    override val streamId: Int
+) : Frame() {
+    override val type: FrameType get() = FrameType.Cancel
     override val flags: Int get() = 0
 
     override fun release(): Unit = Unit