kRPC: Backpressure

build.gradle.kts

@@ -8,7 +8,7 @@ plugins {
     alias(libs.plugins.serialization) apply false
     alias(libs.plugins.kotlinx.rpc) apply false
     alias(libs.plugins.atomicfu) apply false
-    alias(libs.plugins.conventions.kover)
+//    alias(libs.plugins.conventions.kover)
     alias(libs.plugins.conventions.root)
 }
 

detekt/config.yaml

@@ -219,7 +219,7 @@ empty-blocks:
   EmptyForBlock:
     active: true
   EmptyFunctionBlock:
-    active: true
+    active: false
     ignoreOverridden: false
   EmptyIfBlock:
     active: true
@@ -288,7 +288,7 @@ exceptions:
   ThrowingNewInstanceOfSameException:
     active: true
   TooGenericExceptionCaught:
-    active: true
+    active: false
     excludes: ['**/test/**', '**/androidTest/**', '**/commonTest/**', '**/jvmTest/**', '**/androidUnitTest/**', '**/androidInstrumentedTest/**', '**/jsTest/**', '**/iosTest/**']
     exceptionNames:
       - 'ArrayIndexOutOfBoundsException'

docs/pages/kotlinx-rpc/topics/platforms.topic

@@ -73,47 +73,47 @@
 <td>core</td>
 <td>jvm</td>
 <td><list><li>browser</li><li>node</li></list></td>
-<td><list><li>wasmJs<list><li>browser</li><li>d8</li><li>node</li></list></li><li>wasmWasi<list><li>node</li></list></li></list></td>
+<td><list><li>wasmJs<list><li>browser</li><li>node</li></list></li><li>wasmWasi<list><li>node</li></list></li></list></td>
 <td><list><li>apple<list><li>ios<list><li>iosArm64</li><li>iosSimulatorArm64</li><li>iosX64</li></list></li><li>macos<list><li>macosArm64</li><li>macosX64</li></list></li><li>watchos<list><li>watchosArm32</li><li>watchosArm64</li><li>watchosDeviceArm64</li><li>watchosSimulatorArm64</li><li>watchosX64</li></list></li><li>tvos<list><li>tvosArm64</li><li>tvosSimulatorArm64</li><li>tvosX64</li></list></li></list></li><li>linux<list><li>linuxArm64</li><li>linuxX64</li></list></li><li>windows<list><li>mingwX64</li></list></li></list></td>
 </tr>
 
 <tr>
 <td>utils</td>
 <td>jvm</td>
 <td><list><li>browser</li><li>node</li></list></td>
-<td><list><li>wasmJs<list><li>browser</li><li>d8</li><li>node</li></list></li><li>wasmWasi<list><li>node</li></list></li></list></td>
+<td><list><li>wasmJs<list><li>browser</li><li>node</li></list></li><li>wasmWasi<list><li>node</li></list></li></list></td>
 <td><list><li>apple<list><li>ios<list><li>iosArm64</li><li>iosSimulatorArm64</li><li>iosX64</li></list></li><li>macos<list><li>macosArm64</li><li>macosX64</li></list></li><li>watchos<list><li>watchosArm32</li><li>watchosArm64</li><li>watchosDeviceArm64</li><li>watchosSimulatorArm64</li><li>watchosX64</li></list></li><li>tvos<list><li>tvosArm64</li><li>tvosSimulatorArm64</li><li>tvosX64</li></list></li></list></li><li>linux<list><li>linuxArm64</li><li>linuxX64</li></list></li><li>windows<list><li>mingwX64</li></list></li></list></td>
 </tr>
 
 <tr>
 <td>krpc-client</td>
 <td>jvm</td>
 <td><list><li>browser</li><li>node</li></list></td>
-<td><list><li>wasmJs<list><li>browser</li><li>d8</li><li>node</li></list></li></list></td>
+<td><list><li>wasmJs<list><li>browser</li><li>node</li></list></li></list></td>
 <td><list><li>apple<list><li>ios<list><li>iosArm64</li><li>iosSimulatorArm64</li><li>iosX64</li></list></li><li>macos<list><li>macosArm64</li><li>macosX64</li></list></li><li>watchos<list><li>watchosArm64</li><li>watchosSimulatorArm64</li><li>watchosX64</li></list></li><li>tvos<list><li>tvosArm64</li><li>tvosSimulatorArm64</li><li>tvosX64</li></list></li></list></li><li>linux<list><li>linuxArm64</li><li>linuxX64</li></list></li><li>windows<list><li>mingwX64</li></list></li></list></td>
 </tr>
 
 <tr>
 <td>krpc-core</td>
 <td>jvm</td>
 <td><list><li>browser</li><li>node</li></list></td>
