[Android] Coroutine Flow란? MVVM + Flow + Retrofit 예제

최근 안드로이드 진영에서 비동기 처리에 자주 사용했던 라이브러리인 RxJava가 걷어내지고 그 자리를 코루틴의 flow라는 것이 대신한다고 들었다.

그래서 구글에서 Compose를 비롯해 여러 방식으로 구현한 데모 앱인 Sunflow의 커밋 이력을 확인해보니 2020년 말에 data layer를 Flow를 사용하는 방식으로 바꿨다는 커밋 이력이 눈에 들어왔다.

 

 

얼마나 좋은 점이 있어서 안드로이드 진영에서 이런 선택을 한 건지 궁금해져서 flow를 찾아 공부해봤다.

 

먼저 flow의 사전적 정의는 아래와 같다.

 

흐름, 계속적인 공급(생산), 흐르다, 계속 흘러가다

 

RxJava의 데이터 발행과 비슷하다고 보면 되는 건가? 다음은 안드로이드 디벨로퍼에서 설명하는 flow다.

 

https://developer.android.com/kotlin/flow?hl=ko 

Android의 Kotlin 흐름  |  Android 개발자  |  Android Developers

코루틴에서 flow는 단일 값만 반환하는 suspend function(정지 함수)과 달리 여러 값을 순차적으로 내보낼 수 있는 유형이다. 예를 들면 flow를 써서 DB에서 실시간 업데이트를 수신할 수 있다. flow는 코루틴 기반으로 빌드되며 여러 값을 제공할 수 있다. flow는 비동기식으로 계산할 수 있는 데이터 스트림의 개념이다. 내보낸 값은 동일한 유형이어야 한다. 예를 들어 Flow<Int>는 정수값을 내보내는 flow다.
flow는 값 시퀀스를 생성하는 Iterator와 매우 비슷하지만 suspend function을 써서 값을 비동기적으로 생성, 사용한다. 예를 들어 flow는 기본 쓰레드를 차단하지 않고 다음 값을 생성할 네트워크 요청을 안전하게 만들 수 있다.
데이터 스트림에는 3가지 항목이 있다.

- 생산자 : 스트림에 추가되는 데이터 생산. 코루틴 덕분에 flow에서 비동기적으로 데이터가 생산될 수도 있다
- (선택사항) 중개자 : 스트림에 내보내는 값이나 스트림 자체를 수정할 수 있다
- 소비자 : 스트림의 값을 사용한다

 

데이터 스트림 관련 항목. 소비자, 중개자(선택사항), 생산자

안드로이드에서 Repository는 일반적으로 UI 데이터 생산자다. 이 때 UI는 최종적으로 데이터를 표시하는 소비자다. UI 레이어가 사용자 입력 이벤트의 생산자고 계층 구조의 다른 레이어가 이 이벤트를 쓰기도 한다. 생산자, 소비자 사이의 레이어는 일반적으로 다음 레이어의 요구사항에 맞게 조정하기 위해 데이터 스트림을 수정하는 중개자의 역할을 한다...(중략)

 

설명이 어렵다. 다른 사람들이 설명한 내용을 확인해봤다.

 

https://velog.io/@elliot/Android-Coroutine-Flow

[Android] Coroutine Flow

Flow는 비동기로 동작하면서 여러 개의 값을 반환하는 함수를 만들 때 사용하는 코루틴 빌더다

 

https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/-flow/

Flow

Flow는 순차적으로 값을 내보내고 정상적으로 또는 예외로 완료되는 비동기적인 데이터 스트림이다. map, filter, take, zip 등과 같은 Flow의 중간 연산자는 업스트림 Flow나 Flow에 적용돼 추가 연산자를 적용할 수 있는 다운스트림 Flow를 리턴하는 함수다. 중간 연산자는 Flow에서 코드를 실행하지 않고 함수 자체를 일시중단하지 않는다. 이것들은 향후 실행과 신속한 복귀를 위해 일련의 작업을 설정할 뿐이다. 이를 Cold Flow 프로퍼티라고 부른다...(중략)...기본적으로 Flow는 순차적이고 모든 Flow 작업은 동일한 코루틴에서 순차적으로 실행된다

 

자바 8부터는 컬렉션 안에 있는 자료들에 접근할 수 있는 스트림이라는 개념이 등장한다. 그러나 코틀린에는 스트림이 없는데 그 대신 map, filter 같은 스트림 함수들이 존재해서 개발자가 원하는 형태로 데이터를 가공해 원하는 데이터를 얻을 수 있다.

 

