Combine publishers for Apollo iOS client

ApolloPublisher.swift

import Apollo
import Combine

/// Extending ApolloClient to provide a convenient entry point for the publishers
/// I tried to stay as close as possible to the signature of the associated Apollo function calls
extension ApolloClient {
  func queryPublisher<Query: GraphQLQuery>(_ query: Query,
                                           cachePolicy: CachePolicy? = nil,
                                           context: UnsafeMutableRawPointer? = nil,
                                           queue: DispatchQueue? = .main) -> Publishers.ApolloQuery<Query> {
    let config = ApolloQueryConfiguration(client: self, query: query, cachePolicy: cachePolicy, context: context, queue: queue)
    return Publishers.ApolloQuery(with: config)
  }
  
  func mutationPublisher<Mutation: GraphQLMutation>(_ mutation: Mutation,
                                                    context: UnsafeMutableRawPointer? = nil,
                                                    queue: DispatchQueue? = .main) -> Publishers.ApolloMutation<Mutation> {
    let config = ApolloMutationConfiguration(client: self, mutation: mutation, context: context, queue: queue)
    return Publishers.ApolloMutation(with: config)
  }
  
  func uploadPublisher<Operation: GraphQLOperation>(_ operation: Operation,
                                                    context: UnsafeMutableRawPointer? = nil,
                                                    queue: DispatchQueue? = .main,
                                                    files: [GraphQLFile]) -> Publishers.ApolloUpload<Operation> {
    let config = ApolloUploadConfiguration(client: self, operation: operation, context: context, queue: queue, files: files)
    return Publishers.ApolloUpload(with: config)
  }
}

// MARK: -
extension Publishers {
  struct ApolloQuery<Query: GraphQLQuery>: Publisher {
    typealias Output = GraphQLResult<Query.Data>
    typealias Failure = Error
    
    let configuration: ApolloQueryConfiguration<Query>
    
    init(with configuration: ApolloQueryConfiguration<Query>) {
      self.configuration = configuration
    }
    
    func receive<S: Subscriber>(subscriber: S) where Failure == S.Failure, Output == S.Input {
      let subscription = ApolloQuerySubscription(subscriber: subscriber, configuration: configuration)
      subscriber.receive(subscription: subscription)
    }
  }
  
  struct ApolloMutation<Mutation: GraphQLMutation>: Publisher {
    typealias Output = GraphQLResult<Mutation.Data>
    typealias Failure = Error
    
    let configuration: ApolloMutationConfiguration<Mutation>
    
    init(with configuration: ApolloMutationConfiguration<Mutation>) {
      self.configuration = configuration
    }
    
    func receive<S: Subscriber>(subscriber: S) where Failure == S.Failure, Output == S.Input {
      let subscription = ApolloMutationSubscription(subscriber: subscriber, configuration: configuration)
      subscriber.receive(subscription: subscription)
    }
  }
  
  struct ApolloUpload<Operation: GraphQLOperation>: Publisher {
    typealias Output = GraphQLResult<Operation.Data>
    typealias Failure = Error
    
    let configuration: ApolloUploadConfiguration<Operation>
    
    init(with configuration: ApolloUploadConfiguration<Operation>) {
      self.configuration = configuration
    }
    
    func receive<S: Subscriber>(subscriber: S) where Failure == S.Failure, Output == S.Input {
      let subscription = ApolloUploadSubscription(subscriber: subscriber, configuration: configuration)
      subscriber.receive(subscription: subscription)
    }
  }
}

// MARK: -
/// Subscription definitions for each type of call
/// Each subscription has configuration parameters encapsulated in a struct
/// Each subscription also references the Apollo.Cancellable returned from calls, allowing the underlying Apollo task to be cancelled when the subscription is cancelled.
struct ApolloQueryConfiguration<Query: GraphQLQuery> {
  let client: ApolloClient
  let query: Query
  let cachePolicy: CachePolicy?
  let context: UnsafeMutableRawPointer?
  let queue: DispatchQueue?
}

