Using Kotlin Coroutines with Spring
Before Spring 5.2, you can experience Kotlin Coroutines by the effort from community, eg. spring-kotlin-coroutine on Github. There are several features introduced in Spring 5.2, besides the functional programming style introduced in Spring MVC, another attractive feature is that Kotlin Coroutines is finally got official support.
Kotlin coroutines provides an alternative approach to write asynchronous applications with Spring Reactive stack, but coding in an imperative style.
In this post, I will rewrite my previous reactive sample using Kotlin Coroutines with Spring.
Generate a Spring Boot project using Spring initializr. Choose the following options in the Web UI, others accept the default options.
- Language: Kotlin
- Spring Boot version : 2.2.0.BUILD-SNAPSHOT
- Dependencies: Web Reactive
Extract the files into your local disk. Open the pom.xml file in the project root folder, add some modification to get Kotlin Coroutines work in this project.
Add kotlin-coroutines related dependencies in the dependencies section.
<dependency>
<groupId>org.jetbrains.kotlinx</groupId>
<artifactId>kotlinx-coroutines-core</artifactId>
<version>${kotlinx-coroutines.version}</version>
</dependency>
<dependency>
<groupId>org.jetbrains.kotlinx</groupId>
<artifactId>kotlinx-coroutines-reactor</artifactId>
<version>${kotlinx-coroutines.version}</version>
</dependency>Define a kotlin-coroutines.version in the properties using the latest 1.2.0 version.
<kotlinx-coroutines.version>1.2.0</kotlinx-coroutines.version>Kotlin coroutines 1.2.0 is compatible with Kotlin 1.3.30, define a kotlin.version property to override the default value in the parent BOM.
<kotlin.version>1.3.30</kotlin.version>Currently Spring Data project is busy in adding Kotlin Coroutines support. At the moment Spring Data R2DBC got basic coroutines support in its DatabaseClient. In this sample, we use Spring Data R2DBC for data operations.
Add Spring Data R2DBC related dependencies, and use PostgresSQL as the backend database.
<dependency>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-r2dbc</artifactId>
<version>${spring-data-r2dbc.version}</version>
</dependency>
<dependency>
<groupId>org.postgresql</groupId>
<artifactId>postgresql</artifactId>
</dependency>
<dependency>
<groupId>io.r2dbc</groupId>
<artifactId>r2dbc-postgresql</artifactId>
</dependency>Declare a spring-data-r2dbc.version property to use latest Spring Data R2DBC .
<spring-data-r2dbc.version>1.0.0.BUILD-SNAPSHOT</spring-data-r2dbc.version>Enables Data R2dbc support by subclassing AbstractR2dbcConfiguration.
@Configuration
@EnableR2dbcRepositories
class DatabaseConfig : AbstractR2dbcConfiguration() {
override fun connectionFactory(): ConnectionFactory {
return PostgresqlConnectionFactory(
PostgresqlConnectionConfiguration.builder()
.host("localhost")
.database("test")
.username("user")
.password("password")
.build()
)
}
}Create a Kotlin data class, annotate it with @Table which indicates it is mapped to the table posts.
@Table("posts")
data class Post(@Id val id: Long? = null,
@Column("title") val title: String? = null,
@Column("content") val content: String? = null
)Follow the Reactive stack to Kotlin Coroutines translation guide provided in Spring reference documentation, we create a repository class for CRUD operations of Post.
@Component
class PostRepository(private val client: DatabaseClient) {
suspend fun count(): Long =
client.execute().sql("SELECT COUNT(*) FROM posts")
.asType<Long>().fetch().awaitOne()
fun findAll(): Flow<Post> =
client.select().from("posts").asType<Post>().fetch().flow()
suspend fun findOne(id: Long): Post? =
client.execute()
.sql("SELECT * FROM posts WHERE id = \$1")
.bind(0, id).asType<Post>()
.fetch()
.awaitOneOrNull()
suspend fun deleteAll() =
client.execute()
.sql("DELETE FROM posts")
.fetch()
.rowsUpdated()
.awaitSingle()
suspend fun save(post: Post) =
client.insert()
.into<Post>()
.table("posts")
.using(post)
.await()
suspend fun init() {
save(Post(title = "My first post title", content = "Content of my first post"))
save(Post(title = "My second post title", content = "Content of my second post"))
}
}It is easy to understand the above codes, for the return type of these functions.
- The
Flux<T>is changed to Kotlin CoroutinesFlowtype. - The
Monotype is unboxed to parameterized type, and add asuspendmodifier to the function. If the return result can be null, add a?to the return type, eg.Post?. - The
awaitXXXorflow()converts the Reactive APIs to Kotlin Coroutines world.
Create a Controller class for Post, Kotlin Coroutines is also supported in the annotated controllers.
@RestController
@RequestMapping("/posts")
class PostController(
private val postRepository: PostRepository
) {
@GetMapping("")
fun findAll(): Flow<Post> =
postRepository.findAll()
@GetMapping("{id}")
suspend fun findOne(@PathVariable id: Long): Post? =
postRepository.findOne(id) ?: throw PostNotFoundException(id)
@PostMapping("")
suspend fun save(@RequestBody post: Post) =
postRepository.save(post)
@GetMapping("{id}/comments")
fun findCommentsByPostId(@PathVariable id: Long): Flow<Comment> =
commentRepository.findByPostId(id)
}You can also initialize data in a CommandLineRunner bean or listen the @ApplicationReadyEvent, use a runBlocking to wrap coroutines tasks.
runBlocking {
val deleted = postRepository.deleteAll()
println(" $deleted posts was cleared.")
postRepository.init()
}To run the application successfully, make sure there is a running PostgreSQL server. I prepared a docker compose file to simply run a PostgresSQL server and initialize the database schema in a docker container.
docker-compose upRun the application now, it should work well as the previous Reactive examples.
In additional to the annotated controllers, Kotlin Coroutines is also supported in functional RouterFunction DSL using the coRouter to define your routing rules.
@Configuration
class RouterConfiguration {
@Bean
fun routes(postHandler: PostHandler) = coRouter {
"/posts".nest {
GET("", postHandler::all)
GET("/{id}", postHandler::get)
POST("", postHandler::create)
PUT("/{id}", postHandler::update)
DELETE("/{id}", postHandler::delete)
}
}
}Like the changes in the controller, the PostHandler is written in an imperative style.
@Component
class PostHandler(private val posts: PostRepository) {
suspend fun all(req: ServerRequest): ServerResponse {
return ok().bodyAndAwait(this.posts.findAll())
}
suspend fun create(req: ServerRequest): ServerResponse {
val body = req.awaitBody<Post>()
val createdPost = this.posts.save(body)
return created(URI.create("/posts/$createdPost")).buildAndAwait()
}
suspend fun get(req: ServerRequest): ServerResponse {
println("path variable::${req.pathVariable("id")}")
val foundPost = this.posts.findOne(req.pathVariable("id").toLong())
println("found post:$foundPost")
return when {
foundPost != null -> ok().bodyAndAwait(foundPost)
else -> notFound().buildAndAwait()
}
}
suspend fun update(req: ServerRequest): ServerResponse {
val foundPost = this.posts.findOne(req.pathVariable("id").toLong())
val body = req.awaitBody<Post>()
return when {
foundPost != null -> {
this.posts.update(foundPost.copy(title = body.title, content = body.content))
noContent().buildAndAwait()
}
else -> notFound().buildAndAwait()
}
}
suspend fun delete(req: ServerRequest): ServerResponse {
val deletedCount = this.posts.deleteById(req.pathVariable("id").toLong())
println("$deletedCount posts was deleted")
return notFound().buildAndAwait()
}
}Besides annotated controllers and functional router DSL, Spring WebClient also embrace Kotlin Coroutines.
@RestController
@RequestMapping("/posts")
class PostController(private val client: WebClient) {
@GetMapping("")
suspend fun findAll() =
client.get()
.uri("/posts")
.accept(MediaType.APPLICATION_JSON)
.awaitExchange()
.bodyToFlow<Post>()
@GetMapping("/{id}")
suspend fun findOne(@PathVariable id: Long): PostDetails = withDetails(id)
private suspend fun withDetails(id: Long): PostDetails {
val post =
client.get().uri("/posts/$id")
.accept(APPLICATION_JSON)
.awaitExchange().awaitBody<Post>()
val count =
client.get().uri("/posts/$id/comments/count")
.accept(APPLICATION_JSON)
.awaitExchange().awaitBody<Long>()
return PostDetails(post, count)
}
}
In the withDetails method, the post and count call the remote APIs one by one in a sequence.
If you want to perform coroutines in parallel, use async context to wrap every calls, and put all tasks in a coroutineScope context. To build the return result in PostDetails, use await to wait the completion of the remote calls.
private suspend fun withDetails(id: Long): PostDetails = coroutineScope {
val post = async {
client.get().uri("/posts/$id")
.accept(APPLICATION_JSON)
.awaitExchange().awaitBody<Post>()
}
val count = async {
client.get().uri("/posts/$id/comments/count")
.accept(APPLICATION_JSON)
.awaitExchange().awaitBody<Long>()
}
PostDetails(post.await(), count.await())
}Check out the codes from Github.