아무튼 Flow는 비동기로 동작하면서 여러 값을 리턴하는 함수를 만들기 위해 사용하는 Cold Stream 방식의 코루틴 빌더라고 할 수 있겠다.

RxJava로 치면 Cold Observable과 비슷하다. 때문에 Flow 이전에 RxJava를 공부하고 사용해 봤다면 Flow를 이해하기에 엄청나게 어렵지는 않을 것이다.

 

그런데 이 Flow는 왜 쓰는 것인가? Flow 이전에 내가 자주 사용하던 것은 LiveData로 이걸 써서도 충분했었다. Flow가 무엇이 더 좋길래 사용하는 건가 궁금해서 찾아봤다.

 

 

내 경우 Flow는 레트로핏으로 API를 호출해 데이터를 가져올 때 적용해 사용한다. 그 전에 아래의 예시를 보고 가자.

모든 소스코드는 안드로이드 스튜디오에서 작성하고 에뮬레이터로 실행했다.

 

import android.os.Bundle
import android.util.Log
import androidx.appcompat.app.AppCompatActivity
import androidx.lifecycle.Lifecycle
import androidx.lifecycle.repeatOnLifecycle
import com.example.kotlinprac.R
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.launch

class CoroutinePracticeActivity : AppCompatActivity() {

    private val TAG = this.javaClass.simpleName

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_coroutine_practice)

        CoroutineScope(Dispatchers.IO).launch {
            repeatOnLifecycle(Lifecycle.State.STARTED) {
                flowOfStrings.collect { data ->
                    Log.e(TAG, "flowOfStrings : $data")
                }
                printNumbers().collect { data ->
                    Log.e(TAG, ">> printNumbers() : $data")
                }
            }
        }
    }

    private val flowOfStrings = flow {
        for (number in 0..10) {
            emit("Emitting: $number")
        }
    }

    private fun printNumbers(): Flow<Int> = flow {
        for (i in 1..10) {
            emit(i) // emit(value: Int)
            Log.e(TAG, "$i emit됨")
        }
    }

}

 

숫자를 방출하는 Flow를 프로퍼티와 함수 형태로 만들고 onCreate() 안에서 사용하는 예시다.

여기서 repeatOnLifeCycle은 lifecycle-runtime-ktx 라이브러리 의존성을 추가해야 사용할 수 있는 건데, UI 계층에서 Flow를 수집할 때 권장되는 방법이다. 생명주기 단계를 매개변수로 받는 suspend 함수로써 액티비티나 프래그먼트가 생명주기의 특정 단계에 도달하면 새 코루틴을 자동으로 시작한다. 위 코드에선 STARTED를 썼는데 STARTED는 화면이 onStart() 호출 직후 ~ onPause() 호출 직전의 상태일 때 작동한다. 단 onResume() 상태일 때는 작동하지 않으므로 사용 시 주의해야 한다.

 

아무튼 이 예제를 실행하면 아래와 같은 로그가 출력되는 걸 볼 수 있다.

 

 

flowOfStrings 프로퍼티에서 모든 데이터가 방출된 다음 printNumbers()가 작동해서 데이터를 방출하는 걸 볼 수 있다. 이 과정에서 flowOfString과 printNumbers()가 꼬이지도 않는다.

이걸 통해 repeatOnLifeCycle {} 안에서 2개의 Flow를 실행할 경우 소스코드 상 먼저 호출한 Flow의 데이터 흐름이 먼저 시작되고 종료된 다음에야 그 다음 Flow의 데이터 흐름이 시작되는 걸 알 수 있다.

 

그러나 이 예제로 퉁치고 끝낼거면 이 포스팅 안 썼다. 실제로 레트로핏과 Flow를 써서 어떻게 서버 통신을 하는지도 확인해보자. 사용할 API는 깃허브 API다.

매니페스트에 인터넷과 필요한 설정들을 다 넣고 의존성도 다 넣었다고 가정하고 진행한다. 먼저 data class와 ApiService부터 만들어서 깃허브 API를 호출할 BASE URL을 설정한다.

data class의 경우 언더바(_)로 돼 있는 변수들이 있어서 내가 쓰기 편하게 언더바를 없앤 형태로 사용하기 위해 @SerializedName을 쓴 부분이 많다.

 

import com.google.gson.annotations.SerializedName

data class GithubData(
    @SerializedName("incomplete_results") val incompleteResults: Boolean,
    val items: List<Item>,
    @SerializedName("total_count") val totalCount: Int
)

data class Item(
    @SerializedName("allow_forking") val allowForking: Boolean,
    @SerializedName("archiveUrl") val archiveUrl: String,
    val archived: Boolean,
    @SerializedName("assignees_url") val assigneesUrl: String,
    @SerializedName("blobs_url") val blobsUrl: String,
    @SerializedName("branches_url") val branchesUrl: String,
    @SerializedName("clone_url") val cloneUrl: String,
    @SerializedName("collaborators_url") val collaboratorsUrl: String,
    @SerializedName("comments_url") val commentsUrl: String,
    @SerializedName("commits_url") val commitsUrl: String,
    @SerializedName("compare_url") val compareUrl: String,
    @SerializedName("contents_url") val contentsUrl: String,
    @SerializedName("contributors_url") val contributorsUrl: String,
    @SerializedName("created_at") val createdAt: String,
    @SerializedName("default_branch") val defaultBranch: String,
    @SerializedName("deployments_url") val deploymentsUrl: String,
    val description: String?,
    val disabled: Boolean,
    @SerializedName("downloads_url") val downloadsUrl: String,
    @SerializedName("events_url") val eventsUrl: String,
    val fork: Boolean,
    val forks: Int,
    @SerializedName("forks_count") val forksCount: Int,
    @SerializedName("forks_url") val forksUrl: String,
    @SerializedName("full_name") val fullName: String,
    @SerializedName("git_commits_url") val gitCommitsUrl: String,
    @SerializedName("git_refs_url") val gitRefsUrl: String,
    @SerializedName("git_tags_url") val gitTagsUrl: String,
    @SerializedName("git_url") val gitUrl: String,
    @SerializedName("has_downloads") val hasDownloads: Boolean,
    @SerializedName("has_issues") val hasIssues: Boolean,
    @SerializedName("has_pages") val hasPages: Boolean,
    @SerializedName("has_projects") val hasProjects: Boolean,
    @SerializedName("has_wiki") val hasWiki: Boolean,
    val homepage: String?,
    @SerializedName("hooks_url") val hooksUrl: String,
    @SerializedName("html_url") val htmlUrl: String,
    val id: Int,
    @SerializedName("is_template") val isTemplate: Boolean,
    @SerializedName("issue_comment_url") val issueCommentUrl: String,
    @SerializedName("issue_events_url") val issueEventsUrl: String,
    @SerializedName("issues_url") val issuesUrl: String,
    @SerializedName("keys_url") val keysUrl: String,
    @SerializedName("labels_url") val labelsUrl: String,
    val language: String?,
    @SerializedName("languages_url") val languagesUrl: String,
    val license: Licenses?,
    @SerializedName("merges_url") val mergesUrl: String,
    @SerializedName("milestones_url") val milestonesUrl: String,
    @SerializedName("mirror_url") val mirrorUrl: Any,
    val name: String,
    @SerializedName("node_id") val nodeId: String,
    @SerializedName("notifications_url") val notificationsUrl: String,
    @SerializedName("open_issues") val openIssues: Int,
    @SerializedName("open_issues_count") val openIssuesCount: Int,
    val owner: Owner,
    @SerializedName("private") val isPrivate: Boolean,
    @SerializedName("pulls_url") val pullsUrl: String,
    @SerializedName("pushed_at") val pushedAt: String,
    @SerializedName("releases_url") val releasesUrl: String,
    val score: Double,
    val size: Int,
    @SerializedName("ssh_url") val sshUrl: String,
    @SerializedName("stargazers_count") val stargazersCount: Int,
    @SerializedName("stargazers_url") val stargazersUrl: String,
    @SerializedName("statuses_url") val statusesUrl: String,
    @SerializedName("subscribers_url") val subscribersUrl: String,
    @SerializedName("subscription_url") val subscriptionUrl: String,
    @SerializedName("svn_url") val svnUrl: String,
    @SerializedName("tags_url") val tagsUrl: String,
    @SerializedName("teams_url") val teamsUrl: String,
    val topics: ArrayList<String>,
    @SerializedName("trees_url:") val treesUrl: String,
    @SerializedName("updated_at") val updatedAt: String,
    val url: String,
    val visibility: String,
    val watchers: Int,
    @SerializedName("watchers_count") val watchersCount: Int,
    @SerializedName("web_commit_signoff_required") val webCommitSignoffRequired: Boolean
)

data class Owner(
    @SerializedName("avatar_url") val avatarUrl: String,
    @SerializedName("events_url") val eventsUrl: String,
    @SerializedName("followers_url") val followersUrl: String,
    @SerializedName("following_url") val followingUrl: String,
    @SerializedName("gists_url") val gistsUrl: String,
    @SerializedName("gravatar_id") val gravatarId: String,
    @SerializedName("html_url") val htmlUrl: String,
    val id: Int,
    val login: String,
    @SerializedName("node_id") val nodeId: String,
    @SerializedName("organizations_url") val organizationsUrl: String,
    @SerializedName("received_events_url") val receivedEventsUrl: String,
    @SerializedName("repos_url") val reposUrl: String,
    @SerializedName("site_admin") val siteAdmin: Boolean,
    @SerializedName("starred_url") val starredUrl: String,
    @SerializedName("subscriptions_url") val subscriptionsUrl: String,
    val type: String,
    val url: String
)

data class Licenses(
    val key: String?,
    val name: String?,
    @SerializedName("spdx_id") val spdxId: String?,
    val url: String?,
    @SerializedName("node_id") val nodeId: String?
)

import retrofit2.Response
import retrofit2.http.GET
import retrofit2.http.Query

interface GithubService {
    @GET("search/repositories")
    suspend fun getRepositories(
        @Query("q") query: String
    ): Response<GithubData>
}

import retrofit2.Retrofit
import retrofit2.converter.gson.GsonConverterFactory

object ApiService {
    private const val API_URL = "http://api.github.com"

    val client: GithubService = Retrofit.Builder()
        .baseUrl(API_URL)
        .addConverterFactory(GsonConverterFactory.create())
        .build()
        .create(GithubService::class.java)
}

sealed class ApiState<T>(
    val data: T? = null,
    val message: String? = null
) {
    class Success<T>(data: T) : ApiState<T>(data)
    class Error<T>(message: String, data: T? = null) : ApiState<T>(data, message)
    class Loading<T> : ApiState<T>()
}

 

이렇게 하면 나중에 액티비티에서 JSONObject, JSONArray를 만들어 파싱하는 함수가 없어도 내가 원하는 데이터를 가져올 수 있다.

그리고 GithubRepository 클래스를 만든다.

 

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOn

class GithubRepository: BaseFlowResponse() {
    private val githubClient = ApiService.client

    suspend fun getRepositories(queryString: String): Flow<ApiState<GithubData>> = flow {
        emit(flowCall { githubClient.getRepositories(queryString) })
    }.flowOn(Dispatchers.IO)
}

 

BaseFlowResponse는 Flow + 레트로핏을 같이 사용할 때 생기는 보일러 플레이트 코드를 함수 형태로 갖고 있는 추상 클래스다. 이 함수를 클래스에 따로 정리해서 사용하기 전에는 이런 모양새였다.

 

suspend fun getRepositories(queryString: String): Flow<ApiState<GithubData>> = flow {
        try {
            val response = githubClient.getRepositories(queryString)
            if (response.isSuccessful) {
                response.body()?.let {
                    emit(ApiState.Success(it))
                }
            } else {
                try {
                    emit(ApiState.Error(response.errorBody()!!.string()))
                }   catch (e: IOException) {
                    e.printStackTrace()
                }
            }
        }   catch (e: Exception) {
            emit(ApiState.Error(e.message ?: ""))
        } as Unit
    }.flowOn(Dispatchers.IO)

 

추상 클래스 안의 함수(flowCall)로 빼놓으니 코드량도 줄고 가독성도 나름 괜찮아진 게 맘에 들어서 난 이 형태로 사용한다.

그리고 이 레포지토리를 사용하는 뷰모델을 만든다.

 

import androidx.lifecycle.ViewModel
import androidx.lifecycle.viewModelScope
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.launch

class GithubViewModel(
    private val githubRepository: GithubRepository
): ViewModel() {
    var mGithubRepositories: MutableStateFlow<ApiState<GithubData>> = MutableStateFlow(ApiState.Loading())
    var githubRepositories: StateFlow<ApiState<GithubData>> = mGithubRepositories

    fun requestGithubRepositories(query: String) = viewModelScope.launch {
        mGithubRepositories.value = ApiState.Loading()
        githubRepository.getRepositories(query)
            .catch { error ->
                mGithubRepositories.value = ApiState.Error("${error.message}")
            }
            .collect { values ->
                mGithubRepositories.value = values
            }
    }
}

 

Hilt를 사용했다면 필요없지만 지금은 사용하지 않으니 뷰모델 팩토리를 만들어 사용한다. 대충 구글링하면 나오는 코드를 갖다 썼다.

 

import androidx.lifecycle.ViewModel
import androidx.lifecycle.ViewModelProvider

class GithubViewModelFactory(
    private val githubRepository: GithubRepository
): ViewModelProvider.Factory {
    override fun <T : ViewModel> create(modelClass: Class<T>): T {
        return modelClass.getConstructor(GithubRepository::class.java).newInstance(githubRepository)
    }
}

 

그리고 액티비티에서 뷰모델 팩토리로 뷰모델 프로퍼티를 초기화한 뒤, onCreate()에서 뷰모델의 함수를 호출하고 기타 간단한 처리들을 추가한다. 예제기 때문에 로그만 2개 추가했다.

 

import android.os.Bundle
import android.util.Log
import androidx.appcompat.app.AppCompatActivity
import androidx.lifecycle.Lifecycle
import androidx.lifecycle.ViewModelProvider
import androidx.lifecycle.lifecycleScope
import androidx.lifecycle.repeatOnLifecycle
import com.example.kotlinprac.R
import kotlinx.coroutines.launch

class CoroutinePracticeActivity : AppCompatActivity() {

    private val TAG = this.javaClass.simpleName

    private lateinit var viewModel: GithubViewModel
    private lateinit var viewModelFactory: GithubViewModelFactory

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_coroutine_practice)

        viewModelFactory = GithubViewModelFactory(GithubRepository())
        viewModel = ViewModelProvider(this, viewModelFactory)[GithubViewModel::class.java]

        getGithubRepositories("retrofit")
    }

    private fun getGithubRepositories(query: String) {
        lifecycleScope.launch {
            repeatOnLifecycle(Lifecycle.State.STARTED) {
                viewModel.apply {
                    requestGithubRepositories(query)
                    githubRepositories.collect {
                        when (it) {
                            is ApiState.Success -> {
                                it.data?.let { data ->
                                    Log.e(TAG, "data - incomplete_results : ${data.incompleteResults}")
                                    val list = data.items
                                    for (i in list.indices) {
                                        Log.e(TAG, "license : ${list[i].license}")
                                    }
                                }
                                mGithubRepositories.value = ApiState.Loading()
                            }
                            is ApiState.Error -> {
                                Log.e(TAG, "## 에러 : ${it.message}")
                                mGithubRepositories.value = ApiState.Loading()
                            }
                            is ApiState.Loading -> {}
                        }
                    }
                }
            }
        }
    }

}

 

위에서 파싱하는 함수가 없어도 된다고 말했는데 ApiState.Sucess {} 안을 보면 data라 이름 붙인 리시버를 통해 boolean 값과 리스트를 가져온 다음 for 문으로 원하는 데이터(license)를 가져오는 걸 볼 수 있다. 로그캣을 보면 어떻게 가져오는지 볼 수 있다. JSON 형식에 대한 이해가 있으면 쉽게 수작업으로 만들 수 있지만 어렵다면 안드로이드 스튜디오의 플러그인 중 JSON To Kotlin Class(JsonToKotlinClass) 라는 플러그인을 사용하면 편하게 만들 수는 있다. 그러나 null 값의 경우 Any로 퉁치는 경우도 있으니 원하는 타입을 명시해서 사용하고 싶다면 플러그인 사용 후 수정하는 과정이 필요하다.

 

그리고 repeatOnLifeCycle()은 어떻게 쓰냐에 따라서 내가 원하는 대로 작동하지 않을 수 있으니 매개변수로 넘기는 LifeCycle의 종류와 상태, 활용법에 대해서 충분히 확인한 다음 사용하는 게 좋다. 아예 안 쓰고 다른 방법을 쓰는 것도 좋다.

 

이 상태에서 editText와 리사이클러뷰를 추가하면 깃허브 레포지토리 검색 앱도 만들 수 있으니 한번 시도해보자.