Flows

Là một API thuộc Kotlin Coroutines dùng để xử lý stream dữ liệu bất đồng bộ (asynchronous data stream).

1. Các toán tử trong Flow

a. Nhóm toán tử trung gian (Intermediate Operators)

Đây là các toán tử dùng để biến đổi dòng dữ liệu (Upstream) trước khi đến tay người nhận (Downstream). Đặc điểm của chúng là Lazy (chỉ chạy khi có collect).

map: Biến đổi từng giá trị phát ra thành một giá trị khác.

filter: Chỉ cho phép các giá trị thỏa mãn điều kiện đi qua.

onEach: Thực hiện một hành động phụ (như log dữ liệu hoặc lưu cache) mà không làm thay đổi giá trị của dòng dữ liệu.

transform: Toán tử linh hoạt nhất. Có thể emit 0, 1 hoặc nhiều giá trị cho mỗi phần tử đầu vào.

debounce: Chỉ phát giá trị nếu sau một khoảng thời gian nhất định không có giá trị mới nào xuất hiện. Cực kỳ hữu dụng cho tính năng Search.

b. Nhóm toán tử kết hợp (Combining Operrators)

Dùng khi có nhiều nguồn dữ liệu (nhiều Flow) và muốn gộp chúng lại.

zip: Kết hợp các cặp giá trị từ 2 flow theo đúng thứ tự. Nó đợi cả hai flow phát ra giá trị mới rồi mới gộp lại.

fun main() = runBlocking {
    // Luồng Số: Phát ra 1, 2, 3 (mỗi phần tử cách nhau 100ms)
    val numbersFlow = flowOf(1, 2, 3).onEach { delay(100) }
    
    // Luồng Chữ: Phát ra "A", "B", "C", "D" (mỗi phần tử cách nhau 150ms)
    // Lưu ý: Luồng chữ có nhiều phần tử hơn và chạy chậm hơn
    val lettersFlow = flowOf("A", "B", "C", "D").onEach { delay(150) }

    // zip bắt buộc phải có cặp mới chạy. Phần tử thừa ("D") sẽ bị bỏ qua.
    numbersFlow.zip(lettersFlow) { number, letter -> 
        "$number$letter" 
    }.collect { result ->
        println("Zip phát ra: $result")
    }
    // Kết quả: 1A, 2B, 3C (Luôn đi theo cặp tương ứng)
}

combine: Kết hợp giá trị mới nhất của các flow. Chỉ cần 1 trong 2 flow phát ra giá trị mới, nó sẽ lấy giá trị đó gộp với giá trị hiện tại của flow kia.

fun main() = runBlocking {
    // Luồng Số: Phát ra 1, 2, 3 (mỗi phần tử cách nhau 100ms)
    val numbersFlow = flowOf(1, 2, 3).onEach { delay(100) }
    
    // Luồng Chữ: Phát ra "A", "B", "C", "D" (mỗi phần tử cách nhau 150ms)
    val lettersFlow = flowOf("A", "B", "C", "D").onEach { delay(150) }

    numbersFlow.combine(lettersFlow) { number, letter -> 
        "$number$letter" 
    }.collect { result ->
        println("Combine phát ra: $result")
    }
    // Kết quả: 1A, 2A, 2B, 3B, 3C, 3D
}

Ứng dụng: Kết hợp Filter (lọc) và Search Query để hiển thị danh sách sản phẩm.

merge : Gộp các flow lại. Emit data bất cứ khi nào 1 flow emit

fun main() = runBlocking {
    val numbersFlow = flowOf(1, 2, 3).onEach { delay(100) }
    val lettersFlow = flowOf("A", "B", "C", "D").onEach { delay(150) }

    val stringNumbersFlow = numbersFlow.map { it.toString() }
    
    merge(stringNumbersFlow, lettersFlow).collect { result ->
        println("Merge phát ra: $result")
    }
    // Kết quả: 1, A, 2, B, 3, C, D
}

c. Nhóm toán tử đặc thù cho StateFlow & SharedFlow

Chuyển đổi (Conversion)

stateIn: Biến một Cold Flow thành StateFlow.

shareIn: Biến một Cold Flow thành SharedFlow.

Phát dữ liệu (Emission)

.value (Only for StateFlow): Truy cập hoặc cập nhật giá trị hiện tại một cách đồng bộ.

Dùng .value khi quản lý Trạng thái UI (ví dụ: isLoading, userList).

emit(): Là suspend function để phát dữ liệu. Nếu buffer đầy, nó sẽ tạm dừng (suspend) cho đến khi có chỗ trống.

Dùng emit() khi truyền dẫn Luồng dữ liệu hoặc Sự kiện mà tính toàn vẹn của từng dữ liệu là bắt buộc.

tryEmit(): Hàm không suspend. Trả về true nếu phát thành công, false nếu thất bại (do buffer đầy). Thường dùng trong các trường hợp không thể dùng coroutine.

