// ViewModel:

init {
    launch( IO ) { startObserving() }
}

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

    runCatching {

        for ( uiModel in presenter.observe() )
            channelsAvailability.postData( uiModel )

    }.onFailure {
        channelsAvailability.postError( it )
        delay( DEFAULT_ERROR_DELAY )
        startObserving()
    }
}


// Presenter:

/** @return [ReceiveChannel] of [ChannelsAvailabilityUiModel] */
suspend fun observe(): ReceiveChannel<ChannelsAvailabilityUiModel> = coroutineScope {
    val channel = Channel<ChannelsAvailabilityUiModel>( CONFLATED )

    launch( IO ) {
        for ( boolean in hasMovieChannels.observe() ) {
            hadMovieChannels = boolean
            maybeMakeModel()?.let { channel.send( it ) }
        }
    }

    launch( IO ) {
        for ( boolean in hasTvChannels.observe() ) {
            hadTvChannels = boolean
            maybeMakeModel()?.let { channel.send( it ) }
        }
    }

    channel
}

// Use Case:

/** @return [ReceiveChannel] of [Boolean] whether any [TvChannel] is present in [LocalData] */
suspend fun observe() = localData.observeCountTvChannels().map { it > 0 }


// Local Data:

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


// Source:

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


// Query Utils:

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
}

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