Coroutines – a better match for Kotlin than Reactive Streams?

Using Reactive Programming is tempting because it promises to handle higher traffic on the same hardware. Yet this paradigm is also demanding: it increases code complexity and may make bugs harder to notice. Let’s explore how you can use Kotlin and coroutines to reduce the cognitive load of maintaining reactive applications.

Sample code

Imagine you are developing a website that can be used by both logged-in and anonymous users, just like Wikipedia. At the same time, there are proper mechanisms in place to block misbehaving users. Our focus will be the back-end code returning a welcome message displayed on top of the main page:

class WelcomeService {
    fun welcome(usernameOrIp: String): WelcomeMessage {
        val userProfile: UserProfile? = findUserProfile(usernameOrIp)
        val block: Block? = findBlock(usernameOrIp)
        return generateWelcome(usernameOrIp, userProfile, block)
    }
}

The code follows the principles of ’traditional’ blocking style. It’s simple – the business logic allows the user to provide their personal data like the first and last name, and the user can sometimes be blocked by administrators. Nullable types express this clearly. The generateWelcome logic is not relevant to this article, but let’s examine it briefly:

fun generateWelcome(
    usernameOrIp: String,
    userProfile: UserProfile?,
    block: Block?
): WelcomeMessage =
    when {
        block != null -> WelcomeMessage(
            "You are blocked. Reason: ${block.reason}",
            WARNING
        )
        else -> WelcomeMessage(
            "Hello ${userProfile?.fullName ?: usernameOrIp}",
            INFO
        )
    }

Again, null handling in Kotlin allows expressing our intent without too much ceremony.

Reactive code is complex

Now, let’s try to move the service code to the reactive world (using Reactor library):

fun welcome(usernameOrIp: String): Mono<WelcomeMessage> {
    return userProfileRepository.findById(usernameOrIp)
        .zipWith(blockRepository.findById(usernameOrIp))
        .map { tuple ->
            generateWelcome(usernameOrIp, tuple.t1, tuple.t2)
        }
}

It uses reactive repositories, so userProfileRepository.findById() returns Mono<UserProfile>. blockRepository behaves similarly. The two reactive streams are joined using a ‘zip’ operator that emits a Tuple (pair).

One thing that becomes immediately obvious is that even such a simple piece of reactive code becomes complicated and is structured in a completely different way than the initial one. There is another problem with the snippet above. It’s incorrect.

What’s worse, the problem does not manifest itself openly. You probably won’t see any exception. If the result of welcome is returned from a Spring REST controller, sometimes the REST endpoint will respond as expected, but on other occasions it will serve HTTP 200 with an empty body.

What is happening then? The documentation of the ‘zip’ operator says that if any input is empty then the resulting reactive stream is cancelled. If either the user profile or user block is not be found, generateWelcome won’t be executed. Previously, in the blocking and imperative version of this code, the compiler warned us to handle the case when a variable is null. Using reactive streams cancels this Kotlin advantage – the programmer needs to remember about the ’empty’ case, there is no automatic hint.

A fixed version of the code above may look as follows:

fun welcome(usernameOrIp: String): Mono<WelcomeMessage> {
    return userProfileRepository.findById(usernameOrIp)
        .map { Optional.of(it) }
        .defaultIfEmpty(Optional.empty())
        .zipWith(blockRepository.findById(usernameOrIp)
            .map { Optional.of(it) }
            .defaultIfEmpty(Optional.empty()))
        .map { tuple ->
            generateWelcome(
                usernameOrIp, tuple.t1.orElse(null), tuple.t2.orElse(null)
            )
        }
}

It makes the code correct, but at what cost? Most of the contents are now just noise, it’s hard to see the business logic in it. ‘Optional’ and ’t1’ are not words from the domain of this system. This piece isn’t ‘coded in the language of the domain’ (see Kevlin Henney discussing this concept by Dan North more extensively). A person who is not fairly familiar with the system won’t understand what this method does by looking at it – it is not immediately clear what are arguments of generateWelcome, you need to look at the function definition because the call is too cryptic.

You could extract some expressions to local variables:

