compose paginator

Paginator.kt

import kotlin.coroutines.CoroutineContext
import kotlin.coroutines.EmptyCoroutineContext
import kotlinx.collections.immutable.ImmutableList
import kotlinx.collections.immutable.toImmutableList
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.distinctUntilChanged
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.mapNotNull
import kotlinx.coroutines.flow.merge
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withPermit
import kotlinx.coroutines.withContext
import kotlinx.coroutines.yield
import timber.log.Timber

abstract class DefaultPaginator<Key, Item, Model>(
    replayLoadState: Boolean = false,
    private val coroutineContext: CoroutineContext = EmptyCoroutineContext,
    private val maxConcurrentMap: Int = 1
) : Paginator<Key, Item, Model> {

    private var prevKey = getInitialKey()
    private var nextKey = getInitialKey()

    // TODO: should we lock these vars?
    private var isLoadingNextItems = false
    private var isLoadingPrevItems = false
    private var endOfPaginationReachedBefore = MutableStateFlow(false)
    private var endOfPaginationReachedAfter = MutableStateFlow(false)
    private var allowItemModelsUpdate = false

    private val itemModelsFlow = MutableSharedFlow<ImmutableList<Model>>()
    protected val pagingState = MutableSharedFlow<PagingState>(replay = if (replayLoadState) 1 else 0)
    protected val refreshedState = MutableStateFlow(false)

    override fun isEndOfPagination(append: Boolean): Boolean =
        if (append) endOfPaginationReachedAfter.value else endOfPaginationReachedBefore.value

    override fun observeItemModels(): Flow<ImmutableList<Model>> = merge(
        itemModelsFlow,
        itemsChangeFlow()
            .filter { allowItemModelsUpdate }
            .mapNotNull { list ->
                list.mapIndexedNotNull { index, item -> itemModelMapper(item, index) }.toImmutableList()
            }
    ).distinctUntilChanged()

    override fun observePagingState(): Flow<PagingState> = pagingState.asSharedFlow()

    override fun observePagingLoadState(): Flow<PagingState.LoadState> =
        pagingState.asSharedFlow().mapNotNull { it as? PagingState.LoadState }

    override fun observeRefreshedState(): StateFlow<Boolean> = refreshedState.asStateFlow()

    override fun observeEndOfPagination(append: Boolean): StateFlow<Boolean> =
        if (append) endOfPaginationReachedAfter else endOfPaginationReachedBefore

    override fun isRefreshed(): Boolean = refreshedState.value

    // request item list by key
    abstract suspend fun onRequest(key: Key?, isRefresh: Boolean, append: Boolean): Paginator.ResultList<Item>

    // return initial key
    open fun getInitialKey(): Key? = null

    // return key for next loading
    abstract suspend fun getNextKey(pageResultList: Paginator.ResultList<Item>, append: Boolean): Key?

    // return end of pagination reached
    // TODO: we can just return a object containing prevKey, nextKey, endOfPagination values
    abstract suspend fun onSuccessAndCheckEnd(
        items: List<Item>,
        isRefresh: Boolean,
        append: Boolean,
        resultTag: Any?
    ): Boolean

    abstract suspend fun itemModelMapper(item: Item, index: Int): Model?

    abstract suspend fun itemsSnapshot(): List<Item>

    abstract fun itemsChangeFlow(): Flow<List<Item>>

    override suspend fun refresh() {
        reset()
        loadNextItems(refreshing = true)
        refreshedState.value = true
    }

    override suspend fun loadNextItems(refreshing: Boolean) = withContext(coroutineContext) {
        Timber.i("loadNextItems, refreshing = $refreshing, isLoadingNextItems $isLoadingNextItems")
        if (endOfPaginationReachedAfter.value) {
            Timber.i("endOfPaginationReachedAfter == true")
            return@withContext
        }
        if (isLoadingNextItems) {
            Timber.i("isMakingRequest == true")
            return@withContext
        }

        isLoadingNextItems = true

        // let upstream to have a change to re-subscribe state flow before emit state here
        yield()

        // update load state
        if (refreshing) {
            pagingState.emit(PagingState.LoadState.Refresh(true, !isRefreshed()))
        } else {
            pagingState.emit(PagingState.LoadState.Append(true))
        }
        yield()

        // request items
        val result = runSuspendCatching { onRequest(nextKey, refreshing, append = true) }
        isLoadingNextItems = false

        val resultList = result.getOrElse {
            // emit error if any
            Timber.e(it, "Failed to get result")
            pagingState.emit(PagingState.LoadState.Error(append = true, throwable = it))
            return@withContext
        }

        // update key
        if (refreshing) {
            prevKey = getNextKey(resultList, append = false)
        }
        nextKey = getNextKey(resultList, append = true)
        val endOfPagination =
            onSuccessAndCheckEnd(resultList.items, refreshing, append = true, resultTag = resultList.tag)
        endOfPaginationReachedAfter.value = endOfPagination

        // emit items
        val models = runSuspendCatching {
            val semaphore = Semaphore(maxConcurrentMap)
            itemsSnapshot().mapIndexed { index, item ->
                async { semaphore.withPermit { itemModelMapper(item, index) } }
            }.awaitAll().filterNotNull().toImmutableList()
        }.getOrElse {
            Timber.e(it, "Failed to map item")
            pagingState.emit(PagingState.LoadState.Error(append = true, throwable = it))
            return@withContext
        }
        pagingState.emit(PagingState.Result(models, endOfPagination))
        itemModelsFlow.emit(models)
        allowItemModelsUpdate = true
        yield()

        // update load state
        if (refreshing) {
            pagingState.emit(PagingState.LoadState.Refresh(false, !isRefreshed()))
        } else {
            pagingState.emit(PagingState.LoadState.Append(false))
        }
    }

    override suspend fun loadPreviousItems() = withContext(coroutineContext) {
        Timber.i("loadPreviousItems, isLoadingPrevItems $isLoadingPrevItems")
        if (endOfPaginationReachedBefore.value) {
            Timber.i("endOfPaginationReachedBefore == true")
            return@withContext
        }
        if (isLoadingPrevItems) {
            Timber.i("isLoadingPrevItems == true")
            return@withContext
        }
        isLoadingPrevItems = true

        pagingState.emit(PagingState.LoadState.Prepend(true))
        yield()

        // request items
        val result = runSuspendCatching {
            onRequest(prevKey, append = false, isRefresh = false)
        }
        isLoadingPrevItems = false

        val resultList = result.getOrElse {
            // emit error if any
            Timber.e(it, "Failed to get result")
            pagingState.emit(PagingState.LoadState.Error(append = false, throwable = it))
            return@withContext
        }

        // update key
        prevKey = getNextKey(resultList, append = false)
        val endOfPagination =
            onSuccessAndCheckEnd(resultList.items, isRefresh = false, append = false, resultTag = resultList.tag)
        endOfPaginationReachedBefore.value = endOfPagination

        // emit items
        val models = runSuspendCatching {
            val semaphore = Semaphore(maxConcurrentMap)
            itemsSnapshot().mapIndexed { index, item ->
                async { semaphore.withPermit { itemModelMapper(item, index) } }
            }.awaitAll().filterNotNull().toImmutableList()
        }.getOrElse {
            Timber.e(it, "Failed to map item")
            pagingState.emit(PagingState.LoadState.Error(append = false, throwable = it))
            return@withContext
        }
        pagingState.emit(PagingState.Result(models, endOfPagination))
        itemModelsFlow.emit(models)
        allowItemModelsUpdate = true
        yield()

        pagingState.emit(PagingState.LoadState.Prepend(false))
    }

    protected fun reset() {
        prevKey = getInitialKey()
        nextKey = getInitialKey()
        endOfPaginationReachedBefore.value = false
        endOfPaginationReachedAfter.value = false
        isLoadingNextItems = false
        isLoadingPrevItems = false
        allowItemModelsUpdate = false
    }
}

