Coroutine Flow

개념

Flow는 비동기로 동작하면서 여러개의 값을 반환하는 function을 만들때 사용하는 coroutine builder이다.

fun foo(): Flow<Int> = flow { // flow builder
    for (i in 1..3) {
        delay(100) // pretend we are doing something useful here
        emit(i) // emit next value
    }
}

fun main() = runBlocking<Unit> {
    // Launch a concurrent coroutine to check if the main thread is blocked
    launch {
        for (k in 1..3) {
            println("I'm not blocked $k")
            delay(100)
        }
    }
    // Collect the flow
    foo().collect { value -> println(value) }
}

• Flow로 작성한 위 코드는 launch 코루틴과 flow연산이 각각 따로 실행되고 출력된다.

//출력결과
I'm not blocked 1
1
I'm not blocked 2
2
I'm not blocked 3
3

• 결과를 보면 launch내 반복출력과 flow를 활용한 연산이 동시에 이루어지고 있는걸 확인할 수 있다.
• Flow 타입의 생성은 flow{ } 빌더를 사용
• 결과 값들은 flow에서 emit() 함수를 사용해 방출
• flow에서 방출된 값은 collect 함수를 이용해 수집
• foo() 함수 내부의 코드는 foo().collect 함수를 호출 시 실행
• flow { } 블록 내부에서 emit() 메소드는 collect { } 블록으로 값을 전송(방출)

콜드 스트림

• flow{ } 빌더의 코드블럭은 플로우가 수집(Collect)되기 전까지 실행되지 않는다는 것을 의미

// flow 반환 함수
fun foo(): Flow<Int> = flow {
    println("Flow started")
    for (i in 1..3) {
        delay(100)
        emit(i)			// collect에 값 방출
    }
}

fun main() = runBlocking<Unit> {
    println("Calling foo...")
    val flow = foo()
    println("Calling collect...")
    flow.collect { value -> println(value) } 	// flow 수집 1
    println("Calling collect again...")
    flow.collect { value -> println(value) } 	// flow 수집 2
}

//출력결과
Calling foo...
Calling collect...
Flow started 			// collect 수집 (foo() 함수 실행)
1
2
3
Calling collect again...
Flow started 			// collect 수집 (foo() 함수 실행)
1
2
3

• 결과를 보면 flow를 반환하는 foo() 함수가 suspend로 표시되지 않음
• foo() 함수는 호출 시 바로 반환되며 어떤 무엇도 기다리지 않음
• 플로우는 매번 수집(collect) 될 때 마다 새로 시작

플로우 빌더

• 앞에서 본 flow{ } 사용 예시는 가장 기본적인 플로우 빌더
• .asFlow () - 다른 Collection/Sequence들을 -> Flow로 변환
  ex) List / Map / Array -> Flow로 변환
  

// (1..3)는 Int형 배열 의미
(1..3).asFlow().collect { value -> println(value) }

//출력결과
1
2
3

• (1..3)은 1,2,3의 Int형 배열 생성을 의미
• .asFlow()는 다른 컬렉션/시퀸스를 Flow로 변환, (int배열(1..3)을 -> Flow로 변환)
• collect { }는 수집의 개념으로 Flow의 각각 값(1,2,3)을 처리하는 코드 블럭

플로우 중간 연산자

• 플로우는 컬렉션과 시퀸스처럼 중간 연산자를 사용해 변환이 가능
• 중간 연산자도 마찬가지로 플로우와 같이 콜드(Cold) 타입으로 동작
• 중간 연산자는 suspend(중단)함수가 아니기에 새롭게 변환된 플로우를 즉시 반환 (지연x)
• Collection/Sequence와 차이점은 중간 연산자(map, filter등) 블록 내부에서 suspend(중단)함수 호출 가능

suspend fun performRequest(request: Int): String {
    delay(1000) // 1초 대기
    return "response $request" // 플로우 값 매핑 "request -> response $request"
}

fun main() = runBlocking<Unit> {
    (1..3).asFlow() // .asFlow() - 배열 -> Flow
        .map { request -> performRequest(request) } // 대상 플로우를 매핑해서 새로운 플로우로 반환
        .collect { response -> println(response) } // map으로 반환된 새 플로우에 대한 수집(collect)
}

//출력결과
response 1
response 2
response 3

• 1,2,3의 배열을 Flow로 변환 후 중간 연산자(map)를 이용해서 매핑 후 새로운 Flow를 반환
• Flow요소들 매핑 (1,2,3 -> response 1, response2, response3)
• 새로 반환받은 플로우에 대해 수집(collect) 처리 (println)

변환 연산자

• 플로우 변환 연산자들 중에서 가장 일반적인 연산자는 transform 연산자
• transform 연산자는 중간 연산자(map, filter) 같이 단순한 변환 보다 복잡한 변환을 처리위한 연산자
• 복잡한 변환 - 임의의 횟수나 임의의 값으로 변환 하는것을 의미

suspend fun performRequest(request: Int): String {
    delay(1000) // 1초 대기
    return "response $request" // 플로우 값 매핑 "request -> response $request"
}

fun main() = runBlocking<Unit> {
    (1..3).asFlow() // 배열 -> Flow 변환
    	.transform { request ->		// 변환 연산자(transform)
        	emit("Making request $request") 	// emit() - flow 방출
        	emit(performRequest(request)) 		// emit() - flow 방출
    	}.collect { response -> println(response) }	// flow 수집
}

//출력결과
Making request 1 	// Flow 요소 1
response 1
Making request 2 	// Flow 요소 2
response 2
Making request 3 	// Flow 요소 3
response 3

• 중간 연산자(map)은 요소마다 1개의 변환가능
• 변환 연산자(transform)은 예시처럼 emit()을 추가하여 요소마다 여러개의 변환이 가능