Ktor插件读取请求体后终止流程，如何避免取消异常？

你的代码能实现提前响应的需求，但触发CancellationException的核心原因是：调用call.respond发送响应后，Ktor会自动取消当前调用的协程上下文，但接收流水线的后续流程仍在继续执行，后续操作在已取消的协程中运行就会抛出这个异常。另外手动循环读取请求体的写法也存在性能优化空间。

要解决这个问题，核心是在发送提前响应后终止整个调用流水线的后续处理，同时优化请求体读取逻辑，具体有两种实现方式：

方式一：自定义异常终止流水线 #

@KtorDsl
class TestPluginConfig

val TestPlugin: RouteScopedPlugin<TestPluginConfig> = createRouteScopedPlugin(
    "TestPlugin",
    ::TestPluginConfig
) {
    on(ReceiveBytes) { call, body ->
        // 用Ktor内置方法高效读取全部请求体字节
        val bodyBytes = body.readRemaining().readBytes()
        val bodyContent = String(bodyBytes)

        if (bodyContent == "1") {
            call.respond(HttpStatusCode.NotAcceptable, "You wish")
            // 抛出自定义异常终止流水线，阻止后续处理
            throw PipelineTermination("Forbidden request content")
        }

        ByteReadChannel(bodyBytes)
    }
}

private object ReceiveBytes : Hook<suspend (call: ApplicationCall, body: ByteReadChannel) -> ByteReadChannel> {
    override fun install(
        pipeline: ApplicationCallPipeline,
        handler: suspend (call: ApplicationCall, body: ByteReadChannel) -> ByteReadChannel
    ) {
        pipeline.receivePipeline.intercept(ApplicationReceivePipeline.Before) { body ->
            if (body !is ByteReadChannel) return@intercept
            try {
                val newBody = handler(call, body)
                proceedWith(newBody)
            } catch (_: PipelineTermination) {
                // 捕获终止异常，避免向上传播产生无效日志
            }
        }
    }
}

// 自定义异常用于标记流水线终止
class PipelineTermination(message: String) : Exception(message)

方式二：直接调用finish终止流水线 #

这种写法更简洁，无需自定义异常：

@KtorDsl
class TestPluginConfig

val TestPlugin: RouteScopedPlugin<TestPluginConfig> = createRouteScopedPlugin(
    "TestPlugin",
    ::TestPluginConfig
) {
    on(ReceiveBytes) { _, _ ->
        // 这里不需要处理逻辑，直接在拦截器中完成判断和响应
        ByteReadChannel.Empty
    }
}

private object ReceiveBytes : Hook<suspend (call: ApplicationCall, body: ByteReadChannel) -> ByteReadChannel> {
    override fun install(
        pipeline: ApplicationCallPipeline,
        handler: suspend (call: ApplicationCall, body: ByteReadChannel) -> ByteReadChannel
    ) {
        pipeline.receivePipeline.intercept(ApplicationReceivePipeline.Before) { body ->
            if (body !is ByteReadChannel) return@intercept
            
            val bodyBytes = body.readRemaining().readBytes()
            val bodyContent = String(bodyBytes)
            
            if (bodyContent == "1") {
                call.respond(HttpStatusCode.NotAcceptable, "You wish")
                // 直接终止整个调用流水线，后续所有处理步骤都不会执行
                finish()
                return@intercept
            }
            
            proceedWith(ByteReadChannel(bodyBytes))
        }
    }
}

• 终止流水线的核心：无论是抛自定义异常还是调用finish()，本质都是让Ktor停止执行后续的拦截器和处理逻辑，避免在已取消的协程中继续操作
• 请求体读取优化：使用readRemaining().readBytes()替代手动循环拼接字节数组，Ktor内置方法已经做了性能优化，代码更简洁高效

内容的提问来源于stack exchange，提问作者agathis