import androidx.compose.runtime.Immutable
import kotlinx.collections.immutable.ImmutableList
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.emptyFlow

interface Paginator<Key, Item, Model> {
    suspend fun refresh()
    suspend fun loadPreviousItems()
    suspend fun loadNextItems(refreshing: Boolean = false)
    fun observeItemModels(): Flow<ImmutableList<Model>>
    fun observePagingState(): Flow<PagingState>
    fun observePagingLoadState(): Flow<PagingState.LoadState>
    fun observeRefreshedState(): StateFlow<Boolean>
    fun observeEndOfPagination(append: Boolean): StateFlow<Boolean>

    fun isRefreshed(): Boolean
    fun isEndOfPagination(append: Boolean): Boolean

    @Immutable
    sealed class PagingState {
        data class Result<Model>(val itemModels: ImmutableList<Model>, val endOfPagination: Boolean) : PagingState()

        sealed class LoadState(open val loading: Boolean) : PagingState() {
            data class Refresh(override val loading: Boolean, val isInitial: Boolean) : LoadState(loading)
            data class Append(override val loading: Boolean) : LoadState(loading)
            data class Prepend(override val loading: Boolean) : LoadState(loading)

            data class Error(val append: Boolean = false, val throwable: Throwable? = null) : LoadState(loading = false)
        }
        val isInitialLoading: Boolean get() = this is LoadState.Refresh && this.loading && this.isInitial
        val isRefreshing: Boolean get() = this is LoadState.Refresh && this.loading
        val isAppending: Boolean get() = this is LoadState.Append && this.loading
        val isPrepending: Boolean get() = this is LoadState.Prepend && this.loading
        val isErrorState: Boolean get() = this is LoadState.Error
    }

    data class ResultList<T>(val items: List<T>, val tag: Any? = null)

    companion object {
        fun <K, I, M> empty() = object : Paginator<K, I, M> {
            override suspend fun refresh() {}

            override suspend fun loadPreviousItems() {}

            override suspend fun loadNextItems(refreshing: Boolean) {}

            override fun observeItemModels(): Flow<ImmutableList<M>> = emptyFlow()

            override fun observeRefreshedState(): StateFlow<Boolean> = MutableStateFlow(false)

            override fun observeEndOfPagination(append: Boolean): StateFlow<Boolean> = MutableStateFlow(false)

            override fun isRefreshed(): Boolean = false

            override fun isEndOfPagination(append: Boolean): Boolean = false

            override fun observePagingState(): Flow<PagingState> = emptyFlow()

            override fun observePagingLoadState(): Flow<PagingState.LoadState> = emptyFlow()
        }
    }
}