Is there an equivalent of

Observable.concatArrayEager(...)

in Flow?

Hey all! I have been thinkering on this problem for a bit, and I think there’s something I might be missing. I have a data class, MyThing1, which has an ID in it. data class

MyThing1(val id: String)

I have another data class which we don’t care about its content, but let’s call it

MyThing2

I have a Flow, that returns a list of MyThing1

fun myEmitter1(): Flow<List<MyThing1>>

I have another Flow that takes in input an ID and returns a flow of objects, so

fun myEmitter2(id: String): Flow<MyThing2>

For each MyThing1 received from the flow, I want to combine it with the latest MyThing2 that gets emitted by

myEmitter2

and return a

Flow<List<CombinedThing>>

given the CombinedThing is

data class CombinedThing(val myThing1: MyThing1, val myThing2: MyThing2)

. Any thought for doing it efficiently?

This class which basically does validation on input fields. Using

LiveData

to how can i convert the

MediatorLiveData

part to

StateFlow

import androidx.lifecycle.LiveData
import androidx.lifecycle.MediatorLiveData
import androidx.lifecycle.MutableLiveData

abstract class InputFrom {
    private val _isFromValid = MediatorLiveData<Boolean>()
    val isFromValid: LiveData<Boolean>
        get() = _isFromValid

    abstract val fields: List<LiveInputField<*>>

    fun init() {
        fields.forEach { inputField ->
            _isFromValid.addSource(inputField.value) { changedValue ->
                inputField.validate(changedValue)
                _isFromValid.value = fields.all {
                    it.isValid
                }
            }
        }
    }

    inner class LiveInputField<T>(
        private val errorMessage: String? = null,
        private val predicate: (T?) -> Boolean
    ) {

        val value = MutableLiveData<T>()

        private val _error = MutableLiveData<String>()
        val errorText: LiveData<String>
            get() = _error

        var isValid: Boolean = false

        fun validate(value: Any?) {
            @Suppress("UNCHECKED_CAST")
            return if (predicate(value as? T)) {
                _error.value = null
                isValid = true
            } else {
                _error.value = errorMessage
                isValid = false
            }
        }
    }
}

Example Usages:

class LoginFrom : InputFrom() {

    val username =
        LiveInputField<String>("At least 4 characters, only letters, numbers, ., _ allowed.") {
            !it.isNullOrBlank() &&
                Pattern.compile("[a-zA-Z0-9_.]{4,}")
                    .matcher(it)
                    .matches()
        }

    val password =
        LiveInputField<String>(
            "8 characters or more, at least one number, one uppercase, one lowercase."
        ) {
            !it.isNullOrBlank()/* &&
          Pattern.compile("((?=.*\\d)(?=.*[a-z])(?=.*[A-Z]).{8,})")
              .matcher(it)
              .matches()*/
        }

    override val fields: List<LiveInputField<*>>
        get() = listOf(
            username,
            password
        )
}

Hi, can anyone help me figure out how "cancel" an action inside a flow based on incoming values? Ex:

root flow: OperationStart ------ AnotherOperation ------ OperationCancel -------- AnotherBOperation
next flow: OperationStart, perform long running action while !OperationCancel

Hello, I'm looking for a combination of

debounce()

and

distinctUntilChanged()

. I want to debounce elements if they are the same. So

A B B C D

in quick succession will become

A B C D

. I can't just use

distinctUntilChanged()

, because I want to allow sending the same element twice, as long as it's not in quick succession. How would I achieve that?

What is the correct way to unsubscribe from a Flow?

val flow = myFlow()
flow.collect {
  if (it is Success) {
    // I found the value I was waiting for,
    // stop collecting values and go back to executing the function
  }
}

Considering following code snippet

viewLifecycleOwner.lifecycleScope.launch {
  val result = someFlow.first()
  doSomethingInUI(result)
}

where

doSomethingInUI

is not a

suspend

function. What will happen if scope is destroyed/cancelled while there is ongoing operation

someFlow.first()

(eg. database read). Is

first()

cancellable? Will it exit the launch block gracefully or do I need to always call

ensureActive()

before a call to non suspended function in such case? The same stands for

someFlow.onEach {
  doSomnethinginUI(it)
}.launchIn(viewLifecycleOwner.lifecycleScope)

Is there any guarantee

onEach

won’t be called after scope is destroyed/cancelled?

Any advice on how to approach this problem? I want to use

MutableStateFlow.compareAndSet()

, but with a `sealed interface`:

sealed interface State {
  object NotConnected : State
  class Connected(val connection: SomeConnection) : State
}

