Extensions for Kotlin Flow?

[curioustechizen]

Have you considered providing extensions for Kotlin Flows, specifically, Flow<Result<V, E>>? I'm making heavy use of these "flows-of-result" and I have some helpers of my own. However, I feel that it would be much nicer to have binding-style utilities that know about Flow. Here are some examples of helpers that we have in our project (admittedly, some of these can have better names!):

Some variants of Flow's combine operators that know about Flow<Result<V, E>>

/**
 * Variant of Kotlin's [combine] that makes it easier to work with flows of [Result].
 *
 * Use this if [transform] never returns an error. If your transform might return an error, consider using [combineResultWithErr] instead.
 */
fun <T1, T2, E, R> combineResult(
    flow: Flow<Result<T1, E>>,
    flow2: Flow<Result<T2, E>>,
    transform: suspend (T1, T2) -> R
): Flow<Result<R, E>> {
    return combine(flow, flow2) { r1, r2 ->
        binding {
            val s1 = r1.bind()
            val s2 = r2.bind()
            transform(s1, s2)
        }
    }
}

/**
 * Variant of Kotlin's [combine] that makes it easier to work with flows of [Result].
 *
 * Use this if [transform] might return an error. If your transform never returns an error, consider using [combineResult] instead.
 */
fun <T1, T2, E, R> combineResultWithErr(
    flow: Flow<Result<T1, E>>,
    flow2: Flow<Result<T2, E>>,
    transform: suspend (T1, T2) -> Result<R, E>
): Flow<Result<R, E>> {
    return combine(flow, flow2) { r1, r2 ->
        binding {
            val s1 = r1.bind()
            val s2 = r2.bind()
            transform(s1, s2).bind()
        }
    }
}

Variant of andThen that can be applied to a Flow<Result<V, E>>

/**
 * Apply [transform] on each element in this [Flow] if the element is an [Ok] otherwise return the
 *  error as-is into the flow.
 */
fun <V, E, U> Flow<Result<V, E>>.andThen(transform: suspend (V) -> Result<U, E>): Flow<Result<U, E>> {
    return map {
        when (it) {
            is Ok -> transform(it.value)
            is Err -> it
        }
    }
}

Variant of flatMapLatest that makes it simpler to work with Result<V, E>

/**
 * Like [andThen] but allows the [transform] to return a `Flow<Result>`. This method applies the
 *  [flatMapLatest] operator allowing you to return a flow from your transform
 */
fun <V, E, U> Flow<Result<V, E>>.andThenFlow(transform: suspend (V) -> Flow<Result<U, E>>): Flow<Result<U, E>> {
    return flatMapLatest {
        when (it) {
            is Ok -> transform(it.value)
            is Err -> flowOf(it)
        }
    }
}

---

As you can see, these utilities work, but they can be improved a lot. Specifically, they can benefit from the equivalent of binding {} (perhaps something like flowBinding {})

[michaelbull]

This is an interesting idea that I'd accept a PR for. The naming needs work, not a fan of combineResult vs combineResultWithErr - all Results have the potential for an Err so this is a bit confusing (not sure whether we would even need both), and I think andThenFlow should just be andThen given it's already an extension on Flow. Maybe fleshing out more of the common use-cases would get us to more readable naming that remains in line with whatever the naming is in the Flow API.

Given we already have the dependency on kotlin coroutines I presume it could just live in the coroutines submodule? Or does it live in a separate dependency?

[curioustechizen]

In my opinion the snippets that I posted above are not suitable to go into the library as-is. They were just some ideas on what problems could be encountered when using kotlin-result in combinations with Flow and how things could be improved.

Starting with the common use-cases definitely makes sense to me. I see 2 broad categories

1. Applying transformations on a single Flow<Result<V, E>> to return a Flow<Result<U, E>> such that if the upstream flow has an Err element, that element is emitted as-is downstream; and the transformation is only applied if the upstream element is an Ok. The andThen/flatMapLatest examples above belong in this category.
2. Operating on multiple Flow<Result>>s to produce a single Flow<Result> in the end. Here again, if any of the dependent flows emits an Err the final output could be that Err; only the unwrapped Oks could be passed to the actual transformation function. The combineResult snippet above is an example of this.

For me the motivation behind writing those helpers was

