Follow me on Twitter: https://twitter.com/heyitsmohit

Writing a Kotlin Compiler plugin is an uncharted area. The API is undocumented thus far. You may not know where to start. Enter Arrow Meta. It is a library that provides a functional API to write compiler plugins. In this article, I will show you how to write and test a simple compiler plugin with Arrow Meta.

Three years ago, a friend of mine Kevin Most gave a talk at Kotlin Conf on writing compiler plugins. He demonstrated the process with an example plugin DebugLog. I worked on writing an Arrow Meta implementation of this plugin debuglog-meta-arrow. I’ll show you how I wrote it and introduce the API provided by Arrow Meta.

Before we begin, let’s look at an overview of how the Kotlin compiler works as Arrow Meta supplements it.

Kotlin Compiler

As with any compiler, the purpose is to convert your Kotlin code to something that can be executed. From a higher-level perspective, the compiler is divided into two sections — frontend and backend. The front-end performs lexical and grammatical analysis. This produces what is called an intermediate representation that typically takes the form of a tree. An Abstract Syntax Tree (AST) to be precise. Let’s look at an overview of the steps taken in these sections.

Compilation

Your code goes through a sequence of steps to compile. Below is a diagram of the process.

Code

Code is really just strings that we input into our editors. The editor that we primarily use, Intellij IDEA is built on top of the Intellij Community platform. This platform provides reusable UI components for IDEs and a way to hookup a custom grammar for a language, it also has grammar validation, etc. The Kotlin Compiler utilizes this platform to provides Kotlin specific configurations for the IDE. For example, the “Show Kotlin bytecode” option is shown by tapping into the Actions API provided by Intellij platform. This is hooked into the decompiler of Kotlin.

Parsing

Your code is converted from text to a tree-like structure called the Abstract Syntax Tree. This is actually not Kotlin specific. All compilers will transform your code into an AST for any other language. Each node in the AST represents a part of your code. For example, there will be a node for your imports, classes, functions, generic, etc.

Analysis & Resolve

In this phase of the compilation process, the AST is given type information and actual type checking is performed. Every language has a grammar. This is Kotlin’s grammar specification. Your code is parsed against the grammar to validate its correctness. In this phase, it will also type check your program.

Code Generation

The AST is given to the backend of the compiler. With the introduction of intermediate representation, the back-end can now target different platforms. These platforms could be JS, JVM, native.

All compilers have a very similar process. This is a very high-level overview of it. Arrow Meta allows you to hook into each step and provide your own custom implementation. The GIF below made by the Arrow team demonstrates this much better.

Use Case

Let’s look at how to write a plugin that mutates the AST and adds logging information. Assume we have a method that performs some operation and returns the result.

fun prime(n: Int): Long = listOf(1L,2L,3L).take(n).last()

We want to be able to log how long this function takes to execute. We have a program that has a lot of functions. We could use an annotation DebugLog to specify logging for a specific function.

annotation class DebugLog

@DebugLog
fun prime(n: Int): Long = listOf(1L,2L,3L).take(n).last()

Our goal is to transform this function to add logging for how long it takes the function to execute. After transforming our function, it will look like this:

fun prime(n: Int): Long {
  println(“-> prime(n=$n)”)
  val startTime = System.currentTimeMillis()
  val result = listOf(1L,2L,3L).take(n).last()
  val timeToRun = System.currentTimeMillis() — startTime
  println(“<- prime[ran in $timeToRun ms]”)
  return result
}

As you can see above, we want to add logging of the parameter sent to the function, perform a time calculation and print it. How do we accomplish this with Arrow Meta?

Modules

We need to setup Gradle modules for our meta plugin. I will create two modules — create-plugin and use-plugin in our project.

Create-Plugin

This module will contain the main logic for the plugin. Its external dependencies will be:

• Kotlin Embeddable Compiler
• Arrow Meta Compiler Plugin
• Shadow Jar library

Please take look at the build scripts in the project. I want to point out the main points of the configuration.

plugins {
   id “org.jetbrains.kotlin.jvm”
   id “com.github.johnrengelman.shadow”
}

dependencies {
   compileOnly “org.jetbrains.kotlin:
                     kotlin-stdlib:$KOTLIN_VERSION”

   compileOnly “org.jetbrains.kotlin:
                     kotlin-compiler-embeddable:$KOTLIN_VERSION”

   compileOnly “com.github.arrow-kt.arrow-meta:
                     compiler-plugin:-SNAPSHOT”

}

shadowJar {
    configurations = [project.configurations.compileOnly]
    dependencies {
       exclude(“org.jetbrains.kotlin:kotlin-stdlib”)
       exclude(“org.jetbrains.kotlin:kotlin-compiler-embeddable”)
    }
}

In my plugin, I’m using JitPack to get the latest snapshot of Arrow Meta.

Use-Plugin

This module shows how to use the plugin in an example. The create-plugin module creates a jar that is included in freeCompilerArgs . This module will also contain the tests for our plugin. Here is the main configuration from the build script.

compileKotlin {
   kotlinOptions {
      jvmTarget = “$JVM_TARGET_VERSION”
      freeCompilerArgs = [
           “-Xplugin=${project.rootDir}/create
                 plugin/build/libs/create-plugin-all.jar”]
     }
}

For testing, I’m including the test plugin that provides testing utilities. With Jitpack, it’s easier to pull a submodule from the library.

testImplementation 
        ‘com.github.arrow-kt.arrow-meta:testing-plugin:-SNAPSHOT’

If you want to look at this scaffolding in detail, you could check out my plugin.

Compiler Plugin Entry Point

The main entry point for the plugin needs to specify a list of extension phases. The extension phases are our hooks into the compiler we are specifying to perform code gen, type definition, etc.

class DebugLogMetaPlugin : Meta {

    @ExperimentalContracts
    override fun intercept(ctx: CompilerContext): List<Plugin> =
        listOf(
            debugLog
        )
}

In order to define the entry point for our plugin, we need to extend the Meta class and override the intercept method. We will return from this method a list of typePlugin . The debugLog item in the list is our Plugin where we have defined how to perform our desired transformation. Let’s see how the transformation is connected.

Compiler Plugin

val Meta.debugLog: Plugin
 get() =
   "DebugLog" {
     meta(
       namedFunction({ validateFunction() }) { c: KtNamedFunction ->
          Transform.replace(
            replacing = c,
            newDeclaration = replace(c).function
          )
        }
      )
    }

Above is a plugin defined as an extension property. A Plugin is a data class consisting of two fields.

data class Plugin(  
    val name: String,  
    val meta: CompilerContext.() -> List<ExtensionPhase>
)

Source: Meta.kt

The first field specifies the name of our plugin which is “DebugLog”. The second field is a lambda block whose return type is a list of extension phases. We would like to intercept the compilation step where the compiler generates the AST. We need to search the AST for a function that has the DebugLog annotation. Arrow Meta provides an extension on the Meta type that allows you to parse any function defined in the code.

fun Meta.namedFunction(
  match: KtNamedFunction.() -> Boolean,
  map: NamedFunction.(KtNamedFunction) -> Transform<KtNamedFunction>
): ExtensionPhase =
  quote(match, map) { NamedFunction(it) }

Source: MetaExtensions.kt

This method allows us to specify a predicate to match a method and a transformation for it. We need to validate a method that has this signature.

@DebugLog
fun prime(n: Int): Long

I created an extension validateFunction on KtNamedFunction that checks whether the function is annotated with DebugLog , has an Int params and a Long return type.

/**
 * Match a function that fits this scenario.
 *
 * - Function has 1 param that is of Int type.
 * - Function has a Long return type.
 * - Function has a DebugLog annotation.
 */
