Android batch upload example?

Thank youЗдравко and Greg-DB I think I have this working now, thanks for the pointers. Here's my code, in case this is useful for anyone else attempting to do this. I'm using Flutter so the code is littered with my own error handling classes which I can serialize and pass back to dart, but it should be a decent starting template for others. It's not fully tested, and also I'm new to Kotlin coroutines so I'm not sure I'm using coroutines / async etc. correctly!

import com.dropbox.core.InvalidAccessTokenException
import com.dropbox.core.NetworkIOException
import com.dropbox.core.RetryException
import com.dropbox.core.v2.DbxClientV2
import com.dropbox.core.v2.files.CommitInfo
import com.dropbox.core.v2.files.UploadSessionCursor
import com.dropbox.core.v2.files.UploadSessionFinishArg
import com.dropbox.core.v2.files.UploadSessionFinishErrorException
import com.dropbox.core.v2.files.UploadSessionType
import com.dropbox.core.v2.files.WriteError
import com.dropbox.core.v2.files.WriteMode
import kotlinx.coroutines.Deferred
import timber.log.Timber
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.delay
import kotlinx.coroutines.withContext
import java.io.File
import java.io.InputStream


class DropboxWriter {

    companion object {
        private const val BYTES_IN_MEGABYTE = 1048576
        // Must be multiple of 4MB
        // Larger chunk sizes will upload large files faster and usually with fewer network requests
        // but if a chunk upload fails the whole chunk must be re-uploaded
        private const val CHUNKED_UPLOAD_CHUNK_SIZE: Long = 4L * BYTES_IN_MEGABYTE
        // How many times to retry upload (with exponential time backoff) before returning with failure
        private const val MAX_RETRY_ATTEMPTS: Int = 5
    }

