feat: support concurrent chunk uploads

Author: TorstenDittmann

library/src/main/java/io/appwrite/Client.kt

@@ -35,6 +35,7 @@ import java.security.SecureRandom
 import java.security.cert.X509Certificate
 import java.util.concurrent.atomic.AtomicInteger
 import java.util.concurrent.atomic.AtomicLong
+import java.util.concurrent.atomic.AtomicReference
 import javax.net.ssl.SSLContext
 import javax.net.ssl.SSLSocketFactory
 import javax.net.ssl.TrustManager
@@ -590,7 +591,7 @@ class Client @JvmOverloads constructor(
                 responseType = Map::class.java
             )
 
-            if (index == 0 || uploadId == null) {
+            if (index == 0) {
                 uploadId = chunkResult["\$id"].toString()
             }
 
@@ -624,6 +625,9 @@ class Client @JvmOverloads constructor(
             val nextChunk = AtomicInteger(0)
             val completedChunks = AtomicInteger((offset / CHUNK_SIZE).toInt())
             val uploadedBytes = AtomicLong(offset.coerceAtMost(size))
+            val completedResultRef = AtomicReference<Map<*, *>?>(null)
+            val lastResultRef = AtomicReference(result)
+            val progressLock = Any()
 
             coroutineScope {
                 List(MAX_CONCURRENT_UPLOADS.coerceAtMost(chunks.size)) {
@@ -640,23 +644,27 @@ class Client @JvmOverloads constructor(
                             val chunksUploaded = completedChunks.incrementAndGet()
                             val sizeUploaded = uploadedBytes.addAndGet(end - start)
 
+                            lastResultRef.set(chunkResult)
                             if (isUploadComplete(chunkResult)) {
-                                result = chunkResult
+                                completedResultRef.set(chunkResult)
                             }
 
-                            onProgress?.invoke(
-                                UploadProgress(
-                                    id = uploadId ?: chunkResult["\$id"].toString(),
-                                    progress = sizeUploaded.coerceAtMost(size).toDouble() / size * 100,
-                                    sizeUploaded = sizeUploaded.coerceAtMost(size),
-                                    chunksTotal = chunkResult["chunksTotal"].toString().toInt(),
-                                    chunksUploaded = chunksUploaded,
+                            synchronized(progressLock) {
+                                onProgress?.invoke(
+                                    UploadProgress(
+                                        id = uploadId ?: chunkResult["\$id"].toString(),
+                                        progress = sizeUploaded.coerceAtMost(size).toDouble() / size * 100,
+                                        sizeUploaded = sizeUploaded.coerceAtMost(size),
+                                        chunksTotal = chunkResult["chunksTotal"].toString().toInt(),
+                                        chunksUploaded = chunksUploaded,
+                                    )
                                 )
-                            )
+                            }
                         }
                     }
                 }.awaitAll()
             }
+            result = completedResultRef.get() ?: lastResultRef.get()
         }
 
         return converter(result as Map<String, Any>)