feat: project setup and upload file

aws-runtime/aws-http/api/aws-http.api

@@ -171,6 +171,7 @@ public final class aws/sdk/kotlin/runtime/http/interceptors/IgnoreCompositeFlexi
 public final class aws/sdk/kotlin/runtime/http/interceptors/businessmetrics/AwsBusinessMetric : java/lang/Enum, aws/smithy/kotlin/runtime/businessmetrics/BusinessMetric {
 	public static final field DDB_MAPPER Laws/sdk/kotlin/runtime/http/interceptors/businessmetrics/AwsBusinessMetric;
 	public static final field S3_EXPRESS_BUCKET Laws/sdk/kotlin/runtime/http/interceptors/businessmetrics/AwsBusinessMetric;
+	public static final field S3_TRANSFER Laws/sdk/kotlin/runtime/http/interceptors/businessmetrics/AwsBusinessMetric;
 	public static fun getEntries ()Lkotlin/enums/EnumEntries;
 	public fun getIdentifier ()Ljava/lang/String;
 	public fun toString ()Ljava/lang/String;

aws-runtime/aws-http/common/src/aws/sdk/kotlin/runtime/http/interceptors/businessmetrics/AwsBusinessMetricsUtils.kt

@@ -62,6 +62,7 @@ internal fun formatMetrics(metrics: MutableSet<BusinessMetric>, logger: Logger):
 public enum class AwsBusinessMetric(public override val identifier: String) : BusinessMetric {
     S3_EXPRESS_BUCKET("J"),
     DDB_MAPPER("d"),
+    S3_TRANSFER("G"),
     ;
 
     @InternalApi

gradle/libs.versions.toml

@@ -28,6 +28,9 @@ mockk-version = "1.13.13"
 slf4j-version = "2.0.16"
 jsoup-version = "1.20.1"
 
+# s3 transfer manager
+s3-version = "1.5.62"
+
 [libraries]
 aws-kotlin-repo-tools-build-support = { module="aws.sdk.kotlin.gradle:build-support", version.ref = "aws-kotlin-repo-tools-version" }
 
@@ -92,6 +95,8 @@ smithy-kotlin-telemetry-provider-micrometer = { module = "aws.smithy.kotlin:tele
 smithy-kotlin-telemetry-provider-otel = { module = "aws.smithy.kotlin:telemetry-provider-otel", version.ref = "smithy-kotlin-runtime-version" }
 smithy-kotlin-test-suite = { module = "aws.smithy.kotlin:test-suite", version.ref = "smithy-kotlin-runtime-version" }
 smithy-kotlin-testing = { module = "aws.smithy.kotlin:testing", version.ref = "smithy-kotlin-runtime-version" }
+smithy-kotlin-test-jvm = { module = "aws.smithy.kotlin:http-test-jvm", version.ref = "smithy-kotlin-runtime-version" }
+smithy-kotlin-testing-jvm = { module = "aws.smithy.kotlin:testing-jvm", version.ref = "smithy-kotlin-runtime-version" }
 
 smithy-kotlin-codegen = { module = "software.amazon.smithy.kotlin:smithy-kotlin-codegen", version.ref = "smithy-kotlin-codegen-version" }
 smithy-kotlin-codegen-testutils = { module = "software.amazon.smithy.kotlin:smithy-kotlin-codegen-testutils", version.ref = "smithy-kotlin-codegen-version" }
@@ -124,6 +129,7 @@ kotlinx-serialization-json = { module = "org.jetbrains.kotlinx:kotlinx-serializa
 mockk = { module = "io.mockk:mockk", version.ref = "mockk-version" }
 
 ddb-local = { module = "com.amazonaws:DynamoDBLocal", version.ref = "ddb-local-version" }
+s3 = { module = "aws.sdk.kotlin:s3", version.ref = "s3-version" }
 
 [bundles]
 # bundle of smithy-kotlin dependencies all AWS service clients have

hll/build.gradle.kts

@@ -45,7 +45,7 @@ val hllPreviewVersion = if (sdkVersion.contains("-SNAPSHOT")) { // e.g. 1.3.29-b
 
 subprojects {
     group = "aws.sdk.kotlin"
-    version = hllPreviewVersion
+    version = if (name == "s3-transfer-manager") sdkVersion else hllPreviewVersion
     // TODO Use configurePublishing when migrating to Sonatype Publisher API / JReleaser
     configurePublishing("aws-sdk-kotlin")
 }

hll/s3-transfer-manager/build.gradle.kts

@@ -0,0 +1,26 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+description = "S3 Transfer Manager for the AWS SDK for Kotlin"
+extra["displayName"] = "AWS :: SDK :: Kotlin :: HLL :: S3TransferManager"
+extra["moduleName"] = "aws.sdk.kotlin.hll.s3transfermanager"
+
+kotlin {
+    sourceSets {
+        commonMain {
+            dependencies {
+                implementation(project(":aws-runtime:aws-http"))
+                implementation(libs.s3)
+            }
+        }
+        jvmTest {
+            dependencies {
+                implementation(libs.smithy.kotlin.test.jvm)
+                implementation(libs.smithy.kotlin.testing.jvm)
+                implementation(libs.smithy.kotlin.aws.signing.common)
+            }
+        }
+    }
+}

hll/s3-transfer-manager/common/src/aws/sdk/kotlin/hll/s3transfermanager/BusinessMetricInterceptor.kt

@@ -0,0 +1,21 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package aws.sdk.kotlin.hll.s3transfermanager
+
+import aws.sdk.kotlin.runtime.http.interceptors.businessmetrics.AwsBusinessMetric
+import aws.smithy.kotlin.runtime.businessmetrics.emitBusinessMetric
+import aws.smithy.kotlin.runtime.client.RequestInterceptorContext
+import aws.smithy.kotlin.runtime.http.interceptors.HttpInterceptor
+
+/**
+ * An interceptor that emits the S3 Transfer Manager business metric
+ */
+internal object BusinessMetricInterceptor : HttpInterceptor {
+    override suspend fun modifyBeforeSerialization(context: RequestInterceptorContext<Any>): Any {
+        context.executionContext.emitBusinessMetric(AwsBusinessMetric.S3_TRANSFER)
+        return context.request
+    }
+}

hll/s3-transfer-manager/common/src/aws/sdk/kotlin/hll/s3transfermanager/S3TransferManager.kt

@@ -0,0 +1,274 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package aws.sdk.kotlin.hll.s3transfermanager
+
+import aws.sdk.kotlin.hll.s3transfermanager.model.MultiPartDownloadType
+import aws.sdk.kotlin.hll.s3transfermanager.model.Part
+import aws.sdk.kotlin.hll.s3transfermanager.model.UploadFileRequest
+import aws.sdk.kotlin.hll.s3transfermanager.model.UploadFileResponse
+import aws.sdk.kotlin.hll.s3transfermanager.utils.buildCompleteMultipartUploadRequest
+import aws.sdk.kotlin.hll.s3transfermanager.utils.buildUploadPartRequest
+import aws.sdk.kotlin.hll.s3transfermanager.utils.ceilDiv
+import aws.sdk.kotlin.hll.s3transfermanager.utils.getNextPart
+import aws.sdk.kotlin.hll.s3transfermanager.utils.resolvePartSize
+import aws.sdk.kotlin.hll.s3transfermanager.utils.toCreateMultiPartUploadRequest
+import aws.sdk.kotlin.hll.s3transfermanager.utils.toPutObjectRequest
+import aws.sdk.kotlin.hll.s3transfermanager.utils.toUploadFileResponse
+import aws.sdk.kotlin.services.s3.S3Client
+import aws.sdk.kotlin.services.s3.abortMultipartUpload
+import aws.sdk.kotlin.services.s3.model.CompleteMultipartUploadRequest
+import aws.sdk.kotlin.services.s3.model.CompleteMultipartUploadResponse
+import aws.sdk.kotlin.services.s3.model.CompletedPart
+import aws.sdk.kotlin.services.s3.model.CreateMultipartUploadRequest
+import aws.sdk.kotlin.services.s3.model.CreateMultipartUploadResponse
+import aws.sdk.kotlin.services.s3.model.PutObjectRequest
+import aws.sdk.kotlin.services.s3.model.PutObjectResponse
+import aws.sdk.kotlin.services.s3.model.UploadPartRequest
+import aws.sdk.kotlin.services.s3.model.UploadPartResponse
+import aws.sdk.kotlin.services.s3.withConfig
+import aws.smithy.kotlin.runtime.content.ByteStream
+import aws.smithy.kotlin.runtime.io.SdkBuffer
+import aws.smithy.kotlin.runtime.telemetry.logging.logger
+import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.async
+import kotlinx.coroutines.coroutineScope
+
+/**
+ * High level utility for managing transfers to Amazon S3.
+ */
+public class S3TransferManager private constructor(
+    public val client: S3Client,
+    public val targePartSize: Long,
+    public val multipartUploadThreshold: Long,
+    public val multipartDownloadType: MultiPartDownloadType,
+    public val interceptors: MutableList<TransferInterceptor>,
+) {
+    internal var context: TransferContext = TransferContext()
+
+    public companion object {
+        public operator fun invoke(block: Builder.() -> Unit): S3TransferManager =
+            Builder().apply(block).build()
+    }
+
+    public class Builder {
+        public var client: S3Client? = null
+        public var targePartSize: Long = 8_000_000
+        public var multipartUploadThreshold: Long = 16_000_000L
+        public var multipartDownloadType: MultiPartDownloadType = Part
+        public var interceptors: MutableList<TransferInterceptor> = mutableListOf()
+
+        internal fun build(): S3TransferManager =
+            S3TransferManager(
+                client = client?.withConfig { interceptors += BusinessMetricInterceptor } ?: error("client must be set"),
+                targePartSize = targePartSize,
+                multipartUploadThreshold = multipartUploadThreshold,
+                multipartDownloadType = multipartDownloadType,
+                interceptors = interceptors,
+            )
+    }
+
+    /**
+     * Executes a sequence of operations around a hook.
+     *
+     * The execution flow is as follows:
+     * 1. Runs all interceptors scheduled to execute **before** the hook.
+     * 2. Executes the main hook logic.
+     * 3. Runs all interceptors scheduled to execute **after** the hook.
+     */
+    private suspend fun operationHook(hook: TransferHook, block: suspend () -> Any) {
+        when (hook) {
+            is TransferInitiated -> {
+                interceptors.forEach { it.readBeforeTransferInitiated(context) }
+                interceptors.forEach { context = it.modifyBeforeTransferInitiated(context) }
+                block.invoke()
+                interceptors.forEach { it.readAfterTransferInitiated(context) }
+                interceptors.forEach { context = it.modifyAfterTransferInitiated(context) }
+            }
+            is BytesTransferred -> {
+                interceptors.forEach { it.readBeforeBytesTransferred(context) }
+                interceptors.forEach { context = it.modifyBeforeBytesTransferred(context) }
+                block.invoke()
+                interceptors.forEach { it.readAfterBytesTransferred(context) }
+                interceptors.forEach { context = it.modifyAfterBytesTransferred(context) }
+            }
+            is FileTransferred -> {
+                interceptors.forEach { it.readBeforeFileTransferred(context) }
+                interceptors.forEach { context = it.modifyBeforeFileTransferred(context) }
+                block.invoke()
+                interceptors.forEach { it.readAfterFileTransferred(context) }
+                interceptors.forEach { context = it.modifyAfterFileTransferred(context) }
+            }
+            is TransferCompleted -> {
+                interceptors.forEach { it.readBeforeTransferCompleted(context) }
+                interceptors.forEach { context = it.modifyBeforeTransferCompleted(context) }
+                block.invoke()
+                interceptors.forEach { it.readAfterTransferCompleted(context) }
+                interceptors.forEach { context = it.modifyAfterTransferCompleted(context) }
+            }
+            else -> error("TransferHook not implemented: ${hook::class.simpleName}")
+        }
+    }
+
+    /**
+     * Uploads a byte stream to Amazon S3, automatically using multipart uploads
+     * for large objects as needed.
+     *
+     * This function handles the complexity of splitting the data into parts,
+     * uploading each part, and completing the multipart upload. For object smaller than [multipartUploadThreshold],
+     * a standard single-part upload is performed automatically.
+     *
+     * If the specified [targePartSize] for multipart uploads is too small to allow
+     * all parts to fit within S3's limit of 10,000 parts, the part size will be
+     * automatically increased so that exactly 10,000 parts are uploaded.
+     */
+    public suspend fun uploadFile(uploadFileRequest: UploadFileRequest): Deferred<UploadFileResponse> = coroutineScope {
+        val multiPartUpload = uploadFileRequest.contentLength >= multipartUploadThreshold
+        val uploadedParts = mutableListOf<CompletedPart>()
+        var mpuUploadId = "null"
+
+        val logger = coroutineContext.logger<S3TransferManager>()
+
+        /*
+         Handles transfer initiated hook
+         */
+        suspend fun transferInitiated(multiPartUpload: Boolean) {
+            context.transferredBytes = 0L
+            context.transferableBytes = uploadFileRequest.contentLength
+            context.request = if (multiPartUpload) {
+                uploadFileRequest.toCreateMultiPartUploadRequest()
+            } else {
+                uploadFileRequest.toPutObjectRequest()
+            }
+            operationHook(TransferInitiated) {
+                if (multiPartUpload) {
+                    context.response = client.createMultipartUpload(context.request as CreateMultipartUploadRequest)
+                    mpuUploadId = (context.response as CreateMultipartUploadResponse).uploadId ?: throw Exception("Missing upload id in create multipart upload response")
+                }
+            }
+        }
+
+        /*
+         Handles bytes transferred hook
+         */
+        suspend fun transferBytes(multiPartUpload: Boolean) {
+            if (multiPartUpload) {
+                try {
+                    val partSize = resolvePartSize(uploadFileRequest, this@S3TransferManager, logger)
+                    val numberOfParts = ceilDiv(uploadFileRequest.contentLength, partSize)
+                    val partSource = when (uploadFileRequest.body) {
+                        is ByteStream.Buffer -> uploadFileRequest.body.bytes()
+                        is ByteStream.ChannelStream -> uploadFileRequest.body.readFrom()
+                        is ByteStream.SourceStream -> uploadFileRequest.body.readFrom()
+                        else -> error("Unhandled body type: ${uploadFileRequest.body?.let { it::class.simpleName } ?: "null"}")
+                    }
+                    val partBuffer = SdkBuffer()
+                    var currentPartNumber = 1L
+
+                    while (context.transferredBytes!! < context.transferableBytes!!) {
+                        partBuffer.getNextPart(partSource, partSize, this@S3TransferManager)
+                        if (currentPartNumber != numberOfParts) {
+                            check(partBuffer.size == partSize) {
+                                "Part #$currentPartNumber size mismatch detected. Expected $partSize, actual: ${partBuffer.size}"
+                            }
+                        }
+
+                        context.request =
+                            buildUploadPartRequest(
+                                uploadFileRequest,
+                                partBuffer,
+                                currentPartNumber,
+                                mpuUploadId,
+                            )
+
+                        operationHook(BytesTransferred) {
+                            context.response = client.uploadPart(context.request as UploadPartRequest)
+                            context.transferredBytes = context.transferredBytes!! + partSize
+                        }
+
+                        uploadedParts += CompletedPart {
+                            partNumber = currentPartNumber.toInt()
+                            eTag = (context.response as UploadPartResponse).eTag
+                        }
+                        currentPartNumber += 1
+                    }
+
+                    check(uploadedParts.size == numberOfParts.toInt()) {
+                        "The number of uploaded parts does not match the expected count. Expected $numberOfParts, actual: ${uploadedParts.size}"
+                    }
+                } catch (uploadPartThrowable: Throwable) {
+                    try {
+                        client.abortMultipartUpload {
+                            bucket = uploadFileRequest.bucket
+                            expectedBucketOwner = uploadFileRequest.expectedBucketOwner
+                            key = uploadFileRequest.key
+                            requestPayer = uploadFileRequest.requestPayer
+                            uploadId = mpuUploadId
+                        }
+                        throw Exception("Multipart upload failed (ID: $mpuUploadId). One or more parts could not be uploaded", uploadPartThrowable)
+                    } catch (abortThrowable: Throwable) {
+                        throw Exception("Multipart upload failed (ID: $mpuUploadId). Unable to abort multipart upload.", abortThrowable)
+                    }
+                }
+            } else {
+                operationHook(BytesTransferred) {
+                    context.response = client.putObject(context.request as PutObjectRequest)
+                    context.transferredBytes = context.transferableBytes
+                }
+            }
+        }
+
+        /*
+         Handles transfer completed hook
+         */
+        suspend fun transferComplete(multiPartUpload: Boolean) {
+            if (multiPartUpload) {
+                context.request = buildCompleteMultipartUploadRequest(uploadFileRequest, mpuUploadId, uploadedParts)
+            }
+            operationHook(TransferCompleted) {
+                if (multiPartUpload) {
+                    try {
+                        context.response = client.completeMultipartUpload(context.request as CompleteMultipartUploadRequest)
+                    } catch (t: Throwable) {
+                        throw Exception("Unable to complete multipart upload with ID: $mpuUploadId", t)
+                    }
+                }
+            }
+        }
+
+        async {
+            checkNotNull(uploadFileRequest.body?.contentLength) {
+                "UploadFileRequest.body.contentLength must be set"
+            }
+            check(uploadFileRequest.body.contentLength == uploadFileRequest.contentLength) {
+                "contentLength mismatch. uploadFileRequest: ${uploadFileRequest.contentLength}, uploadFileRequest.body.contentLength: ${uploadFileRequest.body.contentLength}"
+            }
+
+            transferInitiated(multiPartUpload)
+            transferBytes(multiPartUpload)
+            transferComplete(multiPartUpload)
+
+            when (context.response) {
+                is PutObjectResponse -> (context.response as PutObjectResponse).toUploadFileResponse()
+                is CompleteMultipartUploadResponse -> (context.response as CompleteMultipartUploadResponse).toUploadFileResponse()
+                else -> error("Unexpected response: ${context.response?.let { it::class.simpleName } ?: "null"}")
+            }
+        }
+    }
+
+    /**
+     * Uploads a byte stream to Amazon S3, automatically using multipart uploads
+     * for large objects as needed.
+     *
+     * This function handles the complexity of splitting the data into parts,
+     * uploading each part, and completing the multipart upload. For object smaller than [multipartUploadThreshold],
+     * a standard single-part upload is performed automatically.
+     *
+     * If the specified [targePartSize] for multipart uploads is too small to allow
+     * all parts to fit within S3's limit of 10,000 parts, the part size will be
+     * automatically increased so that exactly 10,000 parts are uploaded.
+     */
+    public suspend inline fun uploadFile(crossinline block: UploadFileRequest.Builder.() -> Unit): Deferred<UploadFileResponse> = uploadFile(UploadFileRequest.Builder().apply(block).build())
+}