reformat code

olme04 · whyoleg · commit 58ca7978603d · 2022-04-14T09:11:42.000+03:00
diff --git a/benchmarks/src/jvmMain/kotlin/io/rsocket/kotlin/benchmarks/RSocketBenchmark.kt b/benchmarks/src/jvmMain/kotlin/io/rsocket/kotlin/benchmarks/RSocketBenchmark.kt
@@ -113,11 +113,12 @@ abstract class RSocketBenchmark<Payload : Any> {
     }
 
     //Run every request in separate coroutine which will be dispatched on Default dispatcher (threads amount = cores amount)
-    private inline fun parallel(bh: Blackhole, p: Int, crossinline block: suspend (bh: Blackhole) -> Unit): Unit = runBlocking {
-        (0..p).map {
-            GlobalScope.async { block(bh) }
-        }.awaitAll()
-    }
+    private inline fun parallel(bh: Blackhole, p: Int, crossinline block: suspend (bh: Blackhole) -> Unit): Unit =
+        runBlocking {
+            (0..p).map {
+                GlobalScope.async { block(bh) }
+            }.awaitAll()
+        }
 
     //Run every request in separate coroutine, but on single thread dispatcher:
     //  - do request 1
@@ -128,9 +129,10 @@ abstract class RSocketBenchmark<Payload : Any> {
     //  - receive result on request 2
     //  - ....
     //working with requests is single threaded but concurrent
-    private inline fun concurrent(bh: Blackhole, p: Int, crossinline block: suspend (bh: Blackhole) -> Unit): Unit = runBlocking {
-        (0..p).map {
-            async { block(bh) }
-        }.awaitAll()
-    }
+    private inline fun concurrent(bh: Blackhole, p: Int, crossinline block: suspend (bh: Blackhole) -> Unit): Unit =
+        runBlocking {
+            (0..p).map {
+                async { block(bh) }
+            }.awaitAll()
+        }
 }
diff --git a/benchmarks/src/kotlinMain/kotlin/io/rsocket/kotlin/benchmarks/RSocketKotlinBenchmark.kt b/benchmarks/src/kotlinMain/kotlin/io/rsocket/kotlin/benchmarks/RSocketKotlinBenchmark.kt
@@ -81,6 +81,7 @@ class RSocketKotlinBenchmark : RSocketBenchmark<Payload>() {
 
     override suspend fun doRequestStream(): Flow<Payload> = client.requestStream(payloadCopy()).flowOn(requestStrategy)
 
-    override suspend fun doRequestChannel(): Flow<Payload> = client.requestChannel(payloadCopy(), payloadsFlow).flowOn(requestStrategy)
+    override suspend fun doRequestChannel(): Flow<Payload> =
+        client.requestChannel(payloadCopy(), payloadsFlow).flowOn(requestStrategy)
 
 }
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/Connection.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/Connection.kt
@@ -35,7 +35,8 @@ public interface Connection : CoroutineScope {
 }
 
 @OptIn(TransportApi::class)