private fun KtNamedFunction.validateFunction(): Boolean =
        hasOneIntParam() && 
        hasLongReturnType() && 
        hasAnnotation(DEBUG_LOG)

/**
 * Check function has 1 param that is of Int type.
 */
private fun KtNamedFunction.hasOneIntParam(): Boolean =
        valueParameterList?.parameters?.size == 1 &&
          valueParameterList?.parameters?.first()
                             ?.typeReference?.text == INT

/**
 * Check function is returning a Long.
 */
private fun KtNamedFunction.hasLongReturnType(): Boolean =
        hasDeclaredReturnType() && typeReference?.text == LONG

/**
 * Check function has [annotationNames] as an annotation.
 */
fun KtAnnotated.hasAnnotation(
        vararg annotationNames: String
): Boolean {
    val names = annotationNames.toHashSet()
    val predicate: (KtAnnotationEntry) -> Boolean = {
        it.typeReference
                ?.typeElement
                ?.safeAs<KtUserType>()
                ?.referencedName in names
    }
    return annotationEntries.any(predicate)
}

Each requirement that we want to satisfy is specified as an extension. It doesn’t have to be. We check whether there is an annotation on the method by going through any annotationEntries the method may have. We also check if the function has Int param by going through each valueParameterList and checking its typeReference . I could write this to be more general. But, for the sake of simplicity, I constrained it to match a specific signature.

After a method having a DebugLog annotation is found in the AST, we need to transform it. As a reminder, our goal is to transform the function to:

fun prime(n: Int): Long {
  println(“-> prime(n=$n)”)
  val startTime = System.currentTimeMillis()
  val result = listOf(1L,2L,3L).take(n).last()
  val timeToRun = System.currentTimeMillis() — startTime
  println(“<- prime[ran in $timeToRun ms]”)
  return result
}

To accomplish this, I wrote the following method.

fun replace(function: KtNamedFunction): String {
    val functionName = function.name
    val paramName = function.valueParameters.first().name
    val functionBody = function.body()?.bodySourceAsExpression()

    return """
        |//metadebug
        |
        | fun ${functionName}(${paramName}: Int): Long {
        |   println("-> $functionName(${paramName}=$${paramName})")
        |   val startTime = System.currentTimeMillis()
        |   val result = $functionBody
        |   val timeToRun = System.currentTimeMillis() - startTime
        |   println("<- ${functionName}[ran in ${'$'}timeToRun ms]")
        |   return result
        | }"""
}

This method grabs the function name, the first params and the function body first. Then, I return a string that logs the time execution and returns the result from the existing function body.

val Meta.debugLog: Plugin
 get() =
   "DebugLog" {
     meta(
       namedFunction({ validateFunction() }) { c: KtNamedFunction ->
          Transform.replace(
            replacing = c,
            newDeclaration = replace(c).function
          )
        }
      )
    }

We have specified the replacement transformation above. Arrow Meta has support for many types of transformation including removal, new source and transformations could be composed. All of these transforms are extensions on the Transform type.

At this point, we have built our plugin that meets the goal we wanted to accomplish. A function annotated with DebugLog is transformed to log its execution time.

Usage & Output

If we build our project, a jar file will be generated in our create-plugin/build/libs folder. Let’s see our plugin in action. I’m going to create a file with a main method in the use-plugin module.

@DebugLog
fun prime(n: Int): Long = listOf(1L,2L,3L).take(n).last()

fun main() {
  prime(10)
}

If I were to run this method, this will be the output as expected.

-> prime(n=10)
<- prime[ran in 36 ms]

Testing

Arrow meta has thetesting-plugin module that provides utilities for ensuring our plugin functions as we expect it.

class DebugLogMetaPluginTest {

    companion object {

        val debuglog = """
            |annotation class DebugLog
            |
            |@DebugLog
            |fun prime(n: Int): Long =  
            |       listOf(1L,2L,3L).take(n).last()
            | """.trimMargin().trim().source

        val expectedOutput = """
            |annotation class DebugLog
            |
            |//metadebug
            |fun prime(n: Int): Long {
            | println("-> prime(n=${'$'}n)")
            | val startTime = System.currentTimeMillis()
            | val result = listOf(1L,2L,3L).take(n).last()
            | val timeToRun = System.currentTimeMillis() - startTime
            | println("<- prime[ran in ${'$'}timeToRun ms]")
            | return result
            | }""".trimMargin().trim().source
    }
}

For my test above, I have specified two string that presents the input source and expected source after the transformed is applied to the plugin.

@Test
fun `should print function time execution`() {
    assertThis(CompilerTest(
     config = { listOf(addMetaPlugins(DebugLogMetaPlugin())) },
          code = { debuglog },
          assert = { quoteOutputMatches(expectedOutput) }
    ))
}

The testing-plugin module provides a CompileTest that allows you to specify the code you want to transform and what it should assert with. The test above internally creates a file Source.kt that intercepts the analysis phase of the compiler and applies the transformation on my code. Arrow Meta provides helpful logging information as your test runs.

Replacing class arrow.meta.internal.kastree.ast.Node$Decl$Func with [class org.jetbrains.kotlin.psi.KtNamedFunction]: newContents: 
//metadebug
fun prime(n: Int): Long {
   println("-> prime(n=$n)")
   val startTime = System.currentTimeMillis()
   val result = listOf(1L,2L,3L).take(n).last()
   val timeToRun = System.currentTimeMillis() - startTime
   println("<- prime[ran in $timeToRun ms]")
   return result
 }
Transformed file: KtFile: Source.kt. New contents: 
[(KtFile: Source.kt, annotation class DebugLog

//metadebug
fun prime(n: Int): Long {
   println("-> prime(n=$n)")
   val startTime = System.currentTimeMillis()
   val result = listOf(1L,2L,3L).take(n).last()
   val timeToRun = System.currentTimeMillis() - startTime
   println("<- prime[ran in $timeToRun ms]")
   return result
 }
)]
END quote.doAnalysis: [KtFile: Source.kt]

There is a system internally called Quote that drives all the transformation. This is how you test a compiler plugin with Arrow Meta.

Conclusion

I hope that was helpful for you as you create your own Compiler Plugins with Arrow Meta. There is a LOT more you could do. Thanks to Raul Raja, Rachel M. Carmena, Amanda Hinchman and more for creating a very cool tool. This project also contains many examples of how to create a functional API. You could check out my plugin debuglog-arrow-meta. Arrow Meta is still in development. But, I started to use Arrow Meta for complex use cases and I very much enjoy it. If you have any questions, please leave a response.

Resources

• Arrow Meta Examples
• Lambda World 2019 — Arrow Meta — Enabling Functional Programming in the Kotlin Compiler
• DroidCon NYC 2018 — Static Code Analysis with Kotlin

Follow me on Twitter: https://twitter.com/heyitsmohit

Follow me on Twitter: www.twitter.com/heyitsmohit

In February, I gave a presentation at the NY Android Meetup on unit testing Channels and Flows with practical use cases. In this blog post, I will share with you how to test delays, retries, and errors with Flows. I’ve also added useful tips and things to watch out for when unit testing.

Use Case

Before, we could write unit tests. We need a use case. Suppose we are building a feature in an app that displays user details. We may have a setup as shown below.

We have a Repository that goes out to the API and gets the user’s details and gives it to the View Model to display it to the UI. In this article, I will focus on creating the repository and testing it. How would we go about implementing the Repository using a Flow?

Repository

The repository will use an API service to get the user’s details. We could implement this using Retrofit which has support for suspending functions.

interface ApiService {
    @GET("/users/{id}")
    suspend fun userDetails(@Path("id") id: Int): UserDetails
}

