Merge pull request #64 from appwrite/concurrent-chunk-uploads-1-9-x-minimal

feat: support concurrent chunk uploads

Author: TorstenDittmann

Sources/Appwrite/Client.swift

@@ -16,6 +16,7 @@ open class Client {
 
     // MARK: Properties
     public static var chunkSize = 5 * 1024 * 1024 // 5MB
+    public static var maxConcurrentUploads = 8
 
     open var endPoint = "https://cloud.appwrite.io/v1"
 
@@ -583,6 +584,7 @@ open class Client {
 
         var offset = 0
         var result = [String:Any]()
+        var uploadId = idParamName != nil ? params[idParamName!] as? String : nil
 
         if idParamName != nil {
             // Make a request to check if a file already exists
@@ -596,37 +598,113 @@ open class Client {
                 )
                 let chunksUploaded = map["chunksUploaded"] as! Int
                 offset = chunksUploaded * Client.chunkSize
+                result = map
             } catch {
                 // File does not exist yet, swallow exception
             }
         }
 
-        while offset < size {
-            let slice = (input.data as! ByteBuffer).getSlice(at: offset, length: Client.chunkSize)
-                ?? (input.data as! ByteBuffer).getSlice(at: offset, length: Int(size - offset))
+        let totalChunks = Int(ceil(Double(size) / Double(Client.chunkSize)))
+        var nextChunk = offset / Client.chunkSize
+        var completedChunks = nextChunk
+        var uploadedBytes = min(offset, size)
+        var completedResponse: [String: Any]? = nil
+        var lastChunkResponse: [String: Any]? = nil
+        let baseParams = params
+        let baseHeaders = headers
+
+        func isUploadComplete(_ response: [String: Any]) -> Bool {
+            guard let chunksUploaded = response["chunksUploaded"] as? Int else {
+                return false
+            }
+
+            let chunksTotal = response["chunksTotal"] as? Int ?? totalChunks
+            return chunksUploaded >= chunksTotal
+        }
 
-            params[paramName] = InputFile.fromBuffer(slice!, filename: input.filename, mimeType: input.mimeType)
-            headers["content-range"] = "bytes \(offset)-\(min((offset + Client.chunkSize) - 1, size - 1))/\(size)"
+        func uploadChunk(index: Int, uploadId: String?) async throws -> (Int, Int, [String: Any]) {
+            let chunkOffset = index * Client.chunkSize
+            let chunkLength = min(Client.chunkSize, size - chunkOffset)
+            guard let slice = (input.data as! ByteBuffer).getSlice(at: chunkOffset, length: chunkLength) else {
+                throw AppwriteError(message: "Failed to read upload chunk")
+            }
+            var chunkParams = baseParams
+            var chunkHeaders = baseHeaders
+            chunkParams[paramName] = InputFile.fromBuffer(slice, filename: input.filename, mimeType: input.mimeType)
+            chunkHeaders["content-range"] = "bytes \(chunkOffset)-\(chunkOffset + chunkLength - 1)/\(size)"
+            if let uploadId = uploadId {
+                chunkHeaders["x-appwrite-id"] = uploadId
+            }
 
-            result = try await call(
+            let chunkResult = try await call(
                 method: "POST",
                 path: path,
-                headers: headers,
-                params: params,
+                headers: chunkHeaders,
+                params: chunkParams,
                 converter: { return $0 as! [String: Any] }
             )
 
-            offset += Client.chunkSize
-            headers["x-appwrite-id"] = result["$id"] as? String
+            return (index, chunkLength, chunkResult)
+        }
+
+        if nextChunk == 0 {
+            let first = try await uploadChunk(index: 0, uploadId: uploadId)
+            result = first.2
+            uploadId = result["$id"] as? String
+            nextChunk = 1
+            completedChunks = 1
+            uploadedBytes = first.1
             onProgress?(UploadProgress(
-                id: result["$id"] as? String ?? "",
-                progress: Double(min(offset, size))/Double(size) * 100.0,
-                sizeUploaded: min(offset, size),
-                chunksTotal: result["chunksTotal"] as? Int ?? -1,
-                chunksUploaded: result["chunksUploaded"] as? Int ?? -1
+                id: uploadId ?? "",
+                progress: Double(uploadedBytes)/Double(size) * 100.0,
+                sizeUploaded: uploadedBytes,
+                chunksTotal: result["chunksTotal"] as? Int ?? totalChunks,
+                chunksUploaded: result["chunksUploaded"] as? Int ?? completedChunks
             ))
         }
 
+        let maxConcurrency = Client.maxConcurrentUploads
+
+        try await withThrowingTaskGroup(of: (Int, Int, [String: Any]).self) { group in
+            var inFlight = 0
+
+            while inFlight < maxConcurrency && nextChunk < totalChunks {
+                let index = nextChunk
+                let currentUploadId = uploadId
+                group.addTask { try await uploadChunk(index: index, uploadId: currentUploadId) }
+                nextChunk += 1
+                inFlight += 1
+            }
+
+            while let chunk = try await group.next() {
+                inFlight -= 1
+                completedChunks += 1
+                uploadedBytes += chunk.1
+                lastChunkResponse = chunk.2
+                if isUploadComplete(chunk.2) {
+                    completedResponse = chunk.2
+                }
+
+                onProgress?(UploadProgress(
+                    id: uploadId ?? "",
+                    progress: Double(min(uploadedBytes, size))/Double(size) * 100.0,
+                    sizeUploaded: min(uploadedBytes, size),
+                    chunksTotal: chunk.2["chunksTotal"] as? Int ?? totalChunks,
+                    chunksUploaded: chunk.2["chunksUploaded"] as? Int ?? completedChunks
+                ))
+
+                while inFlight < maxConcurrency && nextChunk < totalChunks {
+                    let index = nextChunk
+                    let currentUploadId = uploadId
+                    group.addTask { try await uploadChunk(index: index, uploadId: currentUploadId) }
+                    nextChunk += 1
+                    inFlight += 1
+                }
+            }
+        }
+
+        result = completedResponse ?? lastChunkResponse ?? result
+
         return try converter!(result)
     }