A paginated datasource to be using with RxSwift and ViewModels

RxPaginatedDatasource.swift

//
//  RxPaginatedDatasource.swift
//
//  Created by Marcelo de Aguiar on 11/10/19.
//

import Foundation
import RxSwift
import RxCocoa

/// Represents a source of items that can be refreshed and paginated.
struct RxPaginatedSource<T> {
    private let disposeBag = DisposeBag()

    /// The items generated by the RxPaginatedDataSource.
    let items: Observable<[T]>
    /// The loading state of the datasource
    let isLoading: Observable<Bool>
    /// The errors when loading data if any.
    let errors: Observable<Error>

    /// Constructs a RxPaginatedSource whose pages depends on an index.
    /// - Parameter pageFactory: A factory method to return a page based on the index.
    /// - Parameter refreshTrigger: A observable to trigger a refresh on the RxPaginatedSource.
    /// A refresh start the load at the first page again.
    /// - Parameter nextPageTrigger: A optional observable to trigger the nex page load.
    init(pageFactory: @escaping (Int) -> Single<Page>,
         refreshTrigger: Observable<Void>,
         nextPageTrigger: Observable<Void> = .never()) {

        let pages = BehaviorSubject<[Page]>(value: []) // Our page buffer

        let hasNextTrigger = nextPageTrigger    // When request next page
            .withLatestFrom(pages)              // get last array of pages
            .map { $0.last }                    // get last page
            .compactMap { $0 }                  // remove nulls
            .filter { p in p.hasNext }          // filter if has no next pages
            .map { p in p.index + 1 }           // Get next page index

        let reloadPageIndex = refreshTrigger    // When reload request
            .map { 1 }                          // map to first page index

        let loadTrigger = Observable.merge(reloadPageIndex, hasNextTrigger)    // Both trigger a trip to server

        let nextPageElements = loadTrigger
            .flatMap { pageIndex in             // Map the page index to a page
                pageFactory(pageIndex)
                    .asObservable()
                    .materialize()              // lets filter any onCompleted event (completed and error)...
                    .filter { $0.isCompleted }  //...preventing the observable to end.
            }

        let nextPage = nextPageElements
            .compactMap { $0.element }

        nextPage                                                            // When next page comes
            .withLatestFrom(pages) { (page: $0, pages: $1) }                // Get both the new page with the current pages
            .map { $0.page.index == 1 ? [$0.page] : $0.pages + [$0.page] }  // Concat the new items or replace all if the first page
            .subscribe(pages)                                               // Direct event to pages observable
            .disposed(by: disposeBag)

        items = pages.map { $0.flatMap { p in p.items } } // Flat and expose the items in the pages

        errors = nextPageElements
            .compactMap { $0.error }

        isLoading = Observable.merge(loadTrigger.map { _ in 1 }, nextPage.map { _ in -1 })
            .scan(0, accumulator: +)
            .map { $0 > 0 }
            .distinctUntilChanged()
    }

    /// Constructs a RxPaginatedSource whose pages depends on an index and filter.
    /// - Parameter pageFactory: A factory method to return a new page based on the index and query text.
    /// - Parameter queryTrigger: A observable to trigger the search on the elements;
    /// A query will result in a new list of items based on the query starting at page 1
    /// - Parameter refreshTrigger: A observable to trigger a refresh on the RxPaginatedSource.
    /// A refresh start the load at the first page again.
    /// - Parameter nextPageTrigger: A optional observable to trigger the nex page load.
    init(pageFactory: @escaping (String, Int) -> Single<Page>,
         queryTrigger: Observable<String>,
         refreshTrigger: Observable<Void> = .never(),
         nextPageTrigger: Observable<Void> = .never()) {

        let pages = BehaviorSubject<[Page]>(value: []) // Our page buffer

        let hasNextTrigger = nextPageTrigger    // When request next page
            .withLatestFrom(pages)              // get last array of pages
            .map { $0.last }                    // get last page
            .compactMap { $0 }                  // remove nulls
            .filter { p in p.hasNext }          // filter if has no next pages
            .map { p in p.index + 1 }           // Get next page index

        let reloadPageIndex = refreshTrigger    // When reload request
            .map { 1 }                          // map to first page index

        let queryToPageIndex = queryTrigger
            .map { _ in 1 }

        /// Any of these trigger a new page
        let loadTrigger = Observable.merge(reloadPageIndex, hasNextTrigger, queryToPageIndex)

        // We use this to separate results from errors without completning the observable.
        let nextPageElements = loadTrigger              // Both trigger a trip to server
            .withLatestFrom(queryTrigger) { ($0, $1) }  // joing the index and query text
            .flatMap { index, query in                  // Map the page index and query text to a page
                pageFactory(query, index)               // get a new page
                    .asObservable()
                    .materialize()                      // lets filter any onCompleted event (completed and error)...
                    .filter { $0.isCompleted == false } //...preventing the observable to end.
            }

        let nextPage = nextPageElements
            .compactMap { $0.element }

        nextPage                                                            // When next page comes
            .withLatestFrom(pages) { (page: $0, pages: $1) }                // Get both the new page with the current pages
            .map { $0.page.index == 1 ? [$0.page] : $0.pages + [$0.page] }  // Concat the new items or replace all if the first page
            .subscribe(pages)                                               // Direct event to pages observable
            .disposed(by: disposeBag)

        items = pages.map { $0.flatMap { p in p.items } } // Flat and expose the items in the pages

        errors = nextPageElements
            .compactMap { $0.error }

        isLoading = Observable.merge(loadTrigger.map { _ in 1 }, nextPage.map { _ in -1 })
            .scan(0, accumulator: +)
            .map { $0 > 0 }
            .distinctUntilChanged()
    }
}

// MARK: - RxPaginatedSource+Page
extension RxPaginatedSource {
    /// Represent a  page of items of type T
    struct Page {
        let index: Int
        let hasNext: Bool
        let items: [T]
    }
}