kotlin-ktor-patterns

name kotlin-ktor-patterns description Ktor 服务器模式，包括路由 DSL、插件、身份验证、Koin DI、kotlinx.serialization、WebSockets 和 testApplication 测试。 origin ECC Ktor 服务器模式 使用 Kotlin 协程构建健壮、可维护的 HTTP 服务器的综合 Ktor 模式。 何时启用 构建 Ktor HTTP 服务器 配置 Ktor 插件（Auth、CORS、ContentNegotiation、StatusPages） 使用 Ktor 实现 REST API 使用 Koin 设置依赖注入 使用 testApplication 编写 Ktor 集成测试 在 Ktor 中使用 WebSocket 应用程序结构 标准 Ktor 项目布局 src/main/kotlin/ ├── com/example/ │ ├── Application.kt # 入口点，模块配置 │ ├── plugins/ │ │ ├── Routing.kt # 路由定义 │ │ ├── Serialization.kt # 内容协商设置 │ │ ├── Authentication.kt # 认证配置 │ │ ├── StatusPages.kt # 错误处理 │ │ └── CORS.kt # CORS 配置 │ ├── routes/ │ │ ├── UserRoutes.kt # /users 端点 │ │ ├── AuthRoutes.kt # /auth 端点 │ │ └── HealthRoutes.kt # /health 端点 │ ├── models/ │ │ ├── User.kt # 领域模型 │ │ └── ApiResponse.kt # 响应封装 │ ├── services/ │ │ ├── UserService.kt # 业务逻辑 │ │ └── AuthService.kt # 认证逻辑 │ ├── repositories/ │ │ ├── UserRepository.kt # 数据访问接口 │ │ └── ExposedUserRepository.kt │ └── di/ │ └── AppModule.kt # Koin 模块 src/test/kotlin/ ├── com/example/ │ ├── routes/ │ │ └── UserRoutesTest.kt │ └── services/ │ └── UserServiceTest.kt 应用程序入口点 // Application.kt fun main () { embeddedServer(Netty, port = 8080 , module = Application::module).start(wait = true ) } fun Application. module () { configureSerialization() configureAuthentication() configureStatusPages() configureCORS() configureDI() configureRouting() } 路由 DSL 基本路由 // plugins/Routing.kt fun Application. configureRouting () { routing { userRoutes() authRoutes() healthRoutes() } } // routes/UserRoutes.kt fun Route. userRoutes () { val userService by inject<UserService>() route( "/users" ) { get { val users = userService.getAll() call.respond(users) } get ( "/{id}" ) { val id = call.parameters[ "id" ] ?: return @get call.respond(HttpStatusCode.BadRequest, "Missing id" ) val user = userService.getById(id) ?: return @get call.respond(HttpStatusCode.NotFound) call.respond(user) } post { val request = call.receive<CreateUserRequest>() val user = userService.create(request) call.respond(HttpStatusCode.Created, user) } put( "/{id}" ) { val id = call.parameters[ "id" ] ?: return @put call.respond(HttpStatusCode.BadRequest, "Missing id" ) val request = call.receive<UpdateUserRequest>() val user = userService.update(id, request) ?: return @put call.respond(HttpStatusCode.NotFound) call.respond(user) } delete( "/{id}" ) { val id = call.parameters[ "id" ] ?: return @delete call.respond(HttpStatusCode.BadRequest, "Missing id" ) val deleted = userService.delete(id) if (deleted) call.respond(HttpStatusCode.NoContent) else call.respond(HttpStatusCode.NotFound) } } } 使用认证路由组织路由 fun Route. userRoutes () { route( "/users" ) { // Public routes get { /* list users */ } get ( "/{id}" ) { /* get user */ } // Protected routes authenticate( "jwt" ) { post { /* create user - requires auth */ } put( "/{id}" ) { /* update user - requires auth */ } delete( "/{id}" ) { /* delete user - requires auth */ } } } } 内容协商与序列化 kotlinx.serialization 设置 // plugins/Serialization.kt fun Application. configureSerialization () { install(ContentNegotiation) { json(Json { prettyPrint = true isLenient = false ignoreUnknownKeys = true encodeDefaults = true explicitNulls = false }) } } 可序列化模型 @Serializable data class UserResponse ( val id: String, val name: String, val email: String, val role: Role, @Serializable(with = InstantSerializer::class) val createdAt: Instant, ) @Serializable data class CreateUserRequest ( val name: String, val email: String, val role: Role = Role.USER, ) @Serializable data class ApiResponse < T >( val success: Boolean , val data : T? = null , val error: String? = null , ) { companion object { fun <T> ok ( data : T ) : ApiResponse<T> = ApiResponse(success = true , data = data ) fun <T> error (message: String ) : ApiResponse<T> = ApiResponse(success = false , error = message) } } @Serializable data class PaginatedResponse < T >( val data : List<T>, val total: Long , val page: Int , val limit: Int , ) 自定义序列化器 object InstantSerializer : KSerializer<Instant> { override val descriptor = PrimitiveSerialDescriptor( "Instant" , PrimitiveKind.STRING) override fun serialize (encoder: Encoder , value: Instant ) = encoder.encodeString(value.toString()) override fun deserialize (decoder: Decoder ) : Instant = Instant.parse(decoder.decodeString()) } 身份验证 JWT 身份验证 // plugins/Authentication.kt fun Application. configureAuthentication () { val jwtSecret = environment.config.property( "jwt.secret" ).getString() val jwtIssuer = environment.config.property( "jwt.issuer" ).getString() val jwtAudience = environment.config.property( "jwt.audience" ).getString() val jwtRealm = environment.config.property( "jwt.realm" ).getString() install(Authentication) { jwt( "jwt" ) { realm = jwtRealm verifier( JWT.require(Algorithm.HMAC256(jwtSecret)) .withAudience(jwtAudience) .withIssuer(jwtIssuer) .build() ) validate { credential -> if (credential.payload.audience.contains(jwtAudience)) { JWTPrincipal(credential.payload) } else { null } } challenge { _, _ -> call.respond(HttpStatusCode.Unauthorized, ApiResponse.error< Unit >( "Invalid or expired token" )) } } } } // Extracting user from JWT fun ApplicationCall. userId () : String = principal<JWTPrincipal>() ?.payload ?.getClaim( "userId" ) ?.asString() ?: throw AuthenticationException( "No userId in token" ) 认证路由 fun Route. authRoutes () { val authService by inject<AuthService>() route( "/auth" ) { post( "/login" ) { val request = call.receive<LoginRequest>() val token = authService.login(request.email, request.password) ?: return @post call.respond( HttpStatusCode.Unauthorized, ApiResponse.error< Unit >( "Invalid credentials" ), ) call.respond(ApiResponse.ok(TokenResponse(token))) } post( "/register" ) { val request = call.receive<RegisterRequest>() val user = authService.register(request) call.respond(HttpStatusCode.Created, ApiResponse.ok(user)) } authenticate( "jwt" ) { get ( "/me" ) { val userId = call.userId() val user = authService.getProfile(userId) call.respond(ApiResponse.ok(user)) } } } } 状态页（错误处理） // plugins/StatusPages.kt fun Application. configureStatusPages () { install(StatusPages) { exception<ContentTransformationException> { call, cause -> call.respond( HttpStatusCode.BadRequest, ApiResponse.error< Unit >( "Invalid request body: ${cause.message} " ), ) } exception<IllegalArgumentException> { call, cause -> call.respond( HttpStatusCode.BadRequest, ApiResponse.error< Unit >(cause.message ?: "Bad request" ), ) } exception<AuthenticationException> { call, _ -> call.respond( HttpStatusCode.Unauthorized, ApiResponse.error< Unit >( "Authentication required" ), ) } exception<AuthorizationException> { call, _ -> call.respond( HttpStatusCode.Forbidden, ApiResponse.error< Unit >( "Access denied" ), ) } exception<NotFoundException> { call, cause -> call.respond( HttpStatusCode.NotFound, ApiResponse.error< Unit >(cause.message ?: "Resource not found" ), ) } exception<Throwable> { call, cause -> call.application.log.error( "Unhandled exception" , cause) call.respond( HttpStatusCode.InternalServerError, ApiResponse.error< Unit >( "Internal server error" ), ) } status(HttpStatusCode.NotFound) { call, status -> call.respond(status, ApiResponse.error< Unit >( "Route not found" )) } } } CORS 配置 // plugins/CORS.kt fun Application. configureCORS () { install(CORS) { allowHost( "localhost:3000" ) allowHost( "example.com" , schemes = listOf( "https" )) allowHeader(HttpHeaders.ContentType) allowHeader(HttpHeaders.Authorization) allowMethod(HttpMethod.Put) allowMethod(HttpMethod.Delete) allowMethod(HttpMethod.Patch) allowCredentials = true maxAgeInSeconds = 3600 } } Koin 依赖注入 模块定义 // di/AppModule.kt val appModule = module { // Database single<Database> { DatabaseFactory.create( get ()) } // Repositories single<UserRepository> { ExposedUserRepository( get ()) } single<OrderRepository> { ExposedOrderRepository( get ()) } // Services single { UserService( get ()) } single { OrderService( get (), get ()) } single { AuthService( get (), get ()) } } // Application setup fun Application. configureDI () { install(Koin) { modules(appModule) } } 在路由中使用 Koin fun Route. userRoutes () { val userService by inject<UserService>() route( "/users" ) { get { val users = userService.getAll() call.respond(ApiResponse.ok(users)) } } } 用于测试的 Koin class UserServiceTest : FunSpec (), KoinTest { override fun extensions () = listOf(KoinExtension(testModule)) private val testModule = module { single<UserRepository> { mockk() } single { UserService( get ()) } } private val repository by inject<UserRepository>() private val service by inject<UserService>() init { test( "getUser returns user" ) { coEvery { repository.findById( "1" ) } returns testUser service.getById( "1" ) shouldBe testUser } } } 请求验证 // Validate request data in routes fun Route. userRoutes () { val userService by inject<UserService>() post( "/users" ) { val request = call.receive<CreateUserRequest>() // Validate require(request.name.isNotBlank()) { "Name is required" } require(request.name.length <= 100 ) { "Name must be 100 characters or less" } require(request.email.matches(Regex( ".+@.+\\..+" ))) { "Invalid email format" } val user = userService.create(request) call.respond(HttpStatusCode.Created, ApiResponse.ok(user)) } } // Or use a validation extension fun CreateUserRequest. validate () { require(name.isNotBlank()) { "Name is required" } require(name.length <= 100 ) { "Name must be 100 characters or less" } require(email.matches(Regex( ".+@.+\\..+" ))) { "Invalid email format" } } WebSocket fun Application. configureWebSockets () { install(WebSockets) { pingPeriod = 15. seconds timeout = 15. seconds maxFrameSize = 64 * 1024 // 64 KiB — increase only if your protocol requires larger frames masking = false // Server-to-client frames are unmasked per RFC 6455; client-to-server are always masked by Ktor } } fun Route. chatRoutes () { val connections = Collections.synchronizedSet<Connection>(LinkedHashSet()) webSocket( "/chat" ) { val thisConnection = Connection( this ) connections += thisConnection try { send( "Connected! Users online: ${connections.size} " ) for (frame in incoming) { frame as ? Frame.Text ?: continue val text = frame.readText() val message = ChatMessage(thisConnection.name, text) // Snapshot under lock to avoid ConcurrentModificationException