In this API service, we are going to hit an endpoint that takes an id and return the data in a UserDetails response.

fun <T> flow(
    block: suspend FlowCollector<T>.() -> Unit
): Flow<T>

Source: Builders.kt

Using theflow method, we could implement our Repository as follows.

class UserRepository(val apiService: ApiService) {
 
    fun userDetails(id: Int): Flow<Result<UserDetails>> {
         return flow {
            val users = apiService.userDetails(id)
            emit(Result.success(users))
         }
    }
}

Tips

This Flow is emitting a single item. I could the single method on the Flow and have the userDetails return Result<UserDetails>. However, the restriction on using Result type is that it can’t be returned from a method. This restriction will be lifted in Kotlin 1.4.0.

This could be implemented without using Flows. But, I want to use a simple example to illustrate testing Flows.

Error Handling

What happens if an error occurs in our example? If the method apiService.userDetails(id) throws an exception, it will cause a crash if not handled properly. Here is an example of collecting from a Flow inside of a coroutine launched in a view model scope. It would crash due to the exception.

viewModelScope.launch {
     val flow = repository.getUserDetails("id")
     flow.collect { result: Result<UserDetails>
          showUI(result.data)
     }
}

There are three ways to handle exceptions.

1. Provide an exception handler to the coroutine in the launch method.
2. Use a try and catch around the getUserDetails method.
3. Catch the exception in the Flow.

All of these approaches are valid depending on your use case. I will show you how to use the third approach.

class UserRepository(val apiService: ApiService) {
 
  fun userDetails(id: Int): Flow<Result<UserDetails>> {
       return flow {
           val users = apiService.userDetails(id)
           emit(Result.success(users))
      }

    .catch { throwable ->
         emit(Result.error(throwable))
     }
  }    
}

The catch extension on a Flow intercepts the exception. We then emit it in a wrapper type like a Result. That’s exactly what I’m doing above using Result.error(throwable) . With this approach, the exception is caught and won’t crash our application. Instead, it will be incorporated into our Result type.

Retry

If an error happens, I may want to retry the request a number of times. The coroutines library provides a retry extension that allows you to specify the number of retries and when to perform it. Using the retry extension, our Repository will now look as follows.

class UserRepository(val apiService: ApiService) {
 
  fun userDetails(id: Int): Flow<Result<UserDetails>> {
       return flow {
           val users = apiService.userDetails(id)
           emit(Result.success(users))
      }
      .retry(2) { e ->
         (e is Exception).also { if (it) delay(1000) }
      }
      .catch { throwable ->
         emit(Result.error(throwable))
     }
}

I am specifying that I want to perform two retries when an exception occurs. However, before retrying the request, I will delay the coroutine for one second. Each retry will execute the Flow block again.

Dispatcher

In order to test delays and have control over them, we need to be able to inject a dispatcher. Currently, our repository will use a dispatcher from the coroutine it is called from. But, it is a good practice to explicitly inject into the repository which dispatcher the Flow will use. The coroutines test library provides a test dispatcher that allows us to control virtual time. We could modify our repository to take in a dispatcher in its constructor.

class UserRepository(
    val apiService: ApiService,
    val dispatcher: CoroutineDispatcher
) {
 
  fun userDetails(id: Int): Flow<Result<UserDetails>> {
       return flow {
           val users = apiService.userDetails(id)
           emit(Result.success(users))
      }
      .retry(2) { e ->
         (e is Exception).also { if (it) delay(1000) }
      }
      .catch { throwable ->
         emit(Result.error(throwable))
     }
     .flowOn(dispatcher)   
  }
}

The injected dispatcher is used in the Flow by using the flowOn extension. This means everything before flowOn method which is the retry, catch and the flow block itself will run on the inject dispatcher.

Tips

A common pattern followed in the community is to inject a provider for the dispatcher. The provider could give you a Dispatchers.IO or Dispatchers.Default.

Unit Tests

How do we unit test the repository? There are three cases we want to test in our example.

1. Flow emits successfully.
2. Flow completes all retries with an error.
3. Flow retries and a successful response occurs.

Test Success

How do we test the simplest case of a successful emission? We need to mock the API service to return data. I will use mockito-kotlin for stubbing our API service.

class UserRepositoryTest {

    val testDispatcher = TestCoroutineDispatcher()
    val apiService = mock<ApiService>()
    val repository = UserRepository(apiService, testDispatcher)
}

Create the repository we will test with that takes a test dispatcher and a mocked API service. In order to write our test case, we have to create a coroutine that our test will run in. There are two methods we could use from the coroutine library to do this.

1. runBlocking
2. runBlockingTest

The runBlocking method is good for testing non-delays while runBlockingTest gives you finer control over virtual time if you need it. For the successful case, we will use runBlocking to create a coroutine.

class UserRepositoryTest {

    val testDispatcher = TestCoroutineDispatcher()
    val apiService = mock<ApiService>()
    val repository = UserRepository(apiService, testDispatcher)