-internal suspend inline fun <T> Connection.receiveFrame(block: (frame: Frame) -> T): T = receive().readFrame(pool).closeOnError(block)
+internal suspend inline fun <T> Connection.receiveFrame(block: (frame: Frame) -> T): T =
+    receive().readFrame(pool).closeOnError(block)
 
 @OptIn(TransportApi::class)
 internal suspend fun Connection.sendFrame(frame: Frame) {
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/RSocketRequestHandler.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/RSocketRequestHandler.kt
@@ -27,7 +27,8 @@ public class RSocketRequestHandlerBuilder internal constructor() {
     private var fireAndForget: (suspend RSocket.(payload: Payload) -> Unit)? = null
     private var requestResponse: (suspend RSocket.(payload: Payload) -> Payload)? = null
     private var requestStream: (suspend RSocket.(payload: Payload) -> Flow<Payload>)? = null
-    private var requestChannel: (suspend RSocket.(initPayload: Payload, payloads: Flow<Payload>) -> Flow<Payload>)? = null
+    private var requestChannel: (suspend RSocket.(initPayload: Payload, payloads: Flow<Payload>) -> Flow<Payload>)? =
+        null
 
     public fun metadataPush(block: (suspend RSocket.(metadata: ByteReadPacket) -> Unit)) {
         check(metadataPush == null) { "Metadata Push handler already configured" }
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/RequestStrategy.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/RequestStrategy.kt
@@ -24,7 +24,8 @@ import kotlin.coroutines.*
 private val DefaultStrategy: RequestStrategy = PrefetchStrategy(64, 16)
 
 @ExperimentalStreamsApi
-internal fun CoroutineContext.requestStrategy(): RequestStrategy.Element = (get(RequestStrategy) ?: DefaultStrategy).provide()
+internal fun CoroutineContext.requestStrategy(): RequestStrategy.Element =
+    (get(RequestStrategy) ?: DefaultStrategy).provide()
 
 @ExperimentalStreamsApi
 public interface RequestStrategy : CoroutineContext.Element {
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/core/InterceptorsBuilder.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/core/InterceptorsBuilder.kt
@@ -54,5 +54,6 @@ internal class Interceptors(
     fun wrapRequester(requester: RSocket): RSocket = requesters.fold(requester) { r, i -> i.intercept(r) }
     fun wrapResponder(responder: RSocket): RSocket = responders.fold(responder) { r, i -> i.intercept(r) }
     fun wrapConnection(connection: Connection): Connection = connections.fold(connection) { c, i -> i.intercept(c) }
-    fun wrapAcceptor(connection: ConnectionAcceptor): ConnectionAcceptor = acceptors.fold(connection) { c, i -> i.intercept(c) }
+    fun wrapAcceptor(connection: ConnectionAcceptor): ConnectionAcceptor =
+        acceptors.fold(connection) { c, i -> i.intercept(c) }
 }
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/frame/MetadataPushFrame.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/frame/MetadataPushFrame.kt
@@ -45,4 +45,5 @@ internal class MetadataPushFrame(
     }
 }
 
-internal fun ByteReadPacket.readMetadataPush(pool: ObjectPool<ChunkBuffer>): MetadataPushFrame = MetadataPushFrame(readPacket(pool))
+internal fun ByteReadPacket.readMetadataPush(pool: ObjectPool<ChunkBuffer>): MetadataPushFrame =
+    MetadataPushFrame(readPacket(pool))
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/internal/CloseOperations.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/internal/CloseOperations.kt
@@ -31,7 +31,8 @@ internal inline fun <T : Closeable, R> T.closeOnError(block: (T) -> R): R {
 private val onUndeliveredCloseable: (Closeable) -> Unit = Closeable::close
 
 @Suppress("FunctionName")
-internal fun <E : Closeable> SafeChannel(capacity: Int): Channel<E> = Channel(capacity, onUndeliveredElement = onUndeliveredCloseable)
+internal fun <E : Closeable> SafeChannel(capacity: Int): Channel<E> =
+    Channel(capacity, onUndeliveredElement = onUndeliveredCloseable)
 
 internal fun <E : Closeable> SendChannel<E>.safeTrySend(element: E) {
     trySend(element).onFailure { element.close() }
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/internal/FrameSender.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/internal/FrameSender.kt
@@ -42,7 +42,9 @@ internal class FrameSender(
     suspend fun sendMetadataPush(metadata: ByteReadPacket): Unit = prioritizer.send(MetadataPushFrame(metadata))
 
     suspend fun sendCancel(id: Int): Unit = withContext(NonCancellable) { prioritizer.send(CancelFrame(id)) }
-    suspend fun sendError(id: Int, throwable: Throwable): Unit = withContext(NonCancellable) { prioritizer.send(ErrorFrame(id, throwable)) }
+    suspend fun sendError(id: Int, throwable: Throwable): Unit =
+        withContext(NonCancellable) { prioritizer.send(ErrorFrame(id, throwable)) }
+
     suspend fun sendRequestN(id: Int, n: Int): Unit = prioritizer.send(RequestNFrame(id, n))
 
     suspend fun sendRequestPayload(type: FrameType, streamId: Int, payload: Payload, initialRequest: Int = 0) {
@@ -103,14 +105,24 @@ internal class FrameSender(
             val fType = if (first && type.isRequestType) type else FrameType.Payload
             val fragment = Payload(dataFragment, metadataFragment)
             val follows = metadata != null && metadata.isNotEmpty || data.isNotEmpty
-            prioritizer.send(RequestFrame(fType, streamId, follows, (!follows && complete), !fType.isRequestType, initialRequest, fragment))
+            prioritizer.send(
+                RequestFrame(
+                    type = fType,
+                    streamId = streamId,
+                    follows = follows,
+                    complete = (!follows && complete),
+                    next = !fType.isRequestType,
+                    initialRequest = initialRequest,
+                    payload = fragment
+                )
+            )
             first = false
             remaining = fragmentSize
         } while (follows)
     }
 
     private fun Payload.isFragmentable(hasInitialRequest: Boolean) = when (maxFragmentSize) {
-        0    -> false
+        0 -> false
         else -> when (val meta = metadata) {
             null -> data.remaining > maxFragmentSize - fragmentOffset - (if (hasInitialRequest) Int.SIZE_BYTES else 0)
             else -> data.remaining + meta.remaining > maxFragmentSize - fragmentOffsetWithMetadata - (if (hasInitialRequest) Int.SIZE_BYTES else 0)
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/internal/Limiter.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/internal/Limiter.kt
@@ -23,7 +23,10 @@ import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.*
 import kotlin.coroutines.*
 
-internal suspend inline fun Flow<Payload>.collectLimiting(limiter: Limiter, crossinline action: suspend (value: Payload) -> Unit) {
+internal suspend inline fun Flow<Payload>.collectLimiting(
+    limiter: Limiter,
+    crossinline action: suspend (value: Payload) -> Unit
+) {
     collect { payload ->
         payload.closeOnError {
             limiter.useRequest()
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/internal/RSocketRequester.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/internal/RSocketRequester.kt
@@ -87,29 +87,30 @@ internal class RSocketRequester(
         }
     }
 
-    override fun requestChannel(initPayload: Payload, payloads: Flow<Payload>): Flow<Payload> = requestFlow { strategy, initialRequest ->
-        ensureActiveOrRelease(initPayload)
-
-        val id = streamsStorage.nextId()
-
-        val channel = SafeChannel<Payload>(Channel.UNLIMITED)
-        val limiter = Limiter(0)
-        val payloadsJob = Job(this@RSocketRequester.coroutineContext.job)
-        val handler = RequesterRequestChannelFrameHandler(id, streamsStorage, limiter, payloadsJob, channel, pool)
-        streamsStorage.save(id, handler)
-
-        handler.receiveOrCancel(id, initPayload) {
-            sender.sendRequestPayload(FrameType.RequestChannel, id, initPayload, initialRequest)
-            //TODO lazy?
-            launch(payloadsJob) {
-                handler.sendOrFail(id) {
-                    payloads.collectLimiting(limiter) { sender.sendNextPayload(id, it) }
-                    sender.sendCompletePayload(id)
+    override fun requestChannel(initPayload: Payload, payloads: Flow<Payload>): Flow<Payload> =
+        requestFlow { strategy, initialRequest ->
+            ensureActiveOrRelease(initPayload)
+
+            val id = streamsStorage.nextId()
+
+            val channel = SafeChannel<Payload>(Channel.UNLIMITED)
+            val limiter = Limiter(0)
+            val payloadsJob = Job(this@RSocketRequester.coroutineContext.job)
+            val handler = RequesterRequestChannelFrameHandler(id, streamsStorage, limiter, payloadsJob, channel, pool)
+            streamsStorage.save(id, handler)
+
+            handler.receiveOrCancel(id, initPayload) {
+                sender.sendRequestPayload(FrameType.RequestChannel, id, initPayload, initialRequest)
+                //TODO lazy?
+                launch(payloadsJob) {
+                    handler.sendOrFail(id) {
+                        payloads.collectLimiting(limiter) { sender.sendNextPayload(id, it) }
+                        sender.sendCompletePayload(id)
+                    }
                 }
+                emitAllWithRequestN(channel, strategy) { sender.sendRequestN(id, it) }
             }
-            emitAllWithRequestN(channel, strategy) { sender.sendRequestN(id, it) }
         }
-    }
 
     private suspend inline fun SendFrameHandler.sendOrFail(id: Int, block: () -> Unit) {
         try {
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/internal/RSocketResponder.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/internal/RSocketResponder.kt
@@ -64,7 +64,8 @@ internal class RSocketResponder(
             }
         }
         handler.sendOrFail(id, payload) {
-            requestHandler.requestChannel(payload, payloads).collectLimiting(handler.limiter) { sender.sendNextPayload(id, it) }
+            requestHandler.requestChannel(payload, payloads)
+                .collectLimiting(handler.limiter) { sender.sendNextPayload(id, it) }
             sender.sendCompletePayload(id)
         }
     }.closeOnCompletion(payload)
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/internal/RequestFlow.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/internal/RequestFlow.kt
@@ -27,7 +27,11 @@ import kotlinx.coroutines.flow.*
 internal inline fun requestFlow(
     crossinline block: suspend FlowCollector<Payload>.(strategy: RequestStrategy.Element, initialRequest: Int) -> Unit
 ): Flow<Payload> = object : RequestFlow() {
-    override suspend fun collect(collector: FlowCollector<Payload>, strategy: RequestStrategy.Element, initialRequest: Int) {
+    override suspend fun collect(
+        collector: FlowCollector<Payload>,
+        strategy: RequestStrategy.Element,
+        initialRequest: Int
+    ) {
         collector.block(strategy, initialRequest)
     }
 }
@@ -59,7 +63,11 @@ internal abstract class RequestFlow : Flow<Payload> {
         collect(collector, strategy, initial)
     }
 
-    abstract suspend fun collect(collector: FlowCollector<Payload>, strategy: RequestStrategy.Element, initialRequest: Int)
+    abstract suspend fun collect(
+        collector: FlowCollector<Payload>,
+        strategy: RequestStrategy.Element,
+        initialRequest: Int
+    )
 }
 
 @ExperimentalStreamsApi
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/metadata/CompositeMetadataExtensions.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/metadata/CompositeMetadataExtensions.kt
@@ -26,7 +26,10 @@ import io.rsocket.kotlin.core.*
 public fun CompositeMetadata.Entry.hasMimeTypeOf(reader: MetadataReader<*>): Boolean = mimeType == reader.mimeType
 
 @ExperimentalMetadataApi
-public fun <M : Metadata> CompositeMetadata.Entry.read(reader: MetadataReader<M>, pool: ObjectPool<ChunkBuffer> = ChunkBuffer.Pool): M {
+public fun <M : Metadata> CompositeMetadata.Entry.read(
+    reader: MetadataReader<M>,
+    pool: ObjectPool<ChunkBuffer> = ChunkBuffer.Pool
+): M {
     if (mimeType == reader.mimeType) return content.read(reader, pool)
 
     content.close()
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/metadata/Metadata.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/metadata/Metadata.kt
@@ -41,9 +41,13 @@ public interface MetadataReader<M : Metadata> {
 public fun PayloadBuilder.metadata(metadata: Metadata): Unit = metadata(metadata.toPacket())
 
 @ExperimentalMetadataApi
-public fun <M : Metadata> ByteReadPacket.read(reader: MetadataReader<M>, pool: ObjectPool<ChunkBuffer> = ChunkBuffer.Pool): M = use {
+public fun <M : Metadata> ByteReadPacket.read(
+    reader: MetadataReader<M>,
+    pool: ObjectPool<ChunkBuffer> = ChunkBuffer.Pool
+): M = use {
     with(reader) { read(pool) }
 }
 
 @ExperimentalMetadataApi
-public fun Metadata.toPacket(pool: ObjectPool<ChunkBuffer> = ChunkBuffer.Pool): ByteReadPacket = buildPacket(pool) { writeSelf() }
+public fun Metadata.toPacket(pool: ObjectPool<ChunkBuffer> = ChunkBuffer.Pool): ByteReadPacket =
+    buildPacket(pool) { writeSelf() }
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/metadata/RawMetadata.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/metadata/RawMetadata.kt
@@ -37,7 +37,8 @@ public class RawMetadata(
     }
 
     private class Reader(override val mimeType: MimeType) : MetadataReader<RawMetadata> {
-        override fun ByteReadPacket.read(pool: ObjectPool<ChunkBuffer>): RawMetadata = RawMetadata(mimeType, readPacket(pool))
+        override fun ByteReadPacket.read(pool: ObjectPool<ChunkBuffer>): RawMetadata =
+            RawMetadata(mimeType, readPacket(pool))
     }
 
     public companion object {
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/metadata/ZipkinTracingMetadata.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/metadata/ZipkinTracingMetadata.kt
@@ -91,7 +91,16 @@ public class ZipkinTracingMetadata internal constructor(
             val spanId = readLong()
             val parentSpanId = if (hasParentSpanId) readLong() else 0
 
-            return ZipkinTracingMetadata(kind, true, hasParentSpanId, extendedTraceId, traceId, traceIdHigh, spanId, parentSpanId)
+            return ZipkinTracingMetadata(
+                kind = kind,
+                hasIds = true,
+                hasParentSpanId = hasParentSpanId,
+                extendedTraceId = extendedTraceId,
+                traceId = traceId,
+                traceIdHigh = traceIdHigh,
+                spanId = spanId,
+                parentSpanId = parentSpanId
+            )
         }
     }
 }
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/metadata/security/RawAuthMetadata.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/metadata/security/RawAuthMetadata.kt
@@ -48,7 +48,10 @@ public class RawAuthMetadata(
 public fun RawAuthMetadata.hasAuthTypeOf(reader: AuthMetadataReader<*>): Boolean = type == reader.mimeType
 
 @ExperimentalMetadataApi
-public fun <AM : AuthMetadata> RawAuthMetadata.read(reader: AuthMetadataReader<AM>, pool: ObjectPool<ChunkBuffer> = ChunkBuffer.Pool): AM {
+public fun <AM : AuthMetadata> RawAuthMetadata.read(
+    reader: AuthMetadataReader<AM>,
+    pool: ObjectPool<ChunkBuffer> = ChunkBuffer.Pool
+): AM {
     return readOrNull(reader, pool) ?: run {
         content.close()
         error("Expected auth type '${reader.mimeType}' but was '$mimeType'")
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/payload/PayloadBuilder.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/payload/PayloadBuilder.kt
@@ -38,7 +38,9 @@ public inline fun PayloadBuilder.data(block: BytePacketBuilder.() -> Unit): Unit
 public fun PayloadBuilder.data(value: String): Unit = data { writeText(value) }
 public fun PayloadBuilder.data(value: ByteArray): Unit = data { writeFully(value) }
 
-public inline fun PayloadBuilder.metadata(block: BytePacketBuilder.() -> Unit): Unit = metadata(buildPacket(block = block))
+public inline fun PayloadBuilder.metadata(block: BytePacketBuilder.() -> Unit): Unit =
+    metadata(buildPacket(block = block))
+
 public fun PayloadBuilder.metadata(value: String): Unit = metadata { writeText(value) }
 public fun PayloadBuilder.metadata(value: ByteArray): Unit = metadata { writeFully(value) }
 
diff --git a/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/transport/ClientTransport.kt b/rsocket-core/src/commonMain/kotlin/io/rsocket/kotlin/transport/ClientTransport.kt
@@ -28,9 +28,10 @@ public fun interface ClientTransport : CoroutineScope {
 }
 
 @TransportApi
-public fun ClientTransport(coroutineContext: CoroutineContext, transport: ClientTransport): ClientTransport = object : ClientTransport {
-    override val coroutineContext: CoroutineContext get() = coroutineContext
+public fun ClientTransport(coroutineContext: CoroutineContext, transport: ClientTransport): ClientTransport =
+    object : ClientTransport {
+        override val coroutineContext: CoroutineContext get() = coroutineContext
 
-    @TransportApi
-    override suspend fun connect(): Connection = transport.connect()
-}
+        @TransportApi
+        override suspend fun connect(): Connection = transport.connect()
+    }
diff --git a/rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/core/RSocketTest.kt b/rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/core/RSocketTest.kt
diff --git a/rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/core/ReconnectableRSocketTest.kt b/rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/core/ReconnectableRSocketTest.kt
diff --git a/rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/internal/RSocketResponderRequestNTest.kt b/rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/internal/RSocketResponderRequestNTest.kt
diff --git a/rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/internal/handler/RequesterRequestResponseFrameHandlerTest.kt b/rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/internal/handler/RequesterRequestResponseFrameHandlerTest.kt
diff --git a/rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/metadata/Util.kt b/rsocket-core/src/commonTest/kotlin/io/rsocket/kotlin/metadata/Util.kt
diff --git a/rsocket-core/src/jsMain/kotlin/io/rsocket/kotlin/logging/DefaultLoggerFactory.kt b/rsocket-core/src/jsMain/kotlin/io/rsocket/kotlin/logging/DefaultLoggerFactory.kt
diff --git a/rsocket-transport-local/src/commonMain/kotlin/io/rsocket/kotlin/transport/local/LocalServer.kt b/rsocket-transport-local/src/commonMain/kotlin/io/rsocket/kotlin/transport/local/LocalServer.kt

Original file line number	Diff line number	Diff line change
`@@ -81,6 +81,7 @@ class RSocketKotlinBenchmark : RSocketBenchmark<Payload>() {`
`81`	`81`
`82`	`82`	`override suspend fun doRequestStream(): Flow<Payload> = client.requestStream(payloadCopy()).flowOn(requestStrategy)`
`83`	`83`
`84`		`- override suspend fun doRequestChannel(): Flow<Payload> = client.requestChannel(payloadCopy(), payloadsFlow).flowOn(requestStrategy)`
	`84`	`+ override suspend fun doRequestChannel(): Flow<Payload> =`
	`85`	`+ client.requestChannel(payloadCopy(), payloadsFlow).flowOn(requestStrategy)`
`85`	`86`
`86`	`87`	`}`
Original file line number	Diff line number	Diff line change
`@@ -35,7 +35,8 @@ public interface Connection : CoroutineScope {`
`35`	`35`	`}`
`36`	`36`
`37`	`37`	`@OptIn(TransportApi::class)`
`38`		`-internal suspend inline fun <T> Connection.receiveFrame(block: (frame: Frame) -> T): T = receive().readFrame(pool).closeOnError(block)`
	`38`	`+internal suspend inline fun <T> Connection.receiveFrame(block: (frame: Frame) -> T): T =`
	`39`	`+ receive().readFrame(pool).closeOnError(block)`
`39`	`40`
`40`	`41`	`@OptIn(TransportApi::class)`
`41`	`42`	`internal suspend fun Connection.sendFrame(frame: Frame) {`
Original file line number	Diff line number	Diff line change
`@@ -54,5 +54,6 @@ internal class Interceptors(`
`54`	`54`	`fun wrapRequester(requester: RSocket): RSocket = requesters.fold(requester) { r, i -> i.intercept(r) }`
`55`	`55`	`fun wrapResponder(responder: RSocket): RSocket = responders.fold(responder) { r, i -> i.intercept(r) }`
`56`	`56`	`fun wrapConnection(connection: Connection): Connection = connections.fold(connection) { c, i -> i.intercept(c) }`
`57`		`- fun wrapAcceptor(connection: ConnectionAcceptor): ConnectionAcceptor = acceptors.fold(connection) { c, i -> i.intercept(c) }`
	`57`	`+ fun wrapAcceptor(connection: ConnectionAcceptor): ConnectionAcceptor =`
	`58`	`+ acceptors.fold(connection) { c, i -> i.intercept(c) }`
`58`	`59`	`}`
Original file line number	Diff line number	Diff line change
`@@ -45,4 +45,5 @@ internal class MetadataPushFrame(`
`45`	`45`	`}`
`46`	`46`	`}`
`47`	`47`
`48`		`-internal fun ByteReadPacket.readMetadataPush(pool: ObjectPool<ChunkBuffer>): MetadataPushFrame = MetadataPushFrame(readPacket(pool))`
	`48`	`+internal fun ByteReadPacket.readMetadataPush(pool: ObjectPool<ChunkBuffer>): MetadataPushFrame =`
	`49`	`+ MetadataPushFrame(readPacket(pool))`