    suspend fun writeFilesToDropbox(
        credentialJsonString: String,
        filePaths: List<String>,
        remoteFolderPath: String
     MethodChannelResult<Unit?> = withContext(Dispatchers.IO) {
        assert(filePaths.size <= 1000) { "Max batch size is 1000" }
        Timber.i("Starting batch of ${filePaths.size} upload sessions")

        try {
            val client = DropboxClientFactory.getAuthenticatedClient(credentialJsonString)

            // Tell Dropbox a batch will be uploaded with the given number of files
            val uploadSessionStartBatchResult = client.files().uploadSessionStartBatch(
                filePaths.size.toLong(), UploadSessionType.CONCURRENT)

            // Upload each file in the batch
            val uploadResults: List<MethodChannelResult<UploadSessionFinishArg>> = filePaths.mapIndexed { index, filePath ->
                async {
                    uploadSessionAppend(client, uploadSessionStartBatchResult.sessionIds[index], filePath, remoteFolderPath)
                }
            }.map {it.await()}

            // If there were any failures in uploading
            val failureOrNull = uploadResults.firstOrNull { result -> result is MethodChannelResult.Failure }

            if(failureOrNull != null) {
                // Return the first failure
                return@withContext MethodChannelResult.Failure<Unit?>((failureOrNull as MethodChannelResult.Failure).error)
            }
            else {
                // Else we can now commit the batch using the UploadSessionFinishArgs
                val finishBatchResult = client.files().uploadSessionFinishBatchV2(
                    uploadResults.map{ result -> (result as MethodChannelResult.Success).value }
                )

                // If there were any failures in committing the batch
                val firstCommitFailureOrNull = finishBatchResult.entries.firstOrNull { entry -> entry.isFailure }

                if(firstCommitFailureOrNull != null) {
                    if(firstCommitFailureOrNull.failureValue.isPath
                        && firstCommitFailureOrNull.failureValue.pathValue is WriteError) {

                        // Catch some common errors and return handled error codes
                        if((firstCommitFailureOrNull.failureValue.pathValue as WriteError) == WriteError.INSUFFICIENT_SPACE) {
                            return@withContext MethodChannelResult.Failure(
                                MethodChannelError(BackupErrorCode.INSUFFICIENT_SPACE,"Insufficient space")
                            )
                        }
                        else if((firstCommitFailureOrNull.failureValue.pathValue as WriteError) == WriteError.NO_WRITE_PERMISSION) {
                            return@withContext MethodChannelResult.Failure(
                                MethodChannelError(BackupErrorCode.PERMISSIONS,"No write permission")
                            )
                        }
                    }
                    // Else return the first failure
                    return@withContext MethodChannelResult.Failure<Unit?>(
                        MethodChannelError(
                            BackupErrorCode.UNKNOWN,
                            firstCommitFailureOrNull.failureValue.toString())
                    )
                }
                else {
                    // Upload has succeeded
                    return@withContext MethodChannelResult.Success(null)
                }
            }
        }
        catch (e: Throwable) {
            return@withContext when (e) {
                is NetworkIOException -> {
                    MethodChannelResult.Failure(
                        MethodChannelError(BackupErrorCode.OFFLINE,"Can't reach Dropbox")
                    )
                }
                is InvalidAccessTokenException -> {
                    // Gets thrown when the access token you're using to make API calls is invalid.
                    // A more typical situation is that your access token was valid, but the user has since
                    // "unlinked" your application via the Dropbox website (http://www.dropbox.com/account#applications ).
                    // When a user unlinks your application, your access tokens for that user become invalid.
                    // You can re-run the authorization process to obtain a new access token.
                    MethodChannelResult.Failure(
                        MethodChannelError(
                            BackupErrorCode.AUTHENTICATION_FAILED,
                            e.message ?: "Access token was invalid",
                            e.stackTraceToString())
                    )
                }
                else -> {
                    MethodChannelResult.Failure(
                        MethodChannelError(
                            BackupErrorCode.UNKNOWN,
                            e.message ?: "Unknown error writing to dropbox",
                            e.stackTraceToString())
                    )
                }
            }
        }
    }

    private suspend fun uploadSessionAppend(client: DbxClientV2, sessionId: String,
                                            filePath: String, remoteFolderPath: String): MethodChannelResult<UploadSessionFinishArg> = withContext(Dispatchers.IO) {

        Timber.i("Using upload session with ID '${sessionId}' for file '${filePath}'")
        val file = File(filePath)
        if(file.exists()) {

            val remotePath = "/$remoteFolderPath/${file.name}"

            file.inputStream().buffered().use { bufferedInputStream ->

                val appendTasks: ArrayList<Deferred<Unit>> = arrayListOf()
                val sizeOfFileInBytes = file.length()
                var cursor: UploadSessionCursor? = null

                if(sizeOfFileInBytes > 0L) {
                    var totalNumberOfBytesRead = 0L

                    while(totalNumberOfBytesRead < sizeOfFileInBytes) {
                        cursor = UploadSessionCursor(sessionId, totalNumberOfBytesRead)
                        totalNumberOfBytesRead += CHUNKED_UPLOAD_CHUNK_SIZE
                        val close = totalNumberOfBytesRead >= sizeOfFileInBytes

                        appendTasks.add(
                            async {createAppendChunkTask(
                                client, bufferedInputStream, cursor!!, CHUNKED_UPLOAD_CHUNK_SIZE, sizeOfFileInBytes, close)
                            }
                        )
                    }

                }
                else {
                    // For empty files, just call append once to close the upload session.
                    cursor = UploadSessionCursor(sessionId, 0L)
                    appendTasks.add(
                        async {
                            createAppendChunkTask(
                                client,
                                bufferedInputStream,
                                cursor,
                                chunkSize = 0,
                                sizeOfFileInBytes,
                                close = true
                            )
                        })
                }

                try {
                    awaitAll(*appendTasks.toTypedArray())

                    return@withContext MethodChannelResult.Success(
                        UploadSessionFinishArg(cursor!!, CommitInfo(
                            remotePath,
                            WriteMode.OVERWRITE,
                            false, // autorename
                            null, // clientModified date
                            // Normally, users are made aware of any file modifications in their
                            // Dropbox account via notifications in the client software. If true,
                            // this tells the clients that this modification shouldn't result in a user notification.
                            false,  // mute
                            // List of custom properties to add to file
                            null, // propertyGroups
                            // Be more strict about how each WriteMode detects conflict. For example, always return a conflict error when getMode() = WriteMode.getUpdateValue() and the given "rev" doesn't match the existing file's "rev", even if the existing file has been deleted. This also forces a conflict even when the target path refers to a file with identical contents
                            false // strictConflict
                        ))
                    )
                }
                catch (e: FailedAfterMaxRetryAttemptsException) {
                    return@withContext MethodChannelResult.Failure(
                        MethodChannelError(
                            BackupErrorCode.OFFLINE,
                            e.message!!
                        )
                    )
                }
                catch (e: NetworkIOException) {
                    return@withContext MethodChannelResult.Failure(
                        MethodChannelError(BackupErrorCode.OFFLINE,"Can't reach Dropbox")
                    )
                }
                catch (e: Exception) {
                    return@withContext MethodChannelResult.Failure(
                        MethodChannelError(
                            BackupErrorCode.UNKNOWN,
                            e.message ?: "Unknown error writing to dropbox",
                            e.stackTraceToString())
                    )
                }
            }
        }
        else {
            return@withContext MethodChannelResult.Failure(
                MethodChannelError(
                    BackupErrorCode.UNKNOWN,
                    "Error writing to dropbox: file $filePath does not exist"))
        }
    }


    private suspend fun createAppendChunkTask(
        client: DbxClientV2,
        inputStream: InputStream,
        cursor: UploadSessionCursor,
        chunkSize: Long,
        sizeOfFileInBytes: Long,
        close: Boolean
    ) {
        var mutableCursor = cursor
        var mutableClose = close

        for(i in 0..MAX_RETRY_ATTEMPTS) {
            // Try to upload the chunk
            val result = appendChunkTask(client, inputStream, mutableCursor, chunkSize, mutableClose)
            when(result.type) {
                AppendResult.ResultType.Success -> {
                    return
                }

                // If it fails with a result type of Retry, retry after waiting
                AppendResult.ResultType.Retry -> {
                    // Wait for the specified amount of time
                    delay(result.backoffMillis!!)
                    // and try again next time around the loop
                }

                // If it fails with a result type of RetryWithCorrectedOffset
                AppendResult.ResultType.RetryWithCorrectedOffset -> {
                    // Correct the cursor position
                    mutableCursor = UploadSessionCursor(cursor.sessionId, result.correctedOffset!!)
                    mutableClose = result.correctedOffset + CHUNKED_UPLOAD_CHUNK_SIZE >= sizeOfFileInBytes
                    Timber.w("Append failed because the provided offset ${cursor.offset} " +
                            "should have been ${mutableCursor.offset}, retrying with corrected offset")
                    // and try again next time around the loop
                }
            }
        }
        // If we reach here, uploading the chunk failed after reaching the max
        // number of upload attempts
        throw FailedAfterMaxRetryAttemptsException()
    }

    private fun appendChunkTask(
        client: DbxClientV2,
        inputStream: InputStream,
        cursor: UploadSessionCursor,
        chunkSize: Long,
        close: Boolean
     AppendResult {
        try {
            Timber.d("Appending to upload session with ID '${cursor.sessionId}' " +
                    "at offset: ${cursor.offset}")
            client.files()
                .uploadSessionAppendV2Builder(cursor)
                .withClose(close)
                .uploadAndFinish(inputStream, chunkSize)
            return AppendResult(AppendResult.ResultType.Success)
        }
        catch(e: RetryException) {
            return AppendResult(AppendResult.ResultType.Retry, backoffMillis = e.backoffMillis)
        }
        catch(e: NetworkIOException) {
            return AppendResult(AppendResult.ResultType.Retry)
        }
        catch (e: UploadSessionFinishErrorException) {
            if (e.errorValue.isLookupFailed && e.errorValue.lookupFailedValue.isIncorrectOffset) {
                // server offset into the stream doesn't match our offset (uploaded). Seek to
                // the expected offset according to the server and try again.
                return AppendResult(
                    AppendResult.ResultType.RetryWithCorrectedOffset,
                    correctedOffset = e.errorValue
                        .lookupFailedValue
                        .incorrectOffsetValue
                        .correctOffset)
            } else {
                // some other error occurred
                throw e
            }
        }
    }
}

class FailedAfterMaxRetryAttemptsException() : Exception("Upload failed after reaching maximum number of retries")

class AppendResult(val type: ResultType, val correctedOffset: Long? = null, val backoffMillis: Long? = null) {
    enum class ResultType {
        Success,
        Retry,
        RetryWithCorrectedOffset;
    }
}

enum class BackupErrorCode(val code: Int) {
    UNKNOWN(0),
    OFFLINE(1),
    INSUFFICIENT_SPACE(2),
    PERMISSIONS(3),
    AUTHENTICATION_FAILED(4),
}

sealed class MethodChannelResult<out S> {
    data class Success<out S>(val value: S) : MethodChannelResult<S>()
    data class Failure<out S>(val error: MethodChannelError) : MethodChannelResult<S>()
}

data class MethodChannelError(val code: BackupErrorCode, val message: String, val stackTraceAsString: String? = null)