I was hoping that I can use

stateIn

plus

distinctUntilChanged

to avoid expensive re-computations after a Flow was temporarily cold. Unfortunately that doesn’t work because

distinctUntilChanged

doesn’t maintain state when the Flow turns cold. Does anybody have an idea how I can implement the following scenario with existing operators? Easy part: 1. My shared flow turns hot because of first subscriber 2. Upstream emits its latest state and periodically emits new states 3. Ignore value if same as previous value 4. Perform expensive transformation (recompute a graph index when vertices and/or edges change) 5. Share expensive value with current and future subscribers (i.e. replay = 1) Difficult part: 6. My shared flow turns cold because all subscribers are gone 7. My shared flow turns hot again because of first subscriber 8. Upstream emits its latest state and periodically emits new states 9. Ignore value if same as previous value ⬅️ doesn’t work for initial value b/c Flow was cold and operator was reset 10. Perform expensive transformation ⬅️ unnecessarily applied again for a value that was already transformed and cached in replay buffer 11. Share expensive value with current and future subscribers (i.e. replay = 1) I kinda need either • a

distinctUntilChanged

that survives a Flow turning cold or • a

stateIn(…, replay = false)

that ignores the upstream value because it’s the same as its current state but that doesn’t replay it.

I’ve implemented a

(Mutable)CacheFlow

that extends

(Mutable)StateFlow

. The main difference is that it can be recursive (normal MutableStateFlow) or non-recursive. When it’s non-recursive then providing a new value to the

MutableCacheFlow

will not emit it to the same flow, thereby avoiding recursive scenarios. Collectors of the recursive cache behind the non-recursive cache will still receive newly emitted value. That means if we have a

cache

and two non-recursive `MutableCacheFlow`s

flow1

and

flow2

then updates to

flow1

would only be collected by

flow2

and updates to

flow2

would only be collected by

flow1

. This is especially useful when values coming out of the cache are transformed downstream and then committed to the cache again. With a regular

MutableStateFlow

that would cause an endless recursion. https://gist.github.com/fluidsonic/a06419b1c129f3045558c63cbafecca4 Thoughts?

Do I understand the documentation for

shareIn

in that terminal events are not propagated to subscribers?

Hello, World! Is there an easy way to implement

MutableStateFlow

? For instance, does something similar to

callbackFlow {}

exist?

Whats the flow equivalent of LiveDatas switchMap? i.e being able to change “input” flow depending on a secondary flow. I saw that Flow had a switchMap but that is deprecated

Is there a flow operator that provides previous emission? I would like to emit something based on current and previous emission. Tried

runningReduce

but it’s not 100% what I need as I always want two last values from upstream rather then the current Value and accumalated value.

Hi, can I possibly collect a flow for as long as a certain period? I can't seem to find an operator for that, that is, if I don't cancel this flow before this duration has elapsed, cancel it for me

Is there a flow operator that would let me loop through a list indefinitely? {1, 2, 3, 4} -> 1, 2, 3, 4, 1, 2, 3, and so on. I would then pass on these values to the UI using

LiveData

.

I’m seeing an issue with Kotlin 1.4.10 and coroutines v1.4.1 on Android, where a breakpoint inside

Flow.map

won’t fire properly.

myFlow
   .map { input ->
        someFunction(input)  // breakpoint here doesn't fire
   }

However, if I extract that method, it fires:

val transform: suspend (input: ...) -> ... = {
    someFunction(input)  // breakpoint here fires!
}

myFlow
    .map(transform)

has anyone seen this?

Hi, beginner flow user here, please what operator do I need to conditionally run a flow

flow2

based on the condition of an upstream flow

flow1

? this is what I have tried but I am not sure if I actually should be doing the tagged line

flow1()
 .flatMapLatest { res ->
   if(res.conditionIsFalse) {
     flowOf() // works but wondering if there is an operator that doesn't require this
   } else {
     flow2()
   }
 }
 .onEach { flow2Res -> 
   // do something with flow2Res
 }
 .launchIn(scope)

I have a strange issue with

shareIn

. in an Android Studio project. The editor code-completes it, I can jump to the source of it, but when I compile I get a

Unresolved reference: shareIn

it sounds like the compiler and the editor isn’t in sync, but I don’t know how to solve it.

I want to make a generic class that combines and array of flows, runs a transformer and emits the result (the class should do more than this, so thats the reason to put it in a class rather than a function) this small example explains the problem

class FlowGroup<T>(
    items:List<MutableSharedFlow<T>>,
    transform: suspend (Array<T>) -> T
) {
    val flow = MutableSharedFlow<T>()
    init {
        MainScope().launch {
            combine(items, transform).collect {
                flow.emit(it)
            }
        }
    }
}

combine

wont compile in this code because it uses the

reified

qualifier, and I don’t know how to get that into this function. Any suggestions?

