3부 채널과 플로우 - 2

[lee-ji-hoon]

19, 20, 21

[lee-ji-hoon]

hot vs cold

핫 플로우 (Hot Flow)

핫 플로우는 데이터 스트림이 항상 활성화되어 있으며, 구독자가 없어도 데이터를 방출하는 특성을 갖습니다. 여러 구독자가 있을 때 각 구독자는 스트림의 다른 지점에서 데이터를 받을 수 있습니다.

ex) LiveData, SharedFlow, StateFlow

• 데이터의 생산이 구독자의 존재에 의존하지 않음
• 이미 시작된 스트림에 새로운 구독자가 중간에 참여될 수 있다.

콜드 플로우 (Cold Flow)

콜드 플로우는 구독자가 있어야만 데이터를 방출합니다. 새로운 구독자가 생기면 처음부터 데이터를 방출하기 시작합니다.

ex) Flow

• 구독자가 있을 때만 데이터를 생산.
• 각 구독자는 데이터 스트림의 시작부터 값을 받음.

Capacity와 BufferOverflow

Capacity

capacity는 버퍼의 크기를 설정하는 옵션입니다. 버퍼는 데이터를 일시적으로 저장할 수 있는 메모리 공간으로, 생산자가 데이터를 방출하고 소비자가 이를 처리하는 동안의 차이를 조정합니다.

BufferOverflow

BufferOverflow는 버퍼가 가득 찼을 때 어떻게 처리할지를 정의합니다. 세 가지 옵션이 있습니다:

• SUSPEND: 새 데이터를 방출하려는 시도가 버퍼가 가득 차면 일시 중지합니다.
• DROP_OLDEST: 버퍼에서 가장 오래된 데이터를 버리고 새 데이터를 저장합니다.
• DROP_LATEST: 새 데이터를 버립니다.

CallbackFlow

callbackFlow는 코틀린의 Flow에서 콜백 기반 API를 플로우로 변환할 때 유용하게 사용되는 빌더 함수입니다.

여기서 awaitClose는 callbackFlow 내부에서 정의된 콜백이 닫히거나 취소될 때 호출되는 함수

awaitClose 호출 시점

• 콜백 기반 API에서 콜백을 취소할 때: callbackFlow 내에서 제공하는 콜백 API가 호출되어 플로우를 종료할 때.
• 플로우 컬렉션이 취소될 때: 플로우를 수집하는 코루틴이 취소되거나 완료될 때.
• callbackFlow 내에서 예외가 발생할 때: 플로우 내에서 예외가 발생하여 플로우가 종료될 때.

예제 코드

fun interface LocationCallback {
    fun onLocationChanged(location: String)
}

class LocationManager {
    private var callback: LocationCallback? = null

    fun startLocationUpdates(callback: LocationCallback) {
        this.callback = callback
        GlobalScope.launch {
            repeat(5) {
                delay(1000L)
                println("callback $it")
                callback.onLocationChanged("Location $it")
            }
        }
    }

    fun stopLocationUpdates() {
        callback = null
    }
}

fun locationFlow(locationManager: LocationManager): Flow<String> = callbackFlow {
    val callback = LocationCallback { location ->
        trySend(location)
    }
    locationManager.startLocationUpdates(callback)

    awaitClose {
        println("await close")
        locationManager.stopLocationUpdates()
    }
}

fun main() = runBlocking {
    val locationManager = LocationManager()
    val job = launch {
        locationFlow(locationManager)
            .flowOn(Dispatchers.IO)
            .collect { location ->
                println("Collected: $location")
            }
    }

    delay(7000L)
    println("cancelAndJoin()")
    job.cancelAndJoin()
}

1. LocationCallback 인터페이스: 위치 업데이트를 수신하기 위한 콜백 인터페이스를 정의합니다.

2. LocationManager 클래스: 위치 업데이트를 관리하는 클래스입니다. startLocationUpdates 메서드에서 위치 업데이트를 시뮬레이트하고, stopLocationUpdates 메서드에서 업데이트를 중지합니다.

3. locationFlow 함수: callbackFlow 를 사용하여 LocationManager의 콜백을 플로우로 변환합니다. awaitClose 블록 내에서 콜백을 해제하고 위치 업데이트를 중지합니다.

4. main 함수: locationFlow를 수집하고, 일정 시간이 지난 후 collect 을 취소

[lee-ji-hoon]

다음장에 할 것

• [ ] SafeFlow

https://flowmarbles.com/#combine