Rough Replication of JavaScript's Promise.All using AnyPublisher in Swift's Combine.

anypublisher-all.swift

import Combine
import Foundation

/// AnyPublisher.All
///
/// Implementation of `AnyPublisher.all`. Takes an array of `AnyPublisher`
/// objects which must all conform to the specified `Output` and `Error` types.
///
///   func foo() -> AnyPublisher<Int, Never> {
///     return Deferred {
///       return Future { promise in
///         promise(.success(420))
///       }
///     }
///     .eraseToAnyPublisher()
///   }
///
///   func bar() -> AnyPublisher<Int, Never> {
///     return Deferred {
///       return Future { promise in
///         promise(.success(69))
///       }
///     }
///     .eraseToAnyPublisher()
///   }
///
///
///   var cancellableBag = Set<AnyCancellable>()
///
///   AnyPublisher<Int, MyError>.all([foo(), bar(), biz()])
///    .sink(
///      receiveCompletion: { completion in
///        switch completion {
///          case .finished:
///            print("Completed Successfully")
///            break
///          case let .failure(error):
///            print("Finished with Error: \(error)")
///            break
///        }
///      },
///      receiveValue: { output in
///        print(output)
///      }
///    )
///    .store(in: &cancellableBag)
///
///
/// - Returns: A new instance of AnyPublisher that will result in either an array of the specified `Output`
///            type in input order, or an error of the specified `Error` type.
///
/// - Note: In current implementation will not complete until all publishers have completed, even if one
///         were to fail early!
///
public extension AnyPublisher {
  static func all<Output, Error>(
    _ publishers: [AnyPublisher<Output, Error>],
    queue: DispatchQueue = .main
  ) -> AnyPublisher<[Output], Error> {
    return Deferred {
      return Future { promise in
        var results: [Output?] = Array(repeating: nil, count: publishers.count)
        var firstError: Error?
        var cancellableBag = Set<AnyCancellable>()
        
        queue.async {
          let group = DispatchGroup()
          for (index, publisher) in publishers.enumerated() {
            group.enter()
            
            publisher
              .sink(
                receiveCompletion: { completion in
                  switch completion {
                    case let .failure(error):
                      if firstError == nil {
                        firstError = error
                      }
                      break
                    default:
                      break
                  }
                  
                  group.leave()
                },
                receiveValue: { value in
                  results[index] = value
                }
              )
              .store(in: &cancellableBag)
          }
          
          group.wait()
          
          group.notify(queue: queue) {
            if let firstError = firstError {
              promise(.failure(firstError))
            } else {
              promise(.success(results.compactMap { $0 }))
            }
          }
        }
      }
    }
    .eraseToAnyPublisher()
  }
}

// Usage Example

enum MyError: Error {
  case unknown
}

let myWorkQueue = DispatchQueue.init(label: "myWorkQueue")

func foo() -> AnyPublisher<Int, MyError> {
  return Deferred {
    return Future { promise in
      print("Finished foo")
      promise(.success(420))
    }
  }
  .eraseToAnyPublisher()
}

func bar() -> AnyPublisher<Int, MyError> {
  return Deferred {
    return Future { promise in
      myWorkQueue.asyncAfter(deadline: .now() + 4.0) {
        print("Finished bar")
        promise(.success(69))
      }
    }
  }
  .eraseToAnyPublisher()
}

func biz() -> AnyPublisher<Int, MyError> {
  return Deferred {
    return Future { promise in
      myWorkQueue.asyncAfter(deadline: .now() + 1.0) {
        print("Finished biz")
        promise(.success(666))
      }
    }
  }
  .eraseToAnyPublisher()
}


var cancellableBag = Set<AnyCancellable>()

// Call `AnyPublisher.all` specifying result Output and Error types.
AnyPublisher<Int, MyError>.all([foo(), bar(), biz()])
  .receive(on: DispatchQueue.main)
  .sink(
    receiveCompletion: { completion in
      switch completion {
        case .finished:
          print("Completed Successfully")
          break
        case let .failure(error):
          print("Finished with Error: \(error)")
          break
      }
    },
    receiveValue: { output in
      print(output)
    }
  )
  .store(in: &cancellableBag)

/// Output from Example Above:
///
///    Finished foo
///    Finished biz
///    Finished bar
///    [420, 69, 666]
///    Completed Successfully