private final class ApolloQuerySubscription<S: Subscriber, Query: GraphQLQuery>: Subscription
                      where S.Failure == Error, S.Input == GraphQLResult<Query.Data> {
  let configuration: ApolloQueryConfiguration<Query>
  var subscriber: S?
  var task: Apollo.Cancellable?
  
  init(subscriber: S, configuration: ApolloQueryConfiguration<Query>) {
    self.subscriber = subscriber
    self.configuration = configuration
  }
  
  func request(_ demand: Subscribers.Demand) {
    task = configuration.client.fetch(query: configuration.query,
                                      cachePolicy: configuration.cachePolicy ?? .returnCacheDataElseFetch,
                                      context: configuration.context,
                                      queue: configuration.queue ?? .main) { [weak self] result in
      switch result {
      case .success(let data):
        _ = self?.subscriber?.receive(data)
        self?.subscriber?.receive(completion: .finished)
        
      case .failure(let error):
        self?.subscriber?.receive(completion: .failure(error))
      }
    }
  }
  
  func cancel() {
    print("===> CANCEL QUERY \(configuration.query.operationName)")
    task?.cancel()
    subscriber = nil
  }
}

// MARK: -
struct ApolloMutationConfiguration<Mutation: GraphQLMutation> {
  let client: ApolloClient
  let mutation: Mutation
  let context: UnsafeMutableRawPointer?
  let queue: DispatchQueue?
}

private final class ApolloMutationSubscription<S: Subscriber, Mutation: GraphQLMutation>: Subscription
where S.Failure == Error, S.Input == GraphQLResult<Mutation.Data> {
  let configuration: ApolloMutationConfiguration<Mutation>
  var subscriber: S?
  var task: Apollo.Cancellable?
  
  init(subscriber: S, configuration: ApolloMutationConfiguration<Mutation>) {
    self.subscriber = subscriber
    self.configuration = configuration
  }
  
  func request(_ demand: Subscribers.Demand) {
    task = configuration.client.perform(mutation: configuration.mutation,
                                        context: configuration.context,
                                        queue: configuration.queue ?? .main) { [weak self] result in
      switch result {
      case .success(let data):
        _ = self?.subscriber?.receive(data)
        self?.subscriber?.receive(completion: .finished)
        
      case .failure(let error):
        self?.subscriber?.receive(completion: .failure(error))
      }
    }
  }
  
  func cancel() {
    print("===> CANCEL MUTATION \(configuration.mutation.operationName)")
    task?.cancel()
    subscriber = nil
  }
}

// MARK: -
struct ApolloUploadConfiguration<Operation: GraphQLOperation> {
  let client: ApolloClient
  let operation: Operation
  let context: UnsafeMutableRawPointer?
  let queue: DispatchQueue?
  let files: [GraphQLFile]
}

private final class ApolloUploadSubscription<S: Subscriber, Operation: GraphQLOperation>: Subscription
where S.Failure == Error, S.Input == GraphQLResult<Operation.Data> {
  let configuration: ApolloUploadConfiguration<Operation>
  var subscriber: S?
  var task: Apollo.Cancellable?
  
  init(subscriber: S, configuration: ApolloUploadConfiguration<Operation>) {
    self.subscriber = subscriber
    self.configuration = configuration
  }
  
  func request(_ demand: Subscribers.Demand) {
    task = configuration.client.upload(operation: configuration.operation,
                                       context: configuration.context,
                                       files: configuration.files,
                                       queue: configuration.queue ?? .main) { [weak self] result in
      switch result {
      case .success(let data):
        _ = self?.subscriber?.receive(data)
        self?.subscriber?.receive(completion: .finished)
        
      case .failure(let error):
        self?.subscriber?.receive(completion: .failure(error))
      }
    }
  }
  
  func cancel() {
    print("===> CANCEL UPLOAD \(configuration.operation.operationName)")
    task?.cancel()
    subscriber = nil
  }
}