From synchronous to asynchronous code with Kotlin's coroutines

TL;DR Kotlin provides non-intrusive, blocking-code-friendly primitives to highly parallel or concurrent code. Async code still reads like sequential code, this means less developer cost for writing and maintaining highly-concurrent applications, which means happier, fitter, more social programmers... 😶 Okay...let's not get carried away 😄

Kotlin makes it easy for developers to move from synchronous, blocking code to a more efficient, async code. The language provides many primitives to deal with the different use-cases that you might encounter, for example:

• When we execute procedure and do not care about the result
• When we want to execute many non-blocking procedures at once and wait for all of them to finish.
• Many subroutines updating state. (See channels

The setup

• Create a empty Kotlin project in IntelliJ IDEA
• Create Main.kt file and add main function. (test with hello world)
• Add klaxon-5.4.jar to the project's libraries
• Add kotlinx-coroutines-core-1.2.2.jar

We will be requesting posts and users from a dummy REST endpoint @ https://jsonplaceholder.typicode.com

Synchronous code

Here is the version of the code that blocks whenever we call the REST endpoints.

import com.beust.klaxon.Klaxon
import java.lang.Exception
import java.net.URL
import kotlin.system.measureTimeMillis

data class Geo(
        val lat: String,
        val lng: String
)

data class Company(
        val name: String,
        val catchPhrase: String,
        val bs: String
)

data class Address(
        val street: String,
        val suite: String,
        val city: String,
        val zipcode: String,
        val geo: Geo
)

data class User(
        val id: Int,
        val name: String,
        val username: String,
        val email: String,
        val address: Address,
        val phone: String,
        val website: String,
        val company: Company
)

data class PostDto(
        val user: User?,
        val id: Int,
        val title: String,
        val body: String
)

data class Post(
        val userId: Int,
        val id: Int,
        val title: String,
        val body: String
)

fun main() {
    val baseUrl = "https://jsonplaceholder.typicode.com";

    fun getPosts(url: String): List<Post> {
        val jsonText = URL(url).readText()

        return try {
            Klaxon().parseArray<Post>(jsonText) ?: emptyList()
        } catch (e: Exception) {
            emptyList<Post>()
        }
    }

    fun getUser(url: String): User? {
        val jsonText = URL(url).readText()

        return try {
            Klaxon().parse<User>(jsonText)
        } catch (e: Exception) {
            null
        }
    }

    val postsWithUsers = mutableListOf<PostDto>()

    val time = measureTimeMillis {
        val posts = getPosts("${baseUrl}/posts")

        posts.map { PostDto(
                user = getUser("${baseUrl}/users/${it.userId}"),
                id = it.id,
                title = it.title,
                body = it.body)
        }.forEach { postsWithUsers.add(it) }
    }

    for (postDto in postsWithUsers) {
        println(postDto)
    }
    println("Posts count: " + postsWithUsers.size)
    println("It took ${time/ 1000.0} seconds")
}

Output:

...json...
It took 7.825 seconds

That, as we can see from the time measure, results in a slow completion of the program.
We have 100 posts and we fetch the user for each post. That is 101 requests total. Results may vary based on your internet connection.

Async code

We can observe and deduct from the code that our program is mostly waiting for network IO. Also, each post is independent from the others, so it is not efficient to wait for one to finish before sending another request.

Let's see how we can use Kotlin-style async/await, similar to JavaScript's constructs with the same names.
The operators give us the ability to fan-out many requests and then wait for the response with code structure not very different from the blocking version.
Here is the snipped that changed:

val time = measureTimeMillis {
    val posts = getPosts("${baseUrl}/posts")

    val postsWithUser: List<Pair<Post, Deferred<User?>>> = posts
            .map {
                Pair(it, GlobalScope.async(Dispatchers.IO) {
                    getUser("${baseUrl}/users/${it.userId}")
                })
            }

    runBlocking {
        postsWithUser.map { PostDto(
                user = it.second.await(),
                id = it.first.id,
                title = it.first.title,
                body = it.first.body)
        }.forEach { postsWithUsers.add(it) }
    }
}

Output:

...some json data...
Posts count: 100
It took 1.858 seconds

A lot better!

Let's break it down:

1. First we are mapping each post to a Pair which holds the post itself and a Deferred object, which is the same as a promise. It signifies that we sent a request for some resource but we are not going to wait for the result, we just keep a handle for it.
2. Now that we have sent all requests for users, without waiting for a response (async block), they are taken care by the REST service. We use a coroutine pool optimized for IO operations.
3. We still want to have the results in the same method for our purposes. So we must block the current thread until all user requests are completed. For that we have runBlocking primitive.
4. await() will block the thread until a result is available.

What is next?

What if some of the requests for user fails? We may need retry logic.
Can we limit the parallelism so we do not DDOS the service we are calling?

Stay safe and happy coding! 🔥
P.S. I love Java, but I appreciate what the guys @ JetBrains did in Kotlin to support concurrency.