Flow - Luồng dữ liệu

// suspend function: Trả về MỘT giá trị
suspend fun fetchUser(): User
 
// Flow: Trả về NHIỀU giá trị theo thời gian
fun observeUsers(): Flow<User>

fun simpleFlow(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(100)     // Suspend
        emit(i)        // Phát ra giá trị
    }
}

val numbersFlow = flowOf(1, 2, 3, 4, 5)

val listFlow = listOf("A", "B", "C").asFlow()
val rangeFlow = (1..10).asFlow()

fun locationFlow(): Flow<Location> = callbackFlow {
    val callback = object : LocationCallback() {
        override fun onLocationResult(result: LocationResult) {
            trySend(result.lastLocation)  // Phát ra location
        }
    }
    
    locationClient.requestLocationUpdates(request, callback, Looper.getMainLooper())
    
    awaitClose {
        // Cleanup khi flow bị cancel
        locationClient.removeLocationUpdates(callback)
    }
}

fun main() = runBlocking {
    simpleFlow().collect { value ->
        println("Received: $value")
    }
}
// Output:
// Received: 1
// Received: 2
// Received: 3

val flow = flowOf(1, 2, 3, 4, 5)
 
// Lấy tất cả thành list
val list = flow.toList()  // [1, 2, 3, 4, 5]
 
// Lấy giá trị đầu
val first = flow.first()  // 1
 
// Lấy giá trị cuối
val last = flow.last()    // 5
 
// Đếm
val count = flow.count()  // 5
 
// Reduce
val sum = flow.reduce { acc, value -> acc + value }  // 15

val flow = (1..10).asFlow()
 
// map - Transform từng giá trị
flow.map { it * 2 }  // 2, 4, 6, 8, ...
 
// filter - Lọc giá trị
flow.filter { it % 2 == 0 }  // 2, 4, 6, 8, 10
 
// take - Lấy n giá trị đầu
flow.take(3)  // 1, 2, 3
 
// drop - Bỏ n giá trị đầu
flow.drop(3)  // 4, 5, 6, 7, 8, 9, 10
 
// transform - Custom transformation
flow.transform { value ->
    emit(value)
    emit(value * 10)
}  // 1, 10, 2, 20, 3, 30, ...

(1..100).asFlow()
    .filter { it % 2 == 0 }    // Số chẵn
    .map { it * it }           // Bình phương
    .take(5)                   // Lấy 5 đầu
    .collect { println(it) }   // 4, 16, 36, 64, 100

fun userFlow(): Flow<User> = flow {
    // Runs on IO
    val user = api.getUser()
    emit(user)
}
.flowOn(Dispatchers.IO)  // Upstream chạy trên IO
 
// Collect chạy trên caller's context
viewModelScope.launch {
    userFlow().collect { user ->
        // Runs on Main (caller's context)
        updateUI(user)
    }
}

class CounterViewModel : ViewModel() {
    
    // MutableStateFlow với initial value
    private val _count = MutableStateFlow(0)
    
    // Public read-only StateFlow
    val count: StateFlow<Int> = _count.asStateFlow()
    
    fun increment() {
        _count.value++
    }
}
 
// Collect trong UI
@Composable
fun CounterScreen(viewModel: CounterViewModel) {
    val count by viewModel.count.collectAsState()
    
    Text("Count: $count")
}

class EventViewModel : ViewModel() {
    
    private val _events = MutableSharedFlow<UiEvent>()
    val events: SharedFlow<UiEvent> = _events.asSharedFlow()
    
    fun showToast(message: String) {
        viewModelScope.launch {
            _events.emit(UiEvent.ShowToast(message))
        }
    }
}
 
sealed class UiEvent {
    data class ShowToast(val message: String) : UiEvent()
    object NavigateBack : UiEvent()
}

// ❌ SAI - Tiếp tục collect khi app ở background
lifecycleScope.launch {
    viewModel.uiState.collect { state ->
        updateUI(state)
    }
}
 
// ✅ ĐÚNG - Chỉ collect khi STARTED
lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        viewModel.uiState.collect { state ->
            updateUI(state)
        }
    }
}

@Composable
fun UserScreen(viewModel: UserViewModel) {
    val uiState by viewModel.uiState.collectAsState()
    
    // Hoặc với lifecycle-awareness
    val uiState by viewModel.uiState.collectAsStateWithLifecycle()
    
    when (uiState) {
        is UiState.Loading -> LoadingScreen()
        is UiState.Success -> UserContent(uiState.data)
        is UiState.Error -> ErrorScreen(uiState.message)
    }
}

class SearchViewModel : ViewModel() {
    
    private val searchQuery = MutableStateFlow("")
    
    val searchResults: StateFlow<List<Result>> = searchQuery
        .debounce(300)                     // Đợi 300ms không có input mới
        .filter { it.length >= 2 }          // Ít nhất 2 ký tự
        .distinctUntilChanged()            // Không search lại nếu query giống
        .flatMapLatest { query ->          // Cancel search cũ nếu có query mới
            searchRepository.search(query)
        }
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5000),
            initialValue = emptyList()
        )
    
    fun onQueryChanged(query: String) {
        searchQuery.value = query
    }
}

@Dao
interface UserDao {
    @Query("SELECT * FROM users")
    fun observeAllUsers(): Flow<List<User>>  // Room tự emit khi data thay đổi
}
 
class UserRepository(private val dao: UserDao) {
    
    fun observeUsers(): Flow<List<User>> = dao.observeAllUsers()
        .flowOn(Dispatchers.IO)
}
 
// ViewModel
class UserListViewModel(private val repository: UserRepository) : ViewModel() {
    
    val users: StateFlow<List<User>> = repository.observeUsers()
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(5000),
            initialValue = emptyList()
        )
}