AsyncStream in a background file uploader

[ghost]

Hello TCA Community!

I hope everyone is doing well. I am reaching out to request your valuable help to further enhance the background file upload dependency. I have created a FileUploaderClient that manages the file upload process using URLSession's background configuration.

import SwiftUI
import ComposableArchitecture

struct FileUploaderClient {
  var upload: @Sendable (Data, URL, String) async throws -> AsyncStream<Event>
}

extension FileUploaderClient: DependencyKey {
  static var liveValue: Self {
    return Self(
      upload: { try await FileUploaderActor.shared.uploadFile($0, to: $1, id: $2) }
    )
  }
}

extension DependencyValues {
  var fileUploader: FileUploaderClient {
    get { self[FileUploaderClient.self] }
    set { self[FileUploaderClient.self] = newValue }
  }
}

enum Event: Equatable {
  case didFinish
  case didWrite(Double)
}

final actor FileUploaderActor: GlobalActor {

  static let shared = FileUploaderActor()

  var backgroundEventHandlers: [String: () -> ()] = [:]
  var tasks: [String: Delegate] = [:]

  func uploadFile(_ data: Data, to targetURL: URL, id: String) throws -> AsyncStream<Event> {
    let fileURL = URL(fileURLWithPath: NSTemporaryDirectory()).appendingPathComponent(id)
    try data.write(to: fileURL)

    let delegate = Delegate()
    let uuid = "background-\(id)"

    let config = URLSessionConfiguration.background(withIdentifier: uuid)
    config.isDiscretionary = false
    config.sessionSendsLaunchEvents = true
    let session = URLSession(configuration: config, delegate: delegate, delegateQueue: nil)

    var request = URLRequest(url: targetURL, cachePolicy: .reloadIgnoringLocalCacheData)
    request.httpMethod = "PUT"
    request.setValue("BlockBlob", forHTTPHeaderField: "x-ms-blob-type")

    let task = session.uploadTask(with: request, fromFile: fileURL)
    defer { task.resume() }
    var continuation: AsyncStream<Event>.Continuation!
    let stream = AsyncStream<Event> {
      $0.onTermination = { _ in
        task.cancel()
        Task { await self.removeTask(id: uuid) }
      }
      continuation = $0
    }
    delegate.continuation = continuation
    self.tasks[uuid] = delegate
    return stream
  }

  private func removeTask(id: String) {
    self.tasks[id] = nil
  }

  func handleEventsForBackgroundURLSection(identifier: String, completionHandler: @escaping () -> ()) async {
    backgroundEventHandlers[identifier] = completionHandler
  }

  func sessionDidFinishEventsForBackgroundURLSession(session: URLSession) async {
    let identifier = session.configuration.identifier
    if let identifier, let handler = backgroundEventHandlers[identifier] {
      handler()
      backgroundEventHandlers.removeValue(forKey: identifier)
    }
  }
}

final class Delegate: NSObject, URLSessionTaskDelegate, URLSessionDelegate {
  var continuation: AsyncStream<Event>.Continuation?

  func urlSession(_ session: URLSession, task: URLSessionTask, didCompleteWithError error: Error?) {
    self.continuation?.yield(.didFinish)
    self.continuation?.finish()
  }

  func urlSession(
    _ session: URLSession,
    task: URLSessionTask,
    didSendBodyData bytesSent: Int64,
    totalBytesSent: Int64,
    totalBytesExpectedToSend: Int64
  ) {
    let progress = Double(totalBytesSent) / Double(totalBytesExpectedToSend)
    self.continuation?.yield(.didWrite(progress))
  }

  func urlSessionDidFinishEvents(forBackgroundURLSession session: URLSession) {
    session.getTasksWithCompletionHandler { dataTasks, uploadTasks, downloadTasks in
      if dataTasks.isEmpty && uploadTasks.isEmpty && downloadTasks.isEmpty {
        Task {
          await FileUploaderActor.shared.sessionDidFinishEventsForBackgroundURLSession(session: session)
        }
      }
    }
  }
}

Here's how I am consuming the client

case let .updateImageTapped(imageData):
        return .run { send in
         // ... other code ...
         let actions = try await self.fileUploader.upload(imageData, writeUrl, uuid)
         for await action in actions {
           await send(.uploadedFileChanged(action))
         }
         .cancellable(id: CancelID.uploadFile)

For some reason, calling the action a second time doesn't seem to make the group task work as expected. I would appreciate any help or insights you can provide to resolve this issue

[kamcma]

What is the purpose of the task group? Why not just

for try await action in self.fileUploader.upload(...) {
  await send(.uploadedFileChanged(action)
}

[kamcma]

Not just for me, but for others as well, can you elaborate on what you expect to happen and what is happening that is not what you expect? Because it's not complete, runnable code, describing it as "doesn't work as expected" isn't quite enough to go on.

[ghost]

Thank you for getting back to me. I'm sorry for not providing enough detail in my previous message. The issue is related to the file uploader dependency. When I try to upload a photo for the second time, the stream does not generate any actions. I have attached a video showing the console logs, which demonstrate that the logs only appear during the first upload attempt, and subsequent upload attempts do not produce any logs or actions. 🙏🏻

demo.mov

[kamcma]

If you put a breakpoint just inside the effect, are you reaching the start of effect execution each time you invoke the .updateImageTapped action? I'm trying to determine if the problem is within the reducer and TCA code, or if the problem is within the dependency implementation.

[ghost]

I sorted it out it was a problem with the dependency. Since I’m dealing with multiple URLSession instances each URLSession needed a unique identifier and somehow i had some trouble with that specific ID. I’ll share an updated version of the source code in case anyone is interested!