val stateFlow = MutableStateFlow<State>(State.NotConnected)
...
// Obviously doesn't work, can't use State.Connected in this way
stateFlow.compareAndSet(State.Connected, State.NotConnected)

I'm sure there used to be a

.filterNotNull()

Flow

operator 🤔 where did it go? Edit: IDE autocomplete fail. Explicitly importing it worked.

Hi all! I’m trying to implement so kind of event bus repository using SharedFlow Long story short: System can send events to TriggerRepository via sendEvent method Any component in the system should have an ability to observe trigger. Trigger should fire when there is sequence of events in specific order emmited. I’m struggling with transformation step where I should somehow collect sequence of events and call the trigger. Code snippet and example in thread. Any form of help is appreciated 🙂

Question: Is there a variant of

MutableStateFlow.compareAndSet(expect, update)

where

update

is evaluated lazily?

Why can't we

.map

a

StateFlow

and have it still be a

StateFlow

? Why the conversion to a regular

Flow

?

Has anyone observed any performance impact from accessing

StateFlow.value

very often? For example, every 15-30 ms. I realize I could just keep a reference, but I worry about it getting stale. Thanks for any insight

Hey all… I keep wrestling with this general problem… and I’m hoping someone can point me in the right direction. I think it’s just my lack of understanding. If I do something like this:

val scope = MainScope() + Job()
val scope2 = MainScope() + Job()


val someEmitter = MutableStateFlow("")

fun test3()= runBlocking {
    scope.launch {
        someEmitter.emit("1")
        someEmitter.emit("2")
        someEmitter.emit("3")
    }
    scope2.launch {
        someEmitter.collect {
            println(it)
        }
    }
}

why does scope2 collect operation get blocked by scope? do they have to be on different threads? Am I missing something? I’ve gotten this type of thing to work before, but only by using another thread as the collector.

For 

Flow.buffer()

 there’s 

BufferOverflow

 

.SUSPEND

, 

.DROP_OLDEST

 and 

.DROP_LATEST

. Is there a way to fail if the buffer is full and stop the collector?

Are those correct?

Instead of MutableLiveData with default value to MutableStateFlow
Instead of MutableLiveData without default value to MutableSharedFlow
Instead of SingleLiveEvent to Channel.receiveAsFlow

Hello everyone! 👋 Is there some best practice for using flow to get only one item? Like alternative Single in Rx, when I guaranteed want to get single value and not ready to process stream.

is there an equivalent to

flowOn

that takes a

CoroutineScope

? i basically want a way to have a

flow { }

block that runs inside an existing scope

Is there something like this

pipe

function (which I just made up) available in the coroutines library? (Sends emissions from one

Flow

to another

FlowCollector

) :

suspend fun <T, U> Flow<T>.pipe(destination: FlowCollector<U>, transform: (T) -> U) {
  collect { 
    destination.emit(transform(it))
  }
}

It feels so obvious but I’ve been looking at the available extension functions and haven’t spotted it. Thanks!

Hi guys, can anyone tell me what's the difference between sharedflow and statedflow? I read the post on Android Developer but still didn't get it.

How can I use

CoroutineExceptionHandler

in flow? for example

val flow = flow { emit("data") }
    .shareIn(viewModelScope, SharingStarted.WhileSubscribed(), 1)

Am I doing the right thing with these nested state flows? For example:

data class MyDomainObject(
  val id: String,
  val enabled: StateFlow<Boolean>
)

val myDomainObjects: StateFlow<List<MyDomainObject> = /* This data comes from somewhere */

// I would like a list of objects which have `enabled.value == true`, and
// I would like this to be updated any time any `enabled` state changes
val enabledObjects: Flow<List<MyDomainObject> = myDomainObjects.transform { objects ->
  combine(*objects.map { it.enabled }.toTypedArray()) { enabledStates ->
    emit(objects.filterIndexed { index, _ -> enabledStates[index] })
  }
}

The implementation seems slightly wrong, the way I’m using

index

to emit objects which are enabled. But I don’t see another way to do this. Any advice? Thanks in advance.

if I have this code

private fun SearchScreenEvent.eventToUsecase(): Flow<SearchState> {
    return when (this) {
        is SearchClicked -> searchUsecase(this.query)
        is SearchQueryChanged ->
            flowOf(this.query)
                .debounce(5000)
                .flatMapConcat { searchUsecase(this.query) }
    }
}

why debounce is not working here

