Various small fixes

Author: Mr3zee

docs/pages/kotlinx-rpc/topics/0-8-0.topic

@@ -262,12 +262,9 @@
                         class MyClient(
                             config: KrpcConfig,
                             transport: KrpcTransport,
-                        ) : InitializedKrpcClient(transport, config)
+                        ) : InitializedKrpcClient(config, transport)
                     </code-block>
                 </compare>
-                <note>
-                    Notice that the parameter order is reversed in new <code>InitializedKrpcClient</code>.
-                </note>
             </li>
         </list>
     </chapter>

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

@@ -45,8 +45,8 @@ import kotlin.properties.Delegates
  * See [KrpcClient.initializeTransport].
  */
 public abstract class InitializedKrpcClient(
-    private val transport: KrpcTransport,
     private val config: KrpcConfig.Client,
+    private val transport: KrpcTransport,
 ): KrpcClient() {
     final override suspend fun initializeTransport(): KrpcTransport {
         return transport
@@ -80,6 +80,20 @@ public abstract class KrpcClient : RpcClient, KrpcEndpoint {
      */
     protected abstract fun initializeConfig(): KrpcConfig.Client
 
+    /**
+     * Close this client, removing all the services and stopping accepting messages.
+     */
+    public fun close(message: String? = null) {
+        internalScope.cancel(message ?: "Client closed")
+    }
+
+    /**
+     * Waits until the client is closed.
+     */
+    public suspend fun awaitCompletion() {
+        internalScope.coroutineContext.job.join()
+    }
+
     /*
      * #####################################################################
      * #                                                                   #

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

@@ -37,6 +37,20 @@ public abstract class KrpcServer(
     transport: KrpcTransport,
 ) : RpcServer, KrpcEndpoint {
 
+    /**
+     * Close this server, removing all the services and stopping accepting messages.
+     */
+    public fun close(message: String? = null) {
+        internalScope.cancel(message ?: "Server closed")
+    }
+
+    /**
+     * Waits until the server is closed.
+     */
+    public suspend fun awaitCompletion() {
+        internalScope.coroutineContext.job.join()
+    }
+
     /*
      * #####################################################################
      * #                                                                   #
@@ -119,6 +133,7 @@ public abstract class KrpcServer(
 
     override fun <@Rpc Service : Any> deregisterService(serviceKClass: KClass<Service>) {
         connector.unsubscribeFromServiceMessages(serviceDescriptorOf(serviceKClass).fqName)
+        rpcServices.remove(serviceDescriptorOf(serviceKClass).fqName)
     }
 
     private fun <@Rpc Service : Any> createNewServiceInstance(

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

@@ -21,6 +21,8 @@ interface CancellationService {
 
     suspend fun outgoingStream(stream: Flow<Int>)
 
+    suspend fun outgoingStreamAsync(stream: Flow<Int>)
+
     suspend fun outgoingStreamWithDelayedResponse(stream: Flow<Int>)
 
     suspend fun outgoingStreamWithException(stream: Flow<Int>)
@@ -57,6 +59,14 @@ class CancellationServiceImpl : CancellationService {
         consume(stream)
     }
 
+    @OptIn(DelicateCoroutinesApi::class)
+    override suspend fun outgoingStreamAsync(stream: Flow<Int>) {
+        GlobalScope.launch {
+            consume(stream)
+        }
+        firstIncomingConsumed.await()
+    }
+
     override suspend fun outgoingStreamWithDelayedResponse(stream: Flow<Int>) {
         consume(stream)
 

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.internalScope.cancel()
+        client.close()
         firstRequestJob.join()
         secondRequestJob.join()
 
@@ -85,8 +85,8 @@ class CancellationTest {
 
         assertEquals(0, serverInstances.sumOf { it.delayCounter.value }, "Expected no requests to succeed")
 
-        client.internalScope.join()
-        server.internalScope.join()
+        client.awaitCompletion()
+        server.awaitCompletion()
 
         checkAlive(clientAlive = false, serverAlive = false)
         stopAllAndJoin()
@@ -105,7 +105,7 @@ class CancellationTest {
         }
 
         unskippableDelay(150) // wait for requests to reach server
-        server.internalScope.cancel()
+        server.close()
         firstRequestJob.join()
         secondRequestJob.join()
 
@@ -114,8 +114,8 @@ class CancellationTest {
 
         assertEquals(0, serverInstances.sumOf { it.delayCounter.value }, "Expected no requests to succeed")
 
-        client.internalScope.join()
-        server.internalScope.join()
+        client.awaitCompletion()
+        server.awaitCompletion()
 
         checkAlive(clientAlive = false, serverAlive = false)
         stopAllAndJoin()
@@ -141,6 +141,18 @@ class CancellationTest {
         stopAllAndJoin()
     }
 
+    @Test
+    fun testOutgoingFlowLifetime() = runCancellationTest {
+        val fence = CompletableDeferred<Unit>()
+
+        service.outgoingStreamAsync(resumableFlow(fence))
+
+        serverInstance().consumedAll.await()
+        assertContentEquals(listOf(0), serverInstance().consumedIncomingValues)
+
+        stopAllAndJoin()
+    }
+
     @Test
     fun testStreamIncoming() = runCancellationTest {
         var first: Int = -1
@@ -220,8 +232,8 @@ class CancellationTest {
 
         serverInstance().firstIncomingConsumed.await()
 
-        client.internalScope.cancel("Test request cancelled")
-        client.internalScope.join()
+        client.close("Test request cancelled")
+        client.awaitCompletion()
 
         serverInstance().consumedAll.await()
 
@@ -239,7 +251,7 @@ class CancellationTest {
             caught = it
         }.collect {
             if (it == 0) {
-                client.internalScope.cancel()
+                client.close()
             } else {
                 fail("Expected the request to fail with cancellation of the client")
             }
@@ -254,7 +266,7 @@ class CancellationTest {
     fun testCancelledClientCancelsRequest() = runCancellationTest {
         launch {
             serverInstance().firstIncomingConsumed.await()
-            client.internalScope.cancel("Cancelled by test")
+            client.close("Cancelled by test")
         }
 
         try {
@@ -347,8 +359,5 @@ class CancellationTest {
     private val CoroutineScope.isCompleted get() = coroutineContext.job.isCompleted
     private val CoroutineScope.isCancelled get() = coroutineContext.job.isCancelled
 
-    @Suppress("SuspendFunctionOnCoroutineScope")
-    private suspend fun CoroutineScope.join() = apply { coroutineContext.job.join() }
-
     private suspend fun CancellationToolkit.stopAllAndJoin() = transport.coroutineContext.job.cancelAndJoin()
 }

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

@@ -26,8 +26,6 @@ fun runCancellationTest(body: suspend CancellationToolkit.() -> Unit): TestResul
     return runTest(timeout = 15.seconds) {
         debugCoroutines()
         CancellationToolkit(this).apply {
-            client.initializeTransport()
-
             body()
         }
     }