grpc: Add documentation

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

Author: Jozott00

grpc/grpc-core/src/commonMain/kotlin/kotlinx/rpc/grpc/ClientInterceptor.kt

@@ -8,29 +8,118 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.rpc.grpc.internal.GrpcCallOptions
 import kotlinx.rpc.grpc.internal.MethodDescriptor
 
+/**
+ * The scope of a single outgoing gRPC client call observed by a [ClientInterceptor].
+ *
+ * An interceptor receives this scope instance for every call and can:
+ * - Inspect the RPC [method] being invoked.
+ * - Read or populate [requestHeaders] before the request is sent.
+ * - Read [callOptions] that affect transport-level behavior.
+ * - Register callbacks with [onHeaders] and [onClose] to observe response metadata and final status.
+ * - Cancel the call early via [cancel].
+ * - Continue the call by calling [proceed] with a (possibly transformed) request [Flow].
+ * - Transform the response by modifying the returned [Flow].
+ *
+ * ```kt
+ *  val interceptor = object : ClientInterceptor {
+ *      override fun <Request, Response> ClientCallScope<Request, Response>.intercept(
+ *          request: Flow<Request>
+ *      ): Flow<Response> {
+ *          // Example: add a header before proceeding
+ *          requestHeaders[MyKeys.Authorization] = token
+ *
+ *          // Example: observe response metadata
+ *          onHeaders { headers -> /* inspect headers */ }
+ *          onClose { status, trailers -> /* log status/trailers */ }
+ *
+ *          // IMPORTANT: proceed forwards the call to the next interceptor/transport.
+ *          // If you do not call proceed, no request will be sent and the call is short-circuited.
+ *          return proceed(request)
+ *      }
+ *  }
+ * ```
+ *
+ * @param Request the request message type of the RPC.
+ * @param Response the response message type of the RPC.
+ */
 public interface ClientCallScope<Request, Response> {
+    /** Descriptor of the RPC method (name, marshalling, type) being invoked. */
     public val method: MethodDescriptor<Request, Response>
+
+    /**
+     * Outgoing request headers for this call.
+     *
+     * Interceptors may read and mutate this metadata
+     * before calling [proceed] so the headers are sent to the server. Headers added after
+     * the call has already been proceeded may not be reflected on the wire.
+     */
     public val requestHeaders: GrpcMetadata
+
+    /**
+     * Transport/engine options used for this call (deadlines, compression, etc.).
+     * Modifying this object is only possible before the call is proceeded.
+     */
     public val callOptions: GrpcCallOptions
+
+    /**
+     * Register a callback invoked when the initial response headers are received.
+     * Typical gRPC semantics guarantee headers are delivered at most once per call
+     * and before the first message is received.
+     */
     public fun onHeaders(block: (responseHeaders: GrpcMetadata) -> Unit)
-    public fun onClose(block: (closeStatus: Status, responseTrailers: GrpcMetadata) -> Unit)
+
+    /**
+     * Register a callback invoked when the call completes, successfully or not.
+     * The final `status` and trailing `responseTrailers` are provided.
+     */
+    public fun onClose(block: (status: Status, responseTrailers: GrpcMetadata) -> Unit)
+
+    /**
+     * Cancel the call locally, providing a human-readable [message] and an optional [cause].
+     * This method won't return and abort all further processing.
+     */
     public fun cancel(message: String, cause: Throwable? = null): Nothing
+
+    /**
+     * Continue the invocation by forwarding it to the next interceptor or to the underlying transport.
+     *
+     * This function is the heart of an interceptor:
+     * - It must be called to actually perform the RPC. If you never call [proceed], the request is not sent
+     *   and the call is effectively short-circuited by the interceptor.
+     * - You may transform the [request] flow before passing it to [proceed] (e.g., logging, retry orchestration,
+     *   compression, metrics). The returned [Flow] yields response messages and can also be transformed
+     *   before being returned to the caller.
+     * - Call [proceed] at most once per intercepted call. Calling it multiple times or after cancellation
+     *   is not supported.
+     */
     public fun proceed(request: Flow<Request>): Flow<Response>
 }
 
+/**
+ * Client-side interceptor for gRPC calls.
+ *
+ * Implementations can observe and modify client calls in a structured way. The primary entry point is the
+ * [intercept] extension function on [ClientCallScope], which receives the inbound request [Flow] and must
+ * call [ClientCallScope.proceed] to forward the call.
+ *
+ * Common use-cases include:
+ * - Adding authentication or custom headers.
+ * - Implementing logging/metrics.
+ * - Observing headers/trailers and final status.
+ * - Transforming request/response flows (e.g., mapping, buffering, throttling).
+ */
 public interface ClientInterceptor {
-
     /**
-     * Intercepts and transforms the flow of requests and responses in a client call.
-     * An interceptor can throw an exception at any time to cancel the call.
+     * Intercept a client call.
      *
-     * The interceptor must ensure that it emits an expected number of values.
-     * E.g. if the intercepted method is a unary call, the interceptor's returned flow must emit exactly one value.
+     * You can:
+     * - Inspect [ClientCallScope.method] and [ClientCallScope.callOptions].
+     * - Read or populate [ClientCallScope.requestHeaders].
+     * - Register [ClientCallScope.onHeaders] and [ClientCallScope.onClose] callbacks.
+     * - Transform the [request] flow or wrap the resulting response flow.
      *
-     * @param this The scope of the client call, providing context and methods for managing
-     * the call lifecycle and metadata.
-     * @param request A flow of requests to be sent to the server.
-     * @return A flow of responses received from the server.
+     * IMPORTANT: [ClientCallScope.proceed] must eventually be called to actually execute the RPC and obtain
+     * the response [Flow]. If [ClientCallScope.proceed] is omitted, the call will not reach the server.
      */
     public fun <Request, Response> ClientCallScope<Request, Response>.intercept(
         request: Flow<Request>,

grpc/grpc-core/src/commonMain/kotlin/kotlinx/rpc/grpc/ServerInterceptor.kt

@@ -6,26 +6,120 @@ package kotlinx.rpc.grpc
 
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.FlowCollector
+import kotlinx.rpc.grpc.internal.GrpcContext
 import kotlinx.rpc.grpc.internal.MethodDescriptor
 
+/**
+ * Th scope of a single incoming gRPC server call observed by a [ServerInterceptor].
+ *
+ * An interceptor receives this scope instance for every RPC invocation arriving to the server and can:
+ * - Inspect the target RPC [method].
+ * - Read client-provided [requestHeaders].
+ * - Populate [responseHeaders] (sent before the first response message) and [responseTrailers]
+ *   (sent when the call completes).
+ * - Register a completion callback with [onClose].
+ * - Abort the call early with [close].
+ * - Continue handling by calling [proceed] with the inbound request [Flow] and optionally transform
+ *   the returned response [Flow].
+ *
+ * @param Request the request message type of the RPC.
+ * @param Response the response message type of the RPC.
+ */
 public interface ServerCallScope<Request, Response> {
+    /** Descriptor of the RPC method (name, marshalling, type) being executed. */
     public val method: MethodDescriptor<Request, Response>
+
+    /** Metadata received from the client with the initial request headers. Read-only from the server perspective. */
     public val requestHeaders: GrpcMetadata
+
+    /**
+     * Initial response headers to be sent to the client.
+     * Interceptors and handlers may add entries before the first response element is emitted
+     * (i.e., before proceeding or before producing output), otherwise headers might have already been sent.
+     */
     public val responseHeaders: GrpcMetadata
+
+    /**
+     * Trailing metadata to be sent with the final status when the call completes.
+     * Interceptors can add diagnostics or custom metadata here.
+     */
     public val responseTrailers: GrpcMetadata
 
+    /**
+     * The [GrpcContext] associated with this call.
+     *
+     * It can be used by the interceptor to provide call-scoped information about
+     * the current call, such as the identity of the caller or the current authentication state.
+     */
+    public val grpcContext: GrpcContext
+
+    /**
+     * Register a callback invoked when the call is closed (successfully or exceptionally).
+     * Provides the final [Status] and the sent [GrpcMetadata] trailers.
+     */
     public fun onClose(block: (Status, GrpcMetadata) -> Unit)
+
+    /**
+     * Immediately terminate the call with the given [status] and optional [trailers].
+     *
+     * This method does not return (declared as [Nothing]). After calling it, no further messages will be processed
+     * or sent. Prefer setting [responseHeaders]/[responseTrailers] before closing if you need to include metadata.
+     */
     public fun close(status: Status, trailers: GrpcMetadata = GrpcMetadata()): Nothing
+
+    /**
+     * Continue processing by forwarding the request to the next interceptor or the actual service implementation.
+     *
+     * IMPORTANT:
+     * - You must call [proceed] exactly once to actually handle the RPC; otherwise, the call will be short-circuited
+     *   and the service method will not be invoked.
+     * - You may transform the incoming [request] flow (e.g., validation, logging, metering) before passing it to
+     *   [proceed]. You may also transform the resulting response [Flow] before returning it to the framework.
+     * - The interceptor must ensure to provide and return a valid number of messages, depending on the method type.
+     * - The interceptor must not throw an exception. Use [close] to terminate the call with an error.
+     */
     public fun proceed(request: Flow<Request>): Flow<Response>
 
+    /**
+     * Convenience for flow builders: proceeds with [request] and emits the resulting response elements into this
+     * [FlowCollector]. Useful inside `flow {}` blocks within interceptors.
+     */
     public suspend fun FlowCollector<Response>.proceedFlow(request: Flow<Request>) {
         proceed(request).collect {
             emit(it)
         }
     }
 }
 
+/**
+ * Server-side interceptor for gRPC calls.
+ *
+ * Implementations can observe and modify server handling in a structured way. The entry point is the
+ * [intercept] extension function on [ServerCallScope], which receives the inbound request [Flow] and must
+ * call [ServerCallScope.proceed] to forward the call to the next interceptor or the target service method.
+ *
+ * Common use-cases include:
+ * - Authentication/authorization checks and context propagation.
+ * - Setting response headers and trailers.
+ * - Structured logging and metrics.
+ * - Transforming request/response flows (e.g., validation, mapping, throttling).
+ *
+ * See ServerInterceptorTest for practical usage patterns.
+ */
 public interface ServerInterceptor {
+    /**
+     * Intercept a server call.
+     *
+     * You can:
+     * - Inspect [ServerCallScope.method].
+     * - Read [ServerCallScope.requestHeaders] and populate [ServerCallScope.responseHeaders]/[ServerCallScope.responseTrailers].
+     * - Register [ServerCallScope.onClose] callbacks.
+     * - Transform the [request] flow or wrap the resulting response flow.
+     * - Append information to the [ServerCallScope.grpcContext].
+     *
+     * IMPORTANT: You must eventually call [ServerCallScope.proceed] to actually invoke the service logic and produce
+     * the response [Flow]. If [ServerCallScope.proceed] is omitted, the call will never reach the service.
+     */
     public fun <Request, Response> ServerCallScope<Request, Response>.intercept(
         request: Flow<Request>,
     ): Flow<Response>

grpc/grpc-core/src/commonMain/kotlin/kotlinx/rpc/grpc/internal/GrpcContext.kt

@@ -6,10 +6,13 @@ package kotlinx.rpc.grpc.internal
 
 import kotlin.coroutines.CoroutineContext
 
-internal expect class GrpcContext
+public expect class GrpcContext
+
 internal expect val CurrentGrpcContext: GrpcContext
 
 internal expect class GrpcContextElement : CoroutineContext.Element {
+    val grpcContext: GrpcContext
+
     companion object Key : CoroutineContext.Key<GrpcContextElement> {
         fun current(): GrpcContextElement
     }

grpc/grpc-core/src/commonMain/kotlin/kotlinx/rpc/grpc/internal/suspendServerCalls.kt

@@ -157,15 +157,17 @@ private fun <Request, Response> CoroutineScope.serverCallListenerImpl(
         }
     }
 
+    val context = GrpcContextElement.current()
     val serverCallScope = ServerCallScopeImpl(
         method = descriptor,
         interceptors = interceptors,
         implementation = implementation,
         requestHeaders = requestHeaders,
         serverCall = handler,
+        grpcContext = context.grpcContext,
     )
 
-    val rpcJob = launch(GrpcContextElement.current()) {
+    val rpcJob = launch() {
         val mutex = Mutex()
         val headersSent = AtomicBoolean(false) // enforces only sending headers once
         val failure = runCatching {
@@ -273,6 +275,7 @@ private class ServerCallScopeImpl<Request, Response>(
     val implementation: (Flow<Request>) -> Flow<Response>,
     override val requestHeaders: GrpcMetadata,
     val serverCall: ServerCall<Request, Response>,
+    override val grpcContext: GrpcContext,
 ) : ServerCallScope<Request, Response> {
 
     override val responseHeaders: GrpcMetadata = GrpcMetadata()

grpc/grpc-core/src/jvmMain/kotlin/kotlinx/rpc/grpc/internal/GrpcContext.kt

@@ -8,12 +8,12 @@ import io.grpc.Context
 import kotlinx.coroutines.ThreadContextElement
 import kotlin.coroutines.CoroutineContext
 
-internal actual typealias GrpcContext = Context
+public actual typealias GrpcContext = Context
 
 internal actual val CurrentGrpcContext: GrpcContext
     get() = GrpcContext.current()
 
-internal actual class GrpcContextElement(private val grpcContext: GrpcContext) : ThreadContextElement<GrpcContext> {
+internal actual class GrpcContextElement(actual val grpcContext: GrpcContext) : ThreadContextElement<GrpcContext> {
     actual companion object Key : CoroutineContext.Key<GrpcContextElement> {
         actual fun current(): GrpcContextElement = GrpcContextElement(CurrentGrpcContext)
     }

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

@@ -6,20 +6,20 @@ package kotlinx.rpc.grpc.internal
 
 import kotlin.coroutines.CoroutineContext
 
-internal actual class GrpcContext
+public actual class GrpcContext
 
 private val currentGrpcContext = GrpcContext()
 
 internal actual val CurrentGrpcContext: GrpcContext
     get() = currentGrpcContext
 
-internal actual class GrpcContextElement : CoroutineContext.Element {
+internal actual class GrpcContextElement(actual val grpcContext: GrpcContext) : CoroutineContext.Element {
     actual override val key: CoroutineContext.Key<GrpcContextElement>
         get() = Key
 
     actual companion object Key : CoroutineContext.Key<GrpcContextElement> {
         actual fun current(): GrpcContextElement {
-            return GrpcContextElement()
+            return GrpcContextElement(currentGrpcContext)
         }
     }
 }