private val networkResponse = onContinueClicked.filter { it }.zip(
        combine(
            email.filter { it.count() >= MIN_INPUT_LENGTH },
            password.filter { it.count() >= MIN_INPUT_LENGTH })
        { e, p -> Pair(e, p) }
    ) { _, cred ->
        val (user, pass) = cred
        repository.login(username = user, password = pass)
    }

I’m playing around with flows: aim is to do a simple login network request.

email

and

password

are

MutableStateFlow

, this works fine except I’d like to emit a

Loading

value before the second to last line. The

login

call does not return a flow. I think I’m missing the use of the correct operator here in place of

zip

or I should have my repository return a flow (and

startWith

a given state in the repo).

hey all — i have a pretty common construction where i’m in an Android ViewModel and need to observe a Flow, do some stuff, and then push the result into a 

MutableStateFlow

. Looks something like this:

fun whatever() { 
	viewModelScope.launch { 
		inputFlow
			.map {...}
			.collect { value -> 
				mutableStateFlow.value = value
			}
	}	
}

(Assume that I need to push other values into the 

MSF

 outside this 

collect

 block) One issue I’ve run into with the above construction is that when the observer of the

MSF

cancels collection (e.g. the Fragment observing the ViewModel falls off the screen), it doesn’t cancel this intermediate collection…this makes absolute sense — that cancellation signal is not being propagated upstream! So digging into the

stateIn

sources, I’ve come up with what I think is a way to push the cancellation “upstream” — can y’all give thoughts/comments/suggestions on the following approach?

fun <T> Flow<T>.stateInto(downstream: MutableStateFlow<T>, scope: CoroutineScope) {
    val upstream = this
    scope.launch {
        downstream.subscriptionCount
            .mapLatest { count -> //this is a simplified version of [StartedWhileSubscribed#command]
                if (count > 0) true
                else {
                    delay(800.milliseconds)
                    false
                }
            }
            .dropWhile { active -> !active }
            .distinctUntilChanged()
            .collectLatest { active ->
                when (active) {
                    true -> upstream.collect(downstream) //will be cancelled upon new emission
                    false -> Unit /*just cancel and do nothing*/ 
                }
            }
    }
}

Instead of LiveData, I'm using StateFlow in my ViewModel to emit values to the UI. I realized that StateFlow doesn't emit equal values. How do you approach that? What if you want to emit multiple times the same value to show some message pop-up/toast/smth?

Hi i'm having a hard time converting this RxJS code to Kotlin flow Some how in java script after the buffer we are getting the stream events as array , however in flow i'm not receiving it has an array of events. Is this because buffer() behaves differently from bufferWhen in RxJS ?

var button = document.querySelector('.button');

var label = document.querySelector('h4');

var clickStream = Rx.Observable.fromEvent(button, 'click');

var doubleClickStream = clickStream

  

.bufferWhen(() => clickStream.debounceTime(250))

  

.map(arr => arr.length)

  

.filter(len => len === 2);

doubleClickStream.subscribe(event => {

  

label.textContent = 'double click';

});

doubleClickStream

  

.delay(1000)

  

.subscribe(suggestion => {

    

label.textContent = '-';

  

});

Based on a list of ids I need to retrieve a list of users with detailed informaiton of each user. Below a simplification of my models and API;

data class UserDTO(val id: String)
    data class PetDTO(val name: String)
    
    interface Api {
      suspend fun users(): List<UserDTO>
      suspend fun pets(userId: String): List<PetDTO>
    }
    
    data class User(
      val id: String,
      val pets: List<Pet>
    )
    
    data class Pet(val name: String)
    
    class UsersRepository(api: Api) {
      val users: Flow<List<User>> = TODO()
    }

In RxJava I would do something like this:

val users: Observable<List<User>> = api.users()
      .flatMapIterable { it }
      .concatMap { userDto ->
        api.pets(userDto.id)
          .map { petsListDto ->
            User(userDto.id, petsListDto.map { Pet(it.name) })
          }
      }.toList()

How can I implement

UsersRepository

and return the list of `User`s using Kotlin Flow?

Newbie question: when querying a database using a reactive driver, what are the advantages and differences in behaviour when returning a flow from a repository function, as opposed to returning a list from a suspend function?

I noticed during testing that within a

callbackFlow { ... }

if we throw a plain

Exception

this halts the VM immediately, and is not caught as an in-

Flow

`.catch`able error, while throwing a

RuntimeException

is `.catch`able. While this isn't so surprising given the conventional relationship between

Exception

and

RuntimeException

, I can't find this documented in respect to Flows - should it be? Is it?