Move CoroutineScope from RpcClient/RpcServer to internal implementation details

Author: Mr3zee

core/src/commonMain/kotlin/kotlinx/rpc/RpcClient.kt

@@ -5,9 +5,7 @@
 package kotlinx.rpc
 
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Deferred
 import kotlinx.coroutines.flow.Flow
-import kotlin.coroutines.CoroutineContext
 
 @Deprecated("Use RpcClient instead", ReplaceWith("RpcClient"), level = DeprecationLevel.ERROR)
 public typealias RPCClient = RpcClient
@@ -17,7 +15,7 @@ public typealias RPCClient = RpcClient
  * transform them, send to the server and handle responses and errors.
  * [CoroutineScope] defines the lifetime of the client.
  */
-public interface RpcClient : CoroutineScope {
+public interface RpcClient {
     /**
      * This method is used by generated clients to perform a call to the server.
      *

core/src/commonMain/kotlin/kotlinx/rpc/RpcServer.kt

@@ -4,7 +4,6 @@
 
 package kotlinx.rpc
 
-import kotlinx.coroutines.CoroutineScope
 import kotlinx.rpc.annotations.Rpc
 import kotlin.reflect.KClass
 
@@ -15,7 +14,7 @@ public typealias RPCServer = RpcServer
  * RpcServer is used to accept RPC messages, route them to a specific service, and process given responses.
  * Server may contain multiple services.
  */
-public interface RpcServer : CoroutineScope {
+public interface RpcServer {
     /**
      * Registers new service to the server. Server will route all designated messages to it.
      * Service of any type should be unique on the server, but RpcServer doesn't specify the actual retention policy.

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

@@ -61,7 +61,8 @@ public abstract class KrpcClient(
     private val config: KrpcConfig.Client,
     transport: KrpcTransport,
 ) : RpcClient, KrpcEndpoint {
-    final override val coroutineContext: CoroutineContext = SupervisorJob(transport.coroutineContext.job)
+    @InternalRpcApi
+    public val internalScope: CoroutineScope = CoroutineScope(SupervisorJob(transport.coroutineContext.job))
 
     // we make a child here, so we can send cancellation messages before closing the connection
     private val connector by lazy {
@@ -93,7 +94,7 @@ public abstract class KrpcClient(
     private var clientCancelled = false
 
     init {
-        coroutineContext.job.invokeOnCompletion(onCancelling = true) {
+        internalScope.coroutineContext.job.invokeOnCompletion(onCancelling = true) {
             clientCancelled = true
 
             sendCancellation(CancellationType.ENDPOINT, null, null, closeTransportAfterSending = true)
@@ -102,12 +103,12 @@ public abstract class KrpcClient(
             requestChannels.clear()
         }
 
-        launch(CoroutineName("krpc-client-generic-messages")) {
+        internalScope.launch(CoroutineName("krpc-client-generic-messages")) {
             connector.subscribeToGenericMessages(::handleGenericMessage)
         }
     }
 
-    private val initHandshake: Job = launch(CoroutineName("krpc-client-handshake")) {
+    private val initHandshake: Job = internalScope.launch(CoroutineName("krpc-client-handshake")) {
         connector.sendMessage(KrpcProtocolMessage.Handshake(KrpcPlugin.ALL))
 
         connector.subscribeToProtocolMessages(::handleProtocolMessage)
@@ -284,7 +285,7 @@ public abstract class KrpcClient(
     final override suspend fun handleCancellation(message: KrpcGenericMessage) {
         when (val type = message.cancellationType()) {
             CancellationType.ENDPOINT -> {
-                cancel("Closing client after server cancellation") // we cancel this client
+                internalScope.cancel("Closing client after server cancellation") // we cancel this client
             }
 
             else -> {

krpc/krpc-ktor/krpc-ktor-server/src/commonMain/kotlin/kotlinx/rpc/krpc/ktor/server/KtorServerDsl.kt

@@ -54,6 +54,6 @@ private fun Route.createRpcServer(rpcRouteBuilder: KrpcRoute.() -> Unit) {
             registration(server)
         }
 
-        server.coroutineContext.job.join()
+        server.internalScope.coroutineContext.job.join()
     }
 }

krpc/krpc-server/src/commonMain/kotlin/kotlinx/rpc/krpc/server/KrpcServer.kt

@@ -47,8 +47,8 @@ public abstract class KrpcServer(
     private val config: KrpcConfig.Server,
     transport: KrpcTransport,
 ) : RpcServer, KrpcEndpoint {
-    // we make a child here, so we can send cancellation messages before closing the connection
-    final override val coroutineContext: CoroutineContext = SupervisorJob(transport.coroutineContext.job)
+    @InternalRpcApi
+    public val internalScope: CoroutineScope = CoroutineScope(SupervisorJob(transport.coroutineContext.job))
 
     private val logger = RpcInternalCommonLogger.logger(rpcInternalObjectId())
 
@@ -73,13 +73,13 @@ public abstract class KrpcServer(
     private var cancelledByClient = false
 
     init {
-        coroutineContext.job.invokeOnCompletion(onCancelling = true) {
+        internalScope.coroutineContext.job.invokeOnCompletion(onCancelling = true) {
             if (!cancelledByClient) {
                 sendCancellation(CancellationType.ENDPOINT, null, null, closeTransportAfterSending = true)
             }
         }
 
-        launch(CoroutineName("krpc-server-generic-protocol-messages")) {
+        internalScope.launch(CoroutineName("krpc-server-generic-protocol-messages")) {
             connector.subscribeToProtocolMessages(::handleProtocolMessage)
 
             connector.subscribeToGenericMessages(::handleGenericMessage)
@@ -108,7 +108,7 @@ public abstract class KrpcServer(
     ) {
         val descriptor = serviceDescriptorOf(serviceKClass)
 
-        launch(CoroutineName("krpc-server-service-$descriptor")) {
+        internalScope.launch(CoroutineName("krpc-server-service-$descriptor")) {
             connector.subscribeToServiceMessages(descriptor.fqName) { message ->
                 val rpcServerService = rpcServices.computeIfAbsent(descriptor.fqName) {
                     createNewServiceInstance(descriptor, serviceFactory)
@@ -133,7 +133,7 @@ public abstract class KrpcServer(
             config = config,
             connector = connector,
             supportedPlugins = supportedPlugins,
-            serverScope = this,
+            serverScope = internalScope,
         )
     }
 
@@ -143,7 +143,7 @@ public abstract class KrpcServer(
             CancellationType.ENDPOINT -> {
                 cancelledByClient = true
 
-                cancel("Server cancelled by client")
+                internalScope.cancel("Server cancelled by client")
                 rpcServices.clear()
             }
 

krpc/krpc-test/src/commonTest/kotlin/kotlinx/rpc/krpc/test/ProtocolTestBase.kt

@@ -23,7 +23,6 @@ import kotlinx.rpc.krpc.server.KrpcServer
 import kotlinx.rpc.registerService
 import kotlinx.rpc.withService
 import kotlinx.serialization.BinaryFormat
-import kotlin.coroutines.CoroutineContext
 
 abstract class ProtocolTestBase {
     protected fun runTest(

krpc/krpc-test/src/commonTest/kotlin/kotlinx/rpc/krpc/test/cancellation/CancellationTest.kt

@@ -76,7 +76,7 @@ class CancellationTest {
         }
 
         unskippableDelay(150) // wait for requests to reach server
-        client.cancel()
+        client.internalScope.cancel()
         firstRequestJob.join()
         secondRequestJob.join()
 
@@ -85,8 +85,8 @@ class CancellationTest {
 
         assertEquals(0, serverInstances.sumOf { it.delayCounter.value }, "Expected no requests to succeed")
 
-        client.join()
-        server.join()
+        client.internalScope.join()
+        server.internalScope.join()
 
         checkAlive(clientAlive = false, serverAlive = false)
         stopAllAndJoin()
@@ -105,7 +105,7 @@ class CancellationTest {
         }
 
         unskippableDelay(150) // wait for requests to reach server
-        server.cancel()
+        server.internalScope.cancel()
         firstRequestJob.join()
         secondRequestJob.join()
 
@@ -114,8 +114,8 @@ class CancellationTest {
 
         assertEquals(0, serverInstances.sumOf { it.delayCounter.value }, "Expected no requests to succeed")
 
-        client.join()
-        server.join()
+        client.internalScope.join()
+        server.internalScope.join()
 
         checkAlive(clientAlive = false, serverAlive = false)
         stopAllAndJoin()
@@ -220,8 +220,8 @@ class CancellationTest {
 
         serverInstance().firstIncomingConsumed.await()
 
-        client.cancel("Test request cancelled")
-        client.join()
+        client.internalScope.cancel("Test request cancelled")
+        client.internalScope.join()
 
         serverInstance().consumedAll.await()
 
@@ -239,7 +239,7 @@ class CancellationTest {
             caught = it
         }.collect {
             if (it == 0) {
-                client.cancel()
+                client.internalScope.cancel()
             } else {
                 fail("Expected the request to fail with cancellation of the client")
             }
@@ -254,7 +254,7 @@ class CancellationTest {
     fun testCancelledClientCancelsRequest() = runCancellationTest {
         launch {
             serverInstance().firstIncomingConsumed.await()
-            client.cancel("Cancelled by test")
+            client.internalScope.cancel("Cancelled by test")
         }
 
         try {
@@ -330,8 +330,8 @@ class CancellationTest {
         serverAlive: Boolean = true,
         transportAlive: Boolean = true,
     ) {
-        checkAlive(clientAlive, client, "client")
-        checkAlive(serverAlive, server, "server")
+        checkAlive(clientAlive, client.internalScope, "client")
+        checkAlive(serverAlive, server.internalScope, "server")
         checkAlive(transportAlive, transport, "transport")
     }
 

krpc/krpc-test/src/jvmTest/kotlin/kotlinx/rpc/krpc/test/api/ApiVersioningTest.kt

@@ -89,8 +89,6 @@ class ApiVersioningTest {
         }
     }
 
-    private fun List<List<*>>.join() = joinToString { "[${it.joinToString()}]" }
-
     companion object {
         val LIBRARY_VERSION_DIR = System.getenv("LIBRARY_VERSION")?.versionToDirName()
             ?: error("Expected LIBRARY_VERSION env variable")