Hey Everyone! so is I am getting started with multiprocessing and wanted to know if there is something similar to pythons Pool.map() in kotlin? Or an example showing how can we make a process pool and then run an iterable callback in it.

Guys imagine I have these two sources of data:

val flowA: Flow<String>

suspend fun funB(): Int

How can I combine the result of both into a Flow?

Does

.collect

-in flows inside

.collect

works this ways?

Would love to know if there’s any suggestion/feedback to improve below snippet code ❤️

lifecycleScope.launchWhenCreated {
        val uploadsBody = mutableListOf<File>()
        files
            .asFlow()
            .map { originalFile ->
                originalFile
                    .compressImageAsFlow(this)
                    .collect { compressedImageFile ->
                        uploadsBody.add(compressedImageFile)
                    }
                uploads
            }
            .collect { result ->
                viewModel.uploadFile(this, result)
            }
}

Hello guys 🙂 I try to update the data into a Flow, but I have this (working) solution:

private val _uiState = MutableStateFlow(DeveloperUiState(isLoading = true, sectionsItems = DeveloperSections()))
val uiState: StateFlow<DeveloperUiState> = _uiState.asStateFlow()

...

_uiState.update {
    it.copy(
        sectionsItems = it.sectionsItems.copy(
            userSection = it.sectionsItems.userSection?.copy(
                options = it.sectionsItems.userSection.options.copy(
                    userPremium = it.sectionsItems.userSection.options.userPremium.copy(
                        isSelected = isChecked,
                        summary = if (isChecked) "Yes" else "No"
                    )
                )
            )
        )
    )
}

Is there a way to reduce this path to edit a specific variable from an object? Thanks for your inputs!

Is there an existing Flow method like this? I don’t see one, just curious if I’m missing it.

fun <T> Flow<T>.onChange(action: (oldValue: T, newValue: T) -> Unit): Flow<T>

New to Flow and I'm wondering if anyone can point me towards some resources for a particular use case. I want a Repository that holds a List<Message>. These messages are fetched from a backend service via GET request, but can also be updated at any time by events that come in from a pub/sub service. Currently the result from the backend is a MutableLiveData, but increasingly we're seeing ConcurrentModificationExceptions because we need to read and potentially do some processing on the list while events are coming in and updating the list. It seems like Flow's streams of data might help here but it would be nice to look at some concrete examples.

Hello, is it possible to call a collect in a collect? I have this code:

