How to use Firestore with paging 3 and RxPagingSource

Question

I am creating an app in which i am using RxJava and paging 3 library. I am using RxPagingSource and Retrofit to paginate response coming from Server but i have to fetch some data from Firestore and have to paginate

override fun loadSingle(params: LoadParams<QuerySnapshot>): Single<LoadResult<QuerySnapshot, Notification>> {
        var currentPage : QuerySnapshot
        if(params.key != null){
            currentPage = params.key!!
        }else{
            reference
                .limit(10)
                .get()
                .addOnCompleteListener(OnCompleteListener {
                    if(it.isSuccessful){
                        currentPage = it.result
                        val lastDocumentSnapshot = it.result.documents[it.result.size() - 1]
                        reference
                            .limit(10)
                            .startAfter(lastDocumentSnapshot)
                            .get()
                            .addOnCompleteListener(OnCompleteListener {
                                val nextPage: QuerySnapshot
                                if(it.isSuccessful){
                                    nextPage = it.result
                                    return Single.just(
                                        LoadResult.Page(
                                            data = currentPage.toObjects(Notification::class.java),
                                            prevKey = null,
                                            nextKey = nextPage
                                        )
                                    )
                                    //return
                                }
                            })
                    }
                })
        }

This is code i tried but its not working for me, there is many mistake in this code

How can i paginate Firestore data using RxPagingSource provided by Paging 3 library

Answer 1

PagingDataSource.kt

class NotificationPagingDataSource(val service: AppRepository) :
    RxPagingSource<Int, Notification>() {

    override fun loadSingle(params: LoadParams<Int>): Single<LoadResult<Int, Notification>> {

        val page = params.key ?: 1
        return service.getFireNotification(page)
            .subscribeOn(Schedulers.io())
            .map {
                toLoadResult(it, page)
            }
            .onErrorReturn {
                LoadResult.Error(it)
            }
    }


    private fun toLoadResult(
        data: QuerySnapshot,
        page: Int
    ): LoadResult<Int, Notification> {
     
        Log.i("TAG"," mappingstarted:::  3")
        return LoadResult.Page(
            data = data.toObjects(Notification::class.java),
            prevKey = if (page <= 1) null else page - 1,
            nextKey = if (data.size() == 0) null else page + 1
        )
    }


    override fun getRefreshKey(state: PagingState<Int, Notification>): Int? {
        return state.anchorPosition
    }


}

AppRepository.kt

 var timestamp : String? = null

fun getFireNotification(i: Int): Single<QuerySnapshot> {


        return Single.create<QuerySnapshot>(SingleOnSubscribe { emitter ->

            if (i == 1) {
                FirebaseFirestore.getInstance()
                    .collection(Constant.NOTIFICATION_NODE).document(FirebaseAuth.getInstance().currentUser.uid)
                    .collection(Constant.USER_NOTIFICATION_NODE)
                    .limit(15)
                    .get()
                    .addOnCompleteListener {
                        if (it.isSuccessful) {
                            emitter.onSuccess(it.result)
                            timestamp = it.result.documents.last().data?.get("timestamp").toString()
                        }
                    }
                    .addOnFailureListener {
                        Log.i("TAG"," addOnFailureListener:: $it")
                    }
            } else {
                FirebaseFirestore.getInstance()
                    .collection(Constant.NOTIFICATION_NODE).document(FirebaseAuth.getInstance().currentUser.uid)
                    .collection(Constant.USER_NOTIFICATION_NODE)
                    .orderBy("timestamp",Query.Direction.DESCENDING)
                    .startAfter(timestamp)
                    .limit(15)
                    .get()
                    .addOnCompleteListener {
                        if (it.isSuccessful) {
                            if(it.result.documents.isNotEmpty()) {
                                emitter.onSuccess(it.result)
                                timestamp = it.result.documents.last().data?.get("timestamp").toString()
                            }
                        }
                    }
                    .addOnFailureListener {
                    }
            }
        })
    }

FirebaseViewModel.kt

private val notificationPagingdata: MediatorLiveData<PagingData<NotificationModal>> =
        MediatorLiveData()

    fun getNotification(){
         disposable.add(Pager(
            config =  PagingConfig(
                pageSize = 15,
                enablePlaceholders = false,
                prefetchDistance = 1,
            ),
            pagingSourceFactory = {
                NotificationPagingDataSource(repository)
            }
        ).flowable.subscribe({
           notificationPagingdata.value  = it
           }
       )
     )
}

 fun observeNotificationPagingDataSource():LiveData<PagingData<NotificationModal>>{
        return notificationPagingdata
    }

YourFragment or Activity.kt

private fun setObserver() {
        fireViewModel.observeNotificationPagingDataSource().observe(viewLifecycleOwner, Observer {
            adapter.submitData(lifecycle,it)
        })
    }