PublisherQueue — Serialise the execution of multiple publishers like OperationQueue

AnyEventPublisher.swift

import Combine

public enum AnyError: Error {
  case never(Never)
  case failure(Error)
}

public enum AnyEvent {
  case output(Any)
  case completion(Subscribers.Completion<AnyError>)
}

extension Publisher where Failure == Never {
  public func eraseToAnyEventPublisher() -> AnyPublisher<AnyEvent, Never> {
    map(AnyEvent.output)
      .catch { error in Just(.completion(.failure(.never(error)))) }
      .eraseToAnyPublisher()
  }
}

extension Publisher {
  public func eraseToAnyEventPublisher() -> AnyPublisher<AnyEvent, Never> {
    map(AnyEvent.output)
    .catch { error in Just(.completion(.failure(.failure(error)))) }
    .eraseToAnyPublisher()
  }
}

PublisherQueue.swift

import Combine

public final class PublisherQueue {
  private typealias Operation = Publishers.MakeConnectable<AnyPublisher<AnyEvent, Never>>

  private let operationQueue: PassthroughSubject<Operation, Never>
  private var subscriptions: Set<AnyCancellable>

  public init(size: Int, maxConcurrentPublishers: Subscribers.Demand = .max(1)) {
    self.operationQueue = PassthroughSubject()
    self.subscriptions = []

    operationQueue
      .buffer(size: size, prefetch: .keepFull, whenFull: .dropNewest)
      .flatMap(maxPublishers: maxConcurrentPublishers) { operation in
        operation.autoconnect()
      }
      .sink { _ in }
      .store(in: &subscriptions)
  }

  public func queuedPublisher<P: Publisher>(_ publisher: P) -> AnyPublisher<P.Output, P.Failure>
    where P.Failure == Never
  {
    let operation: Operation = publisher
      .eraseToAnyEventPublisher()
      .makeConnectable()

    return operation.flatMap { event -> AnyPublisher<P.Output, P.Failure> in
      switch event {
      case let .output(value):
        return Result.success(value as! P.Output).publisher.eraseToAnyPublisher()
      case .completion(.finished):
        return Empty().eraseToAnyPublisher()
      case .completion(.failure(.failure)):
        fatalError("unreachable")
      }
    }
    .handleEvents(receiveSubscription: { [operationQueue] _ in
      operationQueue.send(operation)
    })
    .eraseToAnyPublisher()
  }

  public func queuedPublisher<P: Publisher>(_ publisher: P) -> AnyPublisher<P.Output, P.Failure> {
    let operation: Operation = publisher
      .eraseToAnyEventPublisher()
      .makeConnectable()

    return operation.setFailureType(to: P.Failure.self).flatMap { event -> AnyPublisher<P.Output, P.Failure> in
      switch event {
      case let .output(value):
        return Result.success(value as! P.Output).publisher.eraseToAnyPublisher()
      case .completion(.finished):
        return Empty().eraseToAnyPublisher()
      case let .completion(.failure(.failure(error))):
        return Result.failure(error as! P.Failure).publisher.eraseToAnyPublisher()
      }
    }
    .handleEvents(receiveSubscription: { [operationQueue] _ in
      operationQueue.send(operation)
    })
    .eraseToAnyPublisher()
  }
}

Test.swift

import Combine
import Dispatch

func randomInt() -> Deferred<Future<Int, Never>> {
  Deferred {
    Future { promise in
      DispatchQueue.global().asyncAfter(deadline: .now() + 0.01) {
        promise(.success(.random(in: .min ... .max)))
      }
    }
  }
}

let q = PublisherQueue(size: .max)
var subscriptions = Set<AnyCancellable>()
var resultCount = 0
let group = DispatchGroup()

for _ in 1 ... 1000 {
  group.enter()
  q.queuedPublisher(randomInt()).sink(receiveCompletion: { _ in group.leave() }) { number in
    resultCount += 1
  }.store(in: &subscriptions)
}

group.wait()
print(resultCount, "results") // 1000 results

This is now available as a Swift package: https://github.com/sharplet/PublisherQueue.