-<td><list><li>wasmJs<list><li>browser</li><li>d8</li><li>node</li></list></li></list></td>
+<td><list><li>wasmJs<list><li>browser</li><li>node</li></list></li></list></td>
 <td><list><li>apple<list><li>ios<list><li>iosArm64</li><li>iosSimulatorArm64</li><li>iosX64</li></list></li><li>macos<list><li>macosArm64</li><li>macosX64</li></list></li><li>watchos<list><li>watchosArm64</li><li>watchosSimulatorArm64</li><li>watchosX64</li></list></li><li>tvos<list><li>tvosArm64</li><li>tvosSimulatorArm64</li><li>tvosX64</li></list></li></list></li><li>linux<list><li>linuxArm64</li><li>linuxX64</li></list></li><li>windows<list><li>mingwX64</li></list></li></list></td>
 </tr>
 
 <tr>
 <td>krpc-logging</td>
 <td>jvm</td>
 <td><list><li>browser</li><li>node</li></list></td>
-<td><list><li>wasmJs<list><li>browser</li><li>d8</li><li>node</li></list></li></list></td>
+<td><list><li>wasmJs<list><li>browser</li><li>node</li></list></li></list></td>
 <td><list><li>apple<list><li>ios<list><li>iosArm64</li><li>iosSimulatorArm64</li><li>iosX64</li></list></li><li>macos<list><li>macosArm64</li><li>macosX64</li></list></li><li>watchos<list><li>watchosArm64</li><li>watchosSimulatorArm64</li><li>watchosX64</li></list></li><li>tvos<list><li>tvosArm64</li><li>tvosSimulatorArm64</li><li>tvosX64</li></list></li></list></li><li>linux<list><li>linuxArm64</li><li>linuxX64</li></list></li><li>windows<list><li>mingwX64</li></list></li></list></td>
 </tr>
 
 <tr>
 <td>krpc-server</td>
 <td>jvm</td>
 <td><list><li>browser</li><li>node</li></list></td>
-<td><list><li>wasmJs<list><li>browser</li><li>d8</li><li>node</li></list></li></list></td>
+<td><list><li>wasmJs<list><li>browser</li><li>node</li></list></li></list></td>
 <td><list><li>apple<list><li>ios<list><li>iosArm64</li><li>iosSimulatorArm64</li><li>iosX64</li></list></li><li>macos<list><li>macosArm64</li><li>macosX64</li></list></li><li>watchos<list><li>watchosArm64</li><li>watchosSimulatorArm64</li><li>watchosX64</li></list></li><li>tvos<list><li>tvosArm64</li><li>tvosSimulatorArm64</li><li>tvosX64</li></list></li></list></li><li>linux<list><li>linuxArm64</li><li>linuxX64</li></list></li><li>windows<list><li>mingwX64</li></list></li></list></td>
 </tr>
 
@@ -129,55 +129,55 @@
 <td>krpc-ktor-client</td>
 <td>jvm</td>
 <td><list><li>browser</li><li>node</li></list></td>
-<td><list><li>wasmJs<list><li>browser</li><li>d8</li><li>node</li></list></li></list></td>
+<td><list><li>wasmJs<list><li>browser</li><li>node</li></list></li></list></td>
 <td><list><li>apple<list><li>ios<list><li>iosArm64</li><li>iosSimulatorArm64</li><li>iosX64</li></list></li><li>macos<list><li>macosArm64</li><li>macosX64</li></list></li><li>watchos<list><li>watchosArm64</li><li>watchosSimulatorArm64</li><li>watchosX64</li></list></li><li>tvos<list><li>tvosArm64</li><li>tvosSimulatorArm64</li><li>tvosX64</li></list></li></list></li><li>linux<list><li>linuxArm64</li><li>linuxX64</li></list></li><li>windows<list><li>mingwX64</li></list></li></list></td>
 </tr>
 
 <tr>
 <td>krpc-ktor-core</td>
 <td>jvm</td>
 <td><list><li>browser</li><li>node</li></list></td>
