[Kotlin] 코루틴의 callbackFlow와 channelFlow

코루틴의 flow는 RX를 대체할 수 있을 만큼 막강하지만 가끔씩 마주하게 되는 문제들이 있습니다. Thread-Safety와 다운스트림으로의 문제 전달을 막기 위해 flow 블록 내에서는 Context를 하나만 사용할 수 있다는 점입니다.

withContext(Dispatchers.Main) {
         flowWithIOContext()
        .map { ... } // IO에서 작동
        .flowOn(Dispatchers.IO)
        .filter { ... } // Default에서 작동
        .flowOn(Dispatchers.Default)
        .single() // Main에서 작동
}

이런 문제를 위해 flow 파이프라인에서는 rx의 SubscribeOn과 유사하게 flowOn을 통해 context를 바꾸어줄 수 있지만 하나의 블록 안에서 동적으로 바꾸거나 콜백 함수 등에서는 사용할 수 없습니다. 이런 문제를 위해 callbackFlow와 channelFlow가 있습니다.

callbackFlow와 channelFlow에서의 기능상의 차이점은 없으며 사실상 이름만 다른 형태로 사용됩니다. 자세한 내용은 코루틴 깃헙페이지의 이슈에서 확인하실 수 있습니다. 당연히 callbackFlow와 channelFlow는 여러개의 Context를 사용할 수 있습니다.

lifecycleScope.launch {
 flow<Unit>{
  view.setOnClickListener {
    // 에러발생
    //suspend function should be called only from coroutine.
    this.emit(Unit)
  }
 }
}

만약 flow에서 setOnClickListener와 같은 콜백 함수에서 emit을 사용하게 되면 컴파일 에러가 발생하게 됩니다. 콜백 안에서 코 루틴 스코프가 정의되지 않았기 때문인데 만약 코 루틴 스코프를 지정한다고 해도 flow안에서는 스코프를 선언할 수 없기 때문에 또다시 에러가 발생합니다. 이럴 때 callbackFlow를 사용할 수 있습니다.

lifecycleScope.launch {
 callbackFlow<Unit> {
  view.setOnClickListener {
   offer(Unit) //값을 내보냅니다.
   close()
  }

  //callbackFlow안에서 close가 될때까지 기다립니다.
  awaitClose()
 }.collect { 
  //값을 받습니다.
 }
}

callbackFlow 내에서 작동은 BlockingQueue와 유사하게 작동을 하게 됩니다. 동기로 값을 내보낸다면 send 비동기로 값을 내보낸다면 offer를 사용할 수 있고

view.setOnClickListener {
 offer(Unit) //값을 내보냅니다.
 
 launch(Dispatchers.Main){
   send(Unit)
 }
 close(Error("에러원인"))
}.catch{
 // Close에서의 에러값이 전달됩니다.
}

callbackFlow 블록 안에서 Context를 변경할 수도 있습니다. BlockingQueue와의 차이점은 BlockingQueue와는 다르게 close에 원인이되는 문제를 null값이 아닌 Throwable로 넘겨준다면 catch 파이프라인에서 에러값을 핸들링하여 새로운 스트림을 만들어낼 수 있습니다.

또한 awaitClose는 람다형식의 콜백을 인자로받아 close될때의 실행되는 콜백함수를 정의할 수 있습니다.

awaitClose{
 apiData.callbackListern = null
 //또는 
 composeDisposable.dispose()
}

awaitClose를 이용하여 더이상 리스너 스트림이 필요하지않을때 콜백리스너에 null값을 할당하거나 혹은 스트림을 더이상 만들지않기위해 dispose등을 활용할 수 있습니다.