Skip to content

Commit 670d605

Browse files
Jozott00Jozott00
committed
grpc: Add tests
Signed-off-by: Johannes Zottele <[email protected]>
1 parent 0648833 commit 670d605

File tree

4 files changed

+111
-32
lines changed

4 files changed

+111
-32
lines changed

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

Lines changed: 39 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
package kotlinx.rpc.grpc.test.proto
66

77
import kotlinx.coroutines.flow.Flow
8+
import kotlinx.coroutines.flow.flowOf
89
import kotlinx.coroutines.flow.map
910
import kotlinx.rpc.RpcServer
1011
import kotlinx.rpc.grpc.GrpcClient
@@ -16,6 +17,7 @@ import kotlinx.rpc.grpc.StatusCode
1617
import kotlinx.rpc.grpc.StatusException
1718
import kotlinx.rpc.grpc.statusCode
1819
import kotlinx.rpc.grpc.test.EchoRequest
20+
import kotlinx.rpc.grpc.test.EchoResponse
1921
import kotlinx.rpc.grpc.test.EchoService
2022
import kotlinx.rpc.grpc.test.EchoServiceImpl
2123
import kotlinx.rpc.grpc.test.invoke
@@ -61,7 +63,7 @@ class ServerInterceptorTest : GrpcProtoTest() {
6163
}
6264

6365
assertEquals(StatusCode.UNAUTHENTICATED, error.getStatus().statusCode)
64-
assertContains(error.message!!, "Close in interceptor")
66+
assertContains(error.getStatus().getDescription()!!, "Close in interceptor")
6567
}
6668

6769
@Test
@@ -110,6 +112,42 @@ class ServerInterceptorTest : GrpcProtoTest() {
110112
assertContains(error.message!!, "Close in onClose")
111113
}
112114