-<td><list><li>wasmJs<list><li>browser</li><li>d8</li><li>node</li></list></li></list></td>
+<td><list><li>wasmJs<list><li>browser</li><li>node</li></list></li></list></td>
 <td><list><li>apple<list><li>ios<list><li>iosArm64</li><li>iosSimulatorArm64</li><li>iosX64</li></list></li><li>macos<list><li>macosArm64</li><li>macosX64</li></list></li><li>watchos<list><li>watchosArm64</li><li>watchosSimulatorArm64</li><li>watchosX64</li></list></li><li>tvos<list><li>tvosArm64</li><li>tvosSimulatorArm64</li><li>tvosX64</li></list></li></list></li><li>linux<list><li>linuxArm64</li><li>linuxX64</li></list></li><li>windows<list><li>mingwX64</li></list></li></list></td>
 </tr>
 
 <tr>
 <td>krpc-ktor-server</td>
 <td>jvm</td>
 <td><list><li>browser</li><li>node</li></list></td>
-<td><list><li>wasmJs<list><li>browser</li><li>d8</li><li>node</li></list></li></list></td>
+<td><list><li>wasmJs<list><li>browser</li><li>node</li></list></li></list></td>
 <td><list><li>apple<list><li>ios<list><li>iosArm64</li><li>iosSimulatorArm64</li><li>iosX64</li></list></li><li>macos<list><li>macosArm64</li><li>macosX64</li></list></li><li>watchos<list><li>watchosArm64</li><li>watchosSimulatorArm64</li><li>watchosX64</li></list></li><li>tvos<list><li>tvosArm64</li><li>tvosSimulatorArm64</li><li>tvosX64</li></list></li></list></li><li>linux<list><li>linuxArm64</li><li>linuxX64</li></list></li><li>windows<list><li>mingwX64</li></list></li></list></td>
 </tr>
 
 <tr>
 <td>krpc-serialization-cbor</td>
 <td>jvm</td>
 <td><list><li>browser</li><li>node</li></list></td>
-<td><list><li>wasmJs<list><li>browser</li><li>d8</li><li>node</li></list></li></list></td>
+<td><list><li>wasmJs<list><li>browser</li><li>node</li></list></li></list></td>
 <td><list><li>apple<list><li>ios<list><li>iosArm64</li><li>iosSimulatorArm64</li><li>iosX64</li></list></li><li>macos<list><li>macosArm64</li><li>macosX64</li></list></li><li>watchos<list><li>watchosArm64</li><li>watchosSimulatorArm64</li><li>watchosX64</li></list></li><li>tvos<list><li>tvosArm64</li><li>tvosSimulatorArm64</li><li>tvosX64</li></list></li></list></li><li>linux<list><li>linuxArm64</li><li>linuxX64</li></list></li><li>windows<list><li>mingwX64</li></list></li></list></td>
 </tr>
 
 <tr>
 <td>krpc-serialization-core</td>
 <td>jvm</td>
 <td><list><li>browser</li><li>node</li></list></td>
-<td><list><li>wasmJs<list><li>browser</li><li>d8</li><li>node</li></list></li></list></td>
+<td><list><li>wasmJs<list><li>browser</li><li>node</li></list></li></list></td>
 <td><list><li>apple<list><li>ios<list><li>iosArm64</li><li>iosSimulatorArm64</li><li>iosX64</li></list></li><li>macos<list><li>macosArm64</li><li>macosX64</li></list></li><li>watchos<list><li>watchosArm64</li><li>watchosSimulatorArm64</li><li>watchosX64</li></list></li><li>tvos<list><li>tvosArm64</li><li>tvosSimulatorArm64</li><li>tvosX64</li></list></li></list></li><li>linux<list><li>linuxArm64</li><li>linuxX64</li></list></li><li>windows<list><li>mingwX64</li></list></li></list></td>
 </tr>
 
 <tr>
 <td>krpc-serialization-json</td>
 <td>jvm</td>
 <td><list><li>browser</li><li>node</li></list></td>
-<td><list><li>wasmJs<list><li>browser</li><li>d8</li><li>node</li></list></li></list></td>
+<td><list><li>wasmJs<list><li>browser</li><li>node</li></list></li></list></td>
 <td><list><li>apple<list><li>ios<list><li>iosArm64</li><li>iosSimulatorArm64</li><li>iosX64</li></list></li><li>macos<list><li>macosArm64</li><li>macosX64</li></list></li><li>watchos<list><li>watchosArm64</li><li>watchosSimulatorArm64</li><li>watchosX64</li></list></li><li>tvos<list><li>tvosArm64</li><li>tvosSimulatorArm64</li><li>tvosX64</li></list></li></list></li><li>linux<list><li>linuxArm64</li><li>linuxX64</li></list></li><li>windows<list><li>mingwX64</li></list></li></list></td>
 </tr>
 
 <tr>
 <td>krpc-serialization-protobuf</td>
 <td>jvm</td>
 <td><list><li>browser</li><li>node</li></list></td>