Hi everyone, I had a question regarding

suspend functions

vs

Flows

. Assuming I follow a Clean architecture approach for my application, when should i be using

suspend functions

and when should I use

Flows

? Assuming I am only ever handling cold streams of data, when should I use one over the other? Should the presentation layer only be exposed to

Flows

and

suspend functions

be used by data and domain layers? Should the

suspend functions

just be used for “one-off” type functions that need to be done on a separate thread?

I have flow of some ids lets say

flowOf(1, 2, 3)

how would I do some suspended operations on them, write results for each in detabase and then only emit once. rn

collect{}

is called every time new item is emited, which is giving me undesirable results. I tried to use

onCompletion

but it only is called when ViewModel dies.

How should I be replacing LiveData in a database call with flow?

How can I emit emptyFlow when result is null? I have this piece of code :

dao.get(areaAccountingId = areaAccountingId, assetId = assetId)
.filterNotNull()
.map {
     it.toDomainModel()
}.onEmpty {  }

I want onEmpty to get triggered but it doesn't

I want a flow that polls an API indefinately until explicitly canceled. Is there a native/function for that purpose ? Or Shoujld I loop in the flow producer? Or loop calling collect ?

Hi! Is there any blog post/other that explains why flatMapMerge/flatMapConcat are not recommended? It says “most likely, suspending operations in map will be sufficient”, but that doesn't explain how to handle “I have a Flow, and I want to map each element to another flow”

Is there any difference between

.conflate().map { … }

and

.conflate().mapLatest { … }

?

Is there a way for me to be able to create something like a

MutableStateFlow

without an initial state? I.e. I want a flow that I can make into a variable and reference and send data but I don't want to specify an initial state.

Is there a way to suspend an

emit()

invocation of a

MutableSharedFlow

until all subscribers have completed collecting the element? There’s no replay.

I am emitting a flow of

Either

, but it seems like the first emission is being mixed if there are more than one emission, so store and store2 are different instances with should have the same emissions:

store.stream().take(1).toList().also { println(it) }
store2.stream().take(2).toList().also { println(it) }

That snippet prints:

[Right(right=Success(data=[1, 2, 3, 4], isLoading=true))]
[Right(right=Success(data=[1, 2, 3, 4, 7, 8], isLoading=true)), Right(right=Success(data=[1, 2, 3, 4, 7, 8], isLoading=false))]

I'm experimenting with something like this, but now I'm getting warnings about using internal coroutine API:s which doesn't seem very good...

suspend fun <T> collectFromHere(flow: SharedFlow<T>): Flow<T> {
    val channel = Channel<T>()
    flow.collect { channel.send(it) }
    return object: Flow<T> {
        @InternalCoroutinesApi
        override suspend fun collect(collector: FlowCollector<T>) {
            for (v in channel) {
                collector.emit(v)
            }
        }
    }
}

Is there an equivalent Flow construct for the sequence predicates

any

and

all

? I need a way to short circuit a Flow if a value is ever found matching a predicate, similar to `anyMatch`/`allMatch`/`noneMatch` in the Java Streams API

Hello! I try to start/launch a Kotlin Flow from Groovy/Spock. The Flow I want to test is created by a Kotlin function. This is what I roughly try to achieve:

def results = []
BuildersKt.runBlocking(EmptyCoroutineContext.INSTANCE) {
    sut.createFlow(someArg).collect { results << it }
}

I have found this SO post but haven't discovered a simpler solution, yet. Can someone please help me out? Thanks!

Can some one confirm my understanding about emitting from a callback? See thread.

Forgive noob question, but coming from Rx background this tends to be a big issue… If I assign a

SharedFlow

or

StateFlow

initialized from a cold

Flow

(with

stateIn(...Eagerly...)

) to a

var

, but then later assign

null

to the `var`; once there are no subscribers left to the

SharedFlow

(and no references left to it) will it cancel the original cold

Flow

(and otherwise free its resources)?

What is the right way to atomically do What is the right way to atomically do

mutableStateFlow.value += 1

? Is it

var cached = mutableStateFlow.value
while (!mutableStateFlow.compareAndSet(cached, cached + 1)) {
    cached = mutableStateFlow.value
}

Hi all, 99% sure I am using

SharedFlow

incorrectly, but would be good to understand why. My scenario is that I have an Android app with a list. When I click an item in the list, I call a method in my viewModel that emits a sealed class via a SharedFlow. The Android Fragment is collecting this flow and handling the UI based on the sealed class collected from it. However, clicking the same list item multiple times is not changing the UI multiple times - it seems like it happens occasionally and then emits multiple items should I change the item I click on (i.e. return a different instance of the sealed class) My question is, is there a way I can emit items every-time regardless of whether the previous item was the same item to be emitted? I am sure this is a fairly common question, but maybe I have not quite worked out how to google it successfully