Payload releasing (#108)

* add leak detection
* use leak detection in all core tests + transport test for local connection
* fix payload leaks
* mark some more functions as internal
* remove suppress for some functions

Author: whyoleg

README.md

@@ -39,10 +39,10 @@ RSocket interface contains 5 methods:
   `suspend fun metadataPush(metadata: ByteReadPacket)`
 
 ## Using in your projects
-The `master` branch is now dedicated to development of multiplatform rsocket-kotlin.
-For now only snapshots are available via [oss.jfrog.org](oss.jfrog.org) (OJO).
+The `master` branch is now dedicated to development of multiplatform rsocket-kotlin. For now only snapshots are available
+via [oss.jfrog.org](https://oss.jfrog.org/artifactory/oss-snapshot-local/io/rsocket/kotlin/) (OJO).
 
-Make sure, that you use Kotlin 1.4.
+Make sure, that you use Kotlin 1.4.X.
 
 ### Gradle:
 
@@ -225,7 +225,7 @@ val bufferedStream: Flow<Payload> = stream.buffer(10) //here buffer is 10, if `b
 bufferedStream.collect { payload: Payload ->
     println(payload.data.readText())
 }
-``` 
+```
 
 ## Bugs and Feedback
 

build.gradle.kts

@@ -195,7 +195,6 @@ subprojects {
 
         //common configuration
         extensions.configure<KotlinMultiplatformExtension> {
-//        explicitApiWarning() //TODO change to strict before release
             sourceSets.all {
                 languageSettings.apply {
                     progressiveMode = true
@@ -213,11 +212,13 @@ subprojects {
                         useExperimentalAnnotation("kotlinx.coroutines.FlowPreview")
                         useExperimentalAnnotation("io.ktor.util.KtorExperimentalAPI")
                         useExperimentalAnnotation("io.ktor.util.InternalAPI")
+                        useExperimentalAnnotation("io.ktor.utils.io.core.internal.DangerousInternalIoApi")
                     }
                 }
             }
 
             if (project.name != "rsocket-test") {
+                explicitApiWarning() //TODO change to strict before release
                 sourceSets["commonTest"].dependencies {
                     implementation(project(":rsocket-test"))
                 }

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

@@ -17,7 +17,6 @@
 import io.ktor.utils.io.core.*
 import io.rsocket.kotlin.payload.*
 
-@Suppress("FunctionName")
 fun Payload(route: String, packet: ByteReadPacket): Payload = Payload {
     data(packet)
     metadata(route)

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

@@ -17,20 +17,38 @@
 package io.rsocket.kotlin.connection
 
 import io.ktor.utils.io.core.*
+import io.ktor.utils.io.core.internal.*
+import io.ktor.utils.io.pool.*
 import io.rsocket.kotlin.*
 import io.rsocket.kotlin.core.*
+import io.rsocket.kotlin.frame.*
 import kotlinx.coroutines.*
 
+/**
+ * That interface isn't stable for inheritance.
+ */
 interface Connection : Cancelable {
+
+    @DangerousInternalIoApi
+    val pool: ObjectPool<ChunkBuffer>
+        get() = ChunkBuffer.Pool
+
     suspend fun send(packet: ByteReadPacket)
     suspend fun receive(): ByteReadPacket
 }
 
 suspend fun Connection.connectClient(
-    configuration: RSocketConnectorConfiguration = RSocketConnectorConfiguration()
+    configuration: RSocketConnectorConfiguration = RSocketConnectorConfiguration(),
 ): RSocket = RSocketConnector(ConnectionProvider(this), configuration).connect()
 
 suspend fun Connection.startServer(
     configuration: RSocketServerConfiguration = RSocketServerConfiguration(),
-    acceptor: RSocketAcceptor
+    acceptor: RSocketAcceptor,
 ): Job = RSocketServer(ConnectionProvider(this), configuration).start(acceptor)
+
+
+@OptIn(DangerousInternalIoApi::class)
+internal suspend fun Connection.receiveFrame(): Frame = receive().readFrame(pool)
+
+@OptIn(DangerousInternalIoApi::class)
+internal suspend fun Connection.sendFrame(frame: Frame): Unit = send(frame.toPacket(pool))

rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/connection/LoggingConnection.kt

@@ -17,18 +17,23 @@
 package io.rsocket.kotlin.connection
 
 import io.ktor.utils.io.core.*
+import io.ktor.utils.io.core.internal.*
 import io.rsocket.kotlin.frame.*
 import io.rsocket.kotlin.logging.*
-import kotlinx.coroutines.*
 
 internal fun Connection.logging(logger: Logger): Connection =
     if (logger.isLoggable(LoggingLevel.DEBUG)) LoggingConnection(this, logger) else this
 
+@OptIn(DangerousInternalIoApi::class)
 private class LoggingConnection(
     private val delegate: Connection,
     private val logger: Logger,
-) : Connection {
-    override val job: Job get() = delegate.job
+) : Connection by delegate {
+
+    private fun ByteReadPacket.dumpFrameToString(): String {
+        val length = remaining
+        return copy().use { it.readFrame(pool).use { it.dump(length) } }
+    }
 
     override suspend fun send(packet: ByteReadPacket) {
         logger.debug { "Send: ${packet.dumpFrameToString()}" }

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

@@ -46,7 +46,6 @@ class RSocketConnector(
             connection = connection,
             plugin = configuration.plugin,
             setupFrame = setupFrame,
-            ignoredFrameConsumer = configuration.ignoredFrameConsumer,
             acceptor = configuration.acceptor
         )
     }

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

@@ -17,7 +17,6 @@
 package io.rsocket.kotlin.core
 
 import io.rsocket.kotlin.*
-import io.rsocket.kotlin.frame.*
 import io.rsocket.kotlin.keepalive.*
 import io.rsocket.kotlin.logging.*
 import io.rsocket.kotlin.payload.*
@@ -29,6 +28,5 @@ data class RSocketConnectorConfiguration(
     val keepAlive: KeepAlive = KeepAlive(),
     val payloadMimeType: PayloadMimeType = PayloadMimeType(),
     val setupPayload: Payload = Payload.Empty,
-    val ignoredFrameConsumer: (Frame) -> Unit = {},
     val acceptor: RSocketAcceptor = { RSocketRequestHandler { } },
 )

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

@@ -35,7 +35,7 @@ class RSocketServer(
                 .let(configuration.plugin::wrapConnection)
                 .logging(configuration.loggerFactory.logger("io.rsocket.kotlin.frame.Frame"))
 
-        val setupFrame = connection.receive().toFrame()
+        val setupFrame = connection.receiveFrame()
         if (setupFrame !is SetupFrame)
             connection.failSetup(RSocketError.Setup.Invalid("Invalid setup frame: ${setupFrame.type}"))
         if (setupFrame.version != Version.Current)
@@ -46,7 +46,6 @@ class RSocketServer(
                 connection = connection,
                 plugin = configuration.plugin,
                 setupFrame = setupFrame,
-                ignoredFrameConsumer = configuration.ignoredFrameConsumer,
                 acceptor = acceptor
             )
         } catch (e: Throwable) {
@@ -55,7 +54,7 @@ class RSocketServer(
     }
 
     private suspend fun Connection.failSetup(error: RSocketError.Setup): Nothing {
-        send(ErrorFrame(0, error).toPacket())
+        sendFrame(ErrorFrame(0, error))
         cancel("Setup failed", error)
         throw error
     }

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

@@ -16,12 +16,10 @@
 
 package io.rsocket.kotlin.core
 
-import io.rsocket.kotlin.frame.*
 import io.rsocket.kotlin.logging.*
 import io.rsocket.kotlin.plugin.*
 
 data class RSocketServerConfiguration(
     val plugin: Plugin = Plugin(),
     val loggerFactory: LoggerFactory = DefaultLoggerFactory,
-    val ignoredFrameConsumer: (Frame) -> Unit = {},
 )

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

@@ -18,10 +18,13 @@ package io.rsocket.kotlin.frame
 
 import io.ktor.utils.io.core.*
 
-class CancelFrame(
+internal class CancelFrame(
     override val streamId: Int,
 ) : Frame(FrameType.Cancel) {
     override val flags: Int get() = 0
+
+    override fun release(): Unit = Unit
+
     override fun BytePacketBuilder.writeSelf(): Unit = Unit
 
     override fun StringBuilder.appendFlags(): Unit = Unit