-<td><list><li>wasmJs<list><li>browser</li><li>d8</li><li>node</li></list></li></list></td>
+<td><list><li>wasmJs<list><li>browser</li><li>node</li></list></li></list></td>
 <td><list><li>apple<list><li>ios<list><li>iosArm64</li><li>iosSimulatorArm64</li><li>iosX64</li></list></li><li>macos<list><li>macosArm64</li><li>macosX64</li></list></li><li>watchos<list><li>watchosArm64</li><li>watchosSimulatorArm64</li><li>watchosX64</li></list></li><li>tvos<list><li>tvosArm64</li><li>tvosSimulatorArm64</li><li>tvosX64</li></list></li></list></li><li>linux<list><li>linuxArm64</li><li>linuxX64</li></list></li><li>windows<list><li>mingwX64</li></list></li></list></td>
 </tr>
 

gradle-conventions/src/main/kotlin/conventions-common.gradle.kts

@@ -52,10 +52,10 @@ afterEvaluate {
     }
 }
 
-apply(plugin = "org.jetbrains.kotlinx.kover")
+//apply(plugin = "org.jetbrains.kotlinx.kover")
 
 val thisProject = project
 
-rootProject.configurations.matching { it.name == "kover" }.all {
-    rootProject.dependencies.add("kover", thisProject)
-}
+//rootProject.configurations.matching { it.name == "kover" }.all {
+//    rootProject.dependencies.add("kover", thisProject)
+//}

gradle-conventions/src/main/kotlin/conventions-kover.gradle.kts

@@ -7,31 +7,31 @@ import kotlinx.kover.gradle.plugin.dsl.CoverageUnit
 import kotlinx.kover.gradle.plugin.dsl.GroupingEntityType
 
 plugins {
-    id("org.jetbrains.kotlinx.kover")
+//    id("org.jetbrains.kotlinx.kover")
 }
 
-kover {
-    reports {
-        total {
-            html {
-                onCheck.set(false)
-                charset.set("UTF-8")
-                htmlDir.set(rootDir.resolve("kover"))
-            }
-
-            verify {
-                onCheck.set(false)
-
-                rule {
-                    groupBy.set(GroupingEntityType.APPLICATION)
-
-                    bound {
-                        coverageUnits.set(CoverageUnit.LINE)
-                        aggregationForGroup.set(AggregationType.COVERED_PERCENTAGE)
-                        minValue.set(70)
-                    }
-                }
-            }
-        }
-    }
-}
+//kover {
+//    reports {
+//        total {
+//            html {
+//                onCheck.set(false)
+//                charset.set("UTF-8")
+//                htmlDir.set(rootDir.resolve("kover"))
+//            }
+//
+//            verify {
+//                onCheck.set(false)
+//
+//                rule {
+//                    groupBy.set(GroupingEntityType.APPLICATION)
+//
+//                    bound {
+//                        coverageUnits.set(CoverageUnit.LINE)
+//                        aggregationForGroup.set(AggregationType.COVERED_PERCENTAGE)
+//                        minValue.set(70)
+//                    }
+//                }
+//            }
+//        }
+//    }
+//}

gradle-conventions/src/main/kotlin/util/apiValidation.kt

@@ -11,6 +11,7 @@ import org.jetbrains.kotlin.gradle.dsl.abi.ExperimentalAbiValidation
 private val excludedProjects = setOf(
     "krpc-test",
     "krpc-compatibility-tests",
+    "krpc-protocol-compatibility-tests",
     "compiler-plugin-tests",
 )
 

gradle-conventions/src/main/kotlin/util/targets/wasm.kt