    @Test
    fun `flow emits successfully`() = runBlocking {
         
         // Mock API Service
         val userDetails = UserDetails(1, "User 1", "avatar_url")
         userService.stub {
             onBlocking { userDetails(1) } doReturn userDetails
         }

         // Test
         val flow = repository.getUserDetails(id = 1)

         // Verify
         flow.collect { result: Result<UserDetails> ->
              result.isSuccess.shouldBeTrue()
              result.onSuccess {
                 it shouldBeEqualTo userDetails
              }
         }
     }
}

There are three steps in this test — mock, test and verify. In the mocking test,

Mock API Service

I am stubbing out the API service to return a mock UserDetails response. Since the userDetails method is a suspending function, I cannot use
on { userDetails(id = 1) } doReturn userDetails. It will give an error that a suspending method can only be called from another suspending method. The lambda block of the onmethod is not suspending. Mockito-kotlin provides an onBlocking method that takes in a suspending lambda and stubs the method out in a runBlocking coroutine.

Test & Verify

In the test phase, I get the Flow from the repository, collect from it and verify the data in the result. I am using the library Kluent for verification. It provides methods such as shouldBeTrue and shouldBeEqualTo.

Tips

There is a library called MockK that provides methods to verify Flow assertions. In the community, libraries have also built patterns to performs assertions. Check out the SQL Delight’s library FlowAssert pattern.

Test Retry

How do we test that the Flow handles exceptions and retries? There are two retry cases we want to test.

1. Flow completes all retries with an error.
2. Flow retries and a successful response occurs.

Let’s start with the first test case.

1. Flow completes all retries with an error.

@Test
fun `should retry with error`() = 
  testCoroutineDispatcher.runBlockingTest {
      userService.stub {
        onBlocking { userDetails(1) } doAnswer {
            throw IOException()
        }
    }

    val flow = repository.getUserDetails(id = 1)

    flow.collect { result: Result<UserDetails> ->
        result.isFailure.shouldBeTrue()
    }
}

In our Repository, when an error occurs, we delay for 1 second before executing the flow block again.

flow {
    val users = apiService.userDetails(id)
    emit(Result.success(users))
}.retry(2) { e ->
    (e is Exception).also { if (it) delay(1000) }
}.catch { throwable ->
     emit(Result.error(throwable))
}.flowOn(dispatcher)

As we have delays in our code, we will use therunBlockingTest method in our test.

What does runBlockingTest do?

This method creates a coroutine with a TestCoroutineScope that has a TestCoroutineDispatcher and a TestCoroutineExceptionHandler . This method takes in a lambda block testBody and an optional context .

fun runBlockingTest(
   context: CoroutineContext = EmptyCoroutineContext, 
   testBody: suspend TestCoroutineScope.() -> Unit
)

In my test, I am using this method as

testCoroutineDispatcher.runBlockingTest {
    ...
}

This will create a coroutine with our own customtestCoroutineDispatcher.

This is equivalent to:

runBlockingTest(testCoroutineDispatcher) {
    ...
}

However, an extension is provided on runBlockingTest that I am using to provide my own dispatcher.

fun TestCoroutineDispatcher.runBlockingTest(
    block: suspend TestCoroutineScope.() -> Unit
)

Source: TestBuilders.kt

After creating the coroutine my test will run in, I am mocking my API service to throw an exception. Then, I proceed to collect from my Flow and perform an assertion.

@Test
fun `should retry with error`() = 
  testCoroutineDispatcher.runBlockingTest {

      // Mock
      userService.stub {
         onBlocking { userDetails(1) } doAnswer {
             throw IOException()
         }
     }

      // Test
      val flow = repository.getUserDetails(id = 1)

      // Verify
      flow.collect { result: Result<UserDetails> ->
         result.isFailure.shouldBeTrue()
      }
}

Let’s trace how this test runs. If I add logs or add breakpoints in the IDE, it will help visualize what happens.

fun getUserDetails(id: Int): Flow<Result<UserDetails>> {
    return flow {
        println("STARTING FLOW")
        val users = userService.userDetails(id)
        emit(Result.success(users))
    }
    .retry(retries = 2) { t ->
        println("RETRY")
        (t is IOException).also { 
            if (it) { 
                println("BEFORE DELAY")
                delay(DELAY_ONE_SECOND)
                println("AFTER DELAY")
            }
        }
    }
    .catch {
        println("CATCH")
        emit(Result.failure(it))
    }.flowOn(dispatcher)
}

Here is the output of running the test.

// Initial Run of Flow
STARTING FLOW

// 1st Retry
RETRY
BEFORE DELAY
AFTER DELAY
STARTING FLOW

// 2nd Retry
RETRY
BEFORE DELAY
AFTER DELAY
STARTING FLOW

// Both retries failed run catch block
CATCH

Initial Run of Flow

When I collect from the Flow in the test, it will execute the flow block and an exception will be thrown since we used doAnswer to train our Api Service to throw an IOException.

1st and 2nd Retry

It will then go into the retry block and delay for one second. After the delay, it will execute the Flow block again. An exception will occur again. This process will repeat itself for the 2nd retry.

Catch Error

Since we only specified two retries, it will finally halt the retries and go into the catch block since all retries failed and emit the error.

Under the hood

Under the hood, the TestCoroutineDispatcher has a queue which is an event loop.

// The ordered queue for the runnable tasks.
private val queue = ThreadSafeHeap<TimedRunnable>()

Source: TestCoroutineDispatcher

Every time it dispatches, it adds an TimedRunnable object to the queue. This object contains the Runnable task and a time to perform the task.

private fun post(block: Runnable) =              
     queue.addLast(
        TimedRunnable(block, _counter.getAndIncrement())
     )

class TimedRunnable(
    @JvmField val runnable: Runnable,
    private val count: Long = 0,
    @JvmField val time: Long = 0
)

Source: TimedRunnable

As time is advanced forward on the test dispatcher, the Runnable tasks are executed.

When you look at this test, I am not explicitly advancing time forward. How does it move virtual time forward?

@Test
fun `should retry with error`() = 
  testCoroutineDispatcher.runBlockingTest {
      userService.stub {
        onBlocking { userDetails(1) } doAnswer {
            throw IOException()
        }
    }

    val flow = repository.getUserDetails(id = 1)

    flow.collect { result: Result<UserDetails> ->
        result.isFailure.shouldBeTrue()
    }
}

Let’s look under the hood at the implementation of runBlockingTest . You will see that it will advance time forward until there are no more runnables to execute for all of your test.

val startingJobs = safeContext.activeJobs()
val scope = TestCoroutineScope(safeContext)
val deferred = scope.async {
    scope.testBody()
}
dispatcher.advanceUntilIdle()
deferred.getCompletionExceptionOrNull()?.let {
    throw it
}
scope.cleanupTestCoroutines()
val endingJobs = safeContext.activeJobs()
if ((endingJobs - startingJobs).isNotEmpty()) {
    throw UncompletedCoroutinesError(
         "Test finished with active jobs: $endingJobs"
    )
}

Source: TestCoroutineDispatcher

dispatcher.advanceUntilIdle() will advance the delays in our tests. In my tests, I am leveraging this to go through all the retries with delays and emit an error. This is how it works under the hood.

Tips:

The most common error you will get is Test finished with active jobs error when you use runBlockingTest . This happens, because you have an active coroutine running in your test. For example, if you had a producer using a Rendezvous Channel and you had called send and didn’t receive the value, you will get this error. That’s only one example of it. So, watch out for that error.

Test Retry with Success

2. Flow retries and a successful response occurs.

How do we test that a retry actually succeeds and emits the data?

@Test
fun `should retry with success`() 
    testCoroutineDispatcher.runBlockingTest {    
        
       // Mock
       var throwError = true
       val userDetails = UserDetails(1, "User 1", "avatar_url")

       userService.stub {
           onBlocking { userDetails(1) } doAnswer {
            if (throwError) throw IOException() else userDetails
          }
        }

       // Test
       val flow = repository.getUserDetails(id = 1)

       // Verify
       launch {
          flow.collect { result ->
              result.isSuccess.shouldBeTrue()
          }
       }

       // 1st Retry
       advanceTimeBy(DELAY_ONE_SECOND)

       // 2nd Retry
       throwError = false
       advanceTimeBy(DELAY_ONE_SECOND)
}

In this test, I am controlling whether an exception will be thrown or a successful response be given by using thethrowError variable.

var throwError = true
val userDetails = UserDetails(1, "User 1", "avatar_url")

userService.stub {
   onBlocking { userDetails(1) } doAnswer {
      if (throwError) throw IOException() else userDetails
   }
}

I am collecting from the Flow in a child coroutine. This coroutine is the consumer. If I didn’t launch a child coroutine, this test would fail.

launch {
  flow.collect { result ->
     result.isSuccess.shouldBeTrue()
   }
}

I want to be able to advance time forward when it encounters a delay so that the first retry fails. But, I want the second retry to pass. I will setthrowError to false so that when the flowblock is executed it succeeds with data.

// 1st Retry
advanceTimeBy(DELAY_ONE_SECOND)

// 2nd Retry
throwError = false
advanceTimeBy(DELAY_ONE_SECOND)

Tips:

Watch out for an undocumented edge case of runBlockingTest . If you wrote a test that has multiple errors, runBlockingTest will drop the first one. See TestCoroutineExceptionHandler.

Resources

• Unit Testing Channels & Flow in Practice
• Testing with Coroutines

I hope that was helpful to you. If you practice writing lots of tests using runBlockingTest for different scenarios, then it will get easier. If you have any questions, please let me know in the response. I gave more examples in my presentation. Please check out my slides on Unit Testing Channels & Flow in Practice.

Follow me on Twitter: www.twitter.com/heyitsmohit

Follow me on Twitter https://twitter.com/heyitsmohit

How do we combine emissions from multiple Flows? We could use operators such as zip, flattenMerge and combine. This article will explore how to use the combine operator and how it works under the hood. Interestingly, it uses an experimental construct called a Select expression. We will look at how the select expressions work and how they are used internally in the combine operator.

Combining Flows

Use Case

Let’s start with a simple example of combing two streams together.

In the diagram above, we have two Flows that are emitting items at different times. The first Flow is emitting numbers 1, 2 and 3 every second. The second Flow is emitting letters A, B, and C every 2 seconds. How do we combine these emissions?

fun main() = runBlocking {

    val numbersFlow = flowOf(1,2,3).delayEach(1000)
    val lettersFlow = flowOf("A", "B","C").delayEach(2000)
    
    numbersFlow.combine(lettersFlow) { number, letter ->
         "$number$letter"
    }.collect {
        println(it)
    }
    
}

In the code snippet above, I have implemented our example of two Flows each emitting at different times.

val numbersFlow = flowOf(1,2,3).delayEach(1000)
val lettersFlow = flowOf("A", "B","C").delayEach(2000)

Let’s look at the signature of the combine operator. It is an extension on Flow that takes as an argument another Flow to combine with. The second argument it accepts is a lambda which gives you values a: T1 and b: T2 that were emitted most recently both Flows. This lambda represents the transformation you want to perform on the emissions from both Flows.

Flow<T1>.combine(
    flow: Flow<T2>,
    transform: suspend (a: T1, b: T2) -> R
): Flow<R> (source)

This how we could use this operator for our example.

numbersFlow.combine(lettersFlow) { number, letter ->
    "$number$letter"
}.collect {
     println(it)
}

Output
1A 
2A 
3A 
3B 
3C

I have specified the transformation to concatenate the number and letter. We print out the result of the transformation when collecting from the Flow. Let’s understand why the combine operator produces this output.

How combine operator works?

The combine operator returns a Flow whose values are generated with transform function by combining the most recently emitted values by each flow.

• The first two values emitted by the numbers and letters Flow is 1 and A, respectively. These two values are combined together to produce 1A.
• The numbers Flow is first to emit as it emits faster than the letters Flow. It emits the value of 2. At this point, the letters Flow is suspended. It hasn’t emitted anything. What will we combine with? We will combine with the letter Flow’s most recently emitted value of A. This will produce 2A.

• The letters Flow now emits the value of 3. This will follow the same logic and it will combine with the most recently emitted value of A. Thus producing 3A .

• Finally, the letters Flow emits the value of B after delaying for 2 seconds. It will combine with the most recently emitted value of 3 from the numbers Flow producing 3B .

• At this point, everything has been emitted from the numbers Flow. The final emission from the letters Flow is C. It is combined with 3 to get a value of 3C.

This is the trace of how the emissions are combined from each Flow. Feel free to run the code above and familiarize yourself with the combine operator.

How does the combine operator listen to two Flows emitting at different rates? This is accomplished by using select expressions. They are experimental but they are used in the implementations of the zip and the combine operator. Let’s see how it works.

Select Expressions

A select expression is a construct that allows you to listen to multiple suspending coroutines simultaneously.

Use Case

Suppose we have two coroutines that are sending values to a Channel. Producer 1 delays for 1 second and send a value while Producer 2 delays for 2 seconds before sending a value. These coroutines are created by using the produce builder. It provides a Channel for you to send values to it and gives you back a RecieveChannel instance to read values.

fun CoroutineScope.producer1() = produce<String> {
    while (true) {
        delay(1000)
        send("Producer Item 1")
    }
}

fun CoroutineScope.producer2() = produce<String> {
    while (true) {
        delay(2000)
        send("Producer Item 2")
    }
}

We could receive values from either of these producers as shown below.

val channel1: ReceiveChannel = producer1()
val channel2: ReceiveChannel = producer2()

val item1: String = channel1.receive()
val item2: String = channel2.receive()

But, suppose I want to receive values simultaneously from either producer. In order to do this, we will need to use a Select expression.

A Select expression takes a lambda block which allows you to use clauses inside of it to listen for values from multiple Channels.

inline suspend fun <R> select(
    crossinline builder: SelectBuilder<R>.() -> Unit
): R (source)

Source: Select

How do we create a select expression for our example?

val channel1: ReceiveChannel = producer1()
val channel2: ReceiveChannel = producer2()

select<Unit> {
    channel1.onReceive {
        println(it)
    }
    channel2.onReceive {
        println(it)
    }
}

Inside the select expression’s lambda, I am calling the onReceive clause on both Channels. This clause can only be used from inside a select expression. Therefore, I am listening to emissions on both of these Channels simultaneously. The first value received from either Channel will be printed to the console.

Here is a full example that you could play with.

fun main() = runBlocking {
    
    val channel1 = producer1()
    val channel2 = producer2()

    repeat(10) {
        select<Unit> {
            channel1.onReceive {
                println(it)
            }
            channel2.onReceive {
                println(it)
            }
        }
    }

    coroutineContext.cancelChildren()
}

fun CoroutineScope.producer1() = produce<String> {
    while (true) {
        delay(1000)
        send("Producer Item 1")
    }
}

fun CoroutineScope.producer2() = produce<String> {
    while (true) {
        delay(2000)
        send("Producer Item 2")
    }
}

If you run this example, you will see that each execution of the select expression produces a new value that could be from either Channel.

Flow Combine Implementation

Under the hood, the combine operator uses a select expression to listen to emissions from multiple Flows. Let’s look closely at the source to understand the implementation. Here is a diagram showing how it works.

Suppose we want to combine Flow A and Flow B.

1. Two producer coroutines are created for Flow A and Flow B.
2. A select expression is created to listen to the emission of both producers. The select expression runs until a value is received from both producers’ channels. These values emitted from both Flows are stored locally in variables.
3. When emissions are received from both producers, they are combined with the transformation that you specify.
4. Finally, the transformed value is given back to you in a new Flow that you could collect from.

These are the steps that are taken when you use the combine operator. Here is the source for these steps.

suspend fun <T1, T2, R> FlowCollector<R>.combineTransformInternal(
    first: Flow<T1>, second: Flow<T2>,
    transform: suspend FlowCollector<R>.(a: T1, b: T2) -> Unit
) {
    coroutineScope {
        val firstChannel = asFairChannel(first)
        val secondChannel = asFairChannel(second)

        var firstValue: Any? = null
        var secondValue: Any? = null

        var firstIsClosed = false
        var secondIsClosed = false
        while (!firstIsClosed || !secondIsClosed) {

            select<Unit> {
                onReceive(
                    firstIsClosed, 
                    firstChannel, 
                    { firstIsClosed = true }
                ) { value ->
                    firstValue = value
                    if (secondValue !== null) {
                        transform(...)
                    }
                }

                onReceive(
                    secondIsClosed, 
                    secondChannel, 
                    { secondIsClosed = true }
                ) { value ->
                    secondValue = value
                    if (firstValue !== null) {
                        transform(...)
                    }
                }
            }
        }
    }
}

Source: Combine.kt

Let breakdown this source to see the steps in action.

suspend fun <T1, T2, R> FlowCollector<R>.combineTransformInternal(
    first: Flow<T1>, second: Flow<T2>,
    transform: suspend FlowCollector<R>.(a: T1, b: T2) -> Unit
) {

}

This function is internal and private. It takes in two Flows and a transformation that you want to apply.

1. Two producer coroutines are created for Flow A and Flow B.

suspend fun <T1, T2, R> FlowCollector<R>.combineTransformInternal(
    first: Flow<T1>, second: Flow<T2>,
    transform: suspend FlowCollector<R>.(a: T1, b: T2) -> Unit
) {
    coroutineScope {
        val firstChannel = asFairChannel(first)
        val secondChannel = asFairChannel(second)

    }
}

fun CoroutineScope.asFairChannel(
       flow: Flow<*>
): ReceiveChannel<Any> = produce {
    val channel = channel as ChannelCoroutine<Any>
    flow.collect { value ->
        return@collect channel.sendFair(value ?: NULL)
    }
}

A coroutine is created using the produce coroutine builder. It provides you with a Channel that you could send values emitted from a Flow. This is what the asFairChannel method is doing.

2. A select expression is created to listen to the emission of both producers.

suspend fun <T1, T2, R> FlowCollector<R>.combineTransformInternal(
    first: Flow<T1>, second: Flow<T2>,
    transform: suspend FlowCollector<R>.(a: T1, b: T2) -> Unit
) {
    coroutineScope {
        val firstChannel = asFairChannel(first)
        val secondChannel = asFairChannel(second)

        var firstValue: Any? = null
        var secondValue: Any? = null

        var firstIsClosed = false
        var secondIsClosed = false
}

For each channel we store it’s emission into a variable and maintain whether it’s closed with a flag.

suspend fun <T1, T2, R> FlowCollector<R>.combineTransformInternal(
    first: Flow<T1>, second: Flow<T2>,
    transform: suspend FlowCollector<R>.(a: T1, b: T2) -> Unit
) {
    coroutineScope {

        val firstChannel = asFairChannel(first)
        val secondChannel = asFairChannel(second)

        var firstValue: Any? = null
        var secondValue: Any? = null

        var firstIsClosed = false
        var secondIsClosed = false

        while (!firstIsClosed || !secondIsClosed) {
            select<Unit> {
              ...
            }  
        }              
    }
}

Why is the select expression above in a while loop?

So, let’s recall the simple example we looked at for a select expression. You will find that it completes execution when one of the Channels you are listening to emits a value. Depending on the rate of the emissions from the Channels, you may get value from either the first or second Channel in any order. Therefore, we have variables firstValue and secondValue to store the emitted value. Since we need the first emission from both Channels, the select expression is run until either Channel is closed to ensure we read from both Channels.

3. When items are received from both producers, they are combined with the transformation that you specify.

while (!firstIsClosed || !secondIsClosed) {
      select<Unit> {
          onReceive(...) { value ->
              firstValue = value
               if (secondValue !== null) {
                   transform(...)
               }
           }

           onReceive(...) { value ->
               secondValue = value
               if (firstValue !== null) {
                        transform(...)
               }
            }
        }
    }
}

fun SelectBuilder<Unit>.onReceive(
    ...,
    channel: ReceiveChannel<Any>,
    ...,
    onReceive: suspend (value: Any) -> Unit
) {
    if (isClosed) return
    channel.onReceiveOrNull {
        if (it === null) onClosed()
        else onReceive(it)
    }
}

How are values read in from the Channels in the select expression?

onReceiveOrNull is a select clause on Channel. It is being used to read the value emission from the channel above in the onReceive. If there are no more emissions from the Flow, the channel is closed. The emitted values are stored in the variables firstValue and secondValue. These values are used to apply the transformation you had specified.

4. Finally, the transformed value is given back to you in a new Flow that you could collect from.

fun <T1, T2, R> Flow<T1>.combine(
   flow: Flow<T2>, 
   transform: suspend (a: T1, b: T2) -> R
): Flow<R> = flow {
    combineTransformInternal(this@combine, flow) { a, b ->
        emit(transform(a, b))
    }
}

Source: Zip.kt

As you could see in the code snippet above, the combine operator calling the internal method I showed above. It creates a new Flow and emits the transformed values.

This is how the combine operator uses select expressions internally. Although they are experimental, when you look under the hood, they are used in many places. Furthermore, the combine operator is very useful in many use cases such as implementing an MVI architecture. Understanding how combine works internally is valuable when debugging and testing.

I hope this article was helpful. If you have any questions, feel free to respond to this article.

Find me on Twitter.
https://twitter.com/heyitsmohit

Slides from my presentations on Channels & Flows
https://speakerdeck.com/heyitsmohit

Combining Kotlin Flows with Select Expressions was originally published in ProAndroidDev on Medium, where people are continuing the conversation by highlighting and responding to this story.

Use Case

Let’s look at a simple example we could write a test for. The snippet below shows a repository that takes in an API service. The getUserDetails method uses the service to make a request to get UserDetails. The data is returned in a Flow. The example doesn’t have to be implemented with a Flow. However, for the sake of illustrating testing a Flow, I choose the simplest example possible.

class UsersRepository(private val apiService: ApiService) {

