SharedFlow

Channel

SharedFlow

class Navigator : NavigationHandler {

    private val _destination: MutableSharedFlow<ScreenName> = MutableSharedFlow(
        replay = 1,
        onBufferOverflow = BufferOverflow.DROP_OLDEST
    )

    override val destination: SharedFlow<ScreenName> = _destination.asSharedFlow()

    override fun navigate(navigationDestination: ScreenName) {
        _destination.tryEmit(navigationDestination)
    }
}

MutableStateFlow<T>.update/getAndUpdate/updateAndGet

.update

.update

.update

public inline fun <T> MutableStateFlow<T>.update(function: (T) -> T) {
    while (true) {
        val prevValue = value
        val nextValue = function(prevValue)
        if (compareAndSet(prevValue, nextValue)) {
            return
        }
    }
}

function(prevValue)

function may be evaluated multiple times, if value is being concurrently updated.

compareAndSet

fun <T, R> Flow<T>.foldToStateFlow(initial: R, op: (acc: R, next: T) -> R): StateFlow<R>

return channelFlow {
     launch(start = CoroutineStart.UNDISPATCHED) { 
       messageSourceFlow.collect { send(it) 
     }
     try { 
         turnOnMessages()
         awaitCancellation() 
     } finally {
        turnOffMessages() 
     }
}

launch(start = CoroutineStart.UNDISPATCHED) {

turnOnMessages

init

flow {
    val stopTimesFirst = getStopTimesFirst()
    println("preparing first data finished")
    emit(stopTimesFirst)
    
    val stopTimesSecond = getStopTimesSecond()
    println("preparing second data finished")
    emit(stopTimesSecond)
}.flatMapConcat { stopTimes ->
    tickerFlow(FIVE_MINUTES_IN_MS)
        .onEach { now ->
            println("Refreshing data: $now")
            results = refreshData(
                now,
                now.plusMinutes(90),
                stopTimes.toList(),
            )
            println("Refreshing data: done")
        }
}.flowOn(<http://Dispatchers.IO|Dispatchers.IO>)
    .launchIn(GlobalScope)
    .start()

tickerFlow

fun tickerFlow(period: Long, initialDelay: Long = 0) = flow {
    delay(initialDelay)
    while (true) {
        emit(ZonedDateTime.now(zoneId))
        delay(period)
    }
}

flatMapConcat

emit

flow

emit

map

emit

StateFlow

sealed class MyState{
   data class Success(
        val intList: List<Int>
        ....
   ) : MyState()
   data class Success2(
        ....
   ) : MyState()
}

update

myState.update{ currentState ->
   if(currentState is MyState.Success){
      currentState.copy(intList = newIntList())
   } else currentState
} 

fun newIntList(): List<Int> {...}

if/else

myState.updateIf<MyState, MyState.Success>{ mySuccessState -> ... }

fun loadUsersByRegion(regions: List<String>): Flow<List<User>>

fun fetchInfo(user: User): Info

.flatMapConcat { users ->
            flow {
                val usersInfo = users.map { user -> fetchInfo(user) }
                emit(usersInfo)
            }
        }

private var hasFirstBeenEmitted : Boolean = false

merge

flow {
    emit(a)

    while (true) { delay(1_000_000) }
}

awaitCancellation()

private val manualPinKeySynchronizationEvents = MutableSharedFlow<Any>(
        replay = 1,
        extraBufferCapacity = 1,
        onBufferOverflow = BufferOverflow.DROP_OLDEST
    )

replay = 1

merge(failedSynchronizationsSource(), manualSynchronizationSource())
            .distinctUntilChanged()
            .mapLatest { synchronizationRequired ->
                 ...

launchIn

Note that the resulting value of launchIn is not used and the provided scope takes care of cancellation.

launchIn

Job

.cancel()

val monitorDongleEuiJob = settings.dongleEui.takeWhile { dongleEui ->
        dongleEui.isNotEmpty()
    }.onEach { dongleEui ->
        // do something, but nothing happens
    }.catch {
        print(it.message)
    }.launchIn(coroutineScope)

val monitorDongleEuiJob = settings.dongleEui.onEach { dongleEui ->
    if (dongleEui.isEmpty()) {
        return@onEach
    }
    // do something
}.catch {
    print(it.message)
}.launchIn(coroutineScope)

takeWhile

sealed interface SocketResponse<out T> {
    data class Data<T>(val item: T) : SocketResponse<T>
    data class Error(val exception: Throwable) : SocketResponse<Nothing>
    object ConnectionLost : SocketResponse<Nothing>
    object Loading : SocketResponse<Nothing>
}

fun SocketFun() = flow<SocketResponse>{

emit(SocketResonse.Data(SomeClass(variableA, variableB)

emit(SocketResonse.ConnectionLost)

}

fun <T> Flow<T>.asResult(shouldEmit: Boolean = true): Flow<Result<T>> {
    return this
        .map<T, Result<T>> {
            Result.Success(it)
        }
        .onStart { emit(Result.Loading) }
        .catch {
            if (it is Exception) {
                errorLogger("resultException", it.message ?: "Exception in result Flow.")
                if (shouldEmit) {
                    emit(Result.Error(it.getRealException()))
                }
            } else {
                emit(Result.Error(CustomMessage.ExceptionMessage("System not responding.")))
            }
        }
}

SharedFlow

SharedFlow

tryEmit

false

abstract class Parent {
    abstract val flow1: StateFlow<Int>
    
    val flow2 = flow1.mapLatest { it.toString() }
    
}

class Child : Parent() {
    override val flow1 = MutableStateFlow(1).asStateFlow()
}

fun main(){
    val child = Child()
    
    runBlocking {
        child.flow2.collect {
            
        }
    }
}

catch

catch

try-catch

replay

Flow

Flow.buffer(capacity = 1, replay = 1, onBufferOverflowStrategy = DROP_OLDEST)