grpc: Add multi interceptor tests

Signed-off-by: Johannes Zottele <[email protected]>

Author: Jozott00

grpc/grpc-core/src/commonTest/kotlin/kotlinx/rpc/grpc/test/proto/ClientInterceptorTest.kt

@@ -176,6 +176,57 @@ class ClientInterceptorTest : GrpcProtoTest() {
         assertEquals("Cancellation cause", error.cause?.message)
     }
 
+    @Test
+    fun `cancel in two interceptors - should fail with cancellation`() {
+        val error = assertFailsWith<StatusException> {
+            val interceptor1 = interceptor {
+                onClose { _, _ -> cancel("[1] Canceling in onClose", IllegalStateException("Cancellation cause")) }
+                proceed(it)
+            }
+            val interceptor2 = interceptor {
+                onClose { _, _ -> cancel("[2] Canceling in onClose", IllegalStateException("Cancellation cause")) }
+                proceed(it)
+            }
+            runGrpcTest(clientInterceptors = interceptor1 + interceptor2, test = ::unaryCall)
+        }
+
+        assertEquals(StatusCode.CANCELLED, error.getStatus().statusCode)
+        assertContains(error.message!!, "[1] Canceling in onClose")
+        assertIs<IllegalStateException>(error.cause)
+        assertEquals("Cancellation cause", error.cause?.message)
+    }
+
+    @Test
+    fun `cancel in two interceptors withing response stream - should fail with cancellation`() {
+        val error = assertFailsWith<StatusException> {
+            val interceptor1 = interceptor {
+                proceed(it).map {
+                    val msg = it as EchoResponse
+                    if (msg.message == "Echo-3") {
+                        cancel("[1] Canceling in response flow", IllegalStateException("Cancellation cause"))
+                    }
+                    it
+                }
+            }
+            val interceptor2 = interceptor {
+                proceed(it).map {
+                    val msg = it as EchoResponse
+                    // this is cancelled before the first one
+                    if (msg.message == "Echo-2") {
+                        cancel("[2] Canceling in response flow", IllegalStateException("Cancellation cause"))
+                    }
+                    it
+                }
+            }
+            runGrpcTest(clientInterceptors = interceptor1 + interceptor2, test = ::bidiStream)
+        }
+
+        assertEquals(StatusCode.CANCELLED, error.getStatus().statusCode)
+        assertContains(error.message!!, "[2] Canceling in response flow")
+        assertIs<IllegalStateException>(error.cause)
+        assertEquals("Cancellation cause", error.cause?.message)
+    }
+
     @Test
     fun `modify request message - should return modified message`() {
         val interceptor = interceptor {

grpc/grpc-core/src/commonTest/kotlin/kotlinx/rpc/grpc/test/proto/ServerInterceptorTest.kt

@@ -35,6 +35,21 @@ class ServerInterceptorTest : GrpcProtoTest() {
         registerService<EchoService> { EchoServiceImpl() }
     }
 
+    @Test
+    fun `throw during onClosing - should fail propagate the exception to the server root`() {
+        val error = assertFailsWith<IllegalStateException> {
+            val interceptor = interceptor {
+                onClose { _, _ -> throw IllegalStateException("Illegal failing in onClose") }
+                proceed(it)
+            }
+            runGrpcTest(serverInterceptors = interceptor, test = ::unaryCall)
+        }
+
+        assertContains(error.message!!, "Illegal failing in onClose")
+        // check that the error is indeed causing a server crash
+        assertContains(error.stackTraceToString(), "suspendServerCall")
+    }
+
     @Test
     fun `throw during intercept - should fail with unknown status on client`() {
         var cause: Throwable? = null
@@ -112,6 +127,24 @@ class ServerInterceptorTest : GrpcProtoTest() {
         assertContains(error.message!!, "Close in onClose")
     }
 
+    @Test
+    fun `close in two interceptors - should fail with correct status on client`() {
+        val error = assertFailsWith<StatusException> {
+            val interceptor1 = interceptor {
+                onClose { _, _ -> close(Status(StatusCode.UNAUTHENTICATED, "[1] Close in onClose"), GrpcMetadata()) }
+                proceed(it)
+            }
+            val interceptor2 = interceptor {
+                onClose { _, _ -> close(Status(StatusCode.UNAUTHENTICATED, "[2] Close in onClose"), GrpcMetadata()) }
+                proceed(it)
+            }
+            runGrpcTest(serverInterceptors = interceptor1 + interceptor2, test = ::unaryCall)
+        }
+
+        assertEquals(StatusCode.UNAUTHENTICATED, error.getStatus().statusCode)
+        assertContains(error.message!!, "[1] Close in onClose")
+    }
+
     @Test
     fun `dont proceed and return custom message - should succeed on client`() {
         val interceptor = interceptor {

grpc/grpc-core/src/nativeMain/kotlin/kotlinx/rpc/grpc/internal/NativeClientCall.kt

@@ -154,7 +154,7 @@ internal class NativeClientCall<Request, Response>(
      */
     private fun turnReady() {
         if (ready.compareAndSet(expect = false, update = true)) {
-            safeUserCode("Failed to call onClose.") {
+            safeUserCode("Failed to call onReady.") {
                 listener?.onReady()
             }
         }
@@ -200,6 +200,7 @@ internal class NativeClientCall<Request, Response>(
                 callResult.future.onComplete { success ->
                     try {
                         if (success) {
+                            // if the batch doesn't succeed, this is reflected in the recv status op batch.
                             onSuccess()
                         }
                     } finally {

grpc/grpc-core/src/nativeMain/kotlin/kotlinx/rpc/grpc/internal/NativeServerCall.kt

@@ -291,7 +291,6 @@ internal class NativeServerCall<Request, Response>(
 
     override fun close(status: Status, trailers: GrpcMetadata) {
         check(initialized) { internalError("Call not initialized") }
-        closed = true
 
         val arena = Arena()
 
@@ -320,6 +319,7 @@ internal class NativeServerCall<Request, Response>(
             if (details != null) grpc_slice_unref(details)
             arena.clear()
         }) {
+            closed = true
             // nothing to do here
         }
     }