Homegrown fetchMore in Apollo iOS

Apollo+Rx.swift

import Foundation
import RxSwift
import RxRelay
import Apollo

// MARK: - Apollo Support

enum ApolloError: Error {
  case genericError(String)
  case graphqlErrors([GraphQLError])
}

extension ApolloError {
  var description: String {
    switch self {
    case .genericError(let error):
      return error
    case .graphqlErrors(let errors):
      return errors.compactMap { $0.message }.joined(separator: "\n")
    }
  }
}

extension ApolloClient: ReactiveCompatible { }

extension Reactive where Base: ApolloClient {
  func fetch<Query: GraphQLQuery>(
    query: Query,
    cachePolicy: CachePolicy = .fetchIgnoringCacheData,
    queue: DispatchQueue = DispatchQueue.main
  ) -> Single<Query.Data> {
    Single.create { single in
      let cancellable = self.base.fetch(query: query, cachePolicy: cachePolicy, queue: queue) { result in
        switch result {
        case .success(let response) where response.errors != nil:
          single(.error(ApolloError.graphqlErrors(response.errors!)))
        case .success(let response):
          single(.success(response.data!))
        case .failure(let error):
          single(.error(error))
        }
      }

      return Disposables.create { cancellable.cancel() }
    }
  }

  func watch<Query: GraphQLQuery>(
    query: Query,
    cachePolicy: CachePolicy = .fetchIgnoringCacheData,
    queue: DispatchQueue = DispatchQueue.main
  ) -> Observable<Query.Data> {
    Observable.create { observer in
      let watcher = self.base.watch(query: query, cachePolicy: cachePolicy, queue: queue) { result in
        switch result {
        case .success(let response) where response.errors != nil:
          observer.onError(ApolloError.graphqlErrors(response.errors!))
        case .success(let response):
          observer.onNext(response.data!)
        case .failure(let error):
          observer.onError(error)
        }
      }

      return Disposables.create { watcher.cancel() }
    }
  }

  func watchedQuery<Query: GraphQLQuery>(
    query: Query,
    cachePolicy: CachePolicy = .fetchIgnoringCacheData,
    queue: DispatchQueue = DispatchQueue.main
  ) -> WatchedQuery<Query> {
    WatchedQuery(client: self.base, query: query, data: watch(query: query, cachePolicy: cachePolicy))
  }

  func perform<Mutation: GraphQLMutation>(
    mutation: Mutation,
    queue: DispatchQueue = DispatchQueue.main
  ) -> Single<Mutation.Data> {
    Single.create { single in
      let cancellable = self.base.perform(mutation: mutation, queue: queue) { result in
        switch result {
        case .success(let response) where response.errors != nil:
          single(.error(ApolloError.graphqlErrors(response.errors!)))
        case .success(let response):
          single(.success(response.data!))
        case .failure(let error):
          single(.error(ApolloError.genericError("\(L.general.error): \(error.localizedDescription)")))
        }
      }

      return Disposables.create { cancellable.cancel() }
    }
  }

  func load<Query: GraphQLQuery>(query: Query) -> Single<GraphQLResult<Query.Data>> {
    Single.create { single in
      self.base.store.load(query: query) { result in
        switch result {
        case .success(let success): single(.success(success))
        case .failure(let error):   single(.error(error))
        }
      }
      return Disposables.create()
    }
  }
}

NetworkOperationState.swift

import Foundation

enum NetworkOperationState: Equatable {
  case performing
  case failed
  case done
}

WatchedQuery.swift

import Apollo
import RxSwift
import RxRelay

struct WatchedQuery<Query: GraphQLQuery> {
  let data: Observable<Query.Data>
  let query: Query
  let loading: Observable<NetworkOperationState>
  let fetchingMore: Observable<NetworkOperationState>

  private let client: ApolloClient
  private let fetchingMoreRelay = BehaviorRelay<NetworkOperationState>(value: .done)
  private let performFetchMore = PublishSubject<Maybe<Void>>()

  private let disposeBag = DisposeBag()

  init(client: ApolloClient, query: Query, data: Observable<Query.Data>) {
    let loading = ReplaySubject<NetworkOperationState>.create(bufferSize: 1)

    self.query = query
    self.client = client
    self.data = data.share(replay: 1)
      .do(
        onNext: { _ in loading.onNext(.done) },
        onError: { _ in loading.onNext(.failed) },
        onSubscribed: { loading.onNext(.performing) }
      )

    self.loading = loading.distinctUntilChanged()
    self.fetchingMore = fetchingMoreRelay.distinctUntilChanged()

    performFetchMore.flatMapFirst { $0 }.subscribe().disposed(by: disposeBag)
  }

  func duplicate() -> WatchedQuery<Query> {
    client.rx.watchedQuery(query: query)
  }

  func update(_ update: @escaping (inout Query.Data) -> Void) {
    self.updateQuery(update)
  }

  func fetchMore<MoreQuery: GraphQLQuery>(
    query: @escaping (Query.Data) -> MoreQuery?,
    update: @escaping (inout Query.Data, MoreQuery.Data) -> Void
  ) {
    let maybe = client.rx.load(query: self.query)
      .map { $0.data }
      .filter { $0 != nil }
      .map { query($0!) }
      .filter { $0 != nil }
      .flatMap { [client, fetchingMoreRelay] query in
        client.rx.fetch(query: query!)
          .asMaybe()
          .do(
            onError: { _ in fetchingMoreRelay.accept(.failed) },
            onCompleted: { fetchingMoreRelay.accept(.done) },
            onSubscribed: { fetchingMoreRelay.accept(.performing) }
          )
      }
      .do(onNext: { moreData in self.updateQuery { update(&$0, moreData) } })
      .map { _ in () }

    performFetchMore.onNext(maybe)
  }

  private func updateQuery(_ update: @escaping (inout Query.Data) -> Void) {
    _ = client.store.withinReadWriteTransaction({ transaction in
      do {
        try transaction.update(query: self.query) { update(&$0) }
      } catch {
        remoteLog(error)
      }
    })
  }
}