dateFlow.collectLatest { date ->
    repository.getContentFromDate(date).collect { result ->

I can change the date so the content update. But I have a favorite feature on the content. On the default date, the favorite is working well (I collect the new value) but when I update the date, the collect doesn’t work on favorite events. What am I doing wrong? Thanks.

Quick question:

suspend fun waitForNetworkConnection() = isConnectedFlow.filter { it }.single()

isConnectedFlow

is a StateFlow, initially

false

(let’s assume we are in airplane mode in an Android app), then network connection is back,

isConnectedFlow.value = true

, however, the

waitForNetworkConnection

never ends and I don’t know why, I can see it only enters in the

filter

once and the second emission is ignored? I’ve also tried

isConnectedFlow.first { it }

to no avail. Any help would be appreciated

I have a SharedFlow to emit user data to UI layer like below, and the user can click a button to call

refresh()

to update

userFlow

(let

emit(userRepository.getUser())

call again.) How do I implement

refresh()

?

// ViewModel

val userFlow = flow {
    emit(userRepository.getUser())
}.shareIn(viewModelScope, SharingStarted.WhileSubscribed(), replay = 1)

fun refresh() {
    // How to reload user?
}

Is there any kind of flow that mixes state and shared flow? I would like to have a flow that emits every time a new value (even if the value is equal to the last value) but has a

.value

parameter so that I can query the parameter at any point in time? (sort of a stateful share flow 🙂 )

Hi, as far as i know using suspend in function that returns flow is bad practice and you should use CoroutineContext.() insted. But what to do when it is a member function and you need to call suspend fun inside? This is how i would call it with suspend

class KeywordService(val keywordRepo: KeywordRepo) {

    fun getKeywordData(
        keywordId: UUID,
    ): Flow<String> {
        val name = keywordRepo.getKeywordName() // suspend function 
        return keywordRepo.getKeywordData(name) // function that returns flow
    }
}

Currently i am calling it with a flow builder, but i am not sure if this is correct.

class KeywordService(val keywordRepo: KeywordRepo) {

    fun getKeywordData(
        keywordId: UUID,
    ): Flow<String> = flow {
        val name = keywordRepo.getKeywordName() // suspend function 
        val dataFlow = keywordRepo.getKeywordData(name) // function that returns flow
        emitAll(dataFlow)
    }
}

This way it feels little bit clunky, is there any better way to do it?

Got a question regarding collecting a flow inside the scope of mapping some other flow, and I am a bit confused if I could improve the way I am currently approaching this. I have a use case where: I have a stateFlow that holds a Result type of an UUID. With that UUID I am starting a new subscription (a flow) that also returns a result type. Finally with this latest result, if it’s successful, I need to perform another API call to get the final data which I am interested combining the original stateflow data with And all this should at the end maps to the UI state of type ViewState More in thread: 🧵

I am collecting a flow, and inside the scope of that collect function I would like to perform 2 network requests in parallel that can start independently, and simply want the result of both in the end. Inside the scope of

.collectLatest { //here }

I am not under a coroutine scope, but simply a suspend function, therefore I do not have access to

CoroutineScope.async

. I should be able to do this somehow right? ps. I’m using collectLatest with the idea that if a new value comes, the two async calls I am doing should cancel themselves since I will not be needing their result anymore, is this a correct assumption?

I'd like to know the simplest way to make a StateFlow directly depend on another. For example, let's say in my viewmodel I have:

private val _stateFlow1 = MutableStateFlow(0)

Is there a way to define

_stateFlow2

so that

_stateFlow2.value = _stateFlow1.value * _stateFlow1.value

?

Is there no

flatMap-

family operator that ignores new upstream emissions while a mapping operation is being processed (i.e. mapped stream is not completed)? How could this be achieved? I don't want emissions buffered as with

concat

, but thrown away/ignored.

Hi, I have a question We can create LiveData or StateFlow in a similar way as below

val _liveData = MutableLiveData(0)
val _stateFlow = MutableStateFlow(0)

But In LiveData, we can ensure the data is saved and restored by using

val _liveData: MutableLiveData<Int> = savedStateHandle.getLiveData("Key", 0)

For StateFlow, is there a way (or API) to keep the last value and restore it, like what we have in LiveData?

I'm having troubles understanding the exception handling in the flow. Could someone explain why the throwable completes the flow even with the

catch { }

in the chain? If I'd want to have a infinite flow I must handle the exceptions with

try catch

in the lambdas because flow will close after any exception "leaking" to the stream, right? Or is there a way to do some explicit recover from the exception? Example code, only

0

will be emitted, next emit

1

is lost due to flow completion after exception.

flow {
  emit(0)
  emit(1) // this emit is never reached
}
  .onEach { if (it == 0) throw Throwable() }
  .catch { println(it) }
  .collect()

Did anyone try using

getOrAwayValue()

(the helper function from Google's testing samples) on

Flow#asLiveData()

? I'm not getting a value.

Is there a nice way to trigger a restart of a flow that is turned into a StateFlow using

statein

? I’ll show you my use case and try to explain what I mean in the thread 🧵

This message was deleted.

Is there something I can do for the use case of: I am observing a flow. I want to start an infinite loop when I receive a specific type (sealed, limited options) the flow gives me. When I get the same type again, I do not want to re-start this infinite loop but let it keep going. When I get a different type I want to stop that old infinite loop and do nothing until I again get the type I’m looking for. Code:

sealed interface Types {
  object One : Types
  object Two : Types
}
val someFlow: Flow<Types>

someFlow
  .filterIsInstance<Types.One>()
  .distinctUntilChanged()
  .collectLatest() { one ->
    // I want this to be cancelled on an emission of `Two` from `someFlow` but keep on going alive without interruption on a repeat emission of `One` if it was `One` before too
    coroutineScope {
      while(isActive) {
        doWork(one)
      }
    }
  }

can someone lead me to a way to collect a flow until some other flow emits something as a condition?? example:

someFlow.onEach { /*Do something*/}.takeUntil(someOtherFlowAsPredicate).collect()

Hi is there a way to know if there are active collectors on a SharedFlow? e.g. of a use case in Android app deciding how to show a notification if notificationFlow is collected by UI then the UI will show a snackbar if not (there are no active collectors) create a system notification

Hi. Am i understanding correctly that

Channel.receiveAsFlow

can process individual items in parallel by collectors? Or is it an internal "fan out". I just want to process a queue of emissions in parallel. Thanks!

val flow = channel.receiveAsFlow()
repeat(5) {  launch { flow.collect { /*receive channel item in separate coroutine */ } } }

Hello, I would like to pass a flow to another flow and collect the latest one:

var running: Flow<Pair<Event, String?>> = flow {}

    suspend fun execute(node: NodeInterface) {
        running = node.run()
    }

The first

execute()

is working, I collect it from running, but others call to

execute()

are not giving anything. How can I make it working correctly? What do I need to use? I tried with shareFlow, tried to collect the

node.run()

flow and emit to the

running

but nothing is collected 😕 Do I need to use channel instead of flow? Thanks for your input!

Hey all! I'm learning how state flows work and trying to test something with the

StateFlow

. My test looks like this:

@Test
fun `test state flow`() = runBlocking {
    val stateFlow = flowOf(1, 2, 3)
        .onEach {
            println(it)
        }
        .stateIn(
            scope = TestCoroutineScope(),
            started = SharingStarted.WhileSubscribed(),
            initialValue = 0
        )
    
    val results = mutableListOf<Int>()
    val job = stateFlow.collectAsync(TestCoroutineScope()) {
        results.add(it)
    }

    assertEquals(listOf(0, 1, 2, 3, 4), results)
    
    job.cancel()
}

I was expecting this test to pass but it is failing. The actual size of results is

1

and the value it contains is

3

. I debugged my test with

onEach

and in console every value from the source flow is printed which means the value was emitted but it was never received in the collector. Does anybody know why am I only receiving the last value in my collector in the above code?

From the docs:

A mutable state flow is created using MutableStateFlow(value) constructor function with the initial value. The value of mutable state flow can be updated by setting its value property. Updates to the value are always conflated. So a slow collector skips fast updates, but always collects the most recently emitted value.

I was wondering does this behavior

So a slow collector skips fast updates, but always collects the most recently emitted value.

also apply to

StateFlow

instances created with the

stateIn

method?

Is there any particular reason why

StateFlow/MutableStateFlow

doesn’t implement

ReadOnlyProperty/ReadWriteProperty

? It seems like it’s a perfect fit for a thread-safe property.

Look at

callbackFlow