Combine S3 Uploads

Uploads.swift

/// Function that will take an Array of `[S3Uploadable]` objects and returns a publisher that will complete once all uploads are done.
/// - Parameter objects: An array of `S3Uploadable` objects
/// - Returns: An `AnyPublisher` that will publish the array of objects you passed in and a faileur type of `Never`
public func upload(uploadableObjects objects: [S3Uploadable]) -> AnyPublisher<[S3Uploadable], Error> {
    //We will collect some Futures into an array
    var futures: [Future<S3Uploadable, Error>] = []
    
    //Collect the futures
    for object in objects {
        let future = self.upload(objectToUpload: object)
        futures.append(future)
    }
    //Call the MergeMany Struct and collect the publishers finally rease to any publisher.
    return Publishers.MergeMany(futures).collect().eraseToAnyPublisher()
}

/// Upload a single `S3Uploadable` object and get a `Future` back to be fulfilled when that upload completes
/// - Parameter object: The object to upload
/// - Returns: The Future Publisher
public func upload(objectToUpload object: S3Uploadable) -> Future<S3Uploadable, Error> {
  //Create the future object.
  let future: Future<S3Uploadable, Error> = Future<S3Uploadable, Error> { promise in

      //My transfer utility was set up with a custom key. If I can not find it filfill the promise with an error.
      guard let transferUtility = AWSS3TransferUtility.s3TransferUtility(forKey: self.utilityKey.key) else {
          promise(.failure(S3TransferError.couldNotRetrieveTransferUtility))
          return
      }

      let expression = AWSS3TransferUtilityUploadExpression()
      expression.progressBlock = { (_ task: AWSS3TransferUtilityTask, _ progress: Progress) in
          //The progress of an object is a CurrentValueSubject so I should send the new progress value.
          object.progress.send(progress.fractionCompleted)
      }


      let completionHandle: AWSS3TransferUtilityUploadCompletionHandlerBlock = { (task, p_error) -> Void in
          //Complete the promise with an error or success.
          guard let error = p_error else {
              promise(.success(object))
              return
          }
          promise(.failure(error))
      }

      let key: String = object.objectCloudKey

      let contentType: String = object.contentType.rawValue

      //My protocol will help determine the method to use. However I read the S3 source code and saw they save the data to disk first rather then upload from memory.
      switch object.uploadableObjectLocation {
      case .DATA_OBJECT(let data):
          _ = transferUtility.uploadData(data, key: key, contentType: contentType, expression: expression, completionHandler: completionHandle).continueWith { (task) -> Any? in
              //I want the task to be assigned to the object so I can publish aspects of it to the UI
              if let result = task.result {
                  object.add(task: result)
              }
              guard let error = task.error else {
                  return nil
              }
              promise(.failure(error))
              return nil
          } as? AWSTask<AWSS3TransferUtilityTask>
      case .LOCAL_FILE(let url):
          _ = transferUtility.uploadFile(url, key: key, contentType: contentType, expression: expression, completionHandler: completionHandle).continueWith(block: { (task) -> Any? in
              if let result = task.result {
                  object.add(task: result)
              }
              guard let error = task.error else {
                  return nil
              }
              promise(.failure(error))
              return nil
          }) as? AWSTask<AWSS3TransferUtilityTask>
      }
  }
  return future
}