    fun getUserDetails(id: Int): Flow<UserDetails> {
        return flow {
            val userDetails = apiService.userDetails(id)
            emit(userDetails)
        }
    }
}

Test

Let’s write a test for this example. We want to test the Flow emits a UserDetails object.

@Test
fun `should get user details`() = runBlocking {

    val userDetails = UserDetails(
         id = 1,
         name = "User 1",
         address = "SF"
    )

    // Mock API Service
    val apiService = mock<ApiService>() {
        onBlocking { userDetails(id = 1) } doReturn userDetails
    }
    val repository = UsersRepository(apiService)

    // Test & Verify
    val flow: Flow<UserDetails> = repository.getUserDetails(1)
    flow.collect { data ->
        data shouldBeEqualTo userDetails
    }
}

I am creating a coroutine using the runBlocking method to run our test in. I have broken down this test into three steps which are mock, test and verify.

Mocking

val userDetails = UserDetails(
         id = 1,
         name = "User 1",
         address = "SF"
    )

// Mock API Service
val apiService = mock<ApiService>() {
    onBlocking { userDetails(id = 1) } doReturn userDetails
}

I am using mockito-kotlin that provides a lot of helpers for mocking and verifying. Using mock<ApiService>, I am mocking the API service and stubbing out theuserDetails method to return the UserDetails object.
The userDetails method is suspending. Mockito-Kotlin provides a method calledonBlocking that starts a coroutine using runBlocking and stubs the method for you.

Testing & Assertion

val repository = UsersRepository(apiService)

val flow: Flow<UserDetails> = repository.getUserDetails(1)
flow.collect { data ->
    data shouldBeEqualTo userDetails
}

I am injecting the mocked API service into the repository and get the Flow by calling getUserDetails. I am collecting from the Flow and verifying the emission. I could also use the single extension on a Flow for this test.

This is a simple test. Let’s look at how we could perform the assertion on a Flow in a different way. The SQL Delight library has an extension on a Flow called test. It gives you an API to collect items, errors and verify that nothing else has been emitted from your Flow. Although, it is not a standalone library current. It's a nice pattern to follow.

Flow Assert

Here is a diagram of how the test extension works.

All the emissions from the flow you are testing a stored in an unlimited buffered Channel. An API is provided to you to query the channel in the test extension. You could query that there are no more emissions or an error was thrown.

Let’s look at the test extension implementation. Don’t worry about having to understand it all. I’m going to take you through it piece by piece

suspend fun <T> Flow<T>.test(
         timeoutMs: Long = 1000L, 
         validate: suspend FlowAssert<T>.() -> Unit
) {
    coroutineScope {
        val events = Channel<Event<T>>(UNLIMITED)
        val collectJob = launch {
            val terminalEvent = try {
                collect { item ->
                    events.send(Event.Item(item))
                }
                Event.Complete
            } catch (_: CancellationException) {
                null
            } catch (t: Throwable) {
                Event.Error(t)
            }
            if (terminalEvent != null) {
                events.send(terminalEvent)
            }
            events.close()
        }
        val flowAssert = FlowAssert(events, collectJob, timeoutMs)
        val ensureConsumed = try {
            flowAssert.validate()
            true
        } catch (e: CancellationException) {
            if (e !== ignoreRemainingEventsException) {
                throw e
            }
            false
        }
        if (ensureConsumed) {
            flowAssert.expectNoMoreEvents()
        }
    }
}

internal sealed class Event<out T> {
    object Complete : Event<Nothing>()
    data class Error(val throwable: Throwable) : Event<Nothing>()
    data class Item<T>(val item: T) : Event<T>()
}

Source: Flow Assert

• The test extension is creating a coroutine with coroutineScope builder. The builder creates a coroutine that inherits the context of the parent coroutine.

suspend fun <T> Flow<T>.test(
         timeoutMs: Long = 1000L, 
         validate: suspend FlowAssert<T>.() -> Unit
) {
    coroutineScope {

    }

• Inside this coroutine with Channel<Event<T>>(UNLIMITED), we declare an unlimited buffered channel to store all the emissions. The type of stored items in this channel is an Event type which is a sealed class. This specifies whether an item was emitted, an error occurred or emission is complete.

suspend fun <T> Flow<T>.test(
         timeoutMs: Long = 1000L, 
         validate: suspend FlowAssert<T>.() -> Unit
) {
    coroutineScope {
        val events = Channel<Event<T>>(UNLIMITED)

    }

}

internal sealed class Event<out T> {
    object Complete : Event<Nothing>()
    data class Error(val throwable: Throwable) : Event<Nothing>()
    data class Item<T>(val item: T) : Event<T>()
}

• A child coroutine is launched in the coroutineScope builder. In this coroutine, the flow is collected. If an item is emitted, it is sent to the channel. As this is an unlimited channel, the coroutine will not suspend itself when the send method is called. If an exception is thrown, it is caught and an error event is sent to the channel. After the collection, the channel is closed.

  val collectJob = launch {
     val terminalEvent = try {
         collect { item ->
            events.send(Event.Item(item))
         }
         Event.Complete
       } catch (_: CancellationException) {
            null
       } catch (t: Throwable) {
           Event.Error(t)
       }
       if (terminalEvent != null) {
         events.send(terminalEvent)
       }
       events.close()
 }

• After the Flow is collected, an instance of the FlowAssert is created. This is also the receiver of the lambda param of the test extension FlowAssert<T>.() -> Unit. This class provides an API to query the channel.

val flowAssert = FlowAssert(events, collectJob, timeoutMs)

Here are some of the methods the FlowAssert implements.

class FlowAssert<T> internal constructor(
    private val events: Channel<Event<T>>,
    private val collectJob: Job,
    private val timeoutMs: Long
) {
    

    ...

    suspend fun expectItem(): T {
        val event = withTimeout {
            events.receive()
        }
        if (event !is Event.Item<T>) {
            throw AssertionError("Expected item but was $event")
        }
        return event.item
    }

    suspend fun expectComplete() {
        val event = withTimeout {
            events.receive()
        }
        if (event != Event.Complete) {
            throw AssertionError("Expected complete but was $event")
        }
    }

    suspend fun expectError(): Throwable {
        val event = withTimeout {
            events.receive()
        }
        if (event !is Event.Error) {
            throw AssertionError("Expected error but was $event")
        }
        return event.throwable
    }
}

Source: Flow Assert

Each method above is querying the events channel by calling the receive method on it. These methods could be used inside the test extension. This is an overview of how the test extension works. It's a good pattern used in the SQL Delight library.

Usage

How do we modify our example I showed earlier to use this extension?

@Test
fun `should get user details`() = runBlocking {

     val userDetails = UserDetails(
             id = 1, 
             userName = "User 1", 
             city = "SF"
    )

    // Mock API Service
    val apiService = mock<ApiService>() {
        onBlocking { userDetails(id = 1) } doReturn userDetails
    }
    val repository = UsersRepository(apiService)

     // Test & Verify
    val flow: Flow<UserDetails> = repository.getUserDetails(1) 
    flow.test {
        assertEquals(expectItem(), userDetails)
        expectComplete()
    }
}

I am getting an emission with expectItem and performing an assertion. The expectComplete method ensures there were no more emissions.

How about testing errors? Let’s write a test for it.

@Test
fun `should throw error`() = runBlocking {
   
    // Mock
    val apiService = mock<ApiService>()
    whenever(apiService.userDetails()) doAnswer {
        throw IOException()
    }

    // Test and Verify 
    val repository = UsersRepository(apiService)
    val flow: Flow<UserDetails> = repository.getUserDetails()

    flow.test {
        assertThat(
            expectError(),     
            instanceOf(IOException::class.java)
        )
    }
}

In the test above, I have mocked the API service to throw an IOException when theuserDetails method is called.

// Mock
val apiService = mock<ApiService>()
whenever(apiService.userDetails()) doAnswer {
     throw IOException()
}

When I call collect from the Flow, it will throw the exception and FlowAssert will catch it and add it as an event to its internal channel. I am querying the channel with the expectError method to verify an error was thrown.

// Test and Verify 
val repository = UsersRepository(apiService)
val flow: Flow<UserDetails> = repository.getUserDetails()

flow.test {
   assertThat(
       expectError(),     
       instanceOf(IOException::class.java)
   )
}

This is a pattern used in the SQL Delight library for Flow assertions. This test extension is not in a standalone library yet. There are improvements that could be made. But, there is an open issue for it. Please contribute if you would like to do so.

If you have any issues and corrections, please let me know. Thanks for reading!

Kotlin Flow Assert Delight was originally published in ProAndroidDev on Medium, where people are continuing the conversation by highlighting and responding to this story.

The Kotlin coroutine library provides a construct called a Channel which behaves like a BlockingQueue. How do Channels work underneath the hood? That’s what we’ll explore in this blog series. As you create complex use cases, it will involve building producers with Channels. Understanding how Channels work will help you to debug your code. Let’s dive into Channels.

What is a Channel?

A Channel is a construct that allows you to send and receive values. It has an optional buffer to cache the values.

It is implemented as an interface that has two suspending methods — send and receive. There are five types of Channels

• Rendezvous Channel
• Buffered Channel
• Unbuffered Channel
• Conflated Channel
• Broadcast Channel

Each Channel has a distinct property. We will explore the Rendezvous Channel in this blog post as it is the simplest type of Channel.

Rendezvous Channel

A Rendezvous Channel has two properties. It doesn’t have a buffer. If you send a value to it, it will wait until there is a receiver to read the value. Let’s look at this behavior with an example.

runBlocking {
    
    val channel = Channel<Int>()

    launch {
      channel.send(2)
    }
}

In this example, we have created a Rendezvous channel in the runBlocking coroutine. The child coroutine sends a value of 2. When the send method is called, it will suspend the child coroutine. This is because there is no reciever reading the sent value.

We could observe this behavior by using DebugProbes. It allows us to view the state of all the coroutines in our program.

runBlocking {

    DebugProbes.install()
    val channel = Channel<Int>()

    launch {
       channel.send(2)
    }

    delay(2000)
    DebugProbes.dumpCoroutines()
}

Output:
Coroutine StandaloneCoroutine{Active, state: SUSPENDED}

In this example, we have installed DebugProbes and used the dumpCoroutines method to view the state of the child coroutine. Its state is suspended. When a consumer is ready to read the value, the child coroutine is resumed. The value sent to the channel is not cached in an array buffer. This is the underlying behavior of a Rendezvous channel. Let’s explore this behavior further by inspecting the channel’s state.

Channel State

Every channel has state. The state represents whether the channel is empty, a value was sent or a value was buffered. When you create a channel, it initializes a queue. This queue is a doubly linked list.

abstract class AbstractSendChannel<E> {

    protected val queue = LockFreeLinkedListHead()

    ...
}

class LockFreeLinkedListNode {

    val _next = atomic<Any>(this)
    val _prev = atomic<Any>(this)
    ...
}

The implementation of the queue is based on this paper Lock-Free and Practical Doubly Linked List-Based Deques Using Single-Word Compare-and-Swap. The operations on a Channel are applications of lock free algorithms.

Each node added to this queue represents a state. Here are some of the nodes that can be added to the queue:

• SendElement
• SendBuffer
• Receive
• Closed

You could query the state of the Channel at any point of your program. The toString method is overridden in the channel and it returns the number of items in the buffer and the current head node in the queue.

class AbstractSendChannel {

   fun toString() = {$queueDebugStateString}$bufferDebugString”
}

Every channel inherits AbstractSendChannel. As you can see, it provides debugging information in its toString method.

runBlocking {

   val channel = Channel<Int>()

   println(channel)

}

Output:
RendezvousChannel{EmptyQueue}

The Channel is currently empty. When I send a value to it, the state will be updated by adding a node of type SendQueued.

runBlocking {

   val channel = Channel<Int>()

   launch {
      channel.send(1)
   }

    delay(2000)
    println(channel)

    val num = channel.receive()
    println(channel)

}

Output:
RendezvousChannel{SendQueued}
RendezvousChannel{EmptyQueue}

After the runBlocking coroutine has read the value, the queue will be updated to be empty again. Querying the Channel is very helpful as you work through examples and learn about Channels.

Being able to query the Channel’s queue and its buffered items is helpful in debugging. We’ll go more in-depth on the lock-free algorithms in the next blog post. If you have any questions, please feel free to comment below.

Follow me on Twitter: @heyitsmohit

In this article, I will explore how to implement and test polling with Channels and Flow. Prior to reading on, please check out this article for understanding the basics of coroutines.

Problem

Suppose we would like to hit an endpoint every second and gather the data to display it in the UI. This use case fits many contexts such as continuously updating a live feed or updating scores for a sports event. How could we go about implementing this?

Solution

The approach we could take is to create a poller that launches a coroutine that delays for a second, makes an API request and returns the data in a hot Flow. We want the polling to start only when there is a subscriber. Let’s dive into the implementation.

interface Poller {

   fun poll(delay: Long): Flow<Data>

   fun close()

}

I have created an interface with a poll and close method. Thepollmethod takes in a delay in milliseconds and returns a Flow for the given data. The close method stops the polling. The advantage of creating this interface is so that you could provide different implementations of polling if needed.

Let’s begin by creating a coroutine implementation of the poller.

class CoroutinePoller(
    val dataRepository: DataRepository,
    val dispatcher: CoroutineDispatcher
): Poller {

    override fun poll(delay: Long): Flow<Data> { ... }

    override fun close() { ... }

}

In this class, I am injecting a repository and a Dispatcher into the poller. The repository makes the request and gives us back the data. The purpose of the Dispatcher is to specify the thread pool the polling should occur on. It also allows me to inject a custom dispatcher for testing as we’ll see later.

Using channelFlow

In order to implement the actual logic of polling, we need to launch a coroutine that iteratively delays, makes a request and emits data into a stream. We want this process to start when a client has subscribed to the stream. In order to do this, we will use the channelFlow method provided by the coroutines library. It launches a coroutine and creates a channel that we could send items to.

class CoroutinePoller(
    val repository: DataRepository,
    val dispatcher: CoroutineDispatcher
): Poller {

    override fun poll(delay: Long): Flow<Data> {
        return channelFlow {
            while (!isClosedForSend) {
                val data = repository.getData()
                send(data)                 
                delay(delay)
            }
        }.flowOn(dispatcher)
    }

    override fun close() {
        dispatcher.cancel()
    }

}

In the poll method above, I am getting the data and sending it to the channel. This occurs on the injected Dispatcher by using the flowOn extension. I have added a polling delay after the request. The delay and request are happening repeatedly until the channel is closed for send. The polling is halted in the close method by invoking cancel on the injected Dispatcher.

Error Handling

In this logic, how do we handle errors? We may want to log the error and continue polling. One way to handle this is to return an Either type from the getData method. A Either type allows you to represent one of two types. In this case, it is an Exception or Data. Arrow is a library that provides many functional data types like Either. It also provides useful extensions on the Either type.

class CoroutinePoller(
    val repository: DataRepository,
    val dispatcher: CoroutineDispatcher,
    val logger: Logger
): Poller {

    override fun poll(delay: Long): Flow<Data> {
        return channelFlow {
            while (!isClosedForSend) {
                delay(delay)
                val either = repository.getData()
                either.fold({ exception ->
                    logger.log(“Failed to get data”)
                }, {
                    send(data)
                })
            }
        }.flowOn(dispatcher)
    }

    override fun close() {
        dispatcher.cancel()
    }

}

In the poll method, we are using the fold extension on an Either which accepts two lambda parameters. The first lambda is invoked when an error occurs from the API request. We are logging the error in this scenario. In this use case, I am not halting the polling when an error occurs. I want to keep polling and only log the error. The second lambda is invoked when the request was successful. We are sending the data to the channel in this lambda.

Testing

In order to test coroutines, you have to set up a test harness. We need to create and launch a coroutine under which our test code could run. A suspend method cannot be called from a regular method. There are two ways to go about setting up a test harness. One way is to use the runBlocking coroutine builder in each of your tests. The other way to test your code is to use the runBlockingTest method. This is provided by the coroutine test library. runBlockingTest is used for testing delays. The library gives you methods such as advanceTimeBy or advanceTimeUntilIdlto control the passage of time. In our use case, we have a delay of 1 second that we want to control.

Let’s test that our poller emits data every second.

class CoroutinePollerTest {

    val data = mock<Data>()

    val repository = mock<DataRepository>() {
        onBlocking { getData()  } doReturn data
    }

    val testDispatcher = TestCoroutineDispatcher()
    val poller = CoroutinePoller(repository, testDispatcher)

    @Test
    fun `should poll every second`() = runBlockingTest {
        val flow = poller.poll(1_000)

        launch {
            val dataList = flow.take(2).toList()
            assertEquals(dataList.size, 2)
        }

        testDispatcher.advanceTimeBy(2_000)

        poller.close()
    }
}

In this test, I am controlling the delay in the poller by calling advanceTimeBy on the injected Dispatcher. The advanceTimeBy method advances forward 2 seconds and two data items are emitted by the channelFlow. The test is using the take extension on the flow to only collect 2 items and store it in a list. We perform an assertion on the list’s size.

Flow Assertions

RxJava has a useful test method on an Observable. The SQLDelight library has a similar convenient test extension on a Flow. It would be very useful if it was in a standalone library. Here is how we would use it:

class CoroutinePollerTest {

    val data = mock<Data>()

    val repository = mock<DataRepository>() {
        onBlocking { getData()  } doReturn data
    }

    val testDispatcher = TestCoroutineDispatcher()
    val poller = CoroutinePoller(repository, testDispatcher)

    @Test
    fun `should poll every second`() = runBlockingTest {
        val flow = poller.poll(1_000)

        launch {
           flow.test {
             expectItem() assertEquals data
             expectItem() assertEquals data
             expectComplete()
           }
        }

        testDispatcher.advanceTimeBy(2_000)

        poller.close()
    }
}

The test extension on Flow launches a coroutine, collects from the flow, and sends the items to an unlimited channel. The channel’s type is an Event that is a sealed class. It defines whether a value was received from the flow, no more items were sent or an exception occurred. You could query this channel by calling expectItem, expectError and expectComplete methods. It makes your test cleaner to read.

This is how you could implement polling and add unit tests for it. If you have any questions or suggestions, please feel free to comment below.

Resources

• Kotlin Coroutines Library
• Flow Assert

Follow me on Twitter
twitter.com/@heyitsmohit

Polling with Kotlin Channels & Flows was originally published in ProAndroidDev on Medium, where people are continuing the conversation by highlighting and responding to this story.