d. Nhóm toán tử Terminal (Toán tử kết thúc)

Đây là các toán tử "kích hoạt" dòng chảy. Nếu không gọi các hàm này, Flow sẽ không bao giờ thực thi.

collect: Thu nhận toàn bộ dữ liệu.

first(): Chỉ lấy giá trị đầu tiên rồi đóng Flow.

toList() / toSet(): Gom tất cả dữ liệu vào một Collection.

reduce / fold: Kết hợp các giá trị lại thành một giá trị duy nhất (giống như tính tổng danh sách).

e. Kiểm soát ngữ cảnh (Context Preservation)

flowOn: thay đổi Dispatcher cho các toán tử phía trước nó.

flatMapConcat: Xử lý các Flow con một cách tuần tự. Flow con thứ nhất phải hoàn thành (onCompleted) thì Flow con thứ hai mới bắt đầu.

runBlocking {
    val startTime = System.currentTimeMillis()
    getIdsFlow()
        .flatMapConcat { id -> fetchUserData(id) }
        .collect { value -> println(value) }
    println("Tổng thời gian: ${System.currentTimeMillis() - startTime}ms")
}

// KẾT QUẢ:
// Bắt đầu lấy data cho ID: 1
// Hoàn thành ID: 1
// Bắt đầu lấy data cho ID: 2
// Hoàn thành ID: 2
// Bắt đầu lấy data cho ID: 3
// Hoàn thành ID: 3
// Tổng thời gian: ~1800ms (100*3 + 500*3)

- flatMapMerge: Xử lý các Flow con song song. Nó có thể thu thập từ nhiều Flow con cùng lúc.

runBlocking {
    val startTime = System.currentTimeMillis()
    getIdsFlow()
        .flatMapMerge { id -> fetchUserData(id) }
        .collect { value -> println(value) }
    println("Tổng thời gian: ${System.currentTimeMillis() - startTime}ms")
}

// KẾT QUẢ:
// Bắt đầu lấy data cho ID: 1
// Bắt đầu lấy data cho ID: 2
// Bắt đầu lấy data cho ID: 3
// Hoàn thành ID: 1
// Hoàn thành ID: 2
// Hoàn thành ID: 3
// Tổng thời gian: ~600ms (Chỉ tốn khoảng thời gian của 1 lần call API dài nhất)

- flatMapLatest: Khi một giá trị mới được phát ra từ Flow gốc, nó sẽ hủy bỏ Flow con đang chạy trước đó để bắt đầu Flow con mới cho giá trị vừa nhận. Thích hợp dùng cho tính năng Search.

runBlocking {
    val startTime = System.currentTimeMillis()
    getIdsFlow()
        .flatMapLatest { id -> fetchUserData(id) }
        .collect { value -> println(value) }
    println("Tổng thời gian: ${System.currentTimeMillis() - startTime}ms")
}

// KẾT QUẢ:
// Bắt đầu lấy data cho ID: 1
// Bắt đầu lấy data cho ID: 2 (ID 1 bị hủy vì chưa kịp qua 500ms delay)
// Bắt đầu lấy data cho ID: 3 (ID 2 bị hủy)
// Hoàn thành ID: 3
// Tổng thời gian: ~800ms

2. Flow chạy trên Thread nào?

Flow chạy trên context của coroutine gọi collect()

Dùng flowOn() để chuyển đổi context của flow.

Q: Phân biệt Cold Flow và Hot Flow? Dùng trong trường hợp nào?

Cold Flow (Flow):

Chỉ bắt đầu emit data khi được collect.
Mỗi Collector sẽ nhận một dòng dữ liệu độc lập.
Ex: Truy vấn Database qua Room hoặc call API

Host Flow (StateFlow, SharedFlow):

Emit data ngay cả khi không có Collector nào.
Dữ liệu tồn tại ngoài luồng thu nhận và có thể chia sẻ giữa nhiều Collector.
Ex: Cập nhật UI State hoặc các sự kiện điều hướng (Navigation Events)

Q: So sánh StateFlow và SharedFlow.

StateFlow:

Phải có giá trị khởi tạo
Chỉ phát giá trị mới nhất cho collecter mới
Có tính chất Conflation: Nếu giá trị mới giống giá trị cũ (distinctUntilChanged), nó sẽ bỏ qua.

Phù hợp để quản lí Trạng thái (State)

private val _uiState = MutableStateFlow(UiState())
val uiState: StateFlow<UiState> = _uiState

ShareFlow:

Không cần giá trị khởi tạo
Cấu hình được replay (gửi lại bao nhiêu giá trị cho Collector mới)
Phù hợp cho xử lý Sự kiện (Events) như show SnackBar, Navigate, Toast.

1. Các toán tử trong Flow​

1. Các toán tử trong Flow