flow (예외 작업 / 다른 CoroutineContext에서 실행시키기)

Catching unexpected exceptions

생산자는 구현은 다른 라이브러리에서 가져올 수 있습니다.

이는 예기치 않은 예외가 발생할 수 있음을 의미합니다. 이러한 예외를 처리하려면 catch 중간 연산자를 사용.

class LatestNewsViewModel(
    private val newsRepository: NewsRepository
) : ViewModel() {

    init {
        viewModelScope.launch {
            newsRepository.favoriteLatestNews
                // catch 중간연산자 예외가 던저진다면,
                // 처리하고 UI를 처리한다. 
                .catch { exception -> notifyError(exception) }
                .collect { favoriteNews ->
                   
                }
        }
    }
}

예외가 발생하면 새 데이터가 수신되지 않았기 때문에 collect lambda가 호출되지 않는다.

catch는 데이터를 flow에 내보낼 수 있습니다. repository layer는 대신 이전에 캐시 된 값을 내보낼 수 있습니다.

class NewsRepository(...) {
    val favoriteLatestNews: Flow<List<ArticleHeadline>> =
        newsRemoteDataSource.latestNews
            .map { news -> news.filter { userData.isFavoriteTopic(it) } }
            .onEach { news -> saveInCache(news) }
            // 만약 에러가 발생한다면 최신 캐쉬된 값을 내보낸다. 
            .catch { exception -> emit(lastCachedNews()) }
}

 

위에서 예외가 발생하면 예외로 인해 캐쉬되었던 새로운 data가 stream으로 내보내 졌으므로 collect lambda가 호출됩니다.

Executing in a different CoroutineContext

기본적으로 flow builder 의 생산자는 collecting하는 coroutine의 CoroutineContext에서 실행된다.

 flow builder에서는 다른 CoroutineContext에서 값을 내보낼 수 없다.

 

이러한 제한 경우에 따라 불편할 수 있다.

 

Repository layer는 viewModelScope에서 사용하는 Dispatchers.Main에서 작업을 수행하지 않아야합니다

flow의 CoroutineContext를 변경하려면 중간 연산자 flowOn을 사용합니다.

 

flowOn은 upStream flow의 CoroutineContext를 변경합니다.

 

upStream flow 는 이전 (또는 그 이상)에 적용된 생산자와 중간 연산자를 의미한다.

downStream flow는 소비자와 함께 flowOn 이후의 중간 연산자를 의미한다. 

 

Producer(생산자)	intermediate operator	flowOn	intermediate operator	Consumer(소비자)
upStream flow			downStream flow
I/O Thread / Default Thread			MainThread

 

downStream flow 은 영향을 받지 않으며 flow에서 collecting하는 데 사용되는 CoroutineContext에서 실행됩니다.

flowOn 연산자가 여러개 일때,  각 연산자는 현재 위치에서 upStream을 downStream으로 변경합니다.

 

class NewsRepository(
    private val newsRemoteDataSource: NewsRemoteDataSource,
    private val userData: UserData,
    private val defaultDispatcher: CoroutineDispatcher
) {
    val favoriteLatestNews: Flow<List<ArticleHeadline>> =
        newsRemoteDataSource.latestNews
            .map { news -> // filter 중간연산자 DefaultThread에서 실행됨. 
                news.filter { userData.isFavoriteTopic(it) }
            }
            .onEach { news -> // saveInCache 중간연산자 DefaultThread에서 실행됨. 
                saveInCache(news)
            }
            // flowOn은 위의 the upstream flow ↑에 영향을 준다. 
            .flowOn(defaultDispatcher)
            // the downstream flow ↓ 은 영향받지 않는다. 
            .catch { exception ->  // MainThread에서 실행된다. 
                emit(lastCachedNews())
            }
}

 

onEach 및 map 연산자는 defaultDispatcher를 사용하는 반면 catch 연산자와 소비자는 viewModelScope에서 사용하는 Dispatchers.Main에서 실행된다.

 

데이터 소스 계층이 I / O 작업을 수행하므로 I / O 작업에 최적화 된 디스패처를 사용해야합니다.

class NewsRemoteDataSource(
    ...,
    private val ioDispatcher: CoroutineDispatcher
) {
    val latestNews: Flow<List<ArticleHeadline>> = flow {
        // Executes on the IO dispatcher
        ...
    }
        .flowOn(ioDispatcher)
}