[Android] Flow Operator 알아보자

실무에 써볼때마다 적어두는 Flow operater입니다!

CollectLatest()

distinctUntilChangedBy()

동일한 키의 모든 후속 반복이 필터링되고 keySelector 기능으로 키가 추출되는 흐름을 반환합니다.
동일한 매개변수와 함께 distinctUntilChanged 연산자를 반복 적용해도 아무런 효과가 없습니다.

public fun <T, K> Flow<T>.distinctUntilChangedBy(keySelector: (T) -> K): Flow<T> =
    distinctUntilChangedBy(keySelector = keySelector, areEquivalent = defaultAreEquivalent)
    
    
private fun <T> Flow<T>.distinctUntilChangedBy(
    keySelector: (T) -> Any?,
    areEquivalent: (old: Any?, new: Any?) -> Boolean
): Flow<T> = when {
    this is DistinctFlowImpl<*> && this.keySelector === keySelector && this.areEquivalent === areEquivalent -> this // same
    else -> DistinctFlowImpl(this, keySelector, areEquivalent)

코드에 답이 숨겨져 있습니다.
=== 즉, 레퍼런스를 비교하여 다를시에 값을 emit합니다.
DistinctFlowImpl 을 보시면 더욱 자세히 알수 있습니다.

private class DistinctFlowImpl<T>(
    private val upstream: Flow<T>,
    @JvmField val keySelector: (T) -> Any?,
    @JvmField val areEquivalent: (old: Any?, new: Any?) -> Boolean
): Flow<T> {
    override suspend fun collect(collector: FlowCollector<T>) {
        var previousKey: Any? = NULL
        upstream.collect { value ->
            val key = keySelector(value)
            @Suppress("UNCHECKED_CAST")
            if (previousKey === NULL || !areEquivalent(previousKey, key)) {
                previousKey = key
                collector.emit(value)
            }
        }
    }
}

previous가 Null 혹은 동등하지 않다면, emit을 발생시켜 다운스트림합니다.

onStart()

이 흐름이 수집되기 시작 하기 전에 지정된 action 을 호출하는 흐름을 반환합니다.
이 action 은 업스트림 흐름이 시작되기 전에 호출되므로 SharedFlow 와 함께 사용되는 경우 이 onStart 작업 내부 또는 직후에 발생하는 업스트림 흐름의 배출이 수집된다는 보장이 없습니다

flowOf("a", "b", "c")
    .onStart { emit("Begin") }
    .collect { println(it) } // prints Begin, a, b, c

수집이 시작되기전에 onStart{} 내부 람다가 실행됩니다.
중요한점은 Consumer가 collect를 시작했을 시점에 호출됩니다.

// 우리의 예상에는 _searchRequsetFlow가 트리거 될때마다 onStart()가 실행될것 같지만 그렇지 않다.
val uiState: StateFlow<CustomerUiState> =
    _searchRequestFlow.map {
		CustomerUiState.DataLoad(customerRepository.fetchCustomerPagingList(shopId, it).data.customers))
    }.onStart {
    	emit(CustomerUiState.Loading)
    }.catch {
		emit(CustomerUiState.Error(it))
        
    }.stateIn(
        viewModelScope,
        SharingStarted.WhileSubscribed(5000L),
        CustomerUiState.Loading
    )

transform()

주어진 흐름의 각 값에 transform 기능을 적용합니다.
transform 의 수신자는 FlowCollector 이므로 transform 은 방출된 요소를 변환하거나 건너뛰거나 여러 번 방출할 수 있는 유연한 함수입니다.
이 연산자는 filter 및 map 연산자를 일반화하고 다른 연산자의 빌딩 블록으로 사용할 수 있습니다.

fun Flow<Int>.skipOddAndDuplicateEven(): Flow<Int> = transform { value ->
    if (value % 2 == 0) { // Emit only even values, but twice
        emit(value)
        emit(value)
    } // Do nothing if odd

map()이나 onEach()같은 많은 operater에 내부에 쓰인입니다.
수신함수가 FlowCollector()이기에 이곳에서 emit() 호출이 가능합니다.
저는 map()과 onEach()가 같이 필요했을때 transform()연산자를 사용했습니다.

val uiState: StateFlow<CustomerUiState> =

//transform을 이용해 매번 트리거 될때마다 Loading을 호출할 수 있게 되었다.
    _searchRequestFlow.transform<CustomerSearchRequest, CustomerUiState> { it ->
        emit(CustomerUiState.Loading)
        emit(CustomerUiState.DataLoad(customerRepository.fetchCustomerPagingList(shopId, it).data.customers))
    }.catch {
        emit(CustomerUiState.Error(it))
    }.stateIn(
        viewModelScope,
        SharingStarted.WhileSubscribed(5000L),
        CustomerUiState.Loading
    )

scan() ,runningFold()

초기 값을 포함한 모든 중간 결과를 내보내는 작업으로 지정된 흐름을 접습(누적합)니다. 
서로 다른 수집기 간에 공유되는 초기 값은 불변(또는 변형되지 않아야)이어야 한다는 점에 유의하십시오.

public fun <T, R> Flow<T>.runningFold(initial: R, @BuilderInference operation: suspend (accumulator: R, value: T) -> R): Flow<R> = flow {
    var accumulator: R = initial
    emit(accumulator)
    collect { value ->
        accumulator = operation(accumulator, value)
        emit(accumulator)
    }
}

runningReduce()와 다른점은 초기값이 있으며, 리턴타입이 초기값의 타입이라는 것이다.

reducer 패턴을 구현할때 사용하기 좋다.
map을 사용하지 못하는 이유는 초기값에 대한 문제 때문이다.

   val state: StateFlow<MainState> = events.receiveAsFlow()
        .runningFold(MainState(), ::reduceState)
        .stateIn(viewModelScope, SharingStarted.Eagerly, MainState())

    private fun reduceState(current: MainState, event: MainEvent): MainState {
        return when (event) {
            MainEvent.Loading -> {
                current.copy(loading = true)
            }
            is MainEvent.Loaded -> {
                current.copy(loading = false, users = event.users)
            }
        }
    }