Code generatorCopy Anchor Link

protoc-gen-connect-ktor turns service RPCs into Kotlin interfaces and Route helpers. You implement the interface; the generator owns path wiring and the choice between handle, handleClientStream, and handleServerStream.

What it emitsCopy Anchor Link

public interface ElizaServiceHandlerInterface {
    public suspend fun say(
        request: SayRequest,
        call: ApplicationCall,
    ): ResponseMessage<SayResponse>

    public suspend fun upload(
        requests: Flow<UploadRequest>,
        call: ApplicationCall,
    ): ResponseMessage<UploadResponse>

    public suspend fun introduce(
        request: IntroduceRequest,
        call: ApplicationCall,
    ): Flow<IntroduceResponse>

    public object Procedures {
        @Resource("/connectrpc.eliza.v1.ElizaService/Say")
        public class Say

        @Resource("/connectrpc.eliza.v1.ElizaService/Upload")
        public class Upload

        @Resource("/connectrpc.eliza.v1.ElizaService/Introduce")
        public class Introduce
    }
}

public fun Route.elizaService(handler: ElizaServiceHandlerInterface) {
    post<ElizaServiceHandlerInterface.Procedures.Say, SayRequest>(handle(handler::say))
    post<ElizaServiceHandlerInterface.Procedures.Upload>(handleClientStream(handler::upload))
    post<ElizaServiceHandlerInterface.Procedures.Introduce>(handleServerStream(handler::introduce))
}

Unary RPCs receive one message. Client-streaming RPCs (rpc X(stream Req) returns (Res)) receive a cold Flow<Req>. Server-streaming RPCs (rpc X(Req) returns (stream Res)) receive one message and return a cold Flow<Res>.

A server-streaming handler reports its outcome through the flow: normal completion ends the stream, and a thrown ConnectException becomes the error in the end-of-stream frame. Response headers go on call.response.headers before the flow is collected; trailers go on call.connectResponseTrailers(), which is read when the end-of-stream frame is written.

override suspend fun introduce(
    request: IntroduceRequest,
    call: ApplicationCall,
): Flow<IntroduceResponse> {
    call.connectResponseTrailers().append("x-sentence-count", "2")
    return flow {
        emit(introduceResponse { sentence = "Hi ${request.name}!" })
        emit(introduceResponse { sentence = "I'm Eliza." })
    }
}

Bidirectional streamingCopy Anchor Link

Bidirectional RPCs (rpc X(stream Req) returns (stream Res)) receive a Flow<Req> and return a Flow<Res>, bound through handleBidiStream:

override suspend fun converse(
    requests: Flow<ConverseRequest>,
    call: ApplicationCall,
): Flow<ConverseResponse> = requests.map { request ->
    converseResponse { sentence = request.sentence }
}

This interleaves each request with a response and requires HTTP/2 (Netty). For half-duplex, collect requests.toList() before returning the response flow; this works over HTTP/1.1 on both CIO and Netty.

Configure streaming request limits with the route-scoped plugin; generated code and service registration stay unchanged:

routing {
    install(ConnectStreaming) {
        maxMessageSize = 8 * 1024 * 1024
    }
    elizaService(handler)
}

Import io.github.ichizero.connect.ktor.streaming.ConnectStreaming. The limit must be positive and applies to each request message, excluding the 5-byte envelope prefix. It does not limit unary requests, response messages, or the sum of messages in a stream. Without the plugin, handlers use 4 MiB. Child routes can install their own configuration. An explicit maxMessageSize on a manual handleClientStream, handleServerStream, or handleBidiStream binding overrides the plugin. Existing generated source needs no regeneration; recompile it against the updated library because these handlers are inline.

The request flow consumes the channel. Sequential collections resume at the next unread message, so requests.firstOrNull() can read metadata from the first request before setting response headers, followed by collection of the remaining messages. Only one collector may read requests at a time. All collection must finish within the call; use structured coroutines such as channelFlow for concurrent work. Headers, trailers, errors, and deadlines follow the server-streaming rules above.

Keep full-duplex routes outside connectBodyLimit / Ktor RequestBodyLimit, which buffers small chunks until request EOF. The framer enforces a 4 MiB limit per message by default; the route-scoped plugin above customizes that limit.

Regenerating a service with bidi methods adds members to its handler interface. Existing implementations must implement these previously skipped methods.

InstallCopy Anchor Link

brew install ichizero/tap/protoc-gen-connect-ktor

Add a local: protoc-gen-connect-ktor entry to buf.gen.yaml as shown in Getting started.