Kotlin Coroutines use cases for Presentation/API/UI Layer

This is a chapter from the book Kotlin Coroutines. You can find it on LeanPub or Amazon.

The last layer we need to discuss is the Presentation layer, which is where coroutines are typically launched. In some types of applications, this layer is easiest because frameworks like Spring Boot or Ktor do the entire job for us. For example, in Spring Boot with Webflux, you can just add the suspend modifier in front of a controller function, and Spring will start this function in a coroutine.

@Controller
class UserController(
    private val tokenService: TokenService,
    private val userService: UserService,
) {
    @GetMapping("/me")
    suspend fun findUser(
        @PathVariable userId: String,
        @RequestHeader("Authorization") authorization: String
    ): UserJson {
        val userId = tokenService.readUserId(authorization)
        val user = userService.findUserById(userId)
        return user.toJson()
    }
}

Similar support is provided by other libraries. On Android, we use Work Manager to schedule tasks. We can use the CoroutineWorker class and implement its doWork method to specify what should be done by a task. This method is a suspend function, so it will be started in a coroutine by the library, therefore we don't need to do this ourselves.

class CoroutineDownloadWorker(
    context: Context,
    params: WorkerParameters
) : CoroutineWorker(context, params) {

    override suspend fun doWork(): Result {
        val data = downloadSynchronously()
        saveData(data)
        return Result.success()
    }
}

However, in some other situations we need to start coroutines ourselves. For this we typically use launch on a scope object. On Android, thanks to lifecycle-viewmodel-ktx, we can use viewModelScope or lifecycleScope in most cases.

class UserProfileViewModel(
    private val loadProfileUseCase: LoadProfileUseCase,
    private val updateProfileUseCase: UpdateProfileUseCase,
) {
    private val _userProfile =MutableStateFlow<ProfileData?>(null)
    val userProfile: StateFlow<ProfileData> = _userProfile

    fun onCreate() {
        viewModelScope.launch {
            val userProfileData = loadProfileUseCase()
            _userProfile.value = userProfileData
            // ...
        }
    }

    fun onNameChanged(newName: String) {
        viewModelScope.launch {
            val newProfile = _userProfile.updateAndGet { 
                it.copy(name = newName)
            }
            updateProfileUseCase(newProfile)
        }
    }
}

Creating custom scope

When you do not have a library or class that can start a coroutine or create a scope, you might need to make a custom scope and use it to launch a coroutine.

class NotificationsSender(
    private val client: NotificationsClient,
    private val notificationScope: CoroutineScope,
) {
    fun sendNotifications(notifications: List<Notification>) {
        for (notification in notifications) {
            notificationScope.launch {
                client.send(notification)
            }
        }
    }
}

class LatestNewsViewModel(
    private val newsRepository: NewsRepository,
) : BaseViewModel() {
    private val _uiState =MutableStateFlow<NewsState>(LoadingNews)
    val uiState: StateFlow<NewsState> = _uiState

    fun onCreate() {
        scope.launch {
            _uiState.value = NewsLoaded(newsRepository.getNews())
        }
    }
}

We define a custom coroutine scope using the CoroutineScope function[^401_16]. Inside it, it is practically standard practice to use SupervisorJob[^401_17].

val analyticsScope = CoroutineScope(SupervisorJob())

Inside a scope definition, we might specify a dispatcher or an exception handler[^401_18]. Scope objects can also be cancelled. Actually, on Android, most scopes are either cancelled or can cancel their children under some conditions, and the question "What scope should I use to run this process?" can often be simplified to "Under what conditions should this process be cancelled?". View models cancel their scopes when they are cleared. Work managers cancel scopes when the associated tasks are cancelled.

// Android example with cancellation and exception handler
abstract class BaseViewModel : ViewModel() {
    private val _failure = Channel<Throwable>(Channel.UNLIMITED)
    val failure: Flow<Throwable> = _failure.receiveAsFlow()

    private val handler = CoroutineExceptionHandler { _, e ->
        _failure.trySendBlocking(e)
    }

    protected val viewModelScope = CoroutineScope(
        Dispatchers.Main.immediate + SupervisorJob() + handler
    )

    override fun onCleared() {
        viewModelScope.coroutineContext.cancelChildren()
    }
}

// Spring example with custom exception handler
@Configuration
class CoroutineScopeConfiguration {

    @Bean
    fun coroutineDispatcher(): CoroutineDispatcher =
        Dispatchers.IO.limitedParallelism(50)

    @Bean
    fun coroutineExceptionHandler(
        monitoringService: MonitoringService,
    ) = CoroutineExceptionHandler { _, throwable ->
        monitoringService.reportError(throwable)
    }

    @Bean
    fun coroutineScope(
        coroutineDispatcher: CoroutineDispatcher,
        coroutineExceptionHandler: CoroutineExceptionHandler,
    ) = CoroutineScope(
        SupervisorJob() +
            coroutineDispatcher +
            coroutineExceptionHandler
    )
}

Using runBlocking

Instead of starting coroutines on a scope object, we can also use the runBlocking function, which starts a coroutine and blocks the current thread until this coroutine is finished. Therefore, runBlocking should only be used when we want to block a thread. The two most common reasons why it is used are:

To wrap the main function. This is a correct use of runBlocking, because we need to block the thread until the coroutine started by runBlocking is finished.
To wrap test functions. In this case, we also need to block the test thread, so this test doesn't finish execution until the coroutine completes.

import kotlinx.coroutines.runBlocking
annotation class Test

fun main(): Unit = runBlocking {
    // ...
}

class SomeTests {
    @Test
    fun someTest() = runBlocking {
        // ...
    }
}

Both these cases have more-modern alternatives. We can suspend the main function with coroutineScope or runTest in tests. This does not mean we should avoid using runBlocking, in some cases it might be enough for our needs.

import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.test.runTest
annotation class Test

suspend fun main(): Unit = coroutineScope {
    // ...
}

class SomeTests {
    @Test
    fun someTest() = runTest {
        // ...
    }
}

In other cases, we should avoid using runBlocking. Remember that runBlocking blocks the current thread, which should be avoided in Kotlin Coroutines. Use runBlocking only if you intentionally want to block the current thread.

class NotificationsSender(
    private val client: NotificationsClient,
    private val notificationScope: CoroutineScope,
) {
    @Measure
    fun sendNotifications(notifications: List<Notification>){
        val jobs = notifications.map { notification ->
            scope.launch {
                client.send(notification)
            }
        }
        // We block thread here until all notifications are
        // sent to make function execution measurement
        // give us correct execution time
        runBlocking { jobs.joinAll() }
    }
}

Working with Flow

When we observe flows, we often handle changes inside onEach, start our flow in another coroutine using launchIn, invoke some action when the flow is started using onStart, invoke some action when the flow is completed using onCompletion, and catch exceptions using catch. If we want to catch all the exceptions that might occur in a flow, specify catch on the last position[^401_19].

fun updateNews() {
    newsFlow()
        .onStart { showProgressBar() }
        .onCompletion { hideProgressBar() }
        .onEach { view.showNews(it) }
        .catch { view.handleError(it) }
        .launchIn(viewModelScope)
}

On Android, it is popular to represent our application state inside properties of type MutableStateFlow inside view model classes[^401_20]. These properties are observed by coroutines that update the view based on their changes.

class NewsViewModel : BaseViewModel() {
    private val _loading = MutableStateFlow(false)
    val loading: StateFlow<Boolean> = _loading

    private val _news = MutableStateFlow(emptyList<News>())
    val news: StateFlow<List<News>> = _news

    fun onCreate() {
        newsFlow()
            .onStart { _loading.value = true }
            .onCompletion { _loading.value = false }
            .onEach { _news.value = it }
            .catch { _failure.value = it }
            .launchIn(viewModelScope)
    }
}

class LatestNewsActivity : AppCompatActivity() {
    @Inject
    val newsViewModel: NewsViewModel

    override fun onCreate(savedInstanceState: Bundle?) {
        // ...
        launchOnStarted {
            newsViewModel.loading.collect {
                progressBar.visbility =
                    if (it) View.VISIBLE else View.GONE
            }
        }
        launchOnStarted {
            newsViewModel.news.collect {
                newsList.adapter = NewsAdapter(it)
            }
        }
    }
}

When a property representing a state depends only on a single flow, we might use the stateIn method. Depending on the started parameter, this flow will be started eagerly (when this class is initialized), lazily (when the first coroutine starts collecting it), or while subscribed[^401_21].

class NewsViewModel : BaseViewModel() {
    private val _loading = MutableStateFlow(false)
    val loading: StateFlow<Boolean> = _loading

    val newsState: StateFlow<List<News>> = newsFlow()
        .onStart { _loading.value = true }
        .onCompletion { _loading.value = false }
        .catch { _failure.value = it }
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.Lazily,
            initialValue = emptyList(),
        )
}

class LocationsViewModel(
    locationService: LocationService
) : ViewModel() {

    private val location = locationService.observeLocations()
        .map { it.toLocationsDisplay() }
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.Lazily,
            initialValue = LocationsDisplay.Loading,
        )

    // ...
}

StateFlow should be used to represent a state. To have some events or updates observed by multiple coroutines, use SharedFlow.

class UserProfileViewModel {
    private val _userChanges =
        MutableSharedFlow<UserChange>()
    val userChanges: SharedFlow<UserChange> = _userChanges

    fun onCreate() {
        viewModelScope.launch {
            userChanges.collect(::applyUserChange)
        }
    }

    fun onNameChanged(newName: String) {
        // ...
        _userChanges.emit(NameChange(newName))
    }

    fun onPublicKeyChanged(newPublicKey: String) {
        // ...
        _userChanges.emit(PublicKeyChange(newPublicKey))
    }
}

[^401_16]: For details, see the Constructing a coroutine scope chapter.
[^401_17]: To learn more about how SupervisorJob works, see the Exception handling chapter.
[^401_18]: For details, see the Constructing a coroutine scope chapter. Dispatchers and exception handlers are described in the Dispatchers and Exception handling chapters, respectively.
[^401_19]: For details, see the Flow lifecycle functions chapter.
[^401_20]: For details, see the SharedFlow and StateFlow chapter.
[^401_21]: All these options are described in the SharedFlow and StateFlow chapter, subchapters shareIn and stateIn.

Kotlin Coroutines use cases for Presentation/API/UI Layer

Creating custom scope

Using runBlocking

Working with Flow

The author:

Comments