1. Avoiding writing those when(result) blocks for every upstream flow all over (this one is probably obvious)
2. Avoiding wrapping the output in Ok(output) all over the place.

I think there is scope for some more generification and simplification on the lines of SuspendableResultBinding, but I still don't know what it would look like exactly.

Given we already have the dependency on kotlin coroutines I presume it could just live in the coroutines submodule?

That's right. Flow is part of the kotlinx.coroutines dependency.

[michaelbull]

Maybe @Munzey, author of the binding story, has some ideas.

[Munzey]

Im away for the next two weeks but from a quick read i think it looks like it would be a cool addition that would make perfect sense to include in the coroutines module!

As for whether we would need an equivalent flowBinding over an implementation like in your example of combine, im not sure whether there are any performance or readability gains to be made here - not that familiar with flow internals but i think it would be tricky to recreate the bind logic for an indeterminate number of flows you want to combine though maybe there is a use case there you can describe 🤔

[Gregory991]

I'm not really sure the best way to achieve this. It could very well be ignorance on my part. Sorry in advance if its missing the scope of this discussion. This level of generics starts to boggle me too much.

How do you achieve a Flow that is dependent upon a Result? My solution at the moment is to first, unwrap the result and in side its .onSuccess fire the flow. But i was hoping for the flow itself, to first get the Result, check its OK then proceed to flow on the return type.

I have an Android room table i want to Flow on. Something like: Flow<Vehicle>
First i need to get the userVehicleId from the User Repository. That isn't a flow, it returns Result<Int, Unit>

The function i have atm looks like this:

suspend fun getUserVehicle(): Flow<Result<Vehicle, Unit>>{ 
   return userRepo.getUserVehicle()              // returns Result<Int, Unit> 
        .andThen{ id ->
            Ok(vehicleDao.flowOnUserVehicle(id)) // returns Flow<Vehicle, Unit>
        }
}

This isn't valid and won't compile. Since I'm returning Result<Flow<Vehicle, Unit>

Is this a misuse/ understanding on my part or maybe a valid problem? If its valid potentially some further functions with flow and result could assist.

[curioustechizen]

@Gregory991 Your problem is that vehicleDao.flowOnUserVehicle(id) returns a Flow<Vehicle> and you are wrapping that in a Result. What you want to do instead is map instead of wrap (at least for the case where getUserVehicle() was successful)

        .andThen{ id ->
            vehicleDao.flowOnUserVehicle(id).map { Ok(it) } // Wraps each emission of Flow<Vehicle> in an Ok thus yielding a Flow<Result<Vehicle, Unit>>
        }

This might still not compile though. You only fixed the andThen part. If userRepo.getUserVehicle returns an Err, you're still only returning an Err, not a Flow. To fix this, you should be wrapping this case in a Flow like this

// Disclaimer: This is pseudo-code and i haven't actually tried this in an IDE

suspend fun getUserVehicle(): Flow<Result<Vehicle, Unit>>{
    return when(val id = userRepo.getUserVehicle()) {
        is Ok -> vehicleDao.flowOnUserVehicle(id).map { Ok(it) } //Flow<Ok<Vehicle<>
        is Err -> flowOf(id) //Flow<Err<Unit>> since id is an Err
    }
}

There might be better ways to do this mapping than using the when clause. Result offers better APIs like mapBoth which you could take advantage of. But the underlying points are:

1. When you first get the vehicle ID from userRepo, you must account for both the Ok and Err cases there
2. Both cases must return a flow
3. When you get a Flow<Vehicle> from your room DB, you need to wrap each emission in an Ok; not the entire Flow. That's how you get from Flow<Vehicle> into a Flow<Result<Vehicle, Unit>> instead of resulting in a Result<Flow<Vehicle>, Unit>.

[curioustechizen]

@michaelbull I wonder if there is scope for something like this to address @Gregory991's usecase

fun <V, E, R> Result<V, E>.toFlow(flowProducer: (V) -> Flow<R>): Flow<Result<R, E> {
    when(this) {
        is Ok -> flowProducer(this.value).map { Ok(it) }
        is Err -> flowOf(this)
    }
}

// Usage:
suspend fun getUserVehicle(): Flow<Result<Vehicle, Unit>>{
    userRepo.getUserVehicle().toFlow { id ->
        vehicleDao.flowOnUserVehcile(id)
    }
}