115+
@Test
116+
fun `dont proceed and return custom message - should succeed on client`() {
117+
val interceptor = interceptor {
118+
flowOf(EchoResponse { message = "Custom message" })
119+
}
120+
runGrpcTest(serverInterceptors = interceptor) {
121+
val service = it.withService<EchoService>()
122+
val response = service.UnaryEcho(EchoRequest { message = "Hello" })
123+
assertEquals("Custom message", response.message)
124+
}
125+
}
126+
127+
@Test
128+
fun `manipulate request - should succeed on client`() {
129+
val interceptor = interceptor {
130+
proceed(it.map { EchoRequest { message = "Modified" } })
131+
}
132+
runGrpcTest(serverInterceptors = interceptor) {
133+
val service = it.withService<EchoService>()
134+
val response = service.UnaryEcho(EchoRequest { message = "Hello" })
135+
assertEquals("Modified", response.message)
136+
}
137+
}
138+
139+
@Test
140+
fun `manipulate response - should succeed on client`() {
141+
val interceptor = interceptor {
142+
proceed(it).map { EchoResponse { message = "Modified" } }
143+
}
144+
runGrpcTest(serverInterceptors = interceptor) {
145+
val service = it.withService<EchoService>()
146+
val response = service.UnaryEcho(EchoRequest { message = "Hello" })
147+
assertEquals("Modified", response.message)
148+
}
149+
}
150+
113151
private suspend fun unaryCall(grpcClient: GrpcClient) {
114152
val service = grpcClient.withService<EchoService>()
115153
val response = service.UnaryEcho(EchoRequest { message = "Hello" })

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

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -367,7 +367,10 @@ internal class NativeClientCall<Request, Response>(
367367
// user side cancellation must always win over any other status (even if the call is already completed).
368368
// this will also preserve the cancellation cause, which cannot be passed to the grpc-core.
369369
closeInfo.value = Pair(status, GrpcTrailers())
370-
cancelInternal(grpc_status_code.GRPC_STATUS_CANCELLED, message ?: "Call cancelled with cause: ${cause?.message}")
370+
cancelInternal(
371+
grpc_status_code.GRPC_STATUS_CANCELLED,
372+
message ?: "Call cancelled with cause: ${cause?.message}"
373+
)
371374
}
372375

373376
private fun cancelInternal(statusCode: grpc_status_code, message: String) {

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

Lines changed: 67 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -16,11 +16,15 @@ import kotlinx.cinterop.CPointerVar
1616
import kotlinx.cinterop.ExperimentalForeignApi
1717
import kotlinx.cinterop.IntVar
1818
import kotlinx.cinterop.alloc
19+
import kotlinx.cinterop.allocArray
20+
import kotlinx.cinterop.convert
21+
import kotlinx.cinterop.get
1922
import kotlinx.cinterop.ptr
20-
import kotlinx.cinterop.readValue
2123
import kotlinx.cinterop.value
2224
import kotlinx.rpc.grpc.GrpcTrailers
2325
import kotlinx.rpc.grpc.Status
26+
import kotlinx.rpc.grpc.StatusCode
27+
import kotlinx.rpc.grpc.StatusException
2428
import kotlinx.rpc.protobuf.input.stream.asInputStream
2529
import kotlinx.rpc.protobuf.input.stream.asSource
2630
import libkgrpc.GRPC_OP_RECV_CLOSE_ON_SERVER
@@ -33,7 +37,6 @@ import libkgrpc.grpc_byte_buffer_destroy
3337
import libkgrpc.grpc_call_cancel_with_status
3438
import libkgrpc.grpc_call_unref
3539
import libkgrpc.grpc_op
36-
import libkgrpc.grpc_slice
3740
import libkgrpc.grpc_slice_unref
3841
import libkgrpc.grpc_status_code
3942
import kotlin.concurrent.Volatile
@@ -66,7 +69,12 @@ internal class NativeServerCall<Request, Response>(
6669
private var cancelled = false
6770
private val finalized = atomic(false)
6871

69-
// Tracks whether at least one request message has been received on this call.
72+
// tracks whether the initial metadata has been sent.
73+
// this is used to determine if we have to send the initial metadata
74+
// when we try to close the call.
75+
private var sentInitialMetadata = false
76+
77+
// tracks whether at least one request message has been received on this call.
7078
private var receivedFirstMessage = false
7179

7280
// we currently don't buffer messages, so after one `sendMessage` call, ready turns false. (KRPC-192)
@@ -245,6 +253,7 @@ internal class NativeServerCall<Request, Response>(
245253
data.send_initial_metadata.metadata = null
246254
}
247255

256+
sentInitialMetadata = true
248257
runBatch(op.ptr, 1u, cleanup = { arena.clear() }) {
249258
// nothing to do here
250259
}
@@ -256,42 +265,52 @@ internal class NativeServerCall<Request, Response>(
256265
val methodDescriptor = checkNotNull(methodDescriptor) { internalError("Method descriptor not set") }
257266

258267
val arena = Arena()
259-
val inputStream = methodDescriptor.getResponseMarshaller().stream(message)
260-
val byteBuffer = inputStream.asSource().toGrpcByteBuffer()
261-
ready.value = false
262-
263-
val op = arena.alloc<grpc_op> {
264-
op = GRPC_OP_SEND_MESSAGE
265-
data.send_message.send_message = byteBuffer
266-
}
268+
tryRun {
269+
val inputStream = methodDescriptor.getResponseMarshaller().stream(message)
270+
val byteBuffer = inputStream.asSource().toGrpcByteBuffer()
271+
ready.value = false
272+
273+
val op = arena.alloc<grpc_op> {
274+
op = GRPC_OP_SEND_MESSAGE
275+
data.send_message.send_message = byteBuffer
276+
}
267277

268-
runBatch(op.ptr, 1u, cleanup = {
269-
arena.clear()
270-
grpc_byte_buffer_destroy(byteBuffer)
271-
}) {
272-
turnReady()
278+
runBatch(op.ptr, 1u, cleanup = {
279+
arena.clear()
280+
grpc_byte_buffer_destroy(byteBuffer)
281+
}) {
282+
turnReady()
283+
}
273284
}
274285
}
275286

276287
override fun close(status: Status, trailers: GrpcTrailers) {
277288
check(initialized) { internalError("Call not initialized") }
278289
val arena = Arena()
279290

280-
val details = status.getDescription()?.let {
281-
arena.alloc<grpc_slice> {
282-
it.toGrpcSlice()
283-
}
284-
}
285-
val op = arena.alloc<grpc_op> {
286-
op = GRPC_OP_SEND_STATUS_FROM_SERVER
287-
data.send_status_from_server.status = status.statusCode.toRawCallAllocation()
288-
data.send_status_from_server.status_details = details?.ptr
289-
data.send_status_from_server.trailing_metadata_count = 0u
290-
data.send_status_from_server.trailing_metadata = null
291+
val details = status.getDescription()?.toGrpcSlice()
292+
val detailsPtr = details?.getPointer(arena)
293+
294+
val nOps = if (sentInitialMetadata) 1uL else 2uL
295+
296+
val ops = arena.allocArray<grpc_op>(nOps.convert())
297+
298+
ops[0].op = GRPC_OP_SEND_STATUS_FROM_SERVER
299+
ops[0].data.send_status_from_server.status = status.statusCode.toRaw()
300+
ops[0].data.send_status_from_server.status_details = detailsPtr
301+
ops[0].data.send_status_from_server.trailing_metadata_count = 0u
302+
ops[0].data.send_status_from_server.trailing_metadata = null
303+
304+
if (!sentInitialMetadata) {
305+
// if we haven't sent GRPC_OP_SEND_INITIAL_METADATA yet,
306+
// so we must do it together with the close operation.
307+
ops[1].op = GRPC_OP_SEND_INITIAL_METADATA
308+
ops[1].data.send_initial_metadata.count = 0u
309+
ops[1].data.send_initial_metadata.metadata = null
291310
}
292311

293-
runBatch(op.ptr, 1u, cleanup = {
294-
if (details != null) grpc_slice_unref(details.readValue())
312+
runBatch(ops, nOps, cleanup = {
313+
if (details != null) grpc_slice_unref(details)
295314
arena.clear()
296315
}) {
297316
// nothing to do here
@@ -306,6 +325,25 @@ internal class NativeServerCall<Request, Response>(
306325
val methodDescriptor = checkNotNull(methodDescriptor) { internalError("Method descriptor not set") }
307326
return methodDescriptor
308327
}
328+
329+
330+
private fun <T> tryRun(block: () -> T): T {
331+
try {
332+
return block()
333+
} catch (e: Throwable) {
334+
// TODO: Log internal error as warning
335+
val status = when (e) {
336+
is StatusException -> e.getStatus()
337+
else -> Status(
338+
StatusCode.INTERNAL,
339+
description = "Internal error, so canceling the stream",
340+
cause = e
341+
)
342+
}
343+
cancel(status.statusCode.toRaw(), status.getDescription() ?: "Unknown error")
344+
throw StatusException(status, trailers = null)
345+
}
346+
}
309347
}
310348

311349
/**

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -193,7 +193,7 @@ internal fun grpc_status_code.toKotlin(): StatusCode = when (this) {
193193
else -> error("Invalid status code: $this")
194194
}
195195

196-
internal fun StatusCode.toRawCallAllocation(): grpc_status_code = when (this) {
196+
internal fun StatusCode.toRaw(): grpc_status_code = when (this) {
197197
StatusCode.OK -> grpc_status_code.GRPC_STATUS_OK
198198
StatusCode.CANCELLED -> grpc_status_code.GRPC_STATUS_CANCELLED
199199
StatusCode.UNKNOWN -> grpc_status_code.GRPC_STATUS_UNKNOWN

0 commit comments

Comments
 (0)