val userProfile = tuple.t1.orElse(null)

but this would make the code even more Byzantine. Kotlin Destructuring Declaration paired with Reactor Kotlin extensions could alleviate the problem a bit by introducing meaningful names without declaring additional variables:

import reactor.kotlin.core.util.function.component1
import reactor.kotlin.core.util.function.component2

.map { (userProfile, block) ->
    generateWelcome(usernameOrIp, userProfile.orElse(null), block.orElse(null))
}

However, the method still remains quite complicated. In reality, we may have more than two inputs, and each input may be much more verbose than just getting an element from a repository. In this way, using a ‘zip’ operator may produce a method that is orders of magnitude more convoluted than a simplistic example here and such cosmetic changes as properly naming zip inputs or outputs won’t make it any more understandable.

Improving readability with coroutines

Coroutines allow writing concurrent Kotlin code in a different way. The basis is suspended functions, which are programmed like normal functions, without wrapping values in futures or reactive streams, but still allow for non-blocking asynchronous execution. As highlighted by Venkat Subramaniam in his Exploring Coroutines in Kotlin talk, coroutines allow writing concurrent code structured similarly to traditional imperative sequential code.

There is a possibility to bridge reactive streams and coroutines using kotlinx-coroutines-reactive library prepared by the Kotlin team. We have already seen how easy it is to miss an empty Mono. awaitFirstOrNull can translate a Mono instance into an instance of unboxed nullable type:

val userProfile: UserProfile? =
    userProfileRepository.findById(usernameOrIp).awaitFirstOrNull()

The transformation is non-blocking, so the whole welcome function has to become a suspended function:

suspend fun welcome(usernameOrIp: String): WelcomeMessage {
    val userProfile = userProfileRepository.findById(usernameOrIp).awaitFirstOrNull()
    val block = blockRepository.findById(usernameOrIp).awaitFirstOrNull()
    return generateWelcome(usernameOrIp, userProfile, block)
}

Note how similar this code is to the initial, sequential and blocking version:

class WelcomeService {
    fun welcome(usernameOrIp: String): WelcomeMessage {
        val userProfile: UserProfile? = findUserProfile(usernameOrIp)
        val block: Block? = findBlock(usernameOrIp)
        return generateWelcome(usernameOrIp, userProfile, block)
    }
}

Coroutines allow asynchronous code to leverage the power of Kotlin in handling nullability. If you change the generateWelcome signature to take non-null parameters, suspending welcome will fail to compile.

Spring WebFlux already supports receiving results from suspended functions. Here is a sample code:

@RestController
class WelcomeController(
    private val welcomeService: WelcomeService
) {
    @GetMapping("/welcome")
    suspend fun welcome(@RequestParam ip: String) =
        welcomeService.welcome(ip)
}

For detailed information on how coroutines are supported, read the Kotlin support section in Spring Framework documentation.

What if you still want to return a reactive stream? You can use suspending functions internally to be able to easily handle nullability and build a Mono instance using the mono function from the kotlinx-coroutines-reactor library:

fun welcome(usernameOrIp: String): Mono<WelcomeMessage> {
    return mono {
        val userProfile = userProfileRepository.findById(usernameOrIp).awaitFirstOrNull()
        val block = blockRepository.findById(usernameOrIp).awaitFirstOrNull()
        generateWelcome(usernameOrIp, userProfile, block)
    }
}

The kotlinx-coroutines-rx2 library offers similar conversions for RxJava 2 types.

Conclusion

Using reactive streams in Kotlin cancels one of its biggest advantages: nullability handling. Handling the absence of value becomes verbose and prone to errors.

Coroutines offer another way of building asynchronous and non-blocking applications. Suspending functions are written similarly to regular functions, so they don’t require writing code in a completely different way, unlike in reactive streams. Handling exceptions and nullability are done like in a sequential Kotlin code, so you can count on the compiler to catch your mistakes.

A mixed approach is possible: reactive streams can be converted to suspended functions, and reactive streams can be produced by calling suspended functions. Spring WebFlux supports both returning reactive streams from REST controller methods as well as controller methods that are suspended functions.