@@ -27,7 +27,8 @@ fun KmpConfig.configureWasm() {
                 browser()
                 nodejs()
                 if (wasmJsD8) {
-                    d8()
+                    // this platform needs some care KRPC-210
+//                    d8()
                 }
 
                 binaries.library()

krpc/krpc-client/src/commonMain/kotlin/kotlinx/rpc/krpc/client/KrpcClient.kt

@@ -5,12 +5,27 @@
 package kotlinx.rpc.krpc.client
 
 import kotlinx.atomicfu.atomic
-import kotlinx.coroutines.*
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineName
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.DelicateCoroutinesApi
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.InternalCoroutinesApi
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.cancelAndJoin
 import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.currentCoroutineContext
+import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.FlowCollector
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.job
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.supervisorScope
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.rpc.RpcCall
@@ -21,7 +36,8 @@ import kotlinx.rpc.descriptor.RpcInvokator
 import kotlinx.rpc.internal.utils.InternalRpcApi
 import kotlinx.rpc.internal.utils.getOrNull
 import kotlinx.rpc.internal.utils.map.RpcInternalConcurrentHashMap
-import kotlinx.rpc.krpc.*
+import kotlinx.rpc.krpc.KrpcConfig
+import kotlinx.rpc.krpc.KrpcTransport
 import kotlinx.rpc.krpc.client.internal.ClientStreamContext
 import kotlinx.rpc.krpc.client.internal.ClientStreamSerializer
 import kotlinx.rpc.krpc.client.internal.KrpcClientConnector
@@ -35,7 +51,6 @@ import kotlinx.serialization.StringFormat
 import kotlinx.serialization.modules.SerializersModule
 import kotlin.collections.first
 import kotlin.concurrent.Volatile
-import kotlin.coroutines.cancellation.CancellationException
 import kotlin.properties.Delegates
 
 /**
@@ -157,7 +172,7 @@ public abstract class KrpcClient : RpcClient, KrpcEndpoint {
     private val connector by lazy {
         checkTransportReadiness()
 
-        KrpcClientConnector(config.serialFormatInitializer.build(), transport, config.waitForServices)
+        KrpcClientConnector(config.serialFormatInitializer.build(), transport, config.connector)
     }
 
     private var connectionId: Long? = null
@@ -221,9 +236,11 @@ public abstract class KrpcClient : RpcClient, KrpcEndpoint {
                             message.errorMessage
                 }
 
-                serverSupportedPlugins.completeExceptionally(
-                    IllegalStateException("Server failed to process protocol message: ${message.failedMessage}")
-                )
+                if (!serverSupportedPlugins.isCompleted) {
+                    serverSupportedPlugins.completeExceptionally(
+                        IllegalStateException("Server failed to process protocol message: ${message.failedMessage}")
+                    )
+                }
             }
         }
     }
@@ -295,15 +312,15 @@ public abstract class KrpcClient : RpcClient, KrpcEndpoint {
 
                     sendCancellation(CancellationType.REQUEST, call.descriptor.fqName, callId)
 
-                    connector.unsubscribeFromMessages(call.descriptor.fqName, callId) {
-                        cancellingRequests.remove(callId)
-                    }
+                    connector.unsubscribeFromMessages(call.descriptor.fqName, callId)
+                    cancellingRequests.remove(callId)
                 }
 
                 throw e
             } finally {
                 channel.close()
                 requestChannels.remove(callId)
+                connector.unsubscribeFromMessages(call.descriptor.fqName, callId)
             }
         }
     }
@@ -341,17 +358,8 @@ public abstract class KrpcClient : RpcClient, KrpcEndpoint {
             }
 
             is KrpcCallMessage.CallException -> {
-                val cause = runCatching {
-                    message.cause.deserialize()
-                }
-
-                val result = if (cause.isFailure) {
-                    cause.exceptionOrNull()!!
-                } else {
-                    cause.getOrNull()!!
-                }
-
-                channel.close(result)
+                val cause = message.cause.deserialize()
+                channel.close(cause)
             }
 
             is KrpcCallMessage.CallSuccess, is KrpcCallMessage.StreamMessage -> {
@@ -456,8 +464,23 @@ public abstract class KrpcClient : RpcClient, KrpcEndpoint {
                 serviceTypeString = serviceTypeString,
             )
         } catch (e: CancellationException) {
+            currentCoroutineContext().ensureActive()
+
+            val wrapped = ManualCancellationException(e)
+            val serializedReason = serializeException(wrapped)
+            val message = KrpcCallMessage.StreamCancel(
+                callId = outgoingStream.callId,
+                serviceType = serviceTypeString,
+                streamId = outgoingStream.streamId,
+                cause = serializedReason,
+                connectionId = outgoingStream.connectionId,
+                serviceId = outgoingStream.serviceId,
+            )
+            sender.sendMessage(message)
+
+            // stop the flow and its coroutine, other flows are not affected
             throw e
-        } catch (@Suppress("detekt.TooGenericExceptionCaught") cause: Throwable) {
+        } catch (cause: Throwable) {
             val serializedReason = serializeException(cause)
             val message = KrpcCallMessage.StreamCancel(
                 callId = outgoingStream.callId,