feat: project setup and upload file #1712
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| /* | ||
| * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| import org.gradle.kotlin.dsl.dependencies | ||
| import org.gradle.kotlin.dsl.sourceSets | ||
|
|
||
| /* | ||
| * 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 :: S3 Transfer Manager" | ||
| extra["moduleName"] = "aws.sdk.kotlin.hll.s3transfermanager" | ||
|
|
||
| kotlin { | ||
| sourceSets { | ||
| commonMain { | ||
| dependencies { | ||
| implementation(project(":aws-runtime:aws-http")) | ||
| implementation(project(":services:s3")) | ||
| } | ||
| } | ||
| jvmTest { | ||
| dependencies { | ||
| implementation(libs.smithy.kotlin.test.jvm) | ||
| implementation(libs.smithy.kotlin.testing.jvm) | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,269 @@ | ||
| /* | ||
| * 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.S3TransferManagerException | ||
| 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( | ||
|
Comment on lines
+40
to
+43
There was a problem hiding this comment. Style: This class has too much logic inside of it and that won't be sustainable once we add more operations and variants. We should refactor this, possibly so that individual operations are located in a single file and maybe even represented by a |
||
| public val client: S3Client, | ||
| public val partSizeBytes: Long, | ||
| public val multipartUploadThresholdBytes: Long, | ||
| public val multipartDownloadType: MultipartDownloadType, | ||
| public val interceptors: MutableList<TransferInterceptor>, | ||
| ) { | ||
| internal var context: TransferContext = TransferContext() | ||
|
There was a problem hiding this comment. Correctness: We cannot share a single mutable context across an S3TM instance. Multiple transfers may be happening at the same time. Contexts should either be local to an operation in progress just like with the SDK itself (i.e., |
||
|
|
||
| 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 partSizeBytes: Long = 8_000_000 | ||
| public var multipartUploadThresholdBytes: Long = 16_000_000L | ||
| public var multipartDownloadType: MultipartDownloadType = Part | ||
| public var interceptors: MutableList<TransferInterceptor> = mutableListOf() | ||
|
|
||
| internal fun build(): S3TransferManager = | ||
| S3TransferManager( | ||
| client = client?.withConfig { interceptors += S3TransferManagerBusinessMetricInterceptor } ?: error("client must be set"), | ||
| partSizeBytes = partSizeBytes, | ||
| multipartUploadThresholdBytes = multipartUploadThresholdBytes, | ||
| 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}") | ||
| } | ||
| } | ||
|
Comment on lines
+82
to
+114
There was a problem hiding this comment. Correctness: This doesn't follow the same pattern as low-level interceptors or DDB Mapper interceptors in that a single failure in any of the |
||
|
|
||
| /** | ||
| * 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 [multipartUploadThresholdBytes], | ||
| * a standard single-part upload is performed automatically. | ||
| * | ||
| * If the specified [partSizeBytes] 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 { | ||
|
There was a problem hiding this comment. Style: Method is too long and should be refactored. There was a problem hiding this comment. Correctness: This method shouldn't return Note that |
||
| val contentLength = uploadFileRequest.body?.contentLength ?: throw S3TransferManagerException("UploadFileRequest.body.contentLength must be set") | ||
| val multiPartUpload = contentLength >= multipartUploadThresholdBytes | ||
| val uploadedParts = mutableListOf<CompletedPart>() | ||
| lateinit var mpuUploadId: String | ||
|
|
||
| val logger = coroutineContext.logger<S3TransferManager>() | ||
|
There was a problem hiding this comment. Question: How/when is a logger set in the |
||
|
|
||
| /* | ||
| Handles transfer initiated hook | ||
| */ | ||
| suspend fun transferInitiated(multiPartUpload: Boolean) { | ||
| context.transferredBytes = 0L | ||
| context.transferableBytes = 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 S3TransferManagerException("Missing upload id in create multipart upload response") | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /* | ||
| Handles bytes transferred hook | ||
| */ | ||
| suspend fun transferBytes(multiPartUpload: Boolean) { | ||
| if (multiPartUpload) { | ||
| try { | ||
| val partSize = resolvePartSize(contentLength, this@S3TransferManager, logger) | ||
| val numberOfParts = ceilDiv(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 -> throw S3TransferManagerException("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) { | ||
| if (partBuffer.size != partSize) { | ||
| throw S3TransferManagerException("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 | ||
| } | ||
|
|
||
| if (uploadedParts.size != numberOfParts.toInt()) { | ||
| throw S3TransferManagerException("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 S3TransferManagerException("Multipart upload failed (ID: $mpuUploadId). One or more parts could not be uploaded", uploadPartThrowable) | ||
| } catch (abortThrowable: Throwable) { | ||
| throw S3TransferManagerException("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 S3TransferManagerException("Unable to complete multipart upload with ID: $mpuUploadId", t) | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| async { | ||
| transferInitiated(multiPartUpload) | ||
| transferBytes(multiPartUpload) | ||
| transferComplete(multiPartUpload) | ||
|
|
||
| when (context.response) { | ||
| is PutObjectResponse -> (context.response as PutObjectResponse).toUploadFileResponse() | ||
| is CompleteMultipartUploadResponse -> (context.response as CompleteMultipartUploadResponse).toUploadFileResponse() | ||
| else -> throw S3TransferManagerException("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 [multipartUploadThresholdBytes], | ||
| * a standard single-part upload is performed automatically. | ||
| * | ||
| * If the specified [partSizeBytes] 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()) | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 S3TransferManagerBusinessMetricInterceptor : HttpInterceptor { | ||
| override suspend fun modifyBeforeSerialization(context: RequestInterceptorContext<Any>): Any { | ||
| context.executionContext.emitBusinessMetric(AwsBusinessMetric.S3_TRANSFER) | ||
| return context.request | ||
| } | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
There are two new features
S3_TRANSFER_UPLOAD_DIRECTORYandS3_TRANSFER_DOWNLOAD_DIRECTORYwe should add here