Flow

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

// 1) Cold flow — restarts for each collector
fun ticker(): Flow<Int> = flow {
    var i = 0
    while (true) {
        delay(1000)
        emit(i++)
    }
}

scope.launch {
    ticker().take(5).collect { println(it) }   // 0,1,2,3,4 then completes
}

// 2) Operators — map / filter / debounce / distinct
ticker()
    .map { it * 2 }
    .filter { it % 4 == 0 }
    .onEach { println("got $it") }
    .catch { e -> println("error $e") }
    .launchIn(scope)

// 3) Backpressure — collect is sequential by default
flowOf(1, 2, 3, 4, 5)
    .onEach { delay(100) }
    .collect { println(it) }                   // one at a time

// 4) Buffer / conflate when producer is faster
fastFlow.buffer().collect { slowProcess(it) }
fastFlow.conflate().collect { slowProcess(it) }  // skip intermediate values

// 5) flowOn — change context for upstream operators only
val rows: Flow<Row> = flow {
    val data = db.query()        // blocking
    data.forEach { emit(it) }
}.flowOn(Dispatchers.IO)

// 6) Combine flows
val combined = combine(usersFlow, postsFlow) { users, posts ->
    users.map { u -> u to posts.filter { it.userId == u.id } }
}

// 7) flatMapLatest — switchMap behavior
searchQuery
    .debounce(300)
    .distinctUntilChanged()
    .flatMapLatest { q -> if (q.isBlank()) flowOf(emptyList()) else api.searchFlow(q) }
    .collect { showResults(it) }

// 8) Convert a callback API to a flow — callbackFlow
fun locationFlow(client: LocationClient): Flow<Location> = callbackFlow {
    val cb = object : LocationCallback {
        override fun onLocation(loc: Location) {
            trySend(loc)
        }
    }
    client.subscribe(cb)
    awaitClose { client.unsubscribe(cb) }
}

// 9) StateFlow — HOT, holds latest value, perfect for UI state
class UserViewModel : ViewModel() {
    private val _state = MutableStateFlow<UiState>(UiState.Idle)
    val state: StateFlow<UiState> = _state.asStateFlow()

    fun load() = viewModelScope.launch {
        _state.value = UiState.Loading
        _state.value = runCatching { api.fetch() }
            .fold({ UiState.Success(it) }, { UiState.Failure(it.message ?: "") })
    }
}

// In a Compose UI:
// val state by viewModel.state.collectAsState()

// 10) SharedFlow — HOT, configurable replay (events, not state)
val events = MutableSharedFlow<Event>(replay = 0, extraBufferCapacity = 64)

fun fire(e: Event) {
    events.tryEmit(e)
}

// 11) Convert StateFlow to LiveData (Android-only)
// val live = viewModel.state.asLiveData()

// 12) Differences cheat sheet
//   Flow            : cold, restart per collector
//   StateFlow       : hot, single value, deduplicated by ==
//   SharedFlow      : hot, multi-value, configurable replay
//   Channel         : hot, point-to-point, NOT a Flow but composes with one

// 13) Best practices
//   • Expose `StateFlow` from ViewModels for UI state
//   • Expose `SharedFlow` (replay = 0) for one-off events (toast, navigation)
//   • Use `flowOn(IO)` for blocking upstream work
//   • Always collect inside a coroutine scope tied to a lifecycle

fun ticks(): Flow<Int> = flow {
    var n = 0
    while (true) { delay(1000); emit(n++) }
}

ticks().take(3).collect { println("tick $it") }