How to filter a list inside Kotlin Flow

Question

I'm replacing my current implementation using RxJava to Coroutines and Flow. I'm having some trouble using some Flow operators.

I'm trying to filter the list of items inside a Flow before providing it to be collected. (Flow<List<TaskWithCategory>>)

Here is the example on Rx2:

        repository.findAllTasksWithCategory()
            .flatMap {
                Flowable.fromIterable(it)
                    .filter { item -> item.task.completed }
                    .toList()
                    .toFlowable()

In the implementation above, I provide a list of TaskWithCategory filtering by Tasks that are already completed.

How can I achieve this using Flow?

Answer 1

Given that the only operator in use is filter the inner flowable is unnecessary, making the flow implementation quite straightforward:

repository.findAllTasksWithCategoryFlow()
    .map { it.filter { item -> item.task.completed } }

If the inner transformation is more involved (lets use transform: suspend (X) -> TaskWithCategory):

repository.findAllTasksWithCategoryFlow()
    // Pick according to desired backpressure behavior
    .flatMap(Latest/Concat/Merge) {
        // Scope all transformations together
        coroutineScope {
            it.map { item ->
                // Perform transform in parallel
                async {
                    transform(item)
                }
            }.awaitAll() // Return when all async are finished.
        }
    }