Значения, не полученные из каналов Coroutine - PullRequest
Новая
       28

Значения, не полученные из каналов Coroutine

0 голосов
/ 02 марта 2019

Мой кодовый поток выглядит следующим образом: SqlDelight queries ( 2x ) -> Coroutine Channels ( 2x ) -> Coroutine Channel ( 1x )

Я заметил, что Queries 'Channel s излучают, Presenter получает и 1010 * излучает, но ViewModel никогда не получает значения.

Также циклы for, повторяющиеся на Channels, никогда не завершаются, ядумаю, что это хорошо, но я не могу понять, почему значения никогда не принимаются, так: /

ViewModel 

init {
    // When ViewModel is instantiated, start observing for UiModels
    launch( IO ) { startObserving() }
}

private suspend fun startObserving() {
    channelsAvailability.postLoading()

    runCatching {

        // Receive Models
        for ( uiModel in presenter.observe() )
            channelsAvailability.postData( uiModel )

    // If some error occurs, notify it, wait and then try again
    }.onFailure {
        channelsAvailability.postError( it )
        delay( DEFAULT_ERROR_DELAY )
        startObserving()
    }
}

Presenter 

/** 
 * @return [ReceiveChannel] of [ChannelsAvailabilityUiModel]
 * Creates a single [ReceiveChannel] from [hasMovieChannels.observe] and
 * [hasTvChannels.observe] [ReceiveChannel]s
 */
suspend fun observe(): ReceiveChannel<ChannelsAvailabilityUiModel> = coroutineScope {
    val channel = Channel<ChannelsAvailabilityUiModel>( CONFLATED )

    launch( IO ) {
        for ( boolean in hasMovieChannels.observe() ) {
            // Cache the value
            hadMovieChannels = boolean
            // Create and send UiModel if both hadMovieChannels and hadTvChannels are not null
            maybeMakeModel()?.let { channel.send( it ) }
        }
    }

    launch( IO ) {
        for ( boolean in hasTvChannels.observe() ) {
            // Cache the value
            hadTvChannels = boolean
            // Create and send UiModel if both hadMovieChannels and hadTvChannels are not null
            maybeMakeModel()?.let { channel.send( it ) }
        }
    }

    channel
}

Вариант использования 

/** 
 * @return [ReceiveChannel] of [Boolean] whether any [TvChannel] is present in [LocalData]
 * Maps the count of [TvChannels] into a [Boolean] representing whether any Channels is available
 */
suspend fun observe() = localData.observeCountTvChannels().map { it > 0 }

Локальные данные 

/** @return [ReceiveChannel] of the [Int] count of the stored [TvChannel]s */
override suspend fun observeCountTvChannels() = tvChannels.observeCount()

Источник 

/** 
 * @return [ReceiveChannel] of the [Int] count of the stored channels [TvChannelPojo]
 * Maps the [Query] into a [ReceiveChannel]
 */
override suspend fun observeCount() = queries.count().asChannel().mapToOne().map { it.toInt() }

Утилиты запросов 

fun <T : Any> Query<T>.asChannel(): ReceiveChannel<Query<T>> {
    val channel = Channel<Query<T>>( CONFLATED )
    // Ensure consumers immediately run the query.
    channel.offer(this )

    val listener = object : Query.Listener, (Throwable?) -> Unit {
        override fun queryResultsChanged() {
            channel.offer(this@asChannel )
        }

        override fun invoke( cause: Throwable? ) {
            removeListener(this )
        }
    }

    addListener( listener )
    channel.invokeOnClose( listener )

    return channel
}

suspend fun <T : Any> ReceiveChannel<Query<T>>.mapToOne( context: CoroutineContext? = null ) =
    map(context ?: coroutineContext ) { it.executeAsOne() }
...