Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified
📁 ..	-	2025-05-25 12:27
📄 arn.go package s3manager import ( "fmt" "github.com/aws/aws-sdk-go/aws/arn" ) func validateSupportedARNType(bucket string) error { if !arn.IsARN(bucket) { return nil } parsedARN, err := arn.Parse(bucket) if err != nil { return err } if parsedARN.Service == "s3-object-lambda" { return fmt.Errorf("manager does not support s3-object-lambda service ARNs") } return nil }	383 B	2025-05-25 12:27
📄 batch.go package s3manager import ( "bytes" "fmt" "io" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/aws/client" "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/service/s3" "github.com/aws/aws-sdk-go/service/s3/s3iface" ) const ( // DefaultBatchSize is the batch size we initialize when constructing a batch delete client. // This value is used when calling DeleteObjects. This represents how many objects to delete // per DeleteObjects call. DefaultBatchSize = 100 ) // BatchError will contain the key and bucket of the object that failed to // either upload or download. type BatchError struct { Errors Errors code string message string } // Errors is a typed alias for a slice of errors to satisfy the error // interface. type Errors []Error func (errs Errors) Error() string { buf := bytes.NewBuffer(nil) for i, err := range errs { buf.WriteString(err.Error()) if i+1 < len(errs) { buf.WriteString("\n") } } return buf.String() } // Error will contain the original error, bucket, and key of the operation that failed // during batch operations. type Error struct { OrigErr error Bucket string Key string } func newError(err error, bucket, key string) Error { return Error{ err, bucket, key, } } func (err Error) Error() string { origErr := "" if err.OrigErr != nil { origErr = ":\n" + err.OrigErr.Error() } return fmt.Sprintf("failed to perform batch operation on %q to %q%s", aws.StringValue(err.Key), aws.StringValue(err.Bucket), origErr, ) } // NewBatchError will return a BatchError that satisfies the awserr.Error interface. func NewBatchError(code, message string, err []Error) awserr.Error { return &BatchError{ Errors: err, code: code, message: message, } } // Code will return the code associated with the batch error. func (err BatchError) Code() string { return err.code } // Message will return the message associated with the batch error. func (err BatchError) Message() string { return err.message } func (err BatchError) Error() string { return awserr.SprintError(err.Code(), err.Message(), "", err.Errors) } // OrigErr will return the original error. Which, in this case, will always be nil // for batched operations. func (err BatchError) OrigErr() error { return err.Errors } // BatchDeleteIterator is an interface that uses the scanner pattern to // iterate through what needs to be deleted. type BatchDeleteIterator interface { Next() bool Err() error DeleteObject() BatchDeleteObject } // DeleteListIterator is an alternative iterator for the BatchDelete client. This will // iterate through a list of objects and delete the objects. // // Example: // // iter := &s3manager.DeleteListIterator{ // Client: svc, // Input: &s3.ListObjectsInput{ // Bucket: aws.String("bucket"), // MaxKeys: aws.Int64(5), // }, // Paginator: request.Pagination{ // NewRequest: func() (request.Request, error) { // var inCpy ListObjectsInput // if input != nil { // tmp := input // inCpy = &tmp // } // req, _ := c.ListObjectsRequest(inCpy) // return req, nil // }, // }, // } // // batcher := s3manager.NewBatchDeleteWithClient(svc) // if err := batcher.Delete(aws.BackgroundContext(), iter); err != nil { // return err // } type DeleteListIterator struct { Bucket string Paginator request.Pagination objects []s3.Object } // NewDeleteListIterator will return a new DeleteListIterator. func NewDeleteListIterator(svc s3iface.S3API, input s3.ListObjectsInput, opts ...func(DeleteListIterator)) BatchDeleteIterator { iter := &DeleteListIterator{ Bucket: input.Bucket, Paginator: request.Pagination{ NewRequest: func() (request.Request, error) { var inCpy s3.ListObjectsInput if input != nil { tmp := input inCpy = &tmp } req, _ := svc.ListObjectsRequest(inCpy) return req, nil }, }, } for _, opt := range opts { opt(iter) } return iter } // Next will use the S3API client to iterate through a list of objects. func (iter DeleteListIterator) Next() bool { if len(iter.objects) > 0 { iter.objects = iter.objects[1:] } if len(iter.objects) == 0 && iter.Paginator.Next() { iter.objects = iter.Paginator.Page().(s3.ListObjectsOutput).Contents } return len(iter.objects) > 0 } // Err will return the last known error from Next. func (iter DeleteListIterator) Err() error { return iter.Paginator.Err() } // DeleteObject will return the current object to be deleted. func (iter DeleteListIterator) DeleteObject() BatchDeleteObject { return BatchDeleteObject{ Object: &s3.DeleteObjectInput{ Bucket: iter.Bucket, Key: iter.objects[0].Key, }, } } // BatchDelete will use the s3 package's service client to perform a batch // delete. type BatchDelete struct { Client s3iface.S3API BatchSize int } // NewBatchDeleteWithClient will return a new delete client that can delete a batched amount of // objects. // // Example: // // batcher := s3manager.NewBatchDeleteWithClient(client, size) // // objects := []BatchDeleteObject{ // { // Object: &s3.DeleteObjectInput { // Key: aws.String("key"), // Bucket: aws.String("bucket"), // }, // }, // } // // if err := batcher.Delete(aws.BackgroundContext(), &s3manager.DeleteObjectsIterator{ // Objects: objects, // }); err != nil { // return err // } func NewBatchDeleteWithClient(client s3iface.S3API, options ...func(BatchDelete)) BatchDelete { svc := &BatchDelete{ Client: client, BatchSize: DefaultBatchSize, } for _, opt := range options { opt(svc) } return svc } // NewBatchDelete will return a new delete client that can delete a batched amount of // objects. // // Example: // // batcher := s3manager.NewBatchDelete(sess, size) // // objects := []BatchDeleteObject{ // { // Object: &s3.DeleteObjectInput { // Key: aws.String("key"), // Bucket: aws.String("bucket"), // }, // }, // } // // if err := batcher.Delete(aws.BackgroundContext(), &s3manager.DeleteObjectsIterator{ // Objects: objects, // }); err != nil { // return err // } func NewBatchDelete(c client.ConfigProvider, options ...func(BatchDelete)) BatchDelete { client := s3.New(c) return NewBatchDeleteWithClient(client, options...) } // BatchDeleteObject is a wrapper object for calling the batch delete operation. type BatchDeleteObject struct { Object s3.DeleteObjectInput // After will run after each iteration during the batch process. This function will // be executed whether or not the request was successful. After func() error } // DeleteObjectsIterator is an interface that uses the scanner pattern to iterate // through a series of objects to be deleted. type DeleteObjectsIterator struct { Objects []BatchDeleteObject index int inc bool } // Next will increment the default iterator's index and ensure that there // is another object to iterator to. func (iter DeleteObjectsIterator) Next() bool { if iter.inc { iter.index++ } else { iter.inc = true } return iter.index < len(iter.Objects) } // Err will return an error. Since this is just used to satisfy the BatchDeleteIterator interface // this will only return nil. func (iter DeleteObjectsIterator) Err() error { return nil } // DeleteObject will return the BatchDeleteObject at the current batched index. func (iter DeleteObjectsIterator) DeleteObject() BatchDeleteObject { object := iter.Objects[iter.index] return object } // Delete will use the iterator to queue up objects that need to be deleted. // Once the batch size is met, this will call the deleteBatch function. func (d BatchDelete) Delete(ctx aws.Context, iter BatchDeleteIterator) error { var errs []Error objects := []BatchDeleteObject{} var input s3.DeleteObjectsInput for iter.Next() { o := iter.DeleteObject() if input == nil { input = initDeleteObjectsInput(o.Object) } parity := hasParity(input, o) if parity { input.Delete.Objects = append(input.Delete.Objects, &s3.ObjectIdentifier{ Key: o.Object.Key, VersionId: o.Object.VersionId, }) objects = append(objects, o) } if len(input.Delete.Objects) == d.BatchSize \|\| !parity { if err := deleteBatch(ctx, d, input, objects); err != nil { errs = append(errs, err...) } objects = objects[:0] input = nil if !parity { objects = append(objects, o) input = initDeleteObjectsInput(o.Object) input.Delete.Objects = append(input.Delete.Objects, &s3.ObjectIdentifier{ Key: o.Object.Key, VersionId: o.Object.VersionId, }) } } } // iter.Next() could return false (above) plus populate iter.Err() if iter.Err() != nil { errs = append(errs, newError(iter.Err(), nil, nil)) } if input != nil && len(input.Delete.Objects) > 0 { if err := deleteBatch(ctx, d, input, objects); err != nil { errs = append(errs, err...) } } if len(errs) > 0 { return NewBatchError("BatchedDeleteIncomplete", "some objects have failed to be deleted.", errs) } return nil } func initDeleteObjectsInput(o s3.DeleteObjectInput) s3.DeleteObjectsInput { return &s3.DeleteObjectsInput{ Bucket: o.Bucket, MFA: o.MFA, RequestPayer: o.RequestPayer, Delete: &s3.Delete{}, } } const ( // ErrDeleteBatchFailCode represents an error code which will be returned // only when DeleteObjects.Errors has an error that does not contain a code. ErrDeleteBatchFailCode = "DeleteBatchError" errDefaultDeleteBatchMessage = "failed to delete" ) // deleteBatch will delete a batch of items in the objects parameters. func deleteBatch(ctx aws.Context, d BatchDelete, input s3.DeleteObjectsInput, objects []BatchDeleteObject) []Error { errs := []Error{} if result, err := d.Client.DeleteObjectsWithContext(ctx, input); err != nil { for i := 0; i < len(input.Delete.Objects); i++ { errs = append(errs, newError(err, input.Bucket, input.Delete.Objects[i].Key)) } } else if len(result.Errors) > 0 { for i := 0; i < len(result.Errors); i++ { code := ErrDeleteBatchFailCode msg := errDefaultDeleteBatchMessage if result.Errors[i].Message != nil { msg = result.Errors[i].Message } if result.Errors[i].Code != nil { code = result.Errors[i].Code } errs = append(errs, newError(awserr.New(code, msg, err), input.Bucket, result.Errors[i].Key)) } } for _, object := range objects { if object.After == nil { continue } if err := object.After(); err != nil { errs = append(errs, newError(err, object.Object.Bucket, object.Object.Key)) } } return errs } func hasParity(o1 s3.DeleteObjectsInput, o2 BatchDeleteObject) bool { if o1.Bucket != nil && o2.Object.Bucket != nil { if o1.Bucket != o2.Object.Bucket { return false } } else if o1.Bucket != o2.Object.Bucket { return false } if o1.MFA != nil && o2.Object.MFA != nil { if o1.MFA != o2.Object.MFA { return false } } else if o1.MFA != o2.Object.MFA { return false } if o1.RequestPayer != nil && o2.Object.RequestPayer != nil { if o1.RequestPayer != o2.Object.RequestPayer { return false } } else if o1.RequestPayer != o2.Object.RequestPayer { return false } return true } // BatchDownloadIterator is an interface that uses the scanner pattern to iterate // through a series of objects to be downloaded. type BatchDownloadIterator interface { Next() bool Err() error DownloadObject() BatchDownloadObject } // BatchDownloadObject contains all necessary information to run a batch operation once. type BatchDownloadObject struct { Object s3.GetObjectInput Writer io.WriterAt // After will run after each iteration during the batch process. This function will // be executed whether or not the request was successful. After func() error } // DownloadObjectsIterator implements the BatchDownloadIterator interface and allows for batched // download of objects. type DownloadObjectsIterator struct { Objects []BatchDownloadObject index int inc bool } // Next will increment the default iterator's index and ensure that there // is another object to iterator to. func (batcher DownloadObjectsIterator) Next() bool { if batcher.inc { batcher.index++ } else { batcher.inc = true } return batcher.index < len(batcher.Objects) } // DownloadObject will return the BatchDownloadObject at the current batched index. func (batcher DownloadObjectsIterator) DownloadObject() BatchDownloadObject { object := batcher.Objects[batcher.index] return object } // Err will return an error. Since this is just used to satisfy the BatchDeleteIterator interface // this will only return nil. func (batcher DownloadObjectsIterator) Err() error { return nil } // BatchUploadIterator is an interface that uses the scanner pattern to // iterate through what needs to be uploaded. type BatchUploadIterator interface { Next() bool Err() error UploadObject() BatchUploadObject } // UploadObjectsIterator implements the BatchUploadIterator interface and allows for batched // upload of objects. type UploadObjectsIterator struct { Objects []BatchUploadObject index int inc bool } // Next will increment the default iterator's index and ensure that there // is another object to iterator to. func (batcher UploadObjectsIterator) Next() bool { if batcher.inc { batcher.index++ } else { batcher.inc = true } return batcher.index < len(batcher.Objects) } // Err will return an error. Since this is just used to satisfy the BatchUploadIterator interface // this will only return nil. func (batcher UploadObjectsIterator) Err() error { return nil } // UploadObject will return the BatchUploadObject at the current batched index. func (batcher UploadObjectsIterator) UploadObject() BatchUploadObject { object := batcher.Objects[batcher.index] return object } // BatchUploadObject contains all necessary information to run a batch operation once. type BatchUploadObject struct { Object *UploadInput // After will run after each iteration during the batch process. This function will // be executed whether or not the request was successful. After func() error }	13.81 KB	2025-05-25 12:27
📄 batch_1_7_test.go //go:build go1.7 // +build go1.7 package s3manager import ( "context" "fmt" "net/http" "net/http/httptest" "testing" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/service/s3" ) // #1790 bug func TestBatchDeleteContext(t testing.T) { cases := []struct { objects []BatchDeleteObject batchSize int expected int earlyCancel bool checkError func(error) error }{ 0: { objects: []BatchDeleteObject{ { Object: &s3.DeleteObjectInput{ Key: aws.String("1"), Bucket: aws.String("bucket1"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("2"), Bucket: aws.String("bucket2"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("3"), Bucket: aws.String("bucket3"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("4"), Bucket: aws.String("bucket4"), }, }, }, batchSize: 1, expected: 0, earlyCancel: true, checkError: func(err error) error { batchErr, ok := err.(BatchError) if !ok { return fmt.Errorf("expect BatchError, got %T, %v", err, err) } errs := batchErr.Errors if len(errs) != 4 { return fmt.Errorf("expected 1 batch errors, but received %d", len(errs)) } for _, tempErr := range errs { aerr, ok := tempErr.OrigErr.(awserr.Error) if !ok { return fmt.Errorf("expect awserr.Error, got %T, %v", tempErr.OrigErr, tempErr.OrigErr) } if e, a := request.CanceledErrorCode, aerr.Code(); e != a { return fmt.Errorf("expect %q, error code, got %q", e, a) } } return nil }, }, 1: { objects: []BatchDeleteObject{ { Object: &s3.DeleteObjectInput{ Key: aws.String("1"), Bucket: aws.String("bucket1"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("2"), Bucket: aws.String("bucket2"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("3"), Bucket: aws.String("bucket3"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("4"), Bucket: aws.String("bucket4"), }, }, }, batchSize: 1, expected: 4, checkError: func(err error) error { if err != nil { return fmt.Errorf("Expect no error, got %v", err) } return nil }, }, } count := 0 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) count++ })) defer server.Close() svc := &mockS3Client{S3: buildS3SvcClient(server.URL)} for i, c := range cases { ctx, cancel := context.WithCancel(context.Background()) defer cancel() if c.earlyCancel { cancel() } batcher := BatchDelete{ Client: svc, BatchSize: c.batchSize, } err := batcher.Delete(ctx, &DeleteObjectsIterator{Objects: c.objects}) if terr := c.checkError(err); terr != nil { t.Fatalf("%d, %s", i, terr) } if count != c.expected { t.Errorf("Case %d: expected %d, but received %d", i, c.expected, count) } count = 0 } }	3.16 KB	2025-05-25 12:27
📄 batch_test.go package s3manager import ( "bytes" "errors" "io/ioutil" "net/http" "net/http/httptest" "strings" "testing" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/aws/credentials" "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/awstesting/unit" "github.com/aws/aws-sdk-go/service/s3" "github.com/aws/aws-sdk-go/service/s3/s3iface" ) func TestHasParity(t testing.T) { cases := []struct { o1 s3.DeleteObjectsInput o2 BatchDeleteObject expected bool }{ { &s3.DeleteObjectsInput{}, BatchDeleteObject{ Object: &s3.DeleteObjectInput{}, }, true, }, { &s3.DeleteObjectsInput{ Bucket: aws.String("foo"), }, BatchDeleteObject{ Object: &s3.DeleteObjectInput{ Bucket: aws.String("bar"), }, }, false, }, { &s3.DeleteObjectsInput{}, BatchDeleteObject{ Object: &s3.DeleteObjectInput{ Bucket: aws.String("foo"), }, }, false, }, { &s3.DeleteObjectsInput{ Bucket: aws.String("foo"), }, BatchDeleteObject{ Object: &s3.DeleteObjectInput{}, }, false, }, { &s3.DeleteObjectsInput{ MFA: aws.String("foo"), }, BatchDeleteObject{ Object: &s3.DeleteObjectInput{ MFA: aws.String("bar"), }, }, false, }, { &s3.DeleteObjectsInput{}, BatchDeleteObject{ Object: &s3.DeleteObjectInput{ MFA: aws.String("foo"), }, }, false, }, { &s3.DeleteObjectsInput{ MFA: aws.String("foo"), }, BatchDeleteObject{ Object: &s3.DeleteObjectInput{}, }, false, }, { &s3.DeleteObjectsInput{ RequestPayer: aws.String("foo"), }, BatchDeleteObject{ Object: &s3.DeleteObjectInput{ RequestPayer: aws.String("bar"), }, }, false, }, { &s3.DeleteObjectsInput{}, BatchDeleteObject{ Object: &s3.DeleteObjectInput{ RequestPayer: aws.String("foo"), }, }, false, }, { &s3.DeleteObjectsInput{ RequestPayer: aws.String("foo"), }, BatchDeleteObject{ Object: &s3.DeleteObjectInput{}, }, false, }, } for i, c := range cases { if result := hasParity(c.o1, c.o2); result != c.expected { t.Errorf("Case %d: expected %t, but received %t\n", i, c.expected, result) } } } func TestBatchDelete(t testing.T) { cases := []struct { objects []BatchDeleteObject size int expected int }{ { []BatchDeleteObject{ { Object: &s3.DeleteObjectInput{ Key: aws.String("1"), Bucket: aws.String("bucket1"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("2"), Bucket: aws.String("bucket2"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("3"), Bucket: aws.String("bucket3"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("4"), Bucket: aws.String("bucket4"), }, }, }, 1, 4, }, { []BatchDeleteObject{ { Object: &s3.DeleteObjectInput{ Key: aws.String("1"), Bucket: aws.String("bucket1"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("2"), Bucket: aws.String("bucket1"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("3"), Bucket: aws.String("bucket3"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("4"), Bucket: aws.String("bucket3"), }, }, }, 1, 4, }, { []BatchDeleteObject{ { Object: &s3.DeleteObjectInput{ Key: aws.String("1"), Bucket: aws.String("bucket1"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("2"), Bucket: aws.String("bucket1"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("3"), Bucket: aws.String("bucket3"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("4"), Bucket: aws.String("bucket3"), }, }, }, 4, 2, }, { []BatchDeleteObject{ { Object: &s3.DeleteObjectInput{ Key: aws.String("1"), Bucket: aws.String("bucket1"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("2"), Bucket: aws.String("bucket1"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("3"), Bucket: aws.String("bucket3"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("4"), Bucket: aws.String("bucket3"), }, }, }, 10, 2, }, { []BatchDeleteObject{ { Object: &s3.DeleteObjectInput{ Key: aws.String("1"), Bucket: aws.String("bucket1"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("2"), Bucket: aws.String("bucket1"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("3"), Bucket: aws.String("bucket1"), }, }, { Object: &s3.DeleteObjectInput{ Key: aws.String("4"), Bucket: aws.String("bucket3"), }, }, }, 2, 3, }, } count := 0 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r http.Request) { w.WriteHeader(http.StatusNoContent) count++ })) defer server.Close() svc := &mockS3Client{S3: buildS3SvcClient(server.URL)} for i, c := range cases { batcher := BatchDelete{ Client: svc, BatchSize: c.size, } if err := batcher.Delete(aws.BackgroundContext(), &DeleteObjectsIterator{Objects: c.objects}); err != nil { panic(err) } if count != c.expected { t.Errorf("Case %d: expected %d, but received %d", i, c.expected, count) } count = 0 } } func TestBatchDeleteError(t testing.T) { cases := []struct { objects []BatchDeleteObject output s3.DeleteObjectsOutput size int expectedErrCode string expectedErrMessage string }{ { []BatchDeleteObject{ { Object: &s3.DeleteObjectInput{ Key: aws.String("1"), Bucket: aws.String("bucket1"), }, }, }, s3.DeleteObjectsOutput{ Errors: []s3.Error{ { Code: aws.String("foo code"), Message: aws.String("foo error"), }, }, }, 1, "foo code", "foo error", }, { []BatchDeleteObject{ { Object: &s3.DeleteObjectInput{ Key: aws.String("1"), Bucket: aws.String("bucket1"), }, }, }, s3.DeleteObjectsOutput{ Errors: []s3.Error{ {}, }, }, 1, ErrDeleteBatchFailCode, errDefaultDeleteBatchMessage, }, } server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r http.Request) { w.WriteHeader(http.StatusNoContent) })) defer server.Close() index := 0 svc := &mockS3Client{ S3: buildS3SvcClient(server.URL), deleteObjects: func(input s3.DeleteObjectsInput) (s3.DeleteObjectsOutput, error) { output := &cases[index].output index++ return output, nil }, } for _, c := range cases { batcher := BatchDelete{ Client: svc, BatchSize: c.size, } err := batcher.Delete(aws.BackgroundContext(), &DeleteObjectsIterator{Objects: c.objects}) if err == nil { t.Errorf("expected error, but received none") } berr := err.(BatchError) if len(berr.Errors) != 1 { t.Errorf("expected 1 error, but received %d", len(berr.Errors)) } aerr := berr.Errors[0].OrigErr.(awserr.Error) if e, a := c.expectedErrCode, aerr.Code(); e != a { t.Errorf("expected %q, but received %q", e, a) } if e, a := c.expectedErrMessage, aerr.Message(); e != a { t.Errorf("expected %q, but received %q", e, a) } } } type mockS3Client struct { s3.S3 index int objects []s3.ListObjectsOutput deleteObjects func(s3.DeleteObjectsInput) (s3.DeleteObjectsOutput, error) } func (client mockS3Client) ListObjects(input s3.ListObjectsInput) (s3.ListObjectsOutput, error) { object := client.objects[client.index] client.index++ return object, nil } func (client mockS3Client) DeleteObjects(input s3.DeleteObjectsInput) (s3.DeleteObjectsOutput, error) { if client.deleteObjects == nil { return client.S3.DeleteObjectsWithContext(aws.BackgroundContext(), input) } return client.deleteObjects(input) } func (client mockS3Client) DeleteObjectsWithContext(ctx aws.Context, input s3.DeleteObjectsInput, opt ...request.Option) (s3.DeleteObjectsOutput, error) { if client.deleteObjects == nil { return client.S3.DeleteObjectsWithContext(ctx, input) } return client.deleteObjects(input) } func TestNilOrigError(t testing.T) { err := Error{ Bucket: aws.String("bucket"), Key: aws.String("key"), } errStr := err.Error() const expected1 = `failed to perform batch operation on "key" to "bucket"` if errStr != expected1 { t.Errorf("Expected %s, but received %s", expected1, errStr) } err = Error{ OrigErr: errors.New("foo"), Bucket: aws.String("bucket"), Key: aws.String("key"), } errStr = err.Error() const expected2 = "failed to perform batch operation on \"key\" to \"bucket\":\nfoo" if errStr != expected2 { t.Errorf("Expected %s, but received %s", expected2, errStr) } } func TestBatchDeleteList(t testing.T) { count := 0 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r http.Request) { w.WriteHeader(http.StatusNoContent) count++ })) defer server.Close() objects := []s3.ListObjectsOutput{ { Contents: []s3.Object{ { Key: aws.String("1"), }, }, NextMarker: aws.String("marker"), IsTruncated: aws.Bool(true), }, { Contents: []s3.Object{ { Key: aws.String("2"), }, }, NextMarker: aws.String("marker"), IsTruncated: aws.Bool(true), }, { Contents: []s3.Object{ { Key: aws.String("3"), }, }, IsTruncated: aws.Bool(false), }, } svc := &mockS3Client{S3: buildS3SvcClient(server.URL), objects: objects} batcher := BatchDelete{ Client: svc, BatchSize: 1, } input := &s3.ListObjectsInput{ Bucket: aws.String("bucket"), } iter := &DeleteListIterator{ Bucket: input.Bucket, Paginator: request.Pagination{ NewRequest: func() (request.Request, error) { var inCpy s3.ListObjectsInput if input != nil { tmp := input inCpy = &tmp } req, _ := svc.ListObjectsRequest(inCpy) req.Handlers.Clear() output, _ := svc.ListObjects(inCpy) req.Data = output return req, nil }, }, } if err := batcher.Delete(aws.BackgroundContext(), iter); err != nil { t.Error(err) } if count != len(objects) { t.Errorf("Expected %d, but received %d", len(objects), count) } } func buildS3SvcClient(u string) s3.S3 { return s3.New(unit.Session, &aws.Config{ Endpoint: aws.String(u), S3ForcePathStyle: aws.Bool(true), DisableSSL: aws.Bool(true), Credentials: credentials.NewStaticCredentials("AKID", "SECRET", "SESSION"), }) } func TestBatchDeleteList_EmptyListObjects(t testing.T) { count := 0 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r http.Request) { w.WriteHeader(http.StatusNoContent) count++ })) defer server.Close() svc := &mockS3Client{S3: buildS3SvcClient(server.URL)} batcher := BatchDelete{ Client: svc, } input := &s3.ListObjectsInput{ Bucket: aws.String("bucket"), } // Test DeleteListIterator in the case when the ListObjectsRequest responds // with an empty listing. // We need a new iterator with a fresh Pagination since // Pagination.HasNextPage() is always true the first time Pagination.Next() // called on it iter := &DeleteListIterator{ Bucket: input.Bucket, Paginator: request.Pagination{ NewRequest: func() (request.Request, error) { req, _ := svc.ListObjectsRequest(input) // Simulate empty listing req.Data = &s3.ListObjectsOutput{Contents: []s3.Object{}} return req, nil }, }, } if err := batcher.Delete(aws.BackgroundContext(), iter); err != nil { t.Error(err) } if count != 1 { t.Errorf("expect count to be 1, got %d", count) } } func TestBatchDownload(t testing.T) { count := 0 expected := []struct { bucket, key string }{ { key: "1", bucket: "bucket1", }, { key: "2", bucket: "bucket2", }, { key: "3", bucket: "bucket3", }, { key: "4", bucket: "bucket4", }, } received := []struct { bucket, key string }{} payload := []string{ "1", "2", "3", "4", } server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r http.Request) { urlParts := strings.Split(r.URL.String(), "/") received = append(received, struct{ bucket, key string }{urlParts[1], urlParts[2]}) w.Write([]byte(payload[count])) count++ })) defer server.Close() svc := NewDownloaderWithClient(buildS3SvcClient(server.URL)) objects := []BatchDownloadObject{ { Object: &s3.GetObjectInput{ Key: aws.String("1"), Bucket: aws.String("bucket1"), }, Writer: aws.NewWriteAtBuffer(make([]byte, 128)), }, { Object: &s3.GetObjectInput{ Key: aws.String("2"), Bucket: aws.String("bucket2"), }, Writer: aws.NewWriteAtBuffer(make([]byte, 128)), }, { Object: &s3.GetObjectInput{ Key: aws.String("3"), Bucket: aws.String("bucket3"), }, Writer: aws.NewWriteAtBuffer(make([]byte, 128)), }, { Object: &s3.GetObjectInput{ Key: aws.String("4"), Bucket: aws.String("bucket4"), }, Writer: aws.NewWriteAtBuffer(make([]byte, 128)), }, } iter := &DownloadObjectsIterator{Objects: objects} if err := svc.DownloadWithIterator(aws.BackgroundContext(), iter); err != nil { panic(err) } if count != len(objects) { t.Errorf("Expected %d, but received %d", len(objects), count) } if len(expected) != len(received) { t.Errorf("Expected %d, but received %d", len(expected), len(received)) } for i := 0; i < len(expected); i++ { if expected[i].key != received[i].key { t.Errorf("Expected %q, but received %q", expected[i].key, received[i].key) } if expected[i].bucket != received[i].bucket { t.Errorf("Expected %q, but received %q", expected[i].bucket, received[i].bucket) } } for i, p := range payload { b := iter.Objects[i].Writer.(aws.WriteAtBuffer).Bytes() b = bytes.Trim(b, "\x00") if string(b) != p { t.Errorf("Expected %q, but received %q", p, b) } } } func TestBatchUpload(t testing.T) { count := 0 expected := []struct { bucket, key string reqBody string }{ { key: "1", bucket: "bucket1", reqBody: "1", }, { key: "2", bucket: "bucket2", reqBody: "2", }, { key: "3", bucket: "bucket3", reqBody: "3", }, { key: "4", bucket: "bucket4", reqBody: "4", }, } received := []struct { bucket, key, reqBody string }{} payload := []string{ "a", "b", "c", "d", } server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r http.Request) { urlParts := strings.Split(r.URL.String(), "/") b, err := ioutil.ReadAll(r.Body) if err != nil { t.Error(err) } received = append(received, struct{ bucket, key, reqBody string }{urlParts[1], urlParts[2], string(b)}) w.Write([]byte(payload[count])) count++ })) defer server.Close() svc := NewUploaderWithClient(buildS3SvcClient(server.URL)) objects := []BatchUploadObject{ { Object: &UploadInput{ Key: aws.String("1"), Bucket: aws.String("bucket1"), Body: bytes.NewBuffer([]byte("1")), }, }, { Object: &UploadInput{ Key: aws.String("2"), Bucket: aws.String("bucket2"), Body: bytes.NewBuffer([]byte("2")), }, }, { Object: &UploadInput{ Key: aws.String("3"), Bucket: aws.String("bucket3"), Body: bytes.NewBuffer([]byte("3")), }, }, { Object: &UploadInput{ Key: aws.String("4"), Bucket: aws.String("bucket4"), Body: bytes.NewBuffer([]byte("4")), }, }, } iter := &UploadObjectsIterator{Objects: objects} if err := svc.UploadWithIterator(aws.BackgroundContext(), iter); err != nil { panic(err) } if count != len(objects) { t.Errorf("Expected %d, but received %d", len(objects), count) } if len(expected) != len(received) { t.Errorf("Expected %d, but received %d", len(expected), len(received)) } for i := 0; i < len(expected); i++ { if expected[i].key != received[i].key { t.Errorf("Expected %q, but received %q", expected[i].key, received[i].key) } if expected[i].bucket != received[i].bucket { t.Errorf("Expected %q, but received %q", expected[i].bucket, received[i].bucket) } if expected[i].reqBody != received[i].reqBody { t.Errorf("Expected %q, but received %q", expected[i].reqBody, received[i].reqBody) } } } type mockClient struct { s3iface.S3API Put func() (s3.PutObjectOutput, error) Get func() (s3.GetObjectOutput, error) List func() (s3.ListObjectsOutput, error) responses []response } type response struct { out interface{} err error } func (client mockClient) PutObject(input s3.PutObjectInput) (s3.PutObjectOutput, error) { return client.Put() } func (client mockClient) PutObjectRequest(input s3.PutObjectInput) (request.Request, s3.PutObjectOutput) { req, _ := client.S3API.PutObjectRequest(input) req.Handlers.Clear() req.Data, req.Error = client.Put() return req, req.Data.(s3.PutObjectOutput) } func (client mockClient) ListObjects(input s3.ListObjectsInput) (s3.ListObjectsOutput, error) { return client.List() } func (client mockClient) ListObjectsRequest(input s3.ListObjectsInput) (request.Request, s3.ListObjectsOutput) { req, _ := client.S3API.ListObjectsRequest(input) req.Handlers.Clear() req.Data, req.Error = client.List() return req, req.Data.(s3.ListObjectsOutput) } func TestBatchError(t testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r http.Request) { })) defer server.Close() index := 0 responses := []response{ { &s3.PutObjectOutput{}, errors.New("Foo"), }, { &s3.PutObjectOutput{}, nil, }, { &s3.PutObjectOutput{}, nil, }, { &s3.PutObjectOutput{}, errors.New("Bar"), }, } svc := &mockClient{ S3API: buildS3SvcClient(server.URL), Put: func() (s3.PutObjectOutput, error) { resp := responses[index] index++ return resp.out.(s3.PutObjectOutput), resp.err }, List: func() (s3.ListObjectsOutput, error) { resp := responses[index] index++ return resp.out.(s3.ListObjectsOutput), resp.err }, } uploader := NewUploaderWithClient(svc) objects := []BatchUploadObject{ { Object: &UploadInput{ Key: aws.String("1"), Bucket: aws.String("bucket1"), Body: bytes.NewBuffer([]byte("1")), }, }, { Object: &UploadInput{ Key: aws.String("2"), Bucket: aws.String("bucket2"), Body: bytes.NewBuffer([]byte("2")), }, }, { Object: &UploadInput{ Key: aws.String("3"), Bucket: aws.String("bucket3"), Body: bytes.NewBuffer([]byte("3")), }, }, { Object: &UploadInput{ Key: aws.String("4"), Bucket: aws.String("bucket4"), Body: bytes.NewBuffer([]byte("4")), }, }, } iter := &UploadObjectsIterator{Objects: objects} if err := uploader.UploadWithIterator(aws.BackgroundContext(), iter); err != nil { if bErr, ok := err.(BatchError); !ok { t.Error("Expected BatchError, but received other") } else { if len(bErr.Errors) != 2 { t.Errorf("Expected 2 errors, but received %d", len(bErr.Errors)) } expected := []struct { bucket, key string }{ { "bucket1", "1", }, { "bucket4", "4", }, } for i, expect := range expected { if bErr.Errors[i].Bucket != expect.bucket { t.Errorf("Case %d: Invalid bucket expected %s, but received %s", i, expect.bucket, bErr.Errors[i].Bucket) } if bErr.Errors[i].Key != expect.key { t.Errorf("Case %d: Invalid key expected %s, but received %s", i, expect.key, bErr.Errors[i].Key) } } } } else { t.Error("Expected error, but received nil") } if index != len(objects) { t.Errorf("Expected %d, but received %d", len(objects), index) } } type testAfterDeleteIter struct { afterDelete bool afterDownload bool afterUpload bool next bool } func (iter testAfterDeleteIter) Next() bool { next := !iter.next iter.next = !iter.next return next } func (iter testAfterDeleteIter) Err() error { return nil } func (iter testAfterDeleteIter) DeleteObject() BatchDeleteObject { return BatchDeleteObject{ Object: &s3.DeleteObjectInput{ Bucket: aws.String("foo"), Key: aws.String("foo"), }, After: func() error { iter.afterDelete = true return nil }, } } type testAfterDownloadIter struct { afterDownload bool afterUpload bool next bool } func (iter testAfterDownloadIter) Next() bool { next := !iter.next iter.next = !iter.next return next } func (iter testAfterDownloadIter) Err() error { return nil } func (iter testAfterDownloadIter) DownloadObject() BatchDownloadObject { return BatchDownloadObject{ Object: &s3.GetObjectInput{ Bucket: aws.String("foo"), Key: aws.String("foo"), }, Writer: aws.NewWriteAtBuffer([]byte{}), After: func() error { iter.afterDownload = true return nil }, } } type testAfterUploadIter struct { afterUpload bool next bool } func (iter testAfterUploadIter) Next() bool { next := !iter.next iter.next = !iter.next return next } func (iter testAfterUploadIter) Err() error { return nil } func (iter testAfterUploadIter) UploadObject() BatchUploadObject { return BatchUploadObject{ Object: &UploadInput{ Bucket: aws.String("foo"), Key: aws.String("foo"), Body: strings.NewReader("bar"), }, After: func() error { iter.afterUpload = true return nil }, } } func TestAfter(t testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r http.Request) { })) defer server.Close() index := 0 responses := []response{ { &s3.PutObjectOutput{}, nil, }, { &s3.GetObjectOutput{}, nil, }, { &s3.DeleteObjectOutput{}, nil, }, } svc := &mockClient{ S3API: buildS3SvcClient(server.URL), Put: func() (s3.PutObjectOutput, error) { resp := responses[index] index++ return resp.out.(s3.PutObjectOutput), resp.err }, Get: func() (s3.GetObjectOutput, error) { resp := responses[index] index++ return resp.out.(s3.GetObjectOutput), resp.err }, List: func() (s3.ListObjectsOutput, error) { resp := responses[index] index++ return resp.out.(*s3.ListObjectsOutput), resp.err }, } uploader := NewUploaderWithClient(svc) downloader := NewDownloaderWithClient(svc) deleter := NewBatchDeleteWithClient(svc) deleteIter := &testAfterDeleteIter{} downloadIter := &testAfterDownloadIter{} uploadIter := &testAfterUploadIter{} if err := uploader.UploadWithIterator(aws.BackgroundContext(), uploadIter); err != nil { t.Error(err) } if err := downloader.DownloadWithIterator(aws.BackgroundContext(), downloadIter); err != nil { t.Error(err) } if err := deleter.Delete(aws.BackgroundContext(), deleteIter); err != nil { t.Error(err) } if !deleteIter.afterDelete { t.Error("Expected 'afterDelete' to be true, but received false") } if !downloadIter.afterDownload { t.Error("Expected 'afterDownload' to be true, but received false") } if !uploadIter.afterUpload { t.Error("Expected 'afterUpload' to be true, but received false") } }	23.14 KB	2025-05-25 12:27
📄 bucket_region.go package s3manager import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/client" "github.com/aws/aws-sdk-go/aws/corehandlers" "github.com/aws/aws-sdk-go/aws/credentials" "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/service/s3" "github.com/aws/aws-sdk-go/service/s3/s3iface" ) // GetBucketRegion will attempt to get the region for a bucket using the // regionHint to determine which AWS partition to perform the query on. // // The request will not be signed, and will not use your AWS credentials. // // A "NotFound" error code will be returned if the bucket does not exist in the // AWS partition the regionHint belongs to. If the regionHint parameter is an // empty string GetBucketRegion will fallback to the ConfigProvider's region // config. If the regionHint is empty, and the ConfigProvider does not have a // region value, an error will be returned.. // // For example to get the region of a bucket which exists in "eu-central-1" // you could provide a region hint of "us-west-2". // // sess := session.Must(session.NewSession()) // // bucket := "my-bucket" // region, err := s3manager.GetBucketRegion(ctx, sess, bucket, "us-west-2") // if err != nil { // if aerr, ok := err.(awserr.Error); ok && aerr.Code() == "NotFound" { // fmt.Fprintf(os.Stderr, "unable to find bucket %s's region not found\n", bucket) // } // return err // } // fmt.Printf("Bucket %s is in %s region\n", bucket, region) // // By default the request will be made to the Amazon S3 endpoint using the Path // style addressing. // // s3.us-west-2.amazonaws.com/bucketname // // This is not compatible with Amazon S3's FIPS endpoints. To override this // behavior to use Virtual Host style addressing, provide a functional option // that will set the Request's Config.S3ForcePathStyle to aws.Bool(false). // // region, err := s3manager.GetBucketRegion(ctx, sess, "bucketname", "us-west-2", func(r request.Request) { // r.S3ForcePathStyle = aws.Bool(false) // }) // // To configure the GetBucketRegion to make a request via the Amazon // S3 FIPS endpoints directly when a FIPS region name is not available, (e.g. // fips-us-gov-west-1) set the Config.Endpoint on the Session, or client the // utility is called with. The hint region will be ignored if an endpoint URL // is configured on the session or client. // // sess, err := session.NewSession(&aws.Config{ // Endpoint: aws.String("https://s3-fips.us-west-2.amazonaws.com"), // }) // // region, err := s3manager.GetBucketRegion(context.Background(), sess, "bucketname", "") func GetBucketRegion(ctx aws.Context, c client.ConfigProvider, bucket, regionHint string, opts ...request.Option) (string, error) { var cfg aws.Config if len(regionHint) != 0 { cfg.Region = aws.String(regionHint) } svc := s3.New(c, &cfg) return GetBucketRegionWithClient(ctx, svc, bucket, opts...) } const bucketRegionHeader = "X-Amz-Bucket-Region" // GetBucketRegionWithClient is the same as GetBucketRegion with the exception // that it takes a S3 service client instead of a Session. The regionHint is // derived from the region the S3 service client was created in. // // By default the request will be made to the Amazon S3 endpoint using the Path // style addressing. // // s3.us-west-2.amazonaws.com/bucketname // // This is not compatible with Amazon S3's FIPS endpoints. To override this // behavior to use Virtual Host style addressing, provide a functional option // that will set the Request's Config.S3ForcePathStyle to aws.Bool(false). // // region, err := s3manager.GetBucketRegionWithClient(ctx, client, "bucketname", func(r request.Request) { // r.S3ForcePathStyle = aws.Bool(false) // }) // // To configure the GetBucketRegion to make a request via the Amazon // S3 FIPS endpoints directly when a FIPS region name is not available, (e.g. // fips-us-gov-west-1) set the Config.Endpoint on the Session, or client the // utility is called with. The hint region will be ignored if an endpoint URL // is configured on the session or client. // // region, err := s3manager.GetBucketRegionWithClient(context.Background(), // s3.New(sess, &aws.Config{ // Endpoint: aws.String("https://s3-fips.us-west-2.amazonaws.com"), // }), // "bucketname") // // See GetBucketRegion for more information. func GetBucketRegionWithClient(ctx aws.Context, svc s3iface.S3API, bucket string, opts ...request.Option) (string, error) { req, _ := svc.HeadBucketRequest(&s3.HeadBucketInput{ Bucket: aws.String(bucket), }) req.Config.S3ForcePathStyle = aws.Bool(true) req.Config.Credentials = credentials.AnonymousCredentials req.SetContext(ctx) // Disable HTTP redirects to prevent an invalid 301 from eating the response // because Go's HTTP client will fail, and drop the response if an 301 is // received without a location header. S3 will return a 301 without the // location header for HeadObject API calls. req.DisableFollowRedirects = true var bucketRegion string req.Handlers.Send.PushBack(func(r request.Request) { bucketRegion = r.HTTPResponse.Header.Get(bucketRegionHeader) if len(bucketRegion) == 0 { return } r.HTTPResponse.StatusCode = 200 r.HTTPResponse.Status = "OK" r.Error = nil }) // Replace the endpoint validation handler to not require a region if an // endpoint URL was specified. Since these requests are not authenticated, // requiring a region is not needed when an endpoint URL is provided. req.Handlers.Validate.Swap( corehandlers.ValidateEndpointHandler.Name, request.NamedHandler{ Name: "validateEndpointWithoutRegion", Fn: validateEndpointWithoutRegion, }, ) req.ApplyOptions(opts...) if err := req.Send(); err != nil { return "", err } bucketRegion = s3.NormalizeBucketLocation(bucketRegion) return bucketRegion, nil } func validateEndpointWithoutRegion(r request.Request) { // Check if the caller provided an explicit URL instead of one derived by // the SDK's endpoint resolver. For GetBucketRegion, with an explicit // endpoint URL, a region is not needed. If no endpoint URL is provided, // fallback the SDK's standard endpoint validation handler. if len(aws.StringValue(r.Config.Endpoint)) == 0 { corehandlers.ValidateEndpointHandler.Fn(r) } }	6.1 KB	2025-05-25 12:27
📄 bucket_region_test.go package s3manager import ( "net/http" "net/http/httptest" "testing" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/awstesting/unit" "github.com/aws/aws-sdk-go/service/s3" ) func testSetupGetBucketRegionServer(region string, statusCode int, incHeader bool) httptest.Server { return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r http.Request) { if incHeader { w.Header().Set(bucketRegionHeader, region) } w.WriteHeader(statusCode) })) } var testGetBucketRegionCases = []struct { RespRegion string StatusCode int HintRegion string ExpectReqRegion string }{ {"bucket-region", 301, "hint-region", ""}, {"bucket-region", 403, "hint-region", ""}, {"bucket-region", 200, "hint-region", ""}, {"bucket-region", 200, "", "default-region"}, } func TestGetBucketRegion_Exists(t testing.T) { for i, c := range testGetBucketRegionCases { server := testSetupGetBucketRegionServer(c.RespRegion, c.StatusCode, true) defer server.Close() sess := unit.Session.Copy() sess.Config.Region = aws.String("default-region") sess.Config.Endpoint = aws.String(server.URL) sess.Config.DisableSSL = aws.Bool(true) ctx := aws.BackgroundContext() region, err := GetBucketRegion(ctx, sess, "bucket", c.HintRegion) if err != nil { t.Fatalf("%d, expect no error, got %v", i, err) } if e, a := c.RespRegion, region; e != a { t.Errorf("%d, expect %q region, got %q", i, e, a) } } } func TestGetBucketRegion_NotExists(t testing.T) { server := testSetupGetBucketRegionServer("ignore-region", 404, false) defer server.Close() sess := unit.Session.Copy() sess.Config.Endpoint = aws.String(server.URL) sess.Config.DisableSSL = aws.Bool(true) ctx := aws.BackgroundContext() region, err := GetBucketRegion(ctx, sess, "bucket", "hint-region") if err == nil { t.Fatalf("expect error, but did not get one") } aerr := err.(awserr.Error) if e, a := "NotFound", aerr.Code(); e != a { t.Errorf("expect %s error code, got %s", e, a) } if len(region) != 0 { t.Errorf("expect region not to be set, got %q", region) } } func TestGetBucketRegionWithClient(t testing.T) { for i, c := range testGetBucketRegionCases { server := testSetupGetBucketRegionServer(c.RespRegion, c.StatusCode, true) defer server.Close() svc := s3.New(unit.Session, &aws.Config{ Region: aws.String("hint-region"), Endpoint: aws.String(server.URL), DisableSSL: aws.Bool(true), }) ctx := aws.BackgroundContext() region, err := GetBucketRegionWithClient(ctx, svc, "bucket") if err != nil { t.Fatalf("%d, expect no error, got %v", i, err) } if e, a := c.RespRegion, region; e != a { t.Errorf("%d, expect %q region, got %q", i, e, a) } } } func TestGetBucketRegionWithClientWithoutRegion(t testing.T) { for i, c := range testGetBucketRegionCases { server := testSetupGetBucketRegionServer(c.RespRegion, c.StatusCode, true) defer server.Close() svc := s3.New(unit.Session, &aws.Config{ Endpoint: aws.String(server.URL), DisableSSL: aws.Bool(true), }) ctx := aws.BackgroundContext() region, err := GetBucketRegionWithClient(ctx, svc, "bucket") if err != nil { t.Fatalf("%d, expect no error, got %v", i, err) } if e, a := c.RespRegion, region; e != a { t.Errorf("%d, expect %q region, got %q", i, e, a) } } }	3.31 KB	2025-05-25 12:27
📄 buffered_read_seeker.go package s3manager import ( "io" "github.com/aws/aws-sdk-go/internal/sdkio" ) // BufferedReadSeeker is buffered io.ReadSeeker type BufferedReadSeeker struct { r io.ReadSeeker buffer []byte readIdx, writeIdx int } // NewBufferedReadSeeker returns a new BufferedReadSeeker // if len(b) == 0 then the buffer will be initialized to 64 KiB. func NewBufferedReadSeeker(r io.ReadSeeker, b []byte) BufferedReadSeeker { if len(b) == 0 { b = make([]byte, 641024) } return &BufferedReadSeeker{r: r, buffer: b} } func (b BufferedReadSeeker) reset(r io.ReadSeeker) { b.r = r b.readIdx, b.writeIdx = 0, 0 } // Read will read up len(p) bytes into p and will return // the number of bytes read and any error that occurred. // If the len(p) > the buffer size then a single read request // will be issued to the underlying io.ReadSeeker for len(p) bytes. // A Read request will at most perform a single Read to the underlying // io.ReadSeeker, and may return < len(p) if serviced from the buffer. func (b BufferedReadSeeker) Read(p []byte) (n int, err error) { if len(p) == 0 { return n, err } if b.readIdx == b.writeIdx { if len(p) >= len(b.buffer) { n, err = b.r.Read(p) return n, err } b.readIdx, b.writeIdx = 0, 0 n, err = b.r.Read(b.buffer) if n == 0 { return n, err } b.writeIdx += n } n = copy(p, b.buffer[b.readIdx:b.writeIdx]) b.readIdx += n return n, err } // Seek will position then underlying io.ReadSeeker to the given offset // and will clear the buffer. func (b BufferedReadSeeker) Seek(offset int64, whence int) (int64, error) { n, err := b.r.Seek(offset, whence) b.reset(b.r) return n, err } // ReadAt will read up to len(p) bytes at the given file offset. // This will result in the buffer being cleared. func (b BufferedReadSeeker) ReadAt(p []byte, off int64) (int, error) { _, err := b.Seek(off, sdkio.SeekStart) if err != nil { return 0, err } return b.Read(p) }	1.92 KB	2025-05-25 12:27
📄 buffered_read_seeker_test.go package s3manager import ( "bytes" "io" "testing" "github.com/aws/aws-sdk-go/internal/sdkio" ) func TestBufferedReadSeekerRead(t testing.T) { expected := []byte("testData") readSeeker := NewBufferedReadSeeker(bytes.NewReader(expected), make([]byte, 4)) var ( actual []byte buffer = make([]byte, 2) ) for { n, err := readSeeker.Read(buffer) actual = append(actual, buffer[:n]...) if err != nil && err == io.EOF { break } else if err != nil { t.Fatalf("failed to read from reader: %v", err) } } if !bytes.Equal(expected, actual) { t.Errorf("expected %v, got %v", expected, actual) } } func TestBufferedReadSeekerSeek(t testing.T) { content := []byte("testData") readSeeker := NewBufferedReadSeeker(bytes.NewReader(content), make([]byte, 4)) _, err := readSeeker.Seek(4, sdkio.SeekStart) if err != nil { t.Fatalf("failed to seek reader: %v", err) } var ( actual []byte buffer = make([]byte, 4) ) for { n, err := readSeeker.Read(buffer) actual = append(actual, buffer[:n]...) if err != nil && err == io.EOF { break } else if err != nil { t.Fatalf("failed to read from reader: %v", err) } } if e := []byte("Data"); !bytes.Equal(e, actual) { t.Errorf("expected %v, got %v", e, actual) } } func TestBufferedReadSeekerReadAt(t *testing.T) { content := []byte("testData") readSeeker := NewBufferedReadSeeker(bytes.NewReader(content), make([]byte, 2)) buffer := make([]byte, 4) _, err := readSeeker.ReadAt(buffer, 0) if err != nil { t.Fatalf("failed to seek reader: %v", err) } if e := content[:4]; !bytes.Equal(e, buffer) { t.Errorf("expected %v, got %v", e, buffer) } }	1.62 KB	2025-05-25 12:27
📄 default_read_seeker_write_to.go //go:build !windows // +build !windows package s3manager func defaultUploadBufferProvider() ReadSeekerWriteToProvider { return nil }	136 B	2025-05-25 12:27
📄 default_read_seeker_write_to_windows.go package s3manager func defaultUploadBufferProvider() ReadSeekerWriteToProvider { return NewBufferedReadSeekerWriteToPool(1024 * 1024) }	138 B	2025-05-25 12:27
📄 default_writer_read_from.go //go:build !windows // +build !windows package s3manager func defaultDownloadBufferProvider() WriterReadFromProvider { return nil }	135 B	2025-05-25 12:27
📄 default_writer_read_from_windows.go package s3manager func defaultDownloadBufferProvider() WriterReadFromProvider { return NewPooledBufferedWriterReadFromProvider(1024 * 1024) }	144 B	2025-05-25 12:27
📄 doc.go // Package s3manager provides utilities to upload and download objects from // S3 concurrently. Helpful for when working with large objects. package s3manager	159 B	2025-05-25 12:27
📄 download.go package s3manager import ( "fmt" "io" "net/http" "strconv" "strings" "sync" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/aws/awsutil" "github.com/aws/aws-sdk-go/aws/client" "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/service/s3" "github.com/aws/aws-sdk-go/service/s3/s3iface" ) // DefaultDownloadPartSize is the default range of bytes to get at a time when // using Download(). const DefaultDownloadPartSize = 1024 * 1024 * 5 // DefaultDownloadConcurrency is the default number of goroutines to spin up // when using Download(). const DefaultDownloadConcurrency = 5 type errReadingBody struct { err error } func (e errReadingBody) Error() string { return fmt.Sprintf("failed to read part body: %v", e.err) } func (e errReadingBody) Unwrap() error { return e.err } // The Downloader structure that calls Download(). It is safe to call Download() // on this structure for multiple objects and across concurrent goroutines. // Mutating the Downloader's properties is not safe to be done concurrently. type Downloader struct { // The size (in bytes) to request from S3 for each part. // The minimum allowed part size is 5MB, and if this value is set to zero, // the DefaultDownloadPartSize value will be used. // // PartSize is ignored if the Range input parameter is provided. PartSize int64 // The number of goroutines to spin up in parallel when sending parts. // If this is set to zero, the DefaultDownloadConcurrency value will be used. // // Concurrency of 1 will download the parts sequentially. // // Concurrency is ignored if the Range input parameter is provided. Concurrency int // An S3 client to use when performing downloads. S3 s3iface.S3API // List of request options that will be passed down to individual API // operation requests made by the downloader. RequestOptions []request.Option // Defines the buffer strategy used when downloading a part. // // If a WriterReadFromProvider is given the Download manager // will pass the io.WriterAt of the Download request to the provider // and will use the returned WriterReadFrom from the provider as the // destination writer when copying from http response body. BufferProvider WriterReadFromProvider } // WithDownloaderRequestOptions appends to the Downloader's API request options. func WithDownloaderRequestOptions(opts ...request.Option) func(Downloader) { return func(d Downloader) { d.RequestOptions = append(d.RequestOptions, opts...) } } // NewDownloader creates a new Downloader instance to downloads objects from // S3 in concurrent chunks. Pass in additional functional options to customize // the downloader behavior. Requires a client.ConfigProvider in order to create // a S3 service client. The session.Session satisfies the client.ConfigProvider // interface. // // Example: // // // The session the S3 Downloader will use // sess := session.Must(session.NewSession()) // // // Create a downloader with the session and default options // downloader := s3manager.NewDownloader(sess) // // // Create a downloader with the session and custom options // downloader := s3manager.NewDownloader(sess, func(d s3manager.Downloader) { // d.PartSize = 64 1024 * 1024 // 64MB per part // }) func NewDownloader(c client.ConfigProvider, options ...func(Downloader)) Downloader { return newDownloader(s3.New(c), options...) } func newDownloader(client s3iface.S3API, options ...func(Downloader)) Downloader { d := &Downloader{ S3: client, PartSize: DefaultDownloadPartSize, Concurrency: DefaultDownloadConcurrency, BufferProvider: defaultDownloadBufferProvider(), } for _, option := range options { option(d) } return d } // NewDownloaderWithClient creates a new Downloader instance to downloads // objects from S3 in concurrent chunks. Pass in additional functional // options to customize the downloader behavior. Requires a S3 service client // to make S3 API calls. // // Example: // // // The session the S3 Downloader will use // sess := session.Must(session.NewSession()) // // // The S3 client the S3 Downloader will use // s3Svc := s3.New(sess) // // // Create a downloader with the s3 client and default options // downloader := s3manager.NewDownloaderWithClient(s3Svc) // // // Create a downloader with the s3 client and custom options // downloader := s3manager.NewDownloaderWithClient(s3Svc, func(d s3manager.Downloader) { // d.PartSize = 64 1024 * 1024 // 64MB per part // }) func NewDownloaderWithClient(svc s3iface.S3API, options ...func(Downloader)) Downloader { return newDownloader(svc, options...) } type maxRetrier interface { MaxRetries() int } // Download downloads an object in S3 and writes the payload into w using // concurrent GET requests. The n int64 returned is the size of the object downloaded // in bytes. // // Additional functional options can be provided to configure the individual // download. These options are copies of the Downloader instance Download is called from. // Modifying the options will not impact the original Downloader instance. // // It is safe to call this method concurrently across goroutines. // // The w io.WriterAt can be satisfied by an os.File to do multipart concurrent // downloads, or in memory []byte wrapper using aws.WriteAtBuffer. // // Specifying a Downloader.Concurrency of 1 will cause the Downloader to // download the parts from S3 sequentially. // // If the GetObjectInput's Range value is provided that will cause the downloader // to perform a single GetObjectInput request for that object's range. This will // caused the part size, and concurrency configurations to be ignored. func (d Downloader) Download(w io.WriterAt, input s3.GetObjectInput, options ...func(Downloader)) (n int64, err error) { return d.DownloadWithContext(aws.BackgroundContext(), w, input, options...) } // DownloadWithContext downloads an object in S3 and writes the payload into w // using concurrent GET requests. The n int64 returned is the size of the object downloaded // in bytes. // // DownloadWithContext is the same as Download with the additional support for // Context input parameters. The Context must not be nil. A nil Context will // cause a panic. Use the Context to add deadlining, timeouts, etc. The // DownloadWithContext may create sub-contexts for individual underlying // requests. // // Additional functional options can be provided to configure the individual // download. These options are copies of the Downloader instance Download is // called from. Modifying the options will not impact the original Downloader // instance. Use the WithDownloaderRequestOptions helper function to pass in request // options that will be applied to all API operations made with this downloader. // // The w io.WriterAt can be satisfied by an os.File to do multipart concurrent // downloads, or in memory []byte wrapper using aws.WriteAtBuffer. // // Specifying a Downloader.Concurrency of 1 will cause the Downloader to // download the parts from S3 sequentially. // // It is safe to call this method concurrently across goroutines. // // If the GetObjectInput's Range value is provided that will cause the downloader // to perform a single GetObjectInput request for that object's range. This will // caused the part size, and concurrency configurations to be ignored. func (d Downloader) DownloadWithContext(ctx aws.Context, w io.WriterAt, input s3.GetObjectInput, options ...func(Downloader)) (n int64, err error) { if err := validateSupportedARNType(aws.StringValue(input.Bucket)); err != nil { return 0, err } impl := downloader{w: w, in: input, cfg: d, ctx: ctx} for _, option := range options { option(&impl.cfg) } impl.cfg.RequestOptions = append(impl.cfg.RequestOptions, request.WithAppendUserAgent("S3Manager")) if s, ok := d.S3.(maxRetrier); ok { impl.partBodyMaxRetries = s.MaxRetries() } impl.totalBytes = -1 if impl.cfg.Concurrency == 0 { impl.cfg.Concurrency = DefaultDownloadConcurrency } if impl.cfg.PartSize == 0 { impl.cfg.PartSize = DefaultDownloadPartSize } return impl.download() } // DownloadWithIterator will download a batched amount of objects in S3 and writes them // to the io.WriterAt specificed in the iterator. // // Example: // // svc := s3manager.NewDownloader(session) // // fooFile, err := os.Open("/tmp/foo.file") // if err != nil { // return err // } // // barFile, err := os.Open("/tmp/bar.file") // if err != nil { // return err // } // // objects := []s3manager.BatchDownloadObject { // { // Object: &s3.GetObjectInput { // Bucket: aws.String("bucket"), // Key: aws.String("foo"), // }, // Writer: fooFile, // }, // { // Object: &s3.GetObjectInput { // Bucket: aws.String("bucket"), // Key: aws.String("bar"), // }, // Writer: barFile, // }, // } // // iter := &s3manager.DownloadObjectsIterator{Objects: objects} // if err := svc.DownloadWithIterator(aws.BackgroundContext(), iter); err != nil { // return err // } func (d Downloader) DownloadWithIterator(ctx aws.Context, iter BatchDownloadIterator, opts ...func(Downloader)) error { var errs []Error for iter.Next() { object := iter.DownloadObject() if _, err := d.DownloadWithContext(ctx, object.Writer, object.Object, opts...); err != nil { errs = append(errs, newError(err, object.Object.Bucket, object.Object.Key)) } if object.After == nil { continue } if err := object.After(); err != nil { errs = append(errs, newError(err, object.Object.Bucket, object.Object.Key)) } } if len(errs) > 0 { return NewBatchError("BatchedDownloadIncomplete", "some objects have failed to download.", errs) } return nil } // downloader is the implementation structure used internally by Downloader. type downloader struct { ctx aws.Context cfg Downloader in s3.GetObjectInput w io.WriterAt wg sync.WaitGroup m sync.Mutex pos int64 totalBytes int64 written int64 err error partBodyMaxRetries int } // download performs the implementation of the object download across ranged // GETs. func (d downloader) download() (n int64, err error) { // If range is specified fall back to single download of that range // this enables the functionality of ranged gets with the downloader but // at the cost of no multipart downloads. if rng := aws.StringValue(d.in.Range); len(rng) > 0 { d.downloadRange(rng) return d.written, d.err } // Spin off first worker to check additional header information d.getChunk() if total := d.getTotalBytes(); total >= 0 { // Spin up workers ch := make(chan dlchunk, d.cfg.Concurrency) for i := 0; i < d.cfg.Concurrency; i++ { d.wg.Add(1) go d.downloadPart(ch) } // Assign work for d.getErr() == nil { if d.pos >= total { break // We're finished queuing chunks } // Queue the next range of bytes to read. ch <- dlchunk{w: d.w, start: d.pos, size: d.cfg.PartSize} d.pos += d.cfg.PartSize } // Wait for completion close(ch) d.wg.Wait() } else { // Checking if we read anything new for d.err == nil { d.getChunk() } // We expect a 416 error letting us know we are done downloading the // total bytes. Since we do not know the content's length, this will // keep grabbing chunks of data until the range of bytes specified in // the request is out of range of the content. Once, this happens, a // 416 should occur. e, ok := d.err.(awserr.RequestFailure) if ok && e.StatusCode() == http.StatusRequestedRangeNotSatisfiable { d.err = nil } } // Return error return d.written, d.err } // downloadPart is an individual goroutine worker reading from the ch channel // and performing a GetObject request on the data with a given byte range. // // If this is the first worker, this operation also resolves the total number // of bytes to be read so that the worker manager knows when it is finished. func (d downloader) downloadPart(ch chan dlchunk) { defer d.wg.Done() for { chunk, ok := <-ch if !ok { break } if d.getErr() != nil { // Drain the channel if there is an error, to prevent deadlocking // of download producer. continue } if err := d.downloadChunk(chunk); err != nil { d.setErr(err) } } } // getChunk grabs a chunk of data from the body. // Not thread safe. Should only used when grabbing data on a single thread. func (d downloader) getChunk() { if d.getErr() != nil { return } chunk := dlchunk{w: d.w, start: d.pos, size: d.cfg.PartSize} d.pos += d.cfg.PartSize if err := d.downloadChunk(chunk); err != nil { d.setErr(err) } } // downloadRange downloads an Object given the passed in Byte-Range value. // The chunk used down download the range will be configured for that range. func (d downloader) downloadRange(rng string) { if d.getErr() != nil { return } chunk := dlchunk{w: d.w, start: d.pos} // Ranges specified will short circuit the multipart download chunk.withRange = rng if err := d.downloadChunk(chunk); err != nil { d.setErr(err) } // Update the position based on the amount of data received. d.pos = d.written } // downloadChunk downloads the chunk from s3 func (d downloader) downloadChunk(chunk dlchunk) error { in := &s3.GetObjectInput{} awsutil.Copy(in, d.in) // Get the next byte range of data in.Range = aws.String(chunk.ByteRange()) var n int64 var err error for retry := 0; retry <= d.partBodyMaxRetries; retry++ { n, err = d.tryDownloadChunk(in, &chunk) if err == nil { break } // Check if the returned error is an errReadingBody. // If err is errReadingBody this indicates that an error // occurred while copying the http response body. // If this occurs we unwrap the err to set the underlying error // and attempt any remaining retries. if bodyErr, ok := err.(errReadingBody); ok { err = bodyErr.Unwrap() } else { return err } chunk.cur = 0 logMessage(d.cfg.S3, aws.LogDebugWithRequestRetries, fmt.Sprintf("DEBUG: object part body download interrupted %s, err, %v, retrying attempt %d", aws.StringValue(in.Key), err, retry)) } d.incrWritten(n) return err } func (d downloader) tryDownloadChunk(in s3.GetObjectInput, w io.Writer) (int64, error) { cleanup := func() {} if d.cfg.BufferProvider != nil { w, cleanup = d.cfg.BufferProvider.GetReadFrom(w) } defer cleanup() resp, err := d.cfg.S3.GetObjectWithContext(d.ctx, in, d.cfg.RequestOptions...) if err != nil { return 0, err } d.setTotalBytes(resp) // Set total if not yet set. var src io.Reader = resp.Body if d.cfg.BufferProvider != nil { src = &suppressWriterAt{suppressed: src} } n, err := io.Copy(w, src) resp.Body.Close() if err != nil { return n, &errReadingBody{err: err} } return n, nil } func logMessage(svc s3iface.S3API, level aws.LogLevelType, msg string) { s, ok := svc.(s3.S3) if !ok { return } if s.Config.Logger == nil { return } if s.Config.LogLevel.Matches(level) { s.Config.Logger.Log(msg) } } // getTotalBytes is a thread-safe getter for retrieving the total byte status. func (d downloader) getTotalBytes() int64 { d.m.Lock() defer d.m.Unlock() return d.totalBytes } // setTotalBytes is a thread-safe setter for setting the total byte status. // Will extract the object's total bytes from the Content-Range if the file // will be chunked, or Content-Length. Content-Length is used when the response // does not include a Content-Range. Meaning the object was not chunked. This // occurs when the full file fits within the PartSize directive. func (d downloader) setTotalBytes(resp s3.GetObjectOutput) { d.m.Lock() defer d.m.Unlock() if d.totalBytes >= 0 { return } if resp.ContentRange == nil { // ContentRange is nil when the full file contents is provided, and // is not chunked. Use ContentLength instead. if resp.ContentLength != nil { d.totalBytes = resp.ContentLength return } } else { parts := strings.Split(resp.ContentRange, "/") total := int64(-1) var err error // Checking for whether or not a numbered total exists // If one does not exist, we will assume the total to be -1, undefined, // and sequentially download each chunk until hitting a 416 error totalStr := parts[len(parts)-1] if totalStr != "" { total, err = strconv.ParseInt(totalStr, 10, 64) if err != nil { d.err = err return } } d.totalBytes = total } } func (d downloader) incrWritten(n int64) { d.m.Lock() defer d.m.Unlock() d.written += n } // getErr is a thread-safe getter for the error object func (d downloader) getErr() error { d.m.Lock() defer d.m.Unlock() return d.err } // setErr is a thread-safe setter for the error object func (d downloader) setErr(e error) { d.m.Lock() defer d.m.Unlock() d.err = e } // dlchunk represents a single chunk of data to write by the worker routine. // This structure also implements an io.SectionReader style interface for // io.WriterAt, effectively making it an io.SectionWriter (which does not // exist). type dlchunk struct { w io.WriterAt start int64 size int64 cur int64 // specifies the byte range the chunk should be downloaded with. withRange string } // Write wraps io.WriterAt for the dlchunk, writing from the dlchunk's start // position to its end (or EOF). // // If a range is specified on the dlchunk the size will be ignored when writing. // as the total size may not of be known ahead of time. func (c dlchunk) Write(p []byte) (n int, err error) { if c.cur >= c.size && len(c.withRange) == 0 { return 0, io.EOF } n, err = c.w.WriteAt(p, c.start+c.cur) c.cur += int64(n) return } // ByteRange returns a HTTP Byte-Range header value that should be used by the // client to request the chunk's range. func (c dlchunk) ByteRange() string { if len(c.withRange) != 0 { return c.withRange } return fmt.Sprintf("bytes=%d-%d", c.start, c.start+c.size-1) }	17.54 KB	2025-05-25 12:27
📄 download_test.go //go:build go1.7 // +build go1.7 package s3manager_test import ( "bytes" "encoding/xml" "fmt" "io" "io/ioutil" "net/http" "reflect" "regexp" "strconv" "strings" "sync" "sync/atomic" "testing" "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/awstesting" "github.com/aws/aws-sdk-go/awstesting/unit" "github.com/aws/aws-sdk-go/internal/sdkio" "github.com/aws/aws-sdk-go/service/s3" "github.com/aws/aws-sdk-go/service/s3/internal/s3testing" "github.com/aws/aws-sdk-go/service/s3/s3manager" ) func dlLoggingSvc(data []byte) (s3.S3, []string, []string) { var m sync.Mutex names := []string{} ranges := []string{} svc := s3.New(unit.Session) svc.Handlers.Send.Clear() svc.Handlers.Send.PushBack(func(r request.Request) { m.Lock() defer m.Unlock() names = append(names, r.Operation.Name) ranges = append(ranges, r.Params.(s3.GetObjectInput).Range) rerng := regexp.MustCompile(`bytes=(\d+)-(\d+)`) rng := rerng.FindStringSubmatch(r.HTTPRequest.Header.Get("Range")) start, _ := strconv.ParseInt(rng[1], 10, 64) fin, _ := strconv.ParseInt(rng[2], 10, 64) fin++ if fin > int64(len(data)) { fin = int64(len(data)) } bodyBytes := data[start:fin] r.HTTPResponse = &http.Response{ StatusCode: 200, Body: ioutil.NopCloser(bytes.NewReader(bodyBytes)), Header: http.Header{}, } r.HTTPResponse.Header.Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, fin-1, len(data))) r.HTTPResponse.Header.Set("Content-Length", fmt.Sprintf("%d", len(bodyBytes))) }) return svc, &names, &ranges } func dlLoggingSvcNoChunk(data []byte) (s3.S3, []string) { var m sync.Mutex names := []string{} svc := s3.New(unit.Session) svc.Handlers.Send.Clear() svc.Handlers.Send.PushBack(func(r request.Request) { m.Lock() defer m.Unlock() names = append(names, r.Operation.Name) r.HTTPResponse = &http.Response{ StatusCode: 200, Body: ioutil.NopCloser(bytes.NewReader(data[:])), Header: http.Header{}, } r.HTTPResponse.Header.Set("Content-Length", fmt.Sprintf("%d", len(data))) }) return svc, &names } func dlLoggingSvcNoContentRangeLength(data []byte, states []int) (s3.S3, []string) { var m sync.Mutex names := []string{} var index int svc := s3.New(unit.Session) svc.Handlers.Send.Clear() svc.Handlers.Send.PushBack(func(r request.Request) { m.Lock() defer m.Unlock() names = append(names, r.Operation.Name) var body io.Reader if states[index] < 400 { body = bytes.NewReader(data[:]) } else { var buffer bytes.Buffer encoder := xml.NewEncoder(&buffer) _ = encoder.Encode(&mockErrorResponse) body = &buffer } r.HTTPResponse = &http.Response{ StatusCode: states[index], Body: ioutil.NopCloser(body), Header: http.Header{}, } index++ }) return svc, &names } func dlLoggingSvcContentRangeTotalAny(data []byte, states []int) (s3.S3, []string) { var m sync.Mutex names := []string{} ranges := []string{} var index int svc := s3.New(unit.Session) svc.Handlers.Send.Clear() svc.Handlers.Send.PushBack(func(r request.Request) { m.Lock() defer m.Unlock() names = append(names, r.Operation.Name) ranges = append(ranges, r.Params.(s3.GetObjectInput).Range) rerng := regexp.MustCompile(`bytes=(\d+)-(\d+)`) rng := rerng.FindStringSubmatch(r.HTTPRequest.Header.Get("Range")) start, _ := strconv.ParseInt(rng[1], 10, 64) fin, _ := strconv.ParseInt(rng[2], 10, 64) fin++ if fin >= int64(len(data)) { fin = int64(len(data)) } // Setting start and finish to 0 because this state of 1 is suppose to // be an error state of 416 if index == len(states)-1 { start = 0 fin = 0 } bodyBytes := data[start:fin] r.HTTPResponse = &http.Response{ StatusCode: states[index], Body: ioutil.NopCloser(bytes.NewReader(bodyBytes)), Header: http.Header{}, } r.HTTPResponse.Header.Set("Content-Range", fmt.Sprintf("bytes %d-%d/", start, fin-1)) index++ }) return svc, &names } func dlLoggingSvcWithErrReader(cases []testErrReader) (s3.S3, []string) { var m sync.Mutex names := []string{} var index int svc := s3.New(unit.Session, &aws.Config{ MaxRetries: aws.Int(len(cases) - 1), }) svc.Handlers.Send.Clear() svc.Handlers.Send.PushBack(func(r request.Request) { m.Lock() defer m.Unlock() names = append(names, r.Operation.Name) c := cases[index] r.HTTPResponse = &http.Response{ StatusCode: http.StatusOK, Body: ioutil.NopCloser(&c), Header: http.Header{}, } r.HTTPResponse.Header.Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", 0, c.Len-1, c.Len)) r.HTTPResponse.Header.Set("Content-Length", fmt.Sprintf("%d", c.Len)) index++ }) return svc, &names } func TestDownloadOrder(t testing.T) { s, names, ranges := dlLoggingSvc(buf12MB) d := s3manager.NewDownloaderWithClient(s, func(d s3manager.Downloader) { d.Concurrency = 1 }) w := aws.NewWriteAtBuffer(make([]byte, len(buf12MB))) n, err := d.Download(w, &s3.GetObjectInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), }) if err != nil { t.Fatalf("expect no error, got %v", err) } if e, a := int64(len(buf12MB)), n; e != a { t.Errorf("expect %d buffer length, got %d", e, a) } expectCalls := []string{"GetObject", "GetObject", "GetObject"} if e, a := expectCalls, names; !reflect.DeepEqual(e, a) { t.Errorf("expect %v API calls, got %v", e, a) } expectRngs := []string{"bytes=0-5242879", "bytes=5242880-10485759", "bytes=10485760-15728639"} if e, a := expectRngs, ranges; !reflect.DeepEqual(e, a) { t.Errorf("expect %v ranges, got %v", e, a) } } func TestDownloadZero(t testing.T) { s, names, ranges := dlLoggingSvc([]byte{}) d := s3manager.NewDownloaderWithClient(s) w := &aws.WriteAtBuffer{} n, err := d.Download(w, &s3.GetObjectInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), }) if err != nil { t.Fatalf("expect no error, got %v", err) } if n != 0 { t.Errorf("expect 0 bytes read, got %d", n) } expectCalls := []string{"GetObject"} if e, a := expectCalls, names; !reflect.DeepEqual(e, a) { t.Errorf("expect %v API calls, got %v", e, a) } expectRngs := []string{"bytes=0-5242879"} if e, a := expectRngs, ranges; !reflect.DeepEqual(e, a) { t.Errorf("expect %v ranges, got %v", e, a) } } func TestDownloadSetPartSize(t testing.T) { s, names, ranges := dlLoggingSvc([]byte{1, 2, 3}) d := s3manager.NewDownloaderWithClient(s, func(d s3manager.Downloader) { d.Concurrency = 1 d.PartSize = 1 }) w := &aws.WriteAtBuffer{} n, err := d.Download(w, &s3.GetObjectInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), }) if err != nil { t.Fatalf("expect no error, got %v", err) } if e, a := int64(3), n; e != a { t.Errorf("expect %d bytes read, got %d", e, a) } expectCalls := []string{"GetObject", "GetObject", "GetObject"} if e, a := expectCalls, names; !reflect.DeepEqual(e, a) { t.Errorf("expect %v API calls, got %v", e, a) } expectRngs := []string{"bytes=0-0", "bytes=1-1", "bytes=2-2"} if e, a := expectRngs, ranges; !reflect.DeepEqual(e, a) { t.Errorf("expect %v ranges, got %v", e, a) } expectBytes := []byte{1, 2, 3} if e, a := expectBytes, w.Bytes(); !reflect.DeepEqual(e, a) { t.Errorf("expect %v bytes, got %v", e, a) } } func TestDownloadError(t testing.T) { s, names, _ := dlLoggingSvc([]byte{1, 2, 3}) num := 0 s.Handlers.Send.PushBack(func(r request.Request) { num++ if num > 1 { r.HTTPResponse.StatusCode = 400 r.HTTPResponse.Body = ioutil.NopCloser(bytes.NewReader([]byte{})) } }) d := s3manager.NewDownloaderWithClient(s, func(d s3manager.Downloader) { d.Concurrency = 1 d.PartSize = 1 }) w := &aws.WriteAtBuffer{} n, err := d.Download(w, &s3.GetObjectInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), }) if err == nil { t.Fatalf("expect error, got none") } aerr := err.(awserr.Error) if e, a := "BadRequest", aerr.Code(); e != a { t.Errorf("expect %s error code, got %s", e, a) } if e, a := int64(1), n; e != a { t.Errorf("expect %d bytes read, got %d", e, a) } expectCalls := []string{"GetObject", "GetObject"} if e, a := expectCalls, names; !reflect.DeepEqual(e, a) { t.Errorf("expect %v API calls, got %v", e, a) } expectBytes := []byte{1} if e, a := expectBytes, w.Bytes(); !reflect.DeepEqual(e, a) { t.Errorf("expect %v bytes, got %v", e, a) } } func TestDownloadNonChunk(t testing.T) { s, names := dlLoggingSvcNoChunk(buf2MB) d := s3manager.NewDownloaderWithClient(s, func(d s3manager.Downloader) { d.Concurrency = 1 }) w := &aws.WriteAtBuffer{} n, err := d.Download(w, &s3.GetObjectInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), }) if err != nil { t.Fatalf("expect no error, got %v", err) } if e, a := int64(len(buf2MB)), n; e != a { t.Errorf("expect %d bytes read, got %d", e, a) } expectCalls := []string{"GetObject"} if e, a := expectCalls, names; !reflect.DeepEqual(e, a) { t.Errorf("expect %v API calls, got %v", e, a) } count := 0 for _, b := range w.Bytes() { count += int(b) } if count != 0 { t.Errorf("expect 0 count, got %d", count) } } func TestDownloadNoContentRangeLength(t testing.T) { s, names := dlLoggingSvcNoContentRangeLength(buf2MB, []int{200, 416}) d := s3manager.NewDownloaderWithClient(s, func(d s3manager.Downloader) { d.Concurrency = 1 }) w := &aws.WriteAtBuffer{} n, err := d.Download(w, &s3.GetObjectInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), }) if err != nil { t.Fatalf("expect no error, got %v", err) } if e, a := int64(len(buf2MB)), n; e != a { t.Errorf("expect %d bytes read, got %d", e, a) } expectCalls := []string{"GetObject", "GetObject"} if e, a := expectCalls, names; !reflect.DeepEqual(e, a) { t.Errorf("expect %v API calls, got %v", e, a) } count := 0 for _, b := range w.Bytes() { count += int(b) } if count != 0 { t.Errorf("expect 0 count, got %d", count) } } func TestDownloadContentRangeTotalAny(t testing.T) { s, names := dlLoggingSvcContentRangeTotalAny(buf2MB, []int{200, 416}) d := s3manager.NewDownloaderWithClient(s, func(d s3manager.Downloader) { d.Concurrency = 1 }) w := &aws.WriteAtBuffer{} n, err := d.Download(w, &s3.GetObjectInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), }) if err != nil { t.Fatalf("expect no error, got %v", err) } if e, a := int64(len(buf2MB)), n; e != a { t.Errorf("expect %d bytes read, got %d", e, a) } expectCalls := []string{"GetObject", "GetObject"} if e, a := expectCalls, names; !reflect.DeepEqual(e, a) { t.Errorf("expect %v API calls, got %v", e, a) } count := 0 for _, b := range w.Bytes() { count += int(b) } if count != 0 { t.Errorf("expect 0 count, got %d", count) } } func TestDownloadPartBodyRetry_SuccessRetry(t testing.T) { s, names := dlLoggingSvcWithErrReader([]testErrReader{ {Buf: []byte("ab"), Len: 3, Err: io.ErrUnexpectedEOF}, {Buf: []byte("123"), Len: 3, Err: io.EOF}, }) d := s3manager.NewDownloaderWithClient(s, func(d s3manager.Downloader) { d.Concurrency = 1 }) w := &aws.WriteAtBuffer{} n, err := d.Download(w, &s3.GetObjectInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), }) if err != nil { t.Fatalf("expect no error, got %v", err) } if e, a := int64(3), n; e != a { t.Errorf("expect %d bytes read, got %d", e, a) } expectCalls := []string{"GetObject", "GetObject"} if e, a := expectCalls, names; !reflect.DeepEqual(e, a) { t.Errorf("expect %v API calls, got %v", e, a) } if e, a := "123", string(w.Bytes()); e != a { t.Errorf("expect %q response, got %q", e, a) } } func TestDownloadPartBodyRetry_SuccessNoRetry(t testing.T) { s, names := dlLoggingSvcWithErrReader([]testErrReader{ {Buf: []byte("abc"), Len: 3, Err: io.EOF}, }) d := s3manager.NewDownloaderWithClient(s, func(d s3manager.Downloader) { d.Concurrency = 1 }) w := &aws.WriteAtBuffer{} n, err := d.Download(w, &s3.GetObjectInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), }) if err != nil { t.Fatalf("expect no error, got %v", err) } if e, a := int64(3), n; e != a { t.Errorf("expect %d bytes read, got %d", e, a) } expectCalls := []string{"GetObject"} if e, a := expectCalls, names; !reflect.DeepEqual(e, a) { t.Errorf("expect %v API calls, got %v", e, a) } if e, a := "abc", string(w.Bytes()); e != a { t.Errorf("expect %q response, got %q", e, a) } } func TestDownloadPartBodyRetry_FailRetry(t testing.T) { s, names := dlLoggingSvcWithErrReader([]testErrReader{ {Buf: []byte("ab"), Len: 3, Err: io.ErrUnexpectedEOF}, }) d := s3manager.NewDownloaderWithClient(s, func(d s3manager.Downloader) { d.Concurrency = 1 }) w := &aws.WriteAtBuffer{} n, err := d.Download(w, &s3.GetObjectInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), }) if err == nil { t.Fatalf("expect error, got none") } if e, a := "unexpected EOF", err.Error(); !strings.Contains(a, e) { t.Errorf("expect %q error message to be in %q", e, a) } if e, a := int64(2), n; e != a { t.Errorf("expect %d bytes read, got %d", e, a) } expectCalls := []string{"GetObject"} if e, a := expectCalls, names; !reflect.DeepEqual(e, a) { t.Errorf("expect %v API calls, got %v", e, a) } if e, a := "ab", string(w.Bytes()); e != a { t.Errorf("expect %q response, got %q", e, a) } } func TestDownloadWithContextCanceled(t testing.T) { d := s3manager.NewDownloader(unit.Session) params := s3.GetObjectInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), } ctx := &awstesting.FakeContext{DoneCh: make(chan struct{})} ctx.Error = fmt.Errorf("context canceled") close(ctx.DoneCh) w := &aws.WriteAtBuffer{} _, err := d.DownloadWithContext(ctx, w, &params) if err == nil { t.Fatalf("expected error, did not get one") } aerr := err.(awserr.Error) if e, a := request.CanceledErrorCode, aerr.Code(); e != a { t.Errorf("expected error code %q, got %q", e, a) } if e, a := "canceled", aerr.Message(); !strings.Contains(a, e) { t.Errorf("expected error message to contain %q, but did not %q", e, a) } } func TestDownload_WithRange(t testing.T) { s, names, ranges := dlLoggingSvc([]byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}) d := s3manager.NewDownloaderWithClient(s, func(d s3manager.Downloader) { d.Concurrency = 10 // should be ignored d.PartSize = 1 // should be ignored }) w := &aws.WriteAtBuffer{} n, err := d.Download(w, &s3.GetObjectInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), Range: aws.String("bytes=2-6"), }) if err != nil { t.Fatalf("expect no error, got %v", err) } if e, a := int64(5), n; e != a { t.Errorf("expect %d bytes read, got %d", e, a) } expectCalls := []string{"GetObject"} if e, a := expectCalls, names; !reflect.DeepEqual(e, a) { t.Errorf("expect %v API calls, got %v", e, a) } expectRngs := []string{"bytes=2-6"} if e, a := expectRngs, ranges; !reflect.DeepEqual(e, a) { t.Errorf("expect %v ranges, got %v", e, a) } expectBytes := []byte{2, 3, 4, 5, 6} if e, a := expectBytes, w.Bytes(); !reflect.DeepEqual(e, a) { t.Errorf("expect %v bytes, got %v", e, a) } } func TestDownload_WithFailure(t testing.T) { svc := s3.New(unit.Session) svc.Handlers.Send.Clear() reqCount := int64(0) startingByte := 0 svc.Handlers.Send.PushBack(func(r request.Request) { switch atomic.LoadInt64(&reqCount) { case 1: // Give a chance for the multipart chunks to be queued up time.Sleep(1 time.Second) r.HTTPResponse = &http.Response{ Header: http.Header{}, Body: ioutil.NopCloser(&bytes.Buffer{}), } r.Error = awserr.New("ConnectionError", "some connection error", nil) r.Retryable = aws.Bool(false) default: body := bytes.NewReader(make([]byte, s3manager.DefaultDownloadPartSize)) r.HTTPResponse = &http.Response{ StatusCode: http.StatusOK, Status: http.StatusText(http.StatusOK), ContentLength: int64(body.Len()), Body: ioutil.NopCloser(body), Header: http.Header{}, } r.HTTPResponse.Header.Set("Content-Length", strconv.Itoa(body.Len())) r.HTTPResponse.Header.Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", startingByte, body.Len()-1, body.Len()10)) startingByte += body.Len() if reqCount > 0 { // sleep here to ensure context switching between goroutines time.Sleep(25 time.Millisecond) } } atomic.AddInt64(&reqCount, 1) }) d := s3manager.NewDownloaderWithClient(svc, func(d s3manager.Downloader) { d.Concurrency = 2 }) w := &aws.WriteAtBuffer{} params := s3.GetObjectInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), } // Expect this request to exit quickly after failure _, err := d.Download(w, &params) if err == nil { t.Fatalf("expect error, got none") } if atomic.LoadInt64(&reqCount) > 3 { t.Errorf("expect no more than 3 requests, but received %d", reqCount) } } func TestDownloadBufferStrategy(t testing.T) { cases := map[string]struct { partSize int64 strategy recordedWriterReadFromProvider expectedSize int64 }{ "no strategy": { partSize: s3manager.DefaultDownloadPartSize, expectedSize: 10 sdkio.MebiByte, }, "partSize modulo bufferSize == 0": { partSize: 5 * sdkio.MebiByte, strategy: &recordedWriterReadFromProvider{ WriterReadFromProvider: s3manager.NewPooledBufferedWriterReadFromProvider(int(sdkio.MebiByte)), // 1 MiB }, expectedSize: 10 * sdkio.MebiByte, // 10 MiB }, "partSize modulo bufferSize > 0": { partSize: 5 * 1024 * 1204, // 5 MiB strategy: &recordedWriterReadFromProvider{ WriterReadFromProvider: s3manager.NewPooledBufferedWriterReadFromProvider(2 * int(sdkio.MebiByte)), // 2 MiB }, expectedSize: 10 * sdkio.MebiByte, // 10 MiB }, } for name, tCase := range cases { t.Logf("starting case: %v", name) expected := s3testing.GetTestBytes(int(tCase.expectedSize)) svc, _, _ := dlLoggingSvc(expected) d := s3manager.NewDownloaderWithClient(svc, func(d s3manager.Downloader) { d.PartSize = tCase.partSize if tCase.strategy != nil { d.BufferProvider = tCase.strategy } }) buffer := aws.NewWriteAtBuffer(make([]byte, len(expected))) n, err := d.Download(buffer, &s3.GetObjectInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), }) if err != nil { t.Errorf("failed to download: %v", err) } if e, a := len(expected), int(n); e != a { t.Errorf("expected %v, got %v downloaded bytes", e, a) } if e, a := expected, buffer.Bytes(); !bytes.Equal(e, a) { t.Errorf("downloaded bytes did not match expected") } if tCase.strategy != nil { if e, a := tCase.strategy.callbacksVended, tCase.strategy.callbacksExecuted; e != a { t.Errorf("expected %v, got %v", e, a) } } } } type testErrReader struct { Buf []byte Err error Len int64 off int } func (r testErrReader) Read(p []byte) (int, error) { to := len(r.Buf) - r.off n := copy(p, r.Buf[r.off:to]) r.off += n if n < len(p) { return n, r.Err } return n, nil } func TestDownloadBufferStrategy_Errors(t testing.T) { expected := s3testing.GetTestBytes(int(10 sdkio.MebiByte)) svc, _, _ := dlLoggingSvc(expected) strat := &recordedWriterReadFromProvider{ WriterReadFromProvider: s3manager.NewPooledBufferedWriterReadFromProvider(int(2 * sdkio.MebiByte)), } d := s3manager.NewDownloaderWithClient(svc, func(d s3manager.Downloader) { d.PartSize = 5 sdkio.MebiByte d.BufferProvider = strat d.Concurrency = 1 }) seenOps := make(map[string]struct{}) svc.Handlers.Send.PushFront(func(request.Request) {}) svc.Handlers.Send.AfterEachFn = func(item request.HandlerListRunItem) bool { r := item.Request if r.Operation.Name != "GetObject" { return true } input := r.Params.(s3.GetObjectInput) fingerPrint := fmt.Sprintf("%s/%s/%s/%s", r.Operation.Name, input.Bucket, input.Key, input.Range) if _, ok := seenOps[fingerPrint]; ok { return true } seenOps[fingerPrint] = struct{}{} regex := regexp.MustCompile(`bytes=(\d+)-(\d+)`) rng := regex.FindStringSubmatch(input.Range) start, _ := strconv.ParseInt(rng[1], 10, 64) fin, _ := strconv.ParseInt(rng[2], 10, 64) _, _ = io.Copy(ioutil.Discard, r.Body) r.HTTPResponse = &http.Response{ StatusCode: 200, Body: aws.ReadSeekCloser(&badReader{err: io.ErrUnexpectedEOF}), ContentLength: fin - start, } return false } buffer := aws.NewWriteAtBuffer(make([]byte, len(expected))) n, err := d.Download(buffer, &s3.GetObjectInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), }) if err != nil { t.Errorf("failed to download: %v", err) } if e, a := len(expected), int(n); e != a { t.Errorf("expected %v, got %v downloaded bytes", e, a) } if e, a := expected, buffer.Bytes(); !bytes.Equal(e, a) { t.Errorf("downloaded bytes did not match expected") } if e, a := strat.callbacksVended, strat.callbacksExecuted; e != a { t.Errorf("expected %v, got %v", e, a) } } func TestDownloaderValidARN(t testing.T) { cases := map[string]struct { input s3.GetObjectInput wantErr bool }{ "standard bucket": { input: s3.GetObjectInput{ Bucket: aws.String("test-bucket"), Key: aws.String("test-key"), }, }, "accesspoint": { input: s3.GetObjectInput{ Bucket: aws.String("arn:aws:s3:us-west-2:123456789012:accesspoint/myap"), Key: aws.String("test-key"), }, }, "outpost accesspoint": { input: s3.GetObjectInput{ Bucket: aws.String("arn:aws:s3-outposts:us-west-2:012345678901:outpost/op-1234567890123456/accesspoint/myaccesspoint"), Key: aws.String("test-key"), }, }, "s3-object-lambda accesspoint": { input: s3.GetObjectInput{ Bucket: aws.String("arn:aws:s3-object-lambda:us-west-2:123456789012:accesspoint/myap"), }, wantErr: true, }, } for name, tt := range cases { t.Run(name, func(t testing.T) { client, _, _ := dlLoggingSvc(buf2MB) client.Config.Region = aws.String("us-west-2") client.ClientInfo.SigningRegion = "us-west-2" downloader := s3manager.NewDownloaderWithClient(client, func(downloader s3manager.Downloader) { downloader.Concurrency = 1 }) _, err := downloader.Download(&awstesting.DiscardAt{}, &tt.input) if (err != nil) != tt.wantErr { t.Errorf("err: %v, wantErr: %v", err, tt.wantErr) } }) } } type recordedWriterReadFromProvider struct { callbacksVended uint32 callbacksExecuted uint32 s3manager.WriterReadFromProvider } func (r recordedWriterReadFromProvider) GetReadFrom(writer io.Writer) (s3manager.WriterReadFrom, func()) { w, cleanup := r.WriterReadFromProvider.GetReadFrom(writer) atomic.AddUint32(&r.callbacksVended, 1) return w, func() { atomic.AddUint32(&r.callbacksExecuted, 1) cleanup() } } type badReader struct { err error } func (b *badReader) Read(p []byte) (int, error) { tb := s3testing.GetTestBytes(len(p)) copy(p, tb) return len(p), b.err } var mockErrorResponse = struct { XMLName xml.Name `xml:"Error"` Code string `xml:"Code"` Message string `xml:"Message"` }{ Code: "MOCK_S3_ERROR_CODE", Message: "Mocked S3 Error Message", }	22.83 KB	2025-05-25 12:27
📄 examples_test.go package s3manager_test import ( "bytes" "fmt" "net/http" "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/session" "github.com/aws/aws-sdk-go/service/s3/s3manager" ) // ExampleNewUploader_overrideReadSeekerProvider gives an example // on a custom ReadSeekerWriteToProvider can be provided to Uploader // to define how parts will be buffered in memory. func ExampleNewUploader_overrideReadSeekerProvider() { sess := session.Must(session.NewSession()) uploader := s3manager.NewUploader(sess, func(u s3manager.Uploader) { // Define a strategy that will buffer 25 MiB in memory u.BufferProvider = s3manager.NewBufferedReadSeekerWriteToPool(25 1024 * 1024) }) _, err := uploader.Upload(&s3manager.UploadInput{ Bucket: aws.String("examplebucket"), Key: aws.String("largeobject"), Body: bytes.NewReader([]byte("large_multi_part_upload")), }) if err != nil { fmt.Println(err.Error()) } } // ExampleNewUploader_overrideTransport gives an example // on how to override the default HTTP transport. This can // be used to tune timeouts such as response headers, or // write / read buffer usage when writing or reading respectively // from the net/http transport. func ExampleNewUploader_overrideTransport() { // Create Transport tr := &http.Transport{ ResponseHeaderTimeout: 1 * time.Second, // WriteBufferSize: 10241024 // Go 1.13 // ReadBufferSize: 10241024 // Go 1.13 } sess := session.Must(session.NewSession(&aws.Config{ HTTPClient: &http.Client{Transport: tr}, })) uploader := s3manager.NewUploader(sess) _, err := uploader.Upload(&s3manager.UploadInput{ Bucket: aws.String("examplebucket"), Key: aws.String("largeobject"), Body: bytes.NewReader([]byte("large_multi_part_upload")), }) if err != nil { fmt.Println(err.Error()) } }	1.78 KB	2025-05-25 12:27
📄 integ_bucket_region_test.go //go:build integration // +build integration package s3manager_test import ( "testing" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/s3/s3manager" ) func TestGetBucketRegion(t *testing.T) { expectRegion := aws.StringValue(integSess.Config.Region) ctx := aws.BackgroundContext() region, err := s3manager.GetBucketRegion(ctx, integSess, aws.StringValue(bucketName), expectRegion) if err != nil { t.Fatalf("expect no error, got %v", err) } if e, a := expectRegion, region; e != a { t.Errorf("expect %s bucket region, got %s", e, a) } }	577 B	2025-05-25 12:27
📄 integ_shared_test.go //go:build integration // +build integration package s3manager_test import ( "crypto/md5" "fmt" "io" "os" "testing" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/session" "github.com/aws/aws-sdk-go/awstesting/integration" "github.com/aws/aws-sdk-go/awstesting/integration/s3integ" "github.com/aws/aws-sdk-go/service/s3" "github.com/aws/aws-sdk-go/service/s3/s3manager" ) func init() { integSess = integration.SessionWithDefaultRegion("us-west-2") } var integSess session.Session var bucketName string var svc s3.S3 func TestMain(m testing.M) { svc = s3.New(integSess) bucketName = aws.String(s3integ.GenerateBucketName()) if err := s3integ.SetupBucket(svc, bucketName); err != nil { panic(err) } var result int defer func() { if err := s3integ.CleanupBucket(svc, bucketName); err != nil { fmt.Fprintln(os.Stderr, err) } if r := recover(); r != nil { fmt.Fprintln(os.Stderr, "S3 integrationt tests paniced,", r) result = 1 } os.Exit(result) }() result = m.Run() } type dlwriter struct { buf []byte } func newDLWriter(size int) dlwriter { return &dlwriter{buf: make([]byte, size)} } func (d dlwriter) WriteAt(p []byte, pos int64) (n int, err error) { if pos > int64(len(d.buf)) { return 0, io.EOF } written := 0 for i, b := range p { if i >= len(d.buf) { break } d.buf[pos+int64(i)] = b written++ } return written, nil } func validate(t testing.T, key string, md5value string) { mgr := s3manager.NewDownloader(integSess) params := &s3.GetObjectInput{Bucket: bucketName, Key: &key} w := newDLWriter(1024 * 1024 * 20) n, err := mgr.Download(w, params) if err != nil { t.Fatalf("expect no error, got %v", err) } if e, a := md5value, fmt.Sprintf("%x", md5.Sum(w.buf[0:n])); e != a { t.Errorf("expect %s md5 value, got %s", e, a) } }	1.8 KB	2025-05-25 12:27
📄 integ_upload_test.go //go:build integration // +build integration package s3manager_test import ( "bytes" "crypto/md5" "fmt" "regexp" "strings" "testing" "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/service/s3" "github.com/aws/aws-sdk-go/service/s3/s3manager" ) var integBuf12MB = make([]byte, 1024102412) var integMD512MB = fmt.Sprintf("%x", md5.Sum(integBuf12MB)) func TestUploadConcurrently(t testing.T) { key := "12mb-1" mgr := s3manager.NewUploader(integSess) out, err := mgr.Upload(&s3manager.UploadInput{ Bucket: bucketName, Key: &key, Body: bytes.NewReader(integBuf12MB), }) if err != nil { t.Fatalf("expect no error, got %v", err) } if len(out.UploadID) == 0 { t.Errorf("expect upload ID but was empty") } re := regexp.MustCompile(`^https?://.+/` + key + `$`) if e, a := re.String(), out.Location; !re.MatchString(a) { t.Errorf("expect %s to match URL regexp %q, did not", e, a) } validate(t, key, integMD512MB) } func TestUploadFailCleanup(t testing.T) { // Break checksum on 2nd part so it fails part := 0 svc.Handlers.Build.PushBack(func(r request.Request) { if r.Operation.Name == "UploadPart" { if part == 1 { r.HTTPRequest.Header.Set("X-Amz-Content-Sha256", "000") } part++ } }) key := "12mb-leave" mgr := s3manager.NewUploaderWithClient(svc, func(u s3manager.Uploader) { u.LeavePartsOnError = false }) _, err := mgr.Upload(&s3manager.UploadInput{ Bucket: bucketName, Key: &key, Body: bytes.NewReader(integBuf12MB), }) if err == nil { t.Fatalf("expect error, but did not get one") } aerr := err.(awserr.Error) if e, a := "MissingRegion", aerr.Code(); strings.Contains(a, e) { t.Errorf("expect %q to not be in error code %q", e, a) } uploadID := "" merr := err.(s3manager.MultiUploadFailure) if uploadID = merr.UploadID(); len(uploadID) == 0 { t.Errorf("expect upload ID to not be empty, but was") } _, err = svc.ListParts(&s3.ListPartsInput{ Bucket: bucketName, Key: &key, UploadId: &uploadID, }) if err == nil { t.Errorf("expect error for list parts, but got none") } }	2.1 KB	2025-05-25 12:27
📄 pool.go package s3manager import ( "fmt" "sync" "github.com/aws/aws-sdk-go/aws" ) type byteSlicePool interface { Get(aws.Context) ([]byte, error) Put([]byte) ModifyCapacity(int) SliceSize() int64 Close() } type maxSlicePool struct { // allocator is defined as a function pointer to allow // for test cases to instrument custom tracers when allocations // occur. allocator sliceAllocator slices chan []byte allocations chan struct{} capacityChange chan struct{} max int sliceSize int64 mtx sync.RWMutex } func newMaxSlicePool(sliceSize int64) maxSlicePool { p := &maxSlicePool{sliceSize: sliceSize} p.allocator = p.newSlice return p } var errZeroCapacity = fmt.Errorf("get called on zero capacity pool") func (p maxSlicePool) Get(ctx aws.Context) ([]byte, error) { // check if context is canceled before attempting to get a slice // this ensures priority is given to the cancel case first select { case <-ctx.Done(): return nil, ctx.Err() default: } p.mtx.RLock() for { select { case bs, ok := <-p.slices: p.mtx.RUnlock() if !ok { // attempt to get on a zero capacity pool return nil, errZeroCapacity } return bs, nil case <-ctx.Done(): p.mtx.RUnlock() return nil, ctx.Err() default: // pass } select { case _, ok := <-p.allocations: p.mtx.RUnlock() if !ok { // attempt to get on a zero capacity pool return nil, errZeroCapacity } return p.allocator(), nil case <-ctx.Done(): p.mtx.RUnlock() return nil, ctx.Err() default: // In the event that there are no slices or allocations available // This prevents some deadlock situations that can occur around sync.RWMutex // When a lock request occurs on ModifyCapacity, no new readers are allowed to acquire a read lock. // By releasing the read lock here and waiting for a notification, we prevent a deadlock situation where // Get could hold the read lock indefinitely waiting for capacity, ModifyCapacity is waiting for a write lock, // and a Put is blocked trying to get a read-lock which is blocked by ModifyCapacity. // Short-circuit if the pool capacity is zero. if p.max == 0 { p.mtx.RUnlock() return nil, errZeroCapacity } // Since we will be releasing the read-lock we need to take the reference to the channel. // Since channels are references we will still get notified if slices are added, or if // the channel is closed due to a capacity modification. This specifically avoids a data race condition // where ModifyCapacity both closes a channel and initializes a new one while we don't have a read-lock. c := p.capacityChange p.mtx.RUnlock() select { case _ = <-c: p.mtx.RLock() case <-ctx.Done(): return nil, ctx.Err() } } } } func (p maxSlicePool) Put(bs []byte) { p.mtx.RLock() defer p.mtx.RUnlock() if p.max == 0 { return } select { case p.slices <- bs: p.notifyCapacity() default: // If the new channel when attempting to add the slice then we drop the slice. // The logic here is to prevent a deadlock situation if channel is already at max capacity. // Allows us to reap allocations that are returned and are no longer needed. } } func (p maxSlicePool) ModifyCapacity(delta int) { if delta == 0 { return } p.mtx.Lock() defer p.mtx.Unlock() p.max += delta if p.max == 0 { p.empty() return } if p.capacityChange != nil { close(p.capacityChange) } p.capacityChange = make(chan struct{}, p.max) origAllocations := p.allocations p.allocations = make(chan struct{}, p.max) newAllocs := len(origAllocations) + delta for i := 0; i < newAllocs; i++ { p.allocations <- struct{}{} } if origAllocations != nil { close(origAllocations) } origSlices := p.slices p.slices = make(chan []byte, p.max) if origSlices == nil { return } close(origSlices) for bs := range origSlices { select { case p.slices <- bs: default: // If the new channel blocks while adding slices from the old channel // then we drop the slice. The logic here is to prevent a deadlock situation // if the new channel has a smaller capacity then the old. } } } func (p maxSlicePool) notifyCapacity() { select { case p.capacityChange <- struct{}{}: default: // This shouldn't* happen as the channel is both buffered to the max pool capacity size and is resized // on capacity modifications. This is just a safety to ensure that a blocking situation can't occur. } } func (p maxSlicePool) SliceSize() int64 { return p.sliceSize } func (p maxSlicePool) Close() { p.mtx.Lock() defer p.mtx.Unlock() p.empty() } func (p maxSlicePool) empty() { p.max = 0 if p.capacityChange != nil { close(p.capacityChange) p.capacityChange = nil } if p.allocations != nil { close(p.allocations) for range p.allocations { // drain channel } p.allocations = nil } if p.slices != nil { close(p.slices) for range p.slices { // drain channel } p.slices = nil } } func (p maxSlicePool) newSlice() []byte { bs := make([]byte, p.sliceSize) return &bs } type returnCapacityPoolCloser struct { byteSlicePool returnCapacity int } func (n returnCapacityPoolCloser) ModifyCapacity(delta int) { if delta > 0 { n.returnCapacity = -1 * delta } n.byteSlicePool.ModifyCapacity(delta) } func (n returnCapacityPoolCloser) Close() { if n.returnCapacity < 0 { n.byteSlicePool.ModifyCapacity(n.returnCapacity) } } type sliceAllocator func() []byte var newByteSlicePool = func(sliceSize int64) byteSlicePool { return newMaxSlicePool(sliceSize) }	5.47 KB	2025-05-25 12:27
📄 pool_test.go //go:build go1.7 // +build go1.7 package s3manager import ( "context" "sync" "sync/atomic" "testing" "github.com/aws/aws-sdk-go/aws" ) func TestMaxSlicePool(t testing.T) { pool := newMaxSlicePool(0) var wg sync.WaitGroup for i := 0; i < 100; i++ { wg.Add(1) go func() { defer wg.Done() // increase pool capacity by 2 pool.ModifyCapacity(2) // remove 2 items bsOne, err := pool.Get(context.Background()) if err != nil { t.Errorf("failed to get slice from pool: %v", err) } bsTwo, err := pool.Get(context.Background()) if err != nil { t.Errorf("failed to get slice from pool: %v", err) } done := make(chan struct{}) go func() { defer close(done) // attempt to remove a 3rd in parallel bs, err := pool.Get(context.Background()) if err != nil { t.Errorf("failed to get slice from pool: %v", err) } pool.Put(bs) // attempt to remove a 4th that has been canceled ctx, cancel := context.WithCancel(context.Background()) cancel() bs, err = pool.Get(ctx) if err == nil { pool.Put(bs) t.Errorf("expected no slice to be returned") return } }() pool.Put(bsOne) <-done pool.ModifyCapacity(-1) pool.Put(bsTwo) pool.ModifyCapacity(-1) // any excess returns should drop rando := make([]byte, 0) pool.Put(&rando) }() } wg.Wait() if e, a := 0, len(pool.slices); e != a { t.Errorf("expected %v, got %v", e, a) } if e, a := 0, len(pool.allocations); e != a { t.Errorf("expected %v, got %v", e, a) } if e, a := 0, pool.max; e != a { t.Errorf("expected %v, got %v", e, a) } _, err := pool.Get(context.Background()) if err == nil { t.Errorf("expected error on zero capacity pool") } pool.Close() } func TestPoolShouldPreferAllocatedSlicesOverNewAllocations(t testing.T) { pool := newMaxSlicePool(0) defer pool.Close() // Prepare pool: make it so that pool contains 1 allocated slice and 1 allocation permit pool.ModifyCapacity(2) initialSlice, err := pool.Get(context.Background()) if err != nil { t.Errorf("failed to get slice from pool: %v", err) } pool.Put(initialSlice) for i := 0; i < 100; i++ { newSlice, err := pool.Get(context.Background()) if err != nil { t.Errorf("failed to get slice from pool: %v", err) return } if newSlice != initialSlice { t.Errorf("pool allocated a new slice despite it having pre-allocated one") return } pool.Put(newSlice) } } type recordedPartPool struct { recordedAllocs uint64 recordedGets uint64 recordedOutstanding int64 maxSlicePool } func newRecordedPartPool(sliceSize int64) recordedPartPool { sp := newMaxSlicePool(sliceSize) rp := &recordedPartPool{} allocator := sp.allocator sp.allocator = func() []byte { atomic.AddUint64(&rp.recordedAllocs, 1) return allocator() } rp.maxSlicePool = sp return rp } func (r recordedPartPool) Get(ctx aws.Context) ([]byte, error) { atomic.AddUint64(&r.recordedGets, 1) atomic.AddInt64(&r.recordedOutstanding, 1) return r.maxSlicePool.Get(ctx) } func (r recordedPartPool) Put(b []byte) { atomic.AddInt64(&r.recordedOutstanding, -1) r.maxSlicePool.Put(b) } func swapByteSlicePool(f func(sliceSize int64) byteSlicePool) func() { orig := newByteSlicePool newByteSlicePool = f return func() { newByteSlicePool = orig } } type syncSlicePool struct { sync.Pool sliceSize int64 } func newSyncSlicePool(sliceSize int64) syncSlicePool { p := &syncSlicePool{sliceSize: sliceSize} p.New = func() interface{} { bs := make([]byte, p.sliceSize) return &bs } return p } func (s syncSlicePool) Get(ctx aws.Context) ([]byte, error) { select { case <-ctx.Done(): return nil, ctx.Err() default: return s.Pool.Get().([]byte), nil } } func (s syncSlicePool) Put(bs []byte) { s.Pool.Put(bs) } func (s syncSlicePool) ModifyCapacity(_ int) { return } func (s syncSlicePool) SliceSize() int64 { return s.sliceSize } func (s syncSlicePool) Close() { return }	3.92 KB	2025-05-25 12:27
📄 read_seeker_write_to.go package s3manager import ( "io" "sync" ) // ReadSeekerWriteTo defines an interface implementing io.WriteTo and io.ReadSeeker type ReadSeekerWriteTo interface { io.ReadSeeker io.WriterTo } // BufferedReadSeekerWriteTo wraps a BufferedReadSeeker with an io.WriteAt // implementation. type BufferedReadSeekerWriteTo struct { BufferedReadSeeker } // WriteTo writes to the given io.Writer from BufferedReadSeeker until there's no more data to write or // an error occurs. Returns the number of bytes written and any error encountered during the write. func (b BufferedReadSeekerWriteTo) WriteTo(writer io.Writer) (int64, error) { return io.Copy(writer, b.BufferedReadSeeker) } // ReadSeekerWriteToProvider provides an implementation of io.WriteTo for an io.ReadSeeker type ReadSeekerWriteToProvider interface { GetWriteTo(seeker io.ReadSeeker) (r ReadSeekerWriteTo, cleanup func()) } // BufferedReadSeekerWriteToPool uses a sync.Pool to create and reuse // []byte slices for buffering parts in memory type BufferedReadSeekerWriteToPool struct { pool sync.Pool } // NewBufferedReadSeekerWriteToPool will return a new BufferedReadSeekerWriteToPool that will create // a pool of reusable buffers . If size is less then < 64 KiB then the buffer // will default to 64 KiB. Reason: io.Copy from writers or readers that don't support io.WriteTo or io.ReadFrom // respectively will default to copying 32 KiB. func NewBufferedReadSeekerWriteToPool(size int) BufferedReadSeekerWriteToPool { if size < 65536 { size = 65536 } return &BufferedReadSeekerWriteToPool{ pool: sync.Pool{New: func() interface{} { return make([]byte, size) }}, } } // GetWriteTo will wrap the provided io.ReadSeeker with a BufferedReadSeekerWriteTo. // The provided cleanup must be called after operations have been completed on the // returned io.ReadSeekerWriteTo in order to signal the return of resources to the pool. func (p BufferedReadSeekerWriteToPool) GetWriteTo(seeker io.ReadSeeker) (r ReadSeekerWriteTo, cleanup func()) { buffer := p.pool.Get().([]byte) r = &BufferedReadSeekerWriteTo{BufferedReadSeeker: NewBufferedReadSeeker(seeker, buffer)} cleanup = func() { p.pool.Put(buffer) } return r, cleanup }	2.17 KB	2025-05-25 12:27
📁 s3manageriface	-	2025-05-25 12:27
📄 shared_test.go package s3manager_test var buf12MB = make([]byte, 1024102412) var buf2MB = make([]byte, 102410242)	104 B	2025-05-25 12:27
📄 upload.go package s3manager import ( "bytes" "fmt" "io" "sort" "sync" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/aws/awsutil" "github.com/aws/aws-sdk-go/aws/client" "github.com/aws/aws-sdk-go/aws/credentials" "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/service/s3" "github.com/aws/aws-sdk-go/service/s3/s3iface" ) // MaxUploadParts is the maximum allowed number of parts in a multi-part upload // on Amazon S3. const MaxUploadParts = 10000 // MinUploadPartSize is the minimum allowed part size when uploading a part to // Amazon S3. const MinUploadPartSize int64 = 1024 * 1024 * 5 // DefaultUploadPartSize is the default part size to buffer chunks of a // payload into. const DefaultUploadPartSize = MinUploadPartSize // DefaultUploadConcurrency is the default number of goroutines to spin up when // using Upload(). const DefaultUploadConcurrency = 5 // A MultiUploadFailure wraps a failed S3 multipart upload. An error returned // will satisfy this interface when a multi part upload failed to upload all // chucks to S3. In the case of a failure the UploadID is needed to operate on // the chunks, if any, which were uploaded. // // Example: // // u := s3manager.NewUploader(opts) // output, err := u.upload(input) // if err != nil { // if multierr, ok := err.(s3manager.MultiUploadFailure); ok { // // Process error and its associated uploadID // fmt.Println("Error:", multierr.Code(), multierr.Message(), multierr.UploadID()) // } else { // // Process error generically // fmt.Println("Error:", err.Error()) // } // } type MultiUploadFailure interface { awserr.Error // Returns the upload id for the S3 multipart upload that failed. UploadID() string } // So that the Error interface type can be included as an anonymous field // in the multiUploadError struct and not conflict with the error.Error() method. type awsError awserr.Error // A multiUploadError wraps the upload ID of a failed s3 multipart upload. // Composed of BaseError for code, message, and original error // // Should be used for an error that occurred failing a S3 multipart upload, // and a upload ID is available. If an uploadID is not available a more relevant type multiUploadError struct { awsError // ID for multipart upload which failed. uploadID string } // Error returns the string representation of the error. // // # See apierr.BaseError ErrorWithExtra for output format // // Satisfies the error interface. func (m multiUploadError) Error() string { extra := fmt.Sprintf("upload id: %s", m.uploadID) return awserr.SprintError(m.Code(), m.Message(), extra, m.OrigErr()) } // String returns the string representation of the error. // Alias for Error to satisfy the stringer interface. func (m multiUploadError) String() string { return m.Error() } // UploadID returns the id of the S3 upload which failed. func (m multiUploadError) UploadID() string { return m.uploadID } // UploadOutput represents a response from the Upload() call. type UploadOutput struct { // The URL where the object was uploaded to. Location string // The version of the object that was uploaded. Will only be populated if // the S3 Bucket is versioned. If the bucket is not versioned this field // will not be set. VersionID string // The ID for a multipart upload to S3. In the case of an error the error // can be cast to the MultiUploadFailure interface to extract the upload ID. UploadID string // Entity tag of the object. ETag string } // WithUploaderRequestOptions appends to the Uploader's API request options. func WithUploaderRequestOptions(opts ...request.Option) func(Uploader) { return func(u Uploader) { u.RequestOptions = append(u.RequestOptions, opts...) } } // The Uploader structure that calls Upload(). It is safe to call Upload() // on this structure for multiple objects and across concurrent goroutines. // Mutating the Uploader's properties is not safe to be done concurrently. // // The ContentMD5 member for pre-computed MD5 checksums will be ignored for // multipart uploads. Objects that will be uploaded in a single part, the // ContentMD5 will be used. // // The Checksum members for pre-computed checksums will be ignored for // multipart uploads. Objects that will be uploaded in a single part, will // include the checksum member in the request. type Uploader struct { // The buffer size (in bytes) to use when buffering data into chunks and // sending them as parts to S3. The minimum allowed part size is 5MB, and // if this value is set to zero, the DefaultUploadPartSize value will be used. PartSize int64 // The number of goroutines to spin up in parallel per call to Upload when // sending parts. If this is set to zero, the DefaultUploadConcurrency value // will be used. // // The concurrency pool is not shared between calls to Upload. Concurrency int // Setting this value to true will cause the SDK to avoid calling // AbortMultipartUpload on a failure, leaving all successfully uploaded // parts on S3 for manual recovery. // // Note that storing parts of an incomplete multipart upload counts towards // space usage on S3 and will add additional costs if not cleaned up. LeavePartsOnError bool // MaxUploadParts is the max number of parts which will be uploaded to S3. // Will be used to calculate the partsize of the object to be uploaded. // E.g: 5GB file, with MaxUploadParts set to 100, will upload the file // as 100, 50MB parts. With a limited of s3.MaxUploadParts (10,000 parts). // // MaxUploadParts must not be used to limit the total number of bytes uploaded. // Use a type like to io.LimitReader (https://golang.org/pkg/io/#LimitedReader) // instead. An io.LimitReader is helpful when uploading an unbounded reader // to S3, and you know its maximum size. Otherwise the reader's io.EOF returned // error must be used to signal end of stream. // // Defaults to package const's MaxUploadParts value. MaxUploadParts int // The client to use when uploading to S3. S3 s3iface.S3API // List of request options that will be passed down to individual API // operation requests made by the uploader. RequestOptions []request.Option // Defines the buffer strategy used when uploading a part BufferProvider ReadSeekerWriteToProvider // partPool allows for the re-usage of streaming payload part buffers between upload calls partPool byteSlicePool } // NewUploader creates a new Uploader instance to upload objects to S3. Pass In // additional functional options to customize the uploader's behavior. Requires a // client.ConfigProvider in order to create a S3 service client. The session.Session // satisfies the client.ConfigProvider interface. // // Example: // // // The session the S3 Uploader will use // sess := session.Must(session.NewSession()) // // // Create an uploader with the session and default options // uploader := s3manager.NewUploader(sess) // // // Create an uploader with the session and custom options // uploader := s3manager.NewUploader(session, func(u s3manager.Uploader) { // u.PartSize = 64 1024 * 1024 // 64MB per part // }) func NewUploader(c client.ConfigProvider, options ...func(Uploader)) Uploader { return newUploader(s3.New(c), options...) } func newUploader(client s3iface.S3API, options ...func(Uploader)) Uploader { u := &Uploader{ S3: client, PartSize: DefaultUploadPartSize, Concurrency: DefaultUploadConcurrency, LeavePartsOnError: false, MaxUploadParts: MaxUploadParts, BufferProvider: defaultUploadBufferProvider(), } for _, option := range options { option(u) } u.partPool = newByteSlicePool(u.PartSize) return u } // NewUploaderWithClient creates a new Uploader instance to upload objects to S3. Pass in // additional functional options to customize the uploader's behavior. Requires // a S3 service client to make S3 API calls. // // Example: // // // The session the S3 Uploader will use // sess := session.Must(session.NewSession()) // // // S3 service client the Upload manager will use. // s3Svc := s3.New(sess) // // // Create an uploader with S3 client and default options // uploader := s3manager.NewUploaderWithClient(s3Svc) // // // Create an uploader with S3 client and custom options // uploader := s3manager.NewUploaderWithClient(s3Svc, func(u s3manager.Uploader) { // u.PartSize = 64 1024 * 1024 // 64MB per part // }) func NewUploaderWithClient(svc s3iface.S3API, options ...func(Uploader)) Uploader { return newUploader(svc, options...) } // Upload uploads an object to S3, intelligently buffering large files into // smaller chunks and sending them in parallel across multiple goroutines. You // can configure the buffer size and concurrency through the Uploader's parameters. // // Additional functional options can be provided to configure the individual // upload. These options are copies of the Uploader instance Upload is called from. // Modifying the options will not impact the original Uploader instance. // // Use the WithUploaderRequestOptions helper function to pass in request // options that will be applied to all API operations made with this uploader. // // It is safe to call this method concurrently across goroutines. // // Example: // // // Upload input parameters // upParams := &s3manager.UploadInput{ // Bucket: &bucketName, // Key: &keyName, // Body: file, // } // // // Perform an upload. // result, err := uploader.Upload(upParams) // // // Perform upload with options different than the those in the Uploader. // result, err := uploader.Upload(upParams, func(u s3manager.Uploader) { // u.PartSize = 10 1024 * 1024 // 10MB part size // u.LeavePartsOnError = true // Don't delete the parts if the upload fails. // }) func (u Uploader) Upload(input UploadInput, options ...func(Uploader)) (UploadOutput, error) { return u.UploadWithContext(aws.BackgroundContext(), input, options...) } // UploadWithContext uploads an object to S3, intelligently buffering large // files into smaller chunks and sending them in parallel across multiple // goroutines. You can configure the buffer size and concurrency through the // Uploader's parameters. // // UploadWithContext is the same as Upload with the additional support for // Context input parameters. The Context must not be nil. A nil Context will // cause a panic. Use the context to add deadlining, timeouts, etc. The // UploadWithContext may create sub-contexts for individual underlying requests. // // Additional functional options can be provided to configure the individual // upload. These options are copies of the Uploader instance Upload is called from. // Modifying the options will not impact the original Uploader instance. // // Use the WithUploaderRequestOptions helper function to pass in request // options that will be applied to all API operations made with this uploader. // // It is safe to call this method concurrently across goroutines. func (u Uploader) UploadWithContext(ctx aws.Context, input UploadInput, opts ...func(Uploader)) (UploadOutput, error) { i := uploader{in: input, cfg: u, ctx: ctx} for _, opt := range opts { opt(&i.cfg) } i.cfg.RequestOptions = append(i.cfg.RequestOptions, request.WithAppendUserAgent("S3Manager")) return i.upload() } // UploadWithIterator will upload a batched amount of objects to S3. This operation uses // the iterator pattern to know which object to upload next. Since this is an interface this // allows for custom defined functionality. // // Example: // // svc:= s3manager.NewUploader(sess) // // objects := []BatchUploadObject{ // { // Object: &s3manager.UploadInput { // Key: aws.String("key"), // Bucket: aws.String("bucket"), // }, // }, // } // // iter := &s3manager.UploadObjectsIterator{Objects: objects} // if err := svc.UploadWithIterator(aws.BackgroundContext(), iter); err != nil { // return err // } func (u Uploader) UploadWithIterator(ctx aws.Context, iter BatchUploadIterator, opts ...func(Uploader)) error { var errs []Error for iter.Next() { object := iter.UploadObject() if _, err := u.UploadWithContext(ctx, object.Object, opts...); err != nil { s3Err := Error{ OrigErr: err, Bucket: object.Object.Bucket, Key: object.Object.Key, } errs = append(errs, s3Err) } if object.After == nil { continue } if err := object.After(); err != nil { s3Err := Error{ OrigErr: err, Bucket: object.Object.Bucket, Key: object.Object.Key, } errs = append(errs, s3Err) } } if len(errs) > 0 { return NewBatchError("BatchedUploadIncomplete", "some objects have failed to upload.", errs) } return nil } // internal structure to manage an upload to S3. type uploader struct { ctx aws.Context cfg Uploader in UploadInput readerPos int64 // current reader position totalSize int64 // set to -1 if the size is not known } // internal logic for deciding whether to upload a single part or use a // multipart upload. func (u uploader) upload() (UploadOutput, error) { if err := u.init(); err != nil { return nil, awserr.New("ReadRequestBody", "unable to initialize upload", err) } defer u.cfg.partPool.Close() if u.cfg.PartSize < MinUploadPartSize { msg := fmt.Sprintf("part size must be at least %d bytes", MinUploadPartSize) return nil, awserr.New("ConfigError", msg, nil) } // Do one read to determine if we have more than one part reader, _, cleanup, err := u.nextReader() if err == io.EOF { // single part return u.singlePart(reader, cleanup) } else if err != nil { cleanup() return nil, awserr.New("ReadRequestBody", "read upload data failed", err) } mu := multiuploader{uploader: u} return mu.upload(reader, cleanup) } // init will initialize all default options. func (u uploader) init() error { if err := validateSupportedARNType(aws.StringValue(u.in.Bucket)); err != nil { return err } if u.cfg.Concurrency == 0 { u.cfg.Concurrency = DefaultUploadConcurrency } if u.cfg.PartSize == 0 { u.cfg.PartSize = DefaultUploadPartSize } if u.cfg.MaxUploadParts == 0 { u.cfg.MaxUploadParts = MaxUploadParts } // Try to get the total size for some optimizations if err := u.initSize(); err != nil { return err } // If PartSize was changed or partPool was never setup then we need to allocated a new pool // so that we return []byte slices of the correct size poolCap := u.cfg.Concurrency + 1 if u.cfg.partPool == nil \|\| u.cfg.partPool.SliceSize() != u.cfg.PartSize { u.cfg.partPool = newByteSlicePool(u.cfg.PartSize) u.cfg.partPool.ModifyCapacity(poolCap) } else { u.cfg.partPool = &returnCapacityPoolCloser{byteSlicePool: u.cfg.partPool} u.cfg.partPool.ModifyCapacity(poolCap) } return nil } // initSize tries to detect the total stream size, setting u.totalSize. If // the size is not known, totalSize is set to -1. func (u uploader) initSize() error { u.totalSize = -1 switch r := u.in.Body.(type) { case io.Seeker: n, err := aws.SeekerLen(r) if err != nil { return err } u.totalSize = n // Try to adjust partSize if it is too small and account for // integer division truncation. if u.totalSize/u.cfg.PartSize >= int64(u.cfg.MaxUploadParts) { // Add one to the part size to account for remainders // during the size calculation. e.g odd number of bytes. u.cfg.PartSize = (u.totalSize / int64(u.cfg.MaxUploadParts)) + 1 } } return nil } // nextReader returns a seekable reader representing the next packet of data. // This operation increases the shared u.readerPos counter, but note that it // does not need to be wrapped in a mutex because nextReader is only called // from the main thread. func (u uploader) nextReader() (io.ReadSeeker, int, func(), error) { switch r := u.in.Body.(type) { case readerAtSeeker: var err error n := u.cfg.PartSize if u.totalSize >= 0 { bytesLeft := u.totalSize - u.readerPos if bytesLeft <= u.cfg.PartSize { err = io.EOF n = bytesLeft } } var ( reader io.ReadSeeker cleanup func() ) reader = io.NewSectionReader(r, u.readerPos, n) if u.cfg.BufferProvider != nil { reader, cleanup = u.cfg.BufferProvider.GetWriteTo(reader) } else { cleanup = func() {} } u.readerPos += n return reader, int(n), cleanup, err default: part, err := u.cfg.partPool.Get(u.ctx) if err != nil { return nil, 0, func() {}, err } n, err := readFillBuf(r, part) u.readerPos += int64(n) cleanup := func() { u.cfg.partPool.Put(part) } return bytes.NewReader((part)[0:n]), n, cleanup, err } } func readFillBuf(r io.Reader, b []byte) (offset int, err error) { for offset < len(b) && err == nil { var n int n, err = r.Read(b[offset:]) offset += n } return offset, err } // singlePart contains upload logic for uploading a single chunk via // a regular PutObject request. Multipart requests require at least two // parts, or at least 5MB of data. func (u uploader) singlePart(r io.ReadSeeker, cleanup func()) (UploadOutput, error) { defer cleanup() params := &s3.PutObjectInput{} awsutil.Copy(params, u.in) params.Body = r // Need to use request form because URL generated in request is // used in return. req, out := u.cfg.S3.PutObjectRequest(params) req.SetContext(u.ctx) req.ApplyOptions(u.cfg.RequestOptions...) if err := req.Send(); err != nil { return nil, err } url := req.HTTPRequest.URL.String() return &UploadOutput{ Location: url, VersionID: out.VersionId, ETag: out.ETag, }, nil } // internal structure to manage a specific multipart upload to S3. type multiuploader struct { uploader wg sync.WaitGroup m sync.Mutex err error uploadID string parts completedParts } // keeps track of a single chunk of data being sent to S3. type chunk struct { buf io.ReadSeeker num int64 cleanup func() } // completedParts is a wrapper to make parts sortable by their part number, // since S3 required this list to be sent in sorted order. type completedParts []s3.CompletedPart func (a completedParts) Len() int { return len(a) } func (a completedParts) Swap(i, j int) { a[i], a[j] = a[j], a[i] } func (a completedParts) Less(i, j int) bool { return a[i].PartNumber < a[j].PartNumber } // upload will perform a multipart upload using the firstBuf buffer containing // the first chunk of data. func (u multiuploader) upload(firstBuf io.ReadSeeker, cleanup func()) (UploadOutput, error) { params := &s3.CreateMultipartUploadInput{} awsutil.Copy(params, u.in) // Create the multipart resp, err := u.cfg.S3.CreateMultipartUploadWithContext(u.ctx, params, u.cfg.RequestOptions...) if err != nil { cleanup() return nil, err } u.uploadID = resp.UploadId // Create the workers ch := make(chan chunk, u.cfg.Concurrency) for i := 0; i < u.cfg.Concurrency; i++ { u.wg.Add(1) go u.readChunk(ch) } // Send part 1 to the workers var num int64 = 1 ch <- chunk{buf: firstBuf, num: num, cleanup: cleanup} // Read and queue the rest of the parts for u.geterr() == nil && err == nil { var ( reader io.ReadSeeker nextChunkLen int ok bool ) reader, nextChunkLen, cleanup, err = u.nextReader() ok, err = u.shouldContinue(num, nextChunkLen, err) if !ok { cleanup() if err != nil { u.seterr(err) } break } num++ ch <- chunk{buf: reader, num: num, cleanup: cleanup} } // Close the channel, wait for workers, and complete upload close(ch) u.wg.Wait() complete := u.complete() if err := u.geterr(); err != nil { return nil, &multiUploadError{ awsError: awserr.New( "MultipartUpload", "upload multipart failed", err), uploadID: u.uploadID, } } // Create a presigned URL of the S3 Get Object in order to have parity with // single part upload. getReq, _ := u.cfg.S3.GetObjectRequest(&s3.GetObjectInput{ Bucket: u.in.Bucket, Key: u.in.Key, }) getReq.Config.Credentials = credentials.AnonymousCredentials getReq.SetContext(u.ctx) uploadLocation, _, _ := getReq.PresignRequest(1) return &UploadOutput{ Location: uploadLocation, VersionID: complete.VersionId, UploadID: u.uploadID, ETag: complete.ETag, }, nil } func (u multiuploader) shouldContinue(part int64, nextChunkLen int, err error) (bool, error) { if err != nil && err != io.EOF { return false, awserr.New("ReadRequestBody", "read multipart upload data failed", err) } if nextChunkLen == 0 { // No need to upload empty part, if file was empty to start // with empty single part would of been created and never // started multipart upload. return false, nil } part++ // This upload exceeded maximum number of supported parts, error now. if part > int64(u.cfg.MaxUploadParts) \|\| part > int64(MaxUploadParts) { var msg string if part > int64(u.cfg.MaxUploadParts) { msg = fmt.Sprintf("exceeded total allowed configured MaxUploadParts (%d). Adjust PartSize to fit in this limit", u.cfg.MaxUploadParts) } else { msg = fmt.Sprintf("exceeded total allowed S3 limit MaxUploadParts (%d). Adjust PartSize to fit in this limit", MaxUploadParts) } return false, awserr.New("TotalPartsExceeded", msg, nil) } return true, err } // readChunk runs in worker goroutines to pull chunks off of the ch channel // and send() them as UploadPart requests. func (u multiuploader) readChunk(ch chan chunk) { defer u.wg.Done() for { data, ok := <-ch if !ok { break } if u.geterr() == nil { if err := u.send(data); err != nil { u.seterr(err) } } data.cleanup() } } // send performs an UploadPart request and keeps track of the completed // part information. func (u multiuploader) send(c chunk) error { params := &s3.UploadPartInput{ Bucket: u.in.Bucket, Key: u.in.Key, Body: c.buf, UploadId: &u.uploadID, SSECustomerAlgorithm: u.in.SSECustomerAlgorithm, SSECustomerKey: u.in.SSECustomerKey, PartNumber: &c.num, } resp, err := u.cfg.S3.UploadPartWithContext(u.ctx, params, u.cfg.RequestOptions...) if err != nil { return err } n := c.num completed := &s3.CompletedPart{ETag: resp.ETag, PartNumber: &n} u.m.Lock() u.parts = append(u.parts, completed) u.m.Unlock() return nil } // geterr is a thread-safe getter for the error object func (u multiuploader) geterr() error { u.m.Lock() defer u.m.Unlock() return u.err } // seterr is a thread-safe setter for the error object func (u multiuploader) seterr(e error) { u.m.Lock() defer u.m.Unlock() u.err = e } // fail will abort the multipart unless LeavePartsOnError is set to true. func (u multiuploader) fail() { if u.cfg.LeavePartsOnError { return } params := &s3.AbortMultipartUploadInput{ Bucket: u.in.Bucket, Key: u.in.Key, UploadId: &u.uploadID, } _, err := u.cfg.S3.AbortMultipartUploadWithContext(u.ctx, params, u.cfg.RequestOptions...) if err != nil { logMessage(u.cfg.S3, aws.LogDebug, fmt.Sprintf("failed to abort multipart upload, %v", err)) } } // complete successfully completes a multipart upload and returns the response. func (u multiuploader) complete() s3.CompleteMultipartUploadOutput { if u.geterr() != nil { u.fail() return nil } // Parts must be sorted in PartNumber order. sort.Sort(u.parts) params := &s3.CompleteMultipartUploadInput{ Bucket: u.in.Bucket, Key: u.in.Key, UploadId: &u.uploadID, MultipartUpload: &s3.CompletedMultipartUpload{Parts: u.parts}, } resp, err := u.cfg.S3.CompleteMultipartUploadWithContext(u.ctx, params, u.cfg.RequestOptions...) if err != nil { u.seterr(err) u.fail() } return resp } type readerAtSeeker interface { io.ReaderAt io.ReadSeeker }	23.3 KB	2025-05-25 12:27
📄 upload_input.go // Code generated by private/model/cli/gen-api/main.go. DO NOT EDIT. package s3manager import ( "io" "time" ) // UploadInput provides the input parameters for uploading a stream or buffer // to an object in an Amazon S3 bucket. This type is similar to the s3 // package's PutObjectInput with the exception that the Body member is an // io.Reader instead of an io.ReadSeeker. // // The ContentMD5 member for pre-computed MD5 checksums will be ignored for // multipart uploads. Objects that will be uploaded in a single part, the // ContentMD5 will be used. // // The Checksum members for pre-computed checksums will be ignored for // multipart uploads. Objects that will be uploaded in a single part, will // include the checksum member in the request. type UploadInput struct { _ struct{} `locationName:"PutObjectRequest" type:"structure" payload:"Body"` // The canned ACL to apply to the object. For more information, see Canned ACL // (https://docs.aws.amazon.com/AmazonS3/latest/dev/acl-overview.html#CannedACL) // in the Amazon S3 User Guide. // // When adding a new object, you can use headers to grant ACL-based permissions // to individual Amazon Web Services accounts or to predefined groups defined // by Amazon S3. These permissions are then added to the ACL on the object. // By default, all objects are private. Only the owner has full access control. // For more information, see Access Control List (ACL) Overview (https://docs.aws.amazon.com/AmazonS3/latest/dev/acl-overview.html) // and Managing ACLs Using the REST API (https://docs.aws.amazon.com/AmazonS3/latest/dev/acl-using-rest-api.html) // in the Amazon S3 User Guide. // // If the bucket that you're uploading objects to uses the bucket owner enforced // setting for S3 Object Ownership, ACLs are disabled and no longer affect permissions. // Buckets that use this setting only accept PUT requests that don't specify // an ACL or PUT requests that specify bucket owner full control ACLs, such // as the bucket-owner-full-control canned ACL or an equivalent form of this // ACL expressed in the XML format. PUT requests that contain other ACLs (for // example, custom grants to certain Amazon Web Services accounts) fail and // return a 400 error with the error code AccessControlListNotSupported. For // more information, see Controlling ownership of objects and disabling ACLs // (https://docs.aws.amazon.com/AmazonS3/latest/userguide/about-object-ownership.html) // in the Amazon S3 User Guide. // // * This functionality is not supported for directory buckets. // // * This functionality is not supported for Amazon S3 on Outposts. ACL string `location:"header" locationName:"x-amz-acl" type:"string" enum:"ObjectCannedACL"` // The readable body payload to send to S3. Body io.Reader // The bucket name to which the PUT action was initiated. // // Directory buckets - When you use this operation with a directory bucket, // you must use virtual-hosted-style requests in the format Bucket_name.s3express-az_id.region.amazonaws.com. // Path-style requests are not supported. Directory bucket names must be unique // in the chosen Availability Zone. Bucket names must follow the format bucket_base_name--az-id--x-s3 // (for example, DOC-EXAMPLE-BUCKET--usw2-az1--x-s3). For information about // bucket naming restrictions, see Directory bucket naming rules (https://docs.aws.amazon.com/AmazonS3/latest/userguide/directory-bucket-naming-rules.html) // in the Amazon S3 User Guide. // // Access points - When you use this action with an access point, you must provide // the alias of the access point in place of the bucket name or specify the // access point ARN. When using the access point ARN, you must direct requests // to the access point hostname. The access point hostname takes the form AccessPointName-AccountId.s3-accesspoint.Region.amazonaws.com. // When using this action with an access point through the Amazon Web Services // SDKs, you provide the access point ARN in place of the bucket name. For more // information about access point ARNs, see Using access points (https://docs.aws.amazon.com/AmazonS3/latest/userguide/using-access-points.html) // in the Amazon S3 User Guide. // // Access points and Object Lambda access points are not supported by directory // buckets. // // S3 on Outposts - When you use this action with Amazon S3 on Outposts, you // must direct requests to the S3 on Outposts hostname. The S3 on Outposts hostname // takes the form AccessPointName-AccountId.outpostID.s3-outposts.Region.amazonaws.com. // When you use this action with S3 on Outposts through the Amazon Web Services // SDKs, you provide the Outposts access point ARN in place of the bucket name. // For more information about S3 on Outposts ARNs, see What is S3 on Outposts? // (https://docs.aws.amazon.com/AmazonS3/latest/userguide/S3onOutposts.html) // in the Amazon S3 User Guide. // // Bucket is a required field Bucket string `location:"uri" locationName:"Bucket" type:"string" required:"true"` // Specifies whether Amazon S3 should use an S3 Bucket Key for object encryption // with server-side encryption using Key Management Service (KMS) keys (SSE-KMS). // Setting this header to true causes Amazon S3 to use an S3 Bucket Key for // object encryption with SSE-KMS. // // Specifying this header with a PUT action doesn’t affect bucket-level settings // for S3 Bucket Key. // // This functionality is not supported for directory buckets. BucketKeyEnabled bool `location:"header" locationName:"x-amz-server-side-encryption-bucket-key-enabled" type:"boolean"` // Can be used to specify caching behavior along the request/reply chain. For // more information, see http://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html#sec14.9 // (http://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html#sec14.9). CacheControl string `location:"header" locationName:"Cache-Control" type:"string"` // Indicates the algorithm used to create the checksum for the object when you // use the SDK. This header will not provide any additional functionality if // you don't use the SDK. When you send this header, there must be a corresponding // x-amz-checksum-algorithm or x-amz-trailer header sent. Otherwise, Amazon // S3 fails the request with the HTTP status code 400 Bad Request. // // For the x-amz-checksum-algorithm header, replace algorithm with the supported // algorithm from the following list: // // * CRC32 // // * CRC32C // // * SHA1 // // * SHA256 // // For more information, see Checking object integrity (https://docs.aws.amazon.com/AmazonS3/latest/userguide/checking-object-integrity.html) // in the Amazon S3 User Guide. // // If the individual checksum value you provide through x-amz-checksum-algorithm // doesn't match the checksum algorithm you set through x-amz-sdk-checksum-algorithm, // Amazon S3 ignores any provided ChecksumAlgorithm parameter and uses the checksum // algorithm that matches the provided value in x-amz-checksum-algorithm . // // For directory buckets, when you use Amazon Web Services SDKs, CRC32 is the // default checksum algorithm that's used for performance. // // The AWS SDK for Go v1 does not support automatic computing request payload // checksum. This feature is available in the AWS SDK for Go v2. If a value // is specified for this parameter, the matching algorithm's checksum member // must be populated with the algorithm's checksum of the request payload. ChecksumAlgorithm string `location:"header" locationName:"x-amz-sdk-checksum-algorithm" type:"string" enum:"ChecksumAlgorithm"` // This header can be used as a data integrity check to verify that the data // received is the same data that was originally sent. This header specifies // the base64-encoded, 32-bit CRC32 checksum of the object. For more information, // see Checking object integrity (https://docs.aws.amazon.com/AmazonS3/latest/userguide/checking-object-integrity.html) // in the Amazon S3 User Guide. ChecksumCRC32 string `location:"header" locationName:"x-amz-checksum-crc32" type:"string"` // This header can be used as a data integrity check to verify that the data // received is the same data that was originally sent. This header specifies // the base64-encoded, 32-bit CRC32C checksum of the object. For more information, // see Checking object integrity (https://docs.aws.amazon.com/AmazonS3/latest/userguide/checking-object-integrity.html) // in the Amazon S3 User Guide. ChecksumCRC32C string `location:"header" locationName:"x-amz-checksum-crc32c" type:"string"` // This header can be used as a data integrity check to verify that the data // received is the same data that was originally sent. This header specifies // the base64-encoded, 160-bit SHA-1 digest of the object. For more information, // see Checking object integrity (https://docs.aws.amazon.com/AmazonS3/latest/userguide/checking-object-integrity.html) // in the Amazon S3 User Guide. ChecksumSHA1 string `location:"header" locationName:"x-amz-checksum-sha1" type:"string"` // This header can be used as a data integrity check to verify that the data // received is the same data that was originally sent. This header specifies // the base64-encoded, 256-bit SHA-256 digest of the object. For more information, // see Checking object integrity (https://docs.aws.amazon.com/AmazonS3/latest/userguide/checking-object-integrity.html) // in the Amazon S3 User Guide. ChecksumSHA256 string `location:"header" locationName:"x-amz-checksum-sha256" type:"string"` // Specifies presentational information for the object. For more information, // see https://www.rfc-editor.org/rfc/rfc6266#section-4 (https://www.rfc-editor.org/rfc/rfc6266#section-4). ContentDisposition string `location:"header" locationName:"Content-Disposition" type:"string"` // Specifies what content encodings have been applied to the object and thus // what decoding mechanisms must be applied to obtain the media-type referenced // by the Content-Type header field. For more information, see https://www.rfc-editor.org/rfc/rfc9110.html#field.content-encoding // (https://www.rfc-editor.org/rfc/rfc9110.html#field.content-encoding). ContentEncoding string `location:"header" locationName:"Content-Encoding" type:"string"` // The language the content is in. ContentLanguage string `location:"header" locationName:"Content-Language" type:"string"` // The base64-encoded 128-bit MD5 digest of the message (without the headers) // according to RFC 1864. This header can be used as a message integrity check // to verify that the data is the same data that was originally sent. Although // it is optional, we recommend using the Content-MD5 mechanism as an end-to-end // integrity check. For more information about REST request authentication, // see REST Authentication (https://docs.aws.amazon.com/AmazonS3/latest/dev/RESTAuthentication.html). // // The Content-MD5 header is required for any request to upload an object with // a retention period configured using Amazon S3 Object Lock. For more information // about Amazon S3 Object Lock, see Amazon S3 Object Lock Overview (https://docs.aws.amazon.com/AmazonS3/latest/dev/object-lock-overview.html) // in the Amazon S3 User Guide. // // This functionality is not supported for directory buckets. // // If the ContentMD5 is provided for a multipart upload, it will be ignored. // Objects that will be uploaded in a single part, the ContentMD5 will be used. ContentMD5 string `location:"header" locationName:"Content-MD5" type:"string"` // A standard MIME type describing the format of the contents. For more information, // see https://www.rfc-editor.org/rfc/rfc9110.html#name-content-type (https://www.rfc-editor.org/rfc/rfc9110.html#name-content-type). ContentType string `location:"header" locationName:"Content-Type" type:"string"` // The account ID of the expected bucket owner. If the account ID that you provide // does not match the actual owner of the bucket, the request fails with the // HTTP status code 403 Forbidden (access denied). ExpectedBucketOwner string `location:"header" locationName:"x-amz-expected-bucket-owner" type:"string"` // The date and time at which the object is no longer cacheable. For more information, // see https://www.rfc-editor.org/rfc/rfc7234#section-5.3 (https://www.rfc-editor.org/rfc/rfc7234#section-5.3). Expires time.Time `location:"header" locationName:"Expires" type:"timestamp"` // Gives the grantee READ, READ_ACP, and WRITE_ACP permissions on the object. // // * This functionality is not supported for directory buckets. // // * This functionality is not supported for Amazon S3 on Outposts. GrantFullControl string `location:"header" locationName:"x-amz-grant-full-control" type:"string"` // Allows grantee to read the object data and its metadata. // // This functionality is not supported for directory buckets. // // * This functionality is not supported for Amazon S3 on Outposts. GrantRead string `location:"header" locationName:"x-amz-grant-read" type:"string"` // Allows grantee to read the object ACL. // // This functionality is not supported for directory buckets. // // * This functionality is not supported for Amazon S3 on Outposts. GrantReadACP string `location:"header" locationName:"x-amz-grant-read-acp" type:"string"` // Allows grantee to write the ACL for the applicable object. // // This functionality is not supported for directory buckets. // // * This functionality is not supported for Amazon S3 on Outposts. GrantWriteACP string `location:"header" locationName:"x-amz-grant-write-acp" type:"string"` // Object key for which the PUT action was initiated. // // Key is a required field Key string `location:"uri" locationName:"Key" min:"1" type:"string" required:"true"` // A map of metadata to store with the object in S3. Metadata map[string]string `location:"headers" locationName:"x-amz-meta-" type:"map"` // Specifies whether a legal hold will be applied to this object. For more information // about S3 Object Lock, see Object Lock (https://docs.aws.amazon.com/AmazonS3/latest/dev/object-lock.html) // in the Amazon S3 User Guide. // // This functionality is not supported for directory buckets. ObjectLockLegalHoldStatus string `location:"header" locationName:"x-amz-object-lock-legal-hold" type:"string" enum:"ObjectLockLegalHoldStatus"` // The Object Lock mode that you want to apply to this object. // // This functionality is not supported for directory buckets. ObjectLockMode string `location:"header" locationName:"x-amz-object-lock-mode" type:"string" enum:"ObjectLockMode"` // The date and time when you want this object's Object Lock to expire. Must // be formatted as a timestamp parameter. // // This functionality is not supported for directory buckets. ObjectLockRetainUntilDate time.Time `location:"header" locationName:"x-amz-object-lock-retain-until-date" type:"timestamp" timestampFormat:"iso8601"` // Confirms that the requester knows that they will be charged for the request. // Bucket owners need not specify this parameter in their requests. If either // the source or destination S3 bucket has Requester Pays enabled, the requester // will pay for corresponding charges to copy the object. For information about // downloading objects from Requester Pays buckets, see Downloading Objects // in Requester Pays Buckets (https://docs.aws.amazon.com/AmazonS3/latest/dev/ObjectsinRequesterPaysBuckets.html) // in the Amazon S3 User Guide. // // This functionality is not supported for directory buckets. RequestPayer string `location:"header" locationName:"x-amz-request-payer" type:"string" enum:"RequestPayer"` // Specifies the algorithm to use when encrypting the object (for example, AES256). // // This functionality is not supported for directory buckets. SSECustomerAlgorithm string `location:"header" locationName:"x-amz-server-side-encryption-customer-algorithm" type:"string"` // Specifies the customer-provided encryption key for Amazon S3 to use in encrypting // data. This value is used to store the object and then it is discarded; Amazon // S3 does not store the encryption key. The key must be appropriate for use // with the algorithm specified in the x-amz-server-side-encryption-customer-algorithm // header. // // This functionality is not supported for directory buckets. SSECustomerKey string `marshal-as:"blob" location:"header" locationName:"x-amz-server-side-encryption-customer-key" type:"string" sensitive:"true"` // Specifies the 128-bit MD5 digest of the encryption key according to RFC 1321. // Amazon S3 uses this header for a message integrity check to ensure that the // encryption key was transmitted without error. // // This functionality is not supported for directory buckets. SSECustomerKeyMD5 string `location:"header" locationName:"x-amz-server-side-encryption-customer-key-MD5" type:"string"` // Specifies the Amazon Web Services KMS Encryption Context to use for object // encryption. The value of this header is a base64-encoded UTF-8 string holding // JSON with the encryption context key-value pairs. This value is stored as // object metadata and automatically gets passed on to Amazon Web Services KMS // for future GetObject or CopyObject operations on this object. This value // must be explicitly added during CopyObject operations. // // This functionality is not supported for directory buckets. SSEKMSEncryptionContext string `location:"header" locationName:"x-amz-server-side-encryption-context" type:"string" sensitive:"true"` // If x-amz-server-side-encryption has a valid value of aws:kms or aws:kms:dsse, // this header specifies the ID (Key ID, Key ARN, or Key Alias) of the Key Management // Service (KMS) symmetric encryption customer managed key that was used for // the object. If you specify x-amz-server-side-encryption:aws:kms or x-amz-server-side-encryption:aws:kms:dsse, // but do not providex-amz-server-side-encryption-aws-kms-key-id, Amazon S3 // uses the Amazon Web Services managed key (aws/s3) to protect the data. If // the KMS key does not exist in the same account that's issuing the command, // you must use the full ARN and not just the ID. // // This functionality is not supported for directory buckets. SSEKMSKeyId string `location:"header" locationName:"x-amz-server-side-encryption-aws-kms-key-id" type:"string" sensitive:"true"` // The server-side encryption algorithm that was used when you store this object // in Amazon S3 (for example, AES256, aws:kms, aws:kms:dsse). // // General purpose buckets - You have four mutually exclusive options to protect // data using server-side encryption in Amazon S3, depending on how you choose // to manage the encryption keys. Specifically, the encryption key options are // Amazon S3 managed keys (SSE-S3), Amazon Web Services KMS keys (SSE-KMS or // DSSE-KMS), and customer-provided keys (SSE-C). Amazon S3 encrypts data with // server-side encryption by using Amazon S3 managed keys (SSE-S3) by default. // You can optionally tell Amazon S3 to encrypt data at rest by using server-side // encryption with other key options. For more information, see Using Server-Side // Encryption (https://docs.aws.amazon.com/AmazonS3/latest/dev/UsingServerSideEncryption.html) // in the Amazon S3 User Guide. // // Directory buckets - For directory buckets, only the server-side encryption // with Amazon S3 managed keys (SSE-S3) (AES256) value is supported. ServerSideEncryption string `location:"header" locationName:"x-amz-server-side-encryption" type:"string" enum:"ServerSideEncryption"` // By default, Amazon S3 uses the STANDARD Storage Class to store newly created // objects. The STANDARD storage class provides high durability and high availability. // Depending on performance needs, you can specify a different Storage Class. // For more information, see Storage Classes (https://docs.aws.amazon.com/AmazonS3/latest/dev/storage-class-intro.html) // in the Amazon S3 User Guide. // // For directory buckets, only the S3 Express One Zone storage class is // supported to store newly created objects. // // * Amazon S3 on Outposts only uses the OUTPOSTS Storage Class. StorageClass string `location:"header" locationName:"x-amz-storage-class" type:"string" enum:"StorageClass"` // The tag-set for the object. The tag-set must be encoded as URL Query parameters. // (For example, "Key1=Value1") // // This functionality is not supported for directory buckets. Tagging string `location:"header" locationName:"x-amz-tagging" type:"string"` // If the bucket is configured as a website, redirects requests for this object // to another object in the same bucket or to an external URL. Amazon S3 stores // the value of this header in the object metadata. For information about object // metadata, see Object Key and Metadata (https://docs.aws.amazon.com/AmazonS3/latest/dev/UsingMetadata.html) // in the Amazon S3 User Guide. // // In the following example, the request header sets the redirect to an object // (anotherPage.html) in the same bucket: // // x-amz-website-redirect-location: /anotherPage.html // // In the following example, the request header sets the object redirect to // another website: // // x-amz-website-redirect-location: http://www.example.com/ // // For more information about website hosting in Amazon S3, see Hosting Websites // on Amazon S3 (https://docs.aws.amazon.com/AmazonS3/latest/dev/WebsiteHosting.html) // and How to Configure Website Page Redirects (https://docs.aws.amazon.com/AmazonS3/latest/dev/how-to-page-redirect.html) // in the Amazon S3 User Guide. // // This functionality is not supported for directory buckets. WebsiteRedirectLocation *string `location:"header" locationName:"x-amz-website-redirect-location" type:"string"` }	21.52 KB	2025-05-25 12:27
📄 upload_internal_test.go //go:build go1.7 // +build go1.7 package s3manager import ( "bytes" "fmt" "io" "io/ioutil" random "math/rand" "net/http" "strconv" "sync" "sync/atomic" "testing" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/awstesting/unit" "github.com/aws/aws-sdk-go/internal/sdkio" "github.com/aws/aws-sdk-go/service/s3" "github.com/aws/aws-sdk-go/service/s3/internal/s3testing" ) const respBody = `<?xml version="1.0" encoding="UTF-8"?> <CompleteMultipartUploadOutput> <Location>mockValue</Location> <Bucket>mockValue</Bucket> <Key>mockValue</Key> <ETag>mockValue</ETag> </CompleteMultipartUploadOutput>` type testReader struct { br bytes.Reader m sync.Mutex } func (r testReader) Read(p []byte) (n int, err error) { r.m.Lock() defer r.m.Unlock() return r.br.Read(p) } func TestUploadByteSlicePool(t testing.T) { cases := map[string]struct { PartSize int64 FileSize int64 Concurrency int ExAllocations uint64 }{ "single part, single concurrency": { PartSize: sdkio.MebiByte 5, FileSize: sdkio.MebiByte * 5, ExAllocations: 2, Concurrency: 1, }, "multi-part, single concurrency": { PartSize: sdkio.MebiByte * 5, FileSize: sdkio.MebiByte * 10, ExAllocations: 2, Concurrency: 1, }, "multi-part, multiple concurrency": { PartSize: sdkio.MebiByte * 5, FileSize: sdkio.MebiByte * 20, ExAllocations: 3, Concurrency: 2, }, } for name, tt := range cases { t.Run(name, func(t testing.T) { var p recordedPartPool unswap := swapByteSlicePool(func(sliceSize int64) byteSlicePool { p = newRecordedPartPool(sliceSize) return p }) defer unswap() sess := unit.Session.Copy() svc := s3.New(sess) svc.Handlers.Unmarshal.Clear() svc.Handlers.UnmarshalMeta.Clear() svc.Handlers.UnmarshalError.Clear() svc.Handlers.Send.Clear() svc.Handlers.Send.PushFront(func(r request.Request) { if r.Body != nil { io.Copy(ioutil.Discard, r.Body) } r.HTTPResponse = &http.Response{ StatusCode: 200, Body: ioutil.NopCloser(bytes.NewReader([]byte(respBody))), } switch data := r.Data.(type) { case s3.CreateMultipartUploadOutput: data.UploadId = aws.String("UPLOAD-ID") case s3.UploadPartOutput: data.ETag = aws.String(fmt.Sprintf("ETAG%d", random.Int())) case s3.CompleteMultipartUploadOutput: data.Location = aws.String("https://location") data.VersionId = aws.String("VERSION-ID") case s3.PutObjectOutput: data.VersionId = aws.String("VERSION-ID") } }) uploader := NewUploaderWithClient(svc, func(u Uploader) { u.PartSize = tt.PartSize u.Concurrency = tt.Concurrency }) expected := s3testing.GetTestBytes(int(tt.FileSize)) _, err := uploader.Upload(&UploadInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), Body: &testReader{br: bytes.NewReader(expected)}, }) if err != nil { t.Errorf("expected no error, but got %v", err) } if v := atomic.LoadInt64(&p.recordedOutstanding); v != 0 { t.Fatalf("expected zero outsnatding pool parts, got %d", v) } gets, allocs := atomic.LoadUint64(&p.recordedGets), atomic.LoadUint64(&p.recordedAllocs) t.Logf("total gets %v, total allocations %v", gets, allocs) if e, a := tt.ExAllocations, allocs; a > e { t.Errorf("expected %v allocations, got %v", e, a) } }) } } func TestUploadByteSlicePool_Failures(t testing.T) { cases := map[string]struct { PartSize int64 FileSize int64 Operations []string }{ "single part": { PartSize: sdkio.MebiByte 5, FileSize: sdkio.MebiByte * 4, Operations: []string{ "PutObject", }, }, "multi-part": { PartSize: sdkio.MebiByte * 5, FileSize: sdkio.MebiByte * 10, Operations: []string{ "CreateMultipartUpload", "UploadPart", "CompleteMultipartUpload", }, }, } for name, tt := range cases { t.Run(name, func(t testing.T) { for _, operation := range tt.Operations { t.Run(operation, func(t testing.T) { var p recordedPartPool unswap := swapByteSlicePool(func(sliceSize int64) byteSlicePool { p = newRecordedPartPool(sliceSize) return p }) defer unswap() sess := unit.Session.Copy() svc := s3.New(sess) svc.Handlers.Unmarshal.Clear() svc.Handlers.UnmarshalMeta.Clear() svc.Handlers.UnmarshalError.Clear() svc.Handlers.Send.Clear() svc.Handlers.Send.PushFront(func(r request.Request) { if r.Body != nil { io.Copy(ioutil.Discard, r.Body) } if r.Operation.Name == operation { r.Retryable = aws.Bool(false) r.Error = fmt.Errorf("request error") r.HTTPResponse = &http.Response{ StatusCode: 500, Body: ioutil.NopCloser(bytes.NewReader([]byte{})), } return } r.HTTPResponse = &http.Response{ StatusCode: 200, Body: ioutil.NopCloser(bytes.NewReader([]byte(respBody))), } switch data := r.Data.(type) { case s3.CreateMultipartUploadOutput: data.UploadId = aws.String("UPLOAD-ID") case s3.UploadPartOutput: data.ETag = aws.String(fmt.Sprintf("ETAG%d", random.Int())) case s3.CompleteMultipartUploadOutput: data.Location = aws.String("https://location") data.VersionId = aws.String("VERSION-ID") case s3.PutObjectOutput: data.VersionId = aws.String("VERSION-ID") } }) uploader := NewUploaderWithClient(svc, func(u Uploader) { u.Concurrency = 1 u.PartSize = tt.PartSize }) expected := s3testing.GetTestBytes(int(tt.FileSize)) _, err := uploader.Upload(&UploadInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), Body: &testReader{br: bytes.NewReader(expected)}, }) if err == nil { t.Fatalf("expected error but got none") } if v := atomic.LoadInt64(&p.recordedOutstanding); v != 0 { t.Fatalf("expected zero outsnatding pool parts, got %d", v) } }) } }) } } func TestUploadByteSlicePoolConcurrentMultiPartSize(t testing.T) { var ( pools []recordedPartPool mtx sync.Mutex ) unswap := swapByteSlicePool(func(sliceSize int64) byteSlicePool { mtx.Lock() defer mtx.Unlock() b := newRecordedPartPool(sliceSize) pools = append(pools, b) return b }) defer unswap() sess := unit.Session.Copy() svc := s3.New(sess) svc.Handlers.Unmarshal.Clear() svc.Handlers.UnmarshalMeta.Clear() svc.Handlers.UnmarshalError.Clear() svc.Handlers.Send.Clear() svc.Handlers.Send.PushFront(func(r request.Request) { if r.Body != nil { io.Copy(ioutil.Discard, r.Body) } r.HTTPResponse = &http.Response{ StatusCode: 200, Body: ioutil.NopCloser(bytes.NewReader([]byte(respBody))), } switch data := r.Data.(type) { case s3.CreateMultipartUploadOutput: data.UploadId = aws.String("UPLOAD-ID") case s3.UploadPartOutput: data.ETag = aws.String(fmt.Sprintf("ETAG%d", random.Int())) case s3.CompleteMultipartUploadOutput: data.Location = aws.String("https://location") data.VersionId = aws.String("VERSION-ID") case s3.PutObjectOutput: data.VersionId = aws.String("VERSION-ID") } }) uploader := NewUploaderWithClient(svc, func(u Uploader) { u.PartSize = 5 sdkio.MebiByte u.Concurrency = 2 }) var wg sync.WaitGroup for i := 0; i < 2; i++ { wg.Add(2) go func() { defer wg.Done() expected := s3testing.GetTestBytes(int(15 * sdkio.MebiByte)) _, err := uploader.Upload(&UploadInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), Body: &testReader{br: bytes.NewReader(expected)}, }) if err != nil { t.Errorf("expected no error, but got %v", err) } }() go func() { defer wg.Done() expected := s3testing.GetTestBytes(int(15 * sdkio.MebiByte)) _, err := uploader.Upload(&UploadInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), Body: &testReader{br: bytes.NewReader(expected)}, }, func(u Uploader) { u.PartSize = 6 sdkio.MebiByte }) if err != nil { t.Errorf("expected no error, but got %v", err) } }() } wg.Wait() if e, a := 3, len(pools); e != a { t.Errorf("expected %v, got %v", e, a) } for _, p := range pools { if v := atomic.LoadInt64(&p.recordedOutstanding); v != 0 { t.Fatalf("expected zero outsnatding pool parts, got %d", v) } t.Logf("total gets %v, total allocations %v", atomic.LoadUint64(&p.recordedGets), atomic.LoadUint64(&p.recordedAllocs)) } } func BenchmarkPools(b testing.B) { cases := []struct { PartSize int64 FileSize int64 Concurrency int ExAllocations uint64 }{ 0: { PartSize: sdkio.MebiByte 5, FileSize: sdkio.MebiByte * 5, Concurrency: 1, }, 1: { PartSize: sdkio.MebiByte * 5, FileSize: sdkio.MebiByte * 10, Concurrency: 1, }, 2: { PartSize: sdkio.MebiByte * 5, FileSize: sdkio.MebiByte * 20, Concurrency: 2, }, 3: { PartSize: sdkio.MebiByte * 5, FileSize: sdkio.MebiByte * 250, Concurrency: 10, }, } sess := unit.Session.Copy() svc := s3.New(sess) svc.Handlers.Unmarshal.Clear() svc.Handlers.UnmarshalMeta.Clear() svc.Handlers.UnmarshalError.Clear() svc.Handlers.Send.Clear() svc.Handlers.Send.PushFront(func(r request.Request) { if r.Body != nil { io.Copy(ioutil.Discard, r.Body) } r.HTTPResponse = &http.Response{ StatusCode: 200, Body: ioutil.NopCloser(bytes.NewReader([]byte{})), } switch data := r.Data.(type) { case s3.CreateMultipartUploadOutput: data.UploadId = aws.String("UPLOAD-ID") case s3.UploadPartOutput: data.ETag = aws.String(fmt.Sprintf("ETAG%d", random.Int())) case s3.CompleteMultipartUploadOutput: data.Location = aws.String("https://location") data.VersionId = aws.String("VERSION-ID") case s3.PutObjectOutput: data.VersionId = aws.String("VERSION-ID") } }) pools := map[string]func(sliceSize int64) byteSlicePool{ "sync.Pool": func(sliceSize int64) byteSlicePool { return newSyncSlicePool(sliceSize) }, "custom": func(sliceSize int64) byteSlicePool { return newMaxSlicePool(sliceSize) }, } for name, poolFunc := range pools { b.Run(name, func(b testing.B) { unswap := swapByteSlicePool(poolFunc) defer unswap() for i, c := range cases { b.Run(strconv.Itoa(i), func(b testing.B) { uploader := NewUploaderWithClient(svc, func(u Uploader) { u.PartSize = c.PartSize u.Concurrency = c.Concurrency }) expected := s3testing.GetTestBytes(int(c.FileSize)) b.ResetTimer() _, err := uploader.Upload(&UploadInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), Body: &testReader{br: bytes.NewReader(expected)}, }) if err != nil { b.Fatalf("expected no error, but got %v", err) } }) } }) } }	10.78 KB	2025-05-25 12:27
📄 upload_test.go //go:build go1.8 // +build go1.8 package s3manager_test import ( "bytes" "fmt" "io" "io/ioutil" random "math/rand" "net/http" "net/http/httptest" "os" "reflect" "regexp" "sort" "strconv" "strings" "sync" "testing" "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/aws/awsutil" "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/awstesting" "github.com/aws/aws-sdk-go/awstesting/unit" "github.com/aws/aws-sdk-go/service/s3" "github.com/aws/aws-sdk-go/service/s3/internal/s3testing" "github.com/aws/aws-sdk-go/service/s3/s3manager" ) var emptyList = []string{} const respMsg = `<?xml version="1.0" encoding="UTF-8"?> <CompleteMultipartUploadOutput> <Location>mockValue</Location> <Bucket>mockValue</Bucket> <Key>mockValue</Key> <ETag>mockValue</ETag> </CompleteMultipartUploadOutput>` func val(i interface{}, s string) interface{} { v, err := awsutil.ValuesAtPath(i, s) if err != nil \|\| len(v) == 0 { return nil } if _, ok := v[0].(io.Reader); ok { return v[0] } if rv := reflect.ValueOf(v[0]); rv.Kind() == reflect.Ptr { return rv.Elem().Interface() } return v[0] } func contains(src []string, s string) bool { for _, v := range src { if s == v { return true } } return false } func loggingSvc(ignoreOps []string) (s3.S3, []string, []interface{}) { var m sync.Mutex partNum := 0 names := []string{} params := []interface{}{} svc := s3.New(unit.Session) svc.Handlers.Unmarshal.Clear() svc.Handlers.UnmarshalMeta.Clear() svc.Handlers.UnmarshalError.Clear() svc.Handlers.Send.Clear() svc.Handlers.Send.PushBack(func(r request.Request) { m.Lock() defer m.Unlock() if !contains(ignoreOps, r.Operation.Name) { names = append(names, r.Operation.Name) params = append(params, r.Params) } r.HTTPResponse = &http.Response{ StatusCode: 200, Body: ioutil.NopCloser(bytes.NewReader([]byte(respMsg))), } switch data := r.Data.(type) { case s3.CreateMultipartUploadOutput: data.UploadId = aws.String("UPLOAD-ID") case s3.UploadPartOutput: partNum++ data.ETag = aws.String(fmt.Sprintf("ETAG%d", partNum)) case s3.CompleteMultipartUploadOutput: data.Location = aws.String("https://location") data.VersionId = aws.String("VERSION-ID") data.ETag = aws.String("ETAG") case s3.PutObjectOutput: data.VersionId = aws.String("VERSION-ID") data.ETag = aws.String("ETAG") } }) return svc, &names, &params } func buflen(i interface{}) int { r := i.(io.Reader) b, _ := ioutil.ReadAll(r) return len(b) } func TestUploadOrderMulti(t testing.T) { s, ops, args := loggingSvc(emptyList) u := s3manager.NewUploaderWithClient(s) resp, err := u.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key - value"), Body: bytes.NewReader(buf12MB), ServerSideEncryption: aws.String("aws:kms"), SSEKMSKeyId: aws.String("KmsId"), ContentType: aws.String("content/type"), }) if err != nil { t.Errorf("Expected no error but received %v", err) } expected := []string{"CreateMultipartUpload", "UploadPart", "UploadPart", "UploadPart", "CompleteMultipartUpload"} if !reflect.DeepEqual(expected, ops) { t.Errorf("Expected %v, but received %v", expected, ops) } if e, a := `https://s3.mock-region.amazonaws.com/Bucket/Key%20-%20value`, resp.Location; e != a { t.Errorf("Expected %q, but received %q", e, a) } if "UPLOAD-ID" != resp.UploadID { t.Errorf("Expected %q, but received %q", "UPLOAD-ID", resp.UploadID) } if "VERSION-ID" != resp.VersionID { t.Errorf("Expected %q, but received %q", "VERSION-ID", resp.VersionID) } if "ETAG" != resp.ETag { t.Errorf("Expected %q, but received %q", "ETAG", resp.ETag) } // Validate input values // UploadPart for i := 1; i < 5; i++ { v := val((args)[i], "UploadId") if "UPLOAD-ID" != v { t.Errorf("Expected %q, but received %q", "UPLOAD-ID", v) } } // CompleteMultipartUpload v := val((args)[4], "UploadId") if "UPLOAD-ID" != v { t.Errorf("Expected %q, but received %q", "UPLOAD-ID", v) } for i := 0; i < 3; i++ { e := val((args)[4], fmt.Sprintf("MultipartUpload.Parts[%d].PartNumber", i)) if int64(i+1) != e.(int64) { t.Errorf("Expected %d, but received %d", i+1, e) } } vals := []string{ val((args)[4], "MultipartUpload.Parts[0].ETag").(string), val((args)[4], "MultipartUpload.Parts[1].ETag").(string), val((args)[4], "MultipartUpload.Parts[2].ETag").(string), } for _, a := range vals { if matched, err := regexp.MatchString(`^ETAG\d+$`, a); !matched \|\| err != nil { t.Errorf("Failed regexp expression `^ETAG\\d+$`") } } // Custom headers e := val((args)[0], "ServerSideEncryption") if e != "aws:kms" { t.Errorf("Expected %q, but received %q", "aws:kms", e) } e = val((args)[0], "SSEKMSKeyId") if e != "KmsId" { t.Errorf("Expected %q, but received %q", "KmsId", e) } e = val((args)[0], "ContentType") if e != "content/type" { t.Errorf("Expected %q, but received %q", "content/type", e) } } func TestUploadOrderMultiDifferentPartSize(t testing.T) { s, ops, args := loggingSvc(emptyList) mgr := s3manager.NewUploaderWithClient(s, func(u s3manager.Uploader) { u.PartSize = 1024 * 1024 * 7 u.Concurrency = 1 }) _, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: bytes.NewReader(buf12MB), }) if err != nil { t.Errorf("Expected no error but received %v", err) } vals := []string{"CreateMultipartUpload", "UploadPart", "UploadPart", "CompleteMultipartUpload"} if !reflect.DeepEqual(vals, ops) { t.Errorf("Expected %v, but received %v", vals, ops) } // Part lengths if len := buflen(val((args)[1], "Body")); 102410247 != len { t.Errorf("Expected %d, but received %d", 102410247, len) } if len := buflen(val((args)[2], "Body")); 102410245 != len { t.Errorf("Expected %d, but received %d", 102410245, len) } } func TestUploadIncreasePartSize(t testing.T) { s, ops, args := loggingSvc(emptyList) mgr := s3manager.NewUploaderWithClient(s, func(u s3manager.Uploader) { u.Concurrency = 1 u.MaxUploadParts = 2 }) _, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: bytes.NewReader(buf12MB), }) if err != nil { t.Errorf("Expected no error but received %v", err) } if int64(s3manager.DefaultDownloadPartSize) != mgr.PartSize { t.Errorf("Expected %d, but received %d", s3manager.DefaultDownloadPartSize, mgr.PartSize) } vals := []string{"CreateMultipartUpload", "UploadPart", "UploadPart", "CompleteMultipartUpload"} if !reflect.DeepEqual(vals, ops) { t.Errorf("Expected %v, but received %v", vals, ops) } // Part lengths if len := buflen(val((args)[1], "Body")); (102410246)+1 != len { t.Errorf("Expected %d, but received %d", (102410246)+1, len) } if len := buflen(val((args)[2], "Body")); (102410246)-1 != len { t.Errorf("Expected %d, but received %d", (102410246)-1, len) } } func TestUploadFailIfPartSizeTooSmall(t testing.T) { mgr := s3manager.NewUploader(unit.Session, func(u s3manager.Uploader) { u.PartSize = 5 }) resp, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: bytes.NewReader(buf12MB), }) if resp != nil { t.Errorf("Expected response to be nil, but received %v", resp) } if err == nil { t.Errorf("Expected error, but received nil") } aerr := err.(awserr.Error) if e, a := "ConfigError", aerr.Code(); e != a { t.Errorf("Expected %q, but received %q", e, a) } if e, a := "part size must be at least", aerr.Message(); !strings.Contains(a, e) { t.Errorf("expect %v to be in %v", e, a) } } func TestUploadOrderSingle(t testing.T) { s, ops, args := loggingSvc(emptyList) mgr := s3manager.NewUploaderWithClient(s) resp, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key - value"), Body: bytes.NewReader(buf2MB), ServerSideEncryption: aws.String("aws:kms"), SSEKMSKeyId: aws.String("KmsId"), ContentType: aws.String("content/type"), }) if err != nil { t.Errorf("Expected no error but received %v", err) } if vals := []string{"PutObject"}; !reflect.DeepEqual(vals, ops) { t.Errorf("Expected %v, but received %v", vals, ops) } if e, a := `https://s3.mock-region.amazonaws.com/Bucket/Key%20-%20value`, resp.Location; e != a { t.Errorf("Expected %q, but received %q", e, a) } if e := "VERSION-ID"; e != resp.VersionID { t.Errorf("Expected %q, but received %q", e, resp.VersionID) } if "ETAG" != resp.ETag { t.Errorf("Expected %q, but received %q", "ETAG", resp.ETag) } if len(resp.UploadID) > 0 { t.Errorf("Expected empty string, but received %q", resp.UploadID) } if e, a := "aws:kms", val((args)[0], "ServerSideEncryption").(string); e != a { t.Errorf("Expected %q, but received %q", e, a) } if e, a := "KmsId", val((args)[0], "SSEKMSKeyId").(string); e != a { t.Errorf("Expected %q, but received %q", e, a) } if e, a := "content/type", val((args)[0], "ContentType").(string); e != a { t.Errorf("Expected %q, but received %q", e, a) } } func TestUploadOrderSingleFailure(t testing.T) { s, ops, _ := loggingSvc(emptyList) s.Handlers.Send.PushBack(func(r request.Request) { r.HTTPResponse.StatusCode = 400 }) mgr := s3manager.NewUploaderWithClient(s) resp, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: bytes.NewReader(buf2MB), }) if err == nil { t.Error("Expected error, but receievd nil") } if vals := []string{"PutObject"}; !reflect.DeepEqual(vals, ops) { t.Errorf("Expected %v, but received %v", vals, ops) } if resp != nil { t.Errorf("Expected response to be nil, but received %v", resp) } } func TestUploadOrderZero(t testing.T) { s, ops, args := loggingSvc(emptyList) mgr := s3manager.NewUploaderWithClient(s) resp, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: bytes.NewReader(make([]byte, 0)), }) if err != nil { t.Errorf("Expected no error but received %v", err) } if vals := []string{"PutObject"}; !reflect.DeepEqual(vals, ops) { t.Errorf("Expected %v, but received %v", vals, ops) } if len(resp.Location) == 0 { t.Error("Expected Location to not be empty") } if len(resp.UploadID) > 0 { t.Errorf("Expected empty string, but received %q", resp.UploadID) } if e, a := 0, buflen(val((args)[0], "Body")); e != a { t.Errorf("Expected %d, but received %d", e, a) } } func TestUploadOrderMultiFailure(t testing.T) { s, ops, _ := loggingSvc(emptyList) s.Handlers.Send.PushBack(func(r request.Request) { switch t := r.Data.(type) { case s3.UploadPartOutput: if t.ETag == "ETAG2" { r.HTTPResponse.StatusCode = 400 } } }) mgr := s3manager.NewUploaderWithClient(s, func(u s3manager.Uploader) { u.Concurrency = 1 }) _, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: bytes.NewReader(buf12MB), }) if err == nil { t.Error("Expected error, but receievd nil") } if e, a := []string{"CreateMultipartUpload", "UploadPart", "UploadPart", "AbortMultipartUpload"}, ops; !reflect.DeepEqual(e, a) { t.Errorf("Expected %v, but received %v", e, a) } } func TestUploadOrderMultiFailureOnComplete(t testing.T) { s, ops, _ := loggingSvc(emptyList) s.Handlers.Send.PushBack(func(r request.Request) { switch r.Data.(type) { case s3.CompleteMultipartUploadOutput: r.HTTPResponse.StatusCode = 400 } }) mgr := s3manager.NewUploaderWithClient(s, func(u s3manager.Uploader) { u.Concurrency = 1 }) _, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: bytes.NewReader(buf12MB), }) if err == nil { t.Error("Expected error, but receievd nil") } if e, a := []string{"CreateMultipartUpload", "UploadPart", "UploadPart", "UploadPart", "CompleteMultipartUpload", "AbortMultipartUpload"}, ops; !reflect.DeepEqual(e, a) { t.Errorf("Expected %v, but received %v", e, a) } } func TestUploadOrderMultiFailureOnCreate(t testing.T) { s, ops, _ := loggingSvc(emptyList) s.Handlers.Send.PushBack(func(r request.Request) { switch r.Data.(type) { case s3.CreateMultipartUploadOutput: r.HTTPResponse.StatusCode = 400 } }) mgr := s3manager.NewUploaderWithClient(s) _, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: bytes.NewReader(make([]byte, 1024102412)), }) if err == nil { t.Error("Expected error, but receievd nil") } if e, a := []string{"CreateMultipartUpload"}, ops; !reflect.DeepEqual(e, a) { t.Errorf("Expected %v, but received %v", e, a) } } func TestUploadOrderMultiFailureLeaveParts(t testing.T) { s, ops, _ := loggingSvc(emptyList) s.Handlers.Send.PushBack(func(r request.Request) { switch data := r.Data.(type) { case s3.UploadPartOutput: if data.ETag == "ETAG2" { r.HTTPResponse.StatusCode = 400 } } }) mgr := s3manager.NewUploaderWithClient(s, func(u s3manager.Uploader) { u.Concurrency = 1 u.LeavePartsOnError = true }) _, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: bytes.NewReader(make([]byte, 1024102412)), }) if err == nil { t.Error("Expected error, but receievd nil") } if e, a := []string{"CreateMultipartUpload", "UploadPart", "UploadPart"}, ops; !reflect.DeepEqual(e, a) { t.Errorf("Expected %v, but received %v", e, a) } } type failreader struct { times int failCount int } func (f failreader) Read(b []byte) (int, error) { f.failCount++ if f.failCount >= f.times { return 0, fmt.Errorf("random failure") } return len(b), nil } func TestUploadOrderReadFail1(t testing.T) { s, ops, _ := loggingSvc(emptyList) mgr := s3manager.NewUploaderWithClient(s) _, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: &failreader{times: 1}, }) if e, a := "ReadRequestBody", err.(awserr.Error).Code(); e != a { t.Errorf("Expected %q, but received %q", e, a) } if e, a := err.(awserr.Error).OrigErr().Error(), "random failure"; e != a { t.Errorf("Expected %q, but received %q", e, a) } if e, a := []string{}, ops; !reflect.DeepEqual(e, a) { t.Errorf("Expected %v, but received %v", e, a) } } func TestUploadOrderReadFail2(t testing.T) { s, ops, _ := loggingSvc([]string{"UploadPart"}) mgr := s3manager.NewUploaderWithClient(s, func(u s3manager.Uploader) { u.Concurrency = 1 }) _, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: &failreader{times: 2}, }) if e, a := "MultipartUpload", err.(awserr.Error).Code(); e != a { t.Errorf("Expected %q, but received %q", e, a) } if e, a := "ReadRequestBody", err.(awserr.Error).OrigErr().(awserr.Error).Code(); e != a { t.Errorf("Expected %q, but received %q", e, a) } if errStr := err.(awserr.Error).OrigErr().Error(); !strings.Contains(errStr, "random failure") { t.Errorf("Expected error to contains 'random failure', but was %q", errStr) } if e, a := []string{"CreateMultipartUpload", "AbortMultipartUpload"}, ops; !reflect.DeepEqual(e, a) { t.Errorf("Expected %v, but receievd %v", e, a) } } type sizedReader struct { size int cur int err error } func (s sizedReader) Read(p []byte) (n int, err error) { if s.cur >= s.size { if s.err == nil { s.err = io.EOF } return 0, s.err } n = len(p) s.cur += len(p) if s.cur > s.size { n -= s.cur - s.size } return n, err } func TestUploadOrderMultiBufferedReader(t testing.T) { s, ops, args := loggingSvc(emptyList) mgr := s3manager.NewUploaderWithClient(s) _, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: &sizedReader{size: 1024 1024 * 12}, }) if err != nil { t.Errorf("Expected no error but received %v", err) } if e, a := []string{"CreateMultipartUpload", "UploadPart", "UploadPart", "UploadPart", "CompleteMultipartUpload"}, ops; !reflect.DeepEqual(e, a) { t.Errorf("Expected %v, but receievd %v", e, a) } // Part lengths parts := []int{ buflen(val((args)[1], "Body")), buflen(val((args)[2], "Body")), buflen(val((args)[3], "Body")), } sort.Ints(parts) if e, a := []int{1024 * 1024 * 2, 1024 * 1024 * 5, 1024 * 1024 * 5}, parts; !reflect.DeepEqual(e, a) { t.Errorf("Expected %v, but receievd %v", e, a) } } func TestUploadOrderMultiBufferedReaderPartial(t testing.T) { s, ops, args := loggingSvc(emptyList) mgr := s3manager.NewUploaderWithClient(s) _, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: &sizedReader{size: 1024 1024 * 12, err: io.EOF}, }) if err != nil { t.Errorf("Expected no error but received %v", err) } if e, a := []string{"CreateMultipartUpload", "UploadPart", "UploadPart", "UploadPart", "CompleteMultipartUpload"}, ops; !reflect.DeepEqual(e, a) { t.Errorf("Expected %v, but receievd %v", e, a) } // Part lengths parts := []int{ buflen(val((args)[1], "Body")), buflen(val((args)[2], "Body")), buflen(val((args)[3], "Body")), } sort.Ints(parts) if e, a := []int{1024 * 1024 * 2, 1024 * 1024 * 5, 1024 * 1024 * 5}, parts; !reflect.DeepEqual(e, a) { t.Errorf("Expected %v, but receievd %v", e, a) } } // TestUploadOrderMultiBufferedReaderEOF tests the edge case where the // file size is the same as part size. func TestUploadOrderMultiBufferedReaderEOF(t testing.T) { s, ops, args := loggingSvc(emptyList) mgr := s3manager.NewUploaderWithClient(s) _, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: &sizedReader{size: 1024 1024 * 10, err: io.EOF}, }) if err != nil { t.Errorf("Expected no error but received %v", err) } if e, a := []string{"CreateMultipartUpload", "UploadPart", "UploadPart", "CompleteMultipartUpload"}, ops; !reflect.DeepEqual(e, a) { t.Errorf("Expected %v, but receievd %v", e, a) } // Part lengths parts := []int{ buflen(val((args)[1], "Body")), buflen(val((args)[2], "Body")), } sort.Ints(parts) if e, a := []int{1024 1024 * 5, 1024 * 1024 * 5}, parts; !reflect.DeepEqual(e, a) { t.Errorf("Expected %v, but receievd %v", e, a) } } func TestUploadOrderMultiBufferedReaderExceedTotalParts(t testing.T) { s, ops, _ := loggingSvc([]string{"UploadPart"}) mgr := s3manager.NewUploaderWithClient(s, func(u s3manager.Uploader) { u.Concurrency = 1 u.MaxUploadParts = 2 }) resp, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: &sizedReader{size: 1024 * 1024 * 12}, }) if err == nil { t.Error("Expected an error, but received nil") } if resp != nil { t.Errorf("Expected nil, but receievd %v", resp) } if e, a := []string{"CreateMultipartUpload", "AbortMultipartUpload"}, ops; !reflect.DeepEqual(e, a) { t.Errorf("Expected %v, but receievd %v", e, a) } aerr := err.(awserr.Error) if e, a := "MultipartUpload", aerr.Code(); e != a { t.Errorf("Expected %q, but received %q", e, a) } if e, a := "TotalPartsExceeded", aerr.OrigErr().(awserr.Error).Code(); e != a { t.Errorf("Expected %q, but received %q", e, a) } if !strings.Contains(aerr.Error(), "configured MaxUploadParts (2)") { t.Errorf("Expected error to contain 'configured MaxUploadParts (2)', but receievd %q", aerr.Error()) } } func TestUploadOrderSingleBufferedReader(t testing.T) { s, ops, _ := loggingSvc(emptyList) mgr := s3manager.NewUploaderWithClient(s) resp, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: &sizedReader{size: 1024 * 1024 * 2}, }) if err != nil { t.Errorf("Expected no error but received %v", err) } if e, a := []string{"PutObject"}, ops; !reflect.DeepEqual(e, a) { t.Errorf("Expected %v, but received %v", e, a) } if len(resp.Location) == 0 { t.Error("Expected a value in Location but received empty string") } if len(resp.UploadID) > 0 { t.Errorf("Expected empty string but received %q", resp.UploadID) } } func TestUploadZeroLenObject(t testing.T) { requestMade := false server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r http.Request) { requestMade = true w.WriteHeader(http.StatusOK) })) defer server.Close() mgr := s3manager.NewUploaderWithClient(s3.New(unit.Session, &aws.Config{ Endpoint: aws.String(server.URL), })) resp, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: strings.NewReader(""), }) if err != nil { t.Errorf("Expected no error but received %v", err) } if !requestMade { t.Error("Expected request to have been made, but was not") } if len(resp.Location) == 0 { t.Error("Expected a non-empty string value for Location") } if len(resp.UploadID) > 0 { t.Errorf("Expected empty string, but received %q", resp.UploadID) } } func TestUploadInputS3PutObjectInputPairity(t testing.T) { matchings := compareStructType(reflect.TypeOf(s3.PutObjectInput{}), reflect.TypeOf(s3manager.UploadInput{})) aOnly := []string{} bOnly := []string{} for k, c := range matchings { if c == 1 && k != "ContentLength" { aOnly = append(aOnly, k) } else if c == 2 { bOnly = append(bOnly, k) } } if len(aOnly) > 0 { t.Errorf("Expected empty array, but received %v", aOnly) } if len(bOnly) > 0 { t.Errorf("Expected empty array, but received %v", bOnly) } } type testIncompleteReader struct { Size int64 read int64 } func (r testIncompleteReader) Read(p []byte) (n int, err error) { r.read += int64(len(p)) if r.read >= r.Size { return int(r.read - r.Size), io.ErrUnexpectedEOF } return len(p), nil } func TestUploadUnexpectedEOF(t testing.T) { s, ops, _ := loggingSvc(emptyList) mgr := s3manager.NewUploaderWithClient(s, func(u s3manager.Uploader) { u.Concurrency = 1 u.PartSize = s3manager.MinUploadPartSize }) _, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: &testIncompleteReader{ Size: int64(s3manager.MinUploadPartSize + 1), }, }) if err == nil { t.Error("Expected error, but received none") } // Ensure upload started. if e, a := "CreateMultipartUpload", (ops)[0]; e != a { t.Errorf("Expected %q, but received %q", e, a) } // Part may or may not be sent because of timing of sending parts and // reading next part in upload manager. Just check for the last abort. if e, a := "AbortMultipartUpload", (ops)[len(ops)-1]; e != a { t.Errorf("Expected %q, but received %q", e, a) } } func compareStructType(a, b reflect.Type) map[string]int { if a.Kind() != reflect.Struct \|\| b.Kind() != reflect.Struct { panic(fmt.Sprintf("types must both be structs, got %v and %v", a.Kind(), b.Kind())) } aFields := enumFields(a) bFields := enumFields(b) matchings := map[string]int{} for i := 0; i < len(aFields) \|\| i < len(bFields); i++ { if i < len(aFields) { c := matchings[aFields[i].Name] matchings[aFields[i].Name] = c + 1 } if i < len(bFields) { c := matchings[bFields[i].Name] matchings[bFields[i].Name] = c + 2 } } return matchings } func enumFields(v reflect.Type) []reflect.StructField { fields := []reflect.StructField{} for i := 0; i < v.NumField(); i++ { field := v.Field(i) // Ignoreing anon fields if field.PkgPath != "" { // Ignore unexported fields continue } fields = append(fields, field) } return fields } type fooReaderAt struct{} func (r fooReaderAt) Read(p []byte) (n int, err error) { return 12, io.EOF } func (r fooReaderAt) ReadAt(p []byte, off int64) (n int, err error) { return 12, io.EOF } func TestReaderAt(t testing.T) { svc := s3.New(unit.Session) svc.Handlers.Unmarshal.Clear() svc.Handlers.UnmarshalMeta.Clear() svc.Handlers.UnmarshalError.Clear() svc.Handlers.Send.Clear() contentLen := "" svc.Handlers.Send.PushBack(func(r request.Request) { contentLen = r.HTTPRequest.Header.Get("Content-Length") r.HTTPResponse = &http.Response{ StatusCode: 200, Body: ioutil.NopCloser(bytes.NewReader([]byte{})), } }) mgr := s3manager.NewUploaderWithClient(svc, func(u s3manager.Uploader) { u.Concurrency = 1 }) _, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: &fooReaderAt{}, }) if err != nil { t.Errorf("Expected no error but received %v", err) } if e, a := "12", contentLen; e != a { t.Errorf("Expected %q, but received %q", e, a) } } func TestSSE(t testing.T) { svc := s3.New(unit.Session) svc.Handlers.Unmarshal.Clear() svc.Handlers.UnmarshalMeta.Clear() svc.Handlers.UnmarshalError.Clear() svc.Handlers.ValidateResponse.Clear() svc.Handlers.Send.Clear() partNum := 0 mutex := &sync.Mutex{} svc.Handlers.Send.PushBack(func(r request.Request) { mutex.Lock() defer mutex.Unlock() r.HTTPResponse = &http.Response{ StatusCode: 200, Body: ioutil.NopCloser(bytes.NewReader([]byte(respMsg))), } switch data := r.Data.(type) { case s3.CreateMultipartUploadOutput: data.UploadId = aws.String("UPLOAD-ID") case s3.UploadPartOutput: input := r.Params.(s3.UploadPartInput) if input.SSECustomerAlgorithm == nil { t.Fatal("SSECustomerAlgoritm should not be nil") } if input.SSECustomerKey == nil { t.Fatal("SSECustomerKey should not be nil") } partNum++ data.ETag = aws.String(fmt.Sprintf("ETAG%d", partNum)) case s3.CompleteMultipartUploadOutput: data.Location = aws.String("https://location") data.VersionId = aws.String("VERSION-ID") case s3.PutObjectOutput: data.VersionId = aws.String("VERSION-ID") } }) mgr := s3manager.NewUploaderWithClient(svc, func(u s3manager.Uploader) { u.Concurrency = 5 }) _, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), SSECustomerAlgorithm: aws.String("AES256"), SSECustomerKey: aws.String("foo"), Body: bytes.NewBuffer(make([]byte, 1024102410)), }) if err != nil { t.Fatal("Expected no error, but received" + err.Error()) } } func TestUploadWithContextCanceled(t testing.T) { u := s3manager.NewUploader(unit.Session) params := s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: bytes.NewReader(make([]byte, 0)), } ctx := &awstesting.FakeContext{DoneCh: make(chan struct{})} ctx.Error = fmt.Errorf("context canceled") close(ctx.DoneCh) _, err := u.UploadWithContext(ctx, &params) if err == nil { t.Fatalf("expected error, did not get one") } aerr := err.(awserr.Error) if e, a := request.CanceledErrorCode, aerr.Code(); e != a { t.Errorf("expected error code %q, got %q", e, a) } if e, a := "canceled", aerr.Message(); !strings.Contains(a, e) { t.Errorf("expected error message to contain %q, but did not %q", e, a) } } // S3 Uploader incorrectly fails an upload if the content being uploaded // has a size of MinPartSize * MaxUploadParts. // Github: aws/aws-sdk-go#2557 func TestUploadMaxPartsEOF(t testing.T) { s, ops, _ := loggingSvc(emptyList) mgr := s3manager.NewUploaderWithClient(s, func(u s3manager.Uploader) { u.Concurrency = 1 u.PartSize = s3manager.DefaultUploadPartSize u.MaxUploadParts = 2 }) f := bytes.NewReader(make([]byte, int(mgr.PartSize)mgr.MaxUploadParts)) r1 := io.NewSectionReader(f, 0, s3manager.DefaultUploadPartSize) r2 := io.NewSectionReader(f, s3manager.DefaultUploadPartSize, 2s3manager.DefaultUploadPartSize) body := io.MultiReader(r1, r2) _, err := mgr.Upload(&s3manager.UploadInput{ Bucket: aws.String("Bucket"), Key: aws.String("Key"), Body: body, }) if err != nil { t.Fatalf("expect no error, got %v", err) } expectOps := []string{ "CreateMultipartUpload", "UploadPart", "UploadPart", "CompleteMultipartUpload", } if e, a := expectOps, ops; !reflect.DeepEqual(e, a) { t.Errorf("expect %v ops, got %v", e, a) } } func createTempFile(t testing.T, size int64) (os.File, func(testing.T), error) { file, err := ioutil.TempFile(os.TempDir(), aws.SDKName+t.Name()) if err != nil { return nil, nil, err } filename := file.Name() if err := file.Truncate(size); err != nil { return nil, nil, err } return file, func(t testing.T) { if err := file.Close(); err != nil { t.Errorf("failed to close temp file, %s, %v", filename, err) } if err := os.Remove(filename); err != nil { t.Errorf("failed to remove temp file, %s, %v", filename, err) } }, nil } func buildFailHandlers(tb testing.TB, parts, retry int) []http.Handler { handlers := make([]http.Handler, parts) for i := 0; i < len(handlers); i++ { handlers[i] = &failPartHandler{ tb: tb, failsRemaining: retry, successHandler: successPartHandler{tb: tb}, } } return handlers } func TestUploadRetry(t testing.T) { const numParts, retries = 3, 10 testFile, testFileCleanup, err := createTempFile(t, s3manager.DefaultUploadPartSizenumParts) if err != nil { t.Fatalf("failed to create test file, %v", err) } defer testFileCleanup(t) cases := map[string]struct { Body io.Reader PartHandlers func(testing.TB) []http.Handler }{ "bytes.Buffer": { Body: bytes.NewBuffer(make([]byte, s3manager.DefaultUploadPartSizenumParts)), PartHandlers: func(tb testing.TB) []http.Handler { return buildFailHandlers(tb, numParts, retries) }, }, "bytes.Reader": { Body: bytes.NewReader(make([]byte, s3manager.DefaultUploadPartSizenumParts)), PartHandlers: func(tb testing.TB) []http.Handler { return buildFailHandlers(tb, numParts, retries) }, }, "os.File": { Body: testFile, PartHandlers: func(tb testing.TB) []http.Handler { return buildFailHandlers(tb, numParts, retries) }, }, } for name, c := range cases { t.Run(name, func(t testing.T) { mux := newMockS3UploadServer(t, c.PartHandlers(t)) server := httptest.NewServer(mux) defer server.Close() var logger aws.Logger var logLevel aws.LogLevelType if v := os.Getenv("DEBUG_BODY"); len(v) != 0 { logger = t logLevel = aws.LogLevel( aws.LogDebugWithRequestErrors \| aws.LogDebugWithRequestRetries, ) } sess := unit.Session.Copy(&aws.Config{ Endpoint: aws.String(server.URL), S3ForcePathStyle: aws.Bool(true), DisableSSL: aws.Bool(true), MaxRetries: aws.Int(retries + 1), SleepDelay: func(time.Duration) {}, Logger: logger, LogLevel: logLevel, //Credentials: credentials.AnonymousCredentials, }) uploader := s3manager.NewUploader(sess, func(u s3manager.Uploader) { // u.Concurrency = 1 }) _, err := uploader.Upload(&s3manager.UploadInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), Body: c.Body, }) if err != nil { t.Fatalf("expect no error, got %v", err) } }) } } func TestUploadBufferStrategy(t testing.T) { cases := map[string]struct { PartSize int64 Size int64 Strategy s3manager.ReadSeekerWriteToProvider callbacks int }{ "NoBuffer": { PartSize: s3manager.DefaultUploadPartSize, Strategy: nil, }, "SinglePart": { PartSize: s3manager.DefaultUploadPartSize, Size: s3manager.DefaultUploadPartSize, Strategy: &recordedBufferProvider{size: int(s3manager.DefaultUploadPartSize)}, callbacks: 1, }, "MultiPart": { PartSize: s3manager.DefaultUploadPartSize, Size: s3manager.DefaultUploadPartSize 2, Strategy: &recordedBufferProvider{size: int(s3manager.DefaultUploadPartSize)}, callbacks: 2, }, } for name, tCase := range cases { t.Run(name, func(t testing.T) { _ = tCase sess := unit.Session.Copy() svc := s3.New(sess) svc.Handlers.Unmarshal.Clear() svc.Handlers.UnmarshalMeta.Clear() svc.Handlers.UnmarshalError.Clear() svc.Handlers.Send.Clear() svc.Handlers.Send.PushBack(func(r request.Request) { if r.Body != nil { io.Copy(ioutil.Discard, r.Body) } r.HTTPResponse = &http.Response{ StatusCode: 200, Body: ioutil.NopCloser(bytes.NewReader([]byte(respMsg))), } switch data := r.Data.(type) { case s3.CreateMultipartUploadOutput: data.UploadId = aws.String("UPLOAD-ID") case s3.UploadPartOutput: data.ETag = aws.String(fmt.Sprintf("ETAG%d", random.Int())) case s3.CompleteMultipartUploadOutput: data.Location = aws.String("https://location") data.VersionId = aws.String("VERSION-ID") case s3.PutObjectOutput: data.VersionId = aws.String("VERSION-ID") } }) uploader := s3manager.NewUploaderWithClient(svc, func(u s3manager.Uploader) { u.PartSize = tCase.PartSize u.BufferProvider = tCase.Strategy u.Concurrency = 1 }) expected := s3testing.GetTestBytes(int(tCase.Size)) _, err := uploader.Upload(&s3manager.UploadInput{ Bucket: aws.String("bucket"), Key: aws.String("key"), Body: bytes.NewReader(expected), }) if err != nil { t.Fatalf("failed to upload file: %v", err) } switch strat := tCase.Strategy.(type) { case recordedBufferProvider: if !bytes.Equal(expected, strat.content) { t.Errorf("content buffered did not match expected") } if tCase.callbacks != strat.callbackCount { t.Errorf("expected %v, got %v callbacks", tCase.callbacks, strat.callbackCount) } } }) } } func TestUploaderValidARN(t testing.T) { cases := map[string]struct { input s3manager.UploadInput wantErr bool }{ "standard bucket": { input: s3manager.UploadInput{ Bucket: aws.String("test-bucket"), Key: aws.String("test-key"), Body: bytes.NewReader([]byte("test body content")), }, }, "accesspoint": { input: s3manager.UploadInput{ Bucket: aws.String("arn:aws:s3:us-west-2:123456789012:accesspoint/myap"), Key: aws.String("test-key"), Body: bytes.NewReader([]byte("test body content")), }, }, "outpost accesspoint": { input: s3manager.UploadInput{ Bucket: aws.String("arn:aws:s3-outposts:us-west-2:012345678901:outpost/op-1234567890123456/accesspoint/myaccesspoint"), Key: aws.String("test-key"), Body: bytes.NewReader([]byte("test body content")), }, }, "s3-object-lambda accesspoint": { input: s3manager.UploadInput{ Bucket: aws.String("arn:aws:s3-object-lambda:us-west-2:123456789012:accesspoint/myap"), Key: aws.String("test-key"), Body: bytes.NewReader([]byte("test body content")), }, wantErr: true, }, } for name, tt := range cases { t.Run(name, func(t testing.T) { client, _, _ := loggingSvc(nil) client.Config.Region = aws.String("us-west-2") client.ClientInfo.SigningRegion = "us-west-2" uploader := s3manager.NewUploaderWithClient(client) _, err := uploader.Upload(&tt.input) if (err != nil) != tt.wantErr { t.Errorf("err: %v, wantErr: %v", err, tt.wantErr) } }) } } type mockS3UploadServer struct { http.ServeMux tb testing.TB partHandler []http.Handler } func newMockS3UploadServer(tb testing.TB, partHandler []http.Handler) mockS3UploadServer { s := &mockS3UploadServer{ ServeMux: http.NewServeMux(), partHandler: partHandler, tb: tb, } s.HandleFunc("/", s.handleRequest) return s } func (s mockS3UploadServer) handleRequest(w http.ResponseWriter, r http.Request) { defer r.Body.Close() _, hasUploads := r.URL.Query()["uploads"] switch { case r.Method == "POST" && hasUploads: // CreateMultipartUpload w.Header().Set("Content-Length", strconv.Itoa(len(createUploadResp))) w.Write([]byte(createUploadResp)) case r.Method == "PUT": // UploadPart partNumStr := r.URL.Query().Get("partNumber") id, err := strconv.Atoi(partNumStr) if err != nil { failRequest(w, 400, "BadRequest", fmt.Sprintf("unable to parse partNumber, %q, %v", partNumStr, err)) return } id-- if id < 0 \|\| id >= len(s.partHandler) { failRequest(w, 400, "BadRequest", fmt.Sprintf("invalid partNumber %v", id)) return } s.partHandler[id].ServeHTTP(w, r) case r.Method == "POST": // CompleteMultipartUpload w.Header().Set("Content-Length", strconv.Itoa(len(completeUploadResp))) w.Write([]byte(completeUploadResp)) case r.Method == "DELETE": // AbortMultipartUpload w.Header().Set("Content-Length", strconv.Itoa(len(abortUploadResp))) w.WriteHeader(200) w.Write([]byte(abortUploadResp)) default: failRequest(w, 400, "BadRequest", fmt.Sprintf("invalid request %v %v", r.Method, r.URL)) } } func failRequest(w http.ResponseWriter, status int, code, msg string) { msg = fmt.Sprintf(baseRequestErrorResp, code, msg) w.Header().Set("Content-Length", strconv.Itoa(len(msg))) w.WriteHeader(status) w.Write([]byte(msg)) } type successPartHandler struct { tb testing.TB } func (h successPartHandler) ServeHTTP(w http.ResponseWriter, r http.Request) { defer r.Body.Close() n, err := io.Copy(ioutil.Discard, r.Body) if err != nil { failRequest(w, 400, "BadRequest", fmt.Sprintf("failed to read body, %v", err)) return } contLenStr := r.Header.Get("Content-Length") expectLen, err := strconv.ParseInt(contLenStr, 10, 64) if err != nil { h.tb.Logf("expect content-length, got %q, %v", contLenStr, err) failRequest(w, 400, "BadRequest", fmt.Sprintf("unable to get content-length %v", err)) return } if e, a := expectLen, n; e != a { h.tb.Logf("expect %v read, got %v", e, a) failRequest(w, 400, "BadRequest", fmt.Sprintf( "content-length and body do not match, %v, %v", e, a)) return } w.Header().Set("Content-Length", strconv.Itoa(len(uploadPartResp))) w.Write([]byte(uploadPartResp)) } type failPartHandler struct { tb testing.TB failsRemaining int successHandler http.Handler } func (h failPartHandler) ServeHTTP(w http.ResponseWriter, r http.Request) { defer r.Body.Close() if h.failsRemaining == 0 && h.successHandler != nil { h.successHandler.ServeHTTP(w, r) return } io.Copy(ioutil.Discard, r.Body) failRequest(w, 500, "InternalException", fmt.Sprintf("mock error, partNumber %v", r.URL.Query().Get("partNumber"))) h.failsRemaining-- } type recordedBufferProvider struct { content []byte size int callbackCount int } func (r *recordedBufferProvider) GetWriteTo(seeker io.ReadSeeker) (s3manager.ReadSeekerWriteTo, func()) { b := make([]byte, r.size) w := &s3manager.BufferedReadSeekerWriteTo{BufferedReadSeeker: s3manager.NewBufferedReadSeeker(seeker, b)} return w, func() { r.content = append(r.content, b...) r.callbackCount++ } } const createUploadResp = ` <CreateMultipartUploadResponse> <Bucket>bucket</Bucket> <Key>key</Key> <UploadId>abc123</UploadId> </CreateMultipartUploadResponse> ` const uploadPartResp = ` <UploadPartResponse> <ETag>key</ETag> </UploadPartResponse> ` const baseRequestErrorResp = ` <Error> <Code>%s</Code> <Message>%s</Message> <RequestId>request-id</RequestId> <HostId>host-id</HostId> </Error> ` const completeUploadResp = ` <CompleteMultipartUploadResponse> <Bucket>bucket</Bucket> <Key>key</Key> <ETag>key</ETag> <Location>https://bucket.us-west-2.amazonaws.com/key</Location> <UploadId>abc123</UploadId> </CompleteMultipartUploadResponse> ` const abortUploadResp = ` <AbortMultipartUploadResponse> </AbortMultipartUploadResponse> `	39 KB	2025-05-25 12:27
📄 writer_read_from.go package s3manager import ( "bufio" "io" "sync" "github.com/aws/aws-sdk-go/internal/sdkio" ) // WriterReadFrom defines an interface implementing io.Writer and io.ReaderFrom type WriterReadFrom interface { io.Writer io.ReaderFrom } // WriterReadFromProvider provides an implementation of io.ReadFrom for the given io.Writer type WriterReadFromProvider interface { GetReadFrom(writer io.Writer) (w WriterReadFrom, cleanup func()) } type bufferedWriter interface { WriterReadFrom Flush() error Reset(io.Writer) } type bufferedReadFrom struct { bufferedWriter } func (b bufferedReadFrom) ReadFrom(r io.Reader) (int64, error) { n, err := b.bufferedWriter.ReadFrom(r) if flushErr := b.Flush(); flushErr != nil && err == nil { err = flushErr } return n, err } // PooledBufferedReadFromProvider is a WriterReadFromProvider that uses a sync.Pool // to manage allocation and reuse of bufio.Writer structures. type PooledBufferedReadFromProvider struct { pool sync.Pool } // NewPooledBufferedWriterReadFromProvider returns a new PooledBufferedReadFromProvider // Size is used to control the size of the underlying bufio.Writer created for // calls to GetReadFrom. func NewPooledBufferedWriterReadFromProvider(size int) PooledBufferedReadFromProvider { if size < int(32sdkio.KibiByte) { size = int(64 sdkio.KibiByte) } return &PooledBufferedReadFromProvider{ pool: sync.Pool{ New: func() interface{} { return &bufferedReadFrom{bufferedWriter: bufio.NewWriterSize(nil, size)} }, }, } } // GetReadFrom takes an io.Writer and wraps it with a type which satisfies the WriterReadFrom // interface/ Additionally a cleanup function is provided which must be called after usage of the WriterReadFrom // has been completed in order to allow the reuse of the bufio.Writer func (p PooledBufferedReadFromProvider) GetReadFrom(writer io.Writer) (r WriterReadFrom, cleanup func()) { buffer := p.pool.Get().(bufferedReadFrom) buffer.Reset(writer) r = buffer cleanup = func() { buffer.Reset(nil) // Reset to nil writer to release reference p.pool.Put(buffer) } return r, cleanup } type suppressWriterAt struct { suppressed io.Reader } func (s suppressWriterAt) Read(p []byte) (n int, err error) { return s.suppressed.Read(p) }	2.22 KB	2025-05-25 12:27
📄 writer_read_from_test.go package s3manager import ( "fmt" "io" "reflect" "testing" ) type testBufioWriter struct { ReadFromN int64 ReadFromErr error FlushReturn error } func (t testBufioWriter) Write(p []byte) (n int, err error) { panic("unused") } func (t testBufioWriter) ReadFrom(r io.Reader) (n int64, err error) { return t.ReadFromN, t.ReadFromErr } func (t testBufioWriter) Flush() error { return t.FlushReturn } func (t testBufioWriter) Reset(io.Writer) { panic("unused") } func TestBufferedReadFromFlusher_ReadFrom(t testing.T) { cases := map[string]struct { w testBufioWriter expectedErr error }{ "no errors": {}, "error returned from underlying ReadFrom": { w: testBufioWriter{ ReadFromN: 42, ReadFromErr: fmt.Errorf("readfrom"), }, expectedErr: fmt.Errorf("readfrom"), }, "error returned from Flush": { w: testBufioWriter{ ReadFromN: 7, FlushReturn: fmt.Errorf("flush"), }, expectedErr: fmt.Errorf("flush"), }, "error returned from ReadFrom and Flush": { w: testBufioWriter{ ReadFromN: 1337, ReadFromErr: fmt.Errorf("readfrom"), FlushReturn: fmt.Errorf("flush"), }, expectedErr: fmt.Errorf("readfrom"), }, } for name, tCase := range cases { t.Log(name) readFromFlusher := bufferedReadFrom{bufferedWriter: &tCase.w} n, err := readFromFlusher.ReadFrom(nil) if e, a := tCase.w.ReadFromN, n; e != a { t.Errorf("expected %v bytes, got %v", e, a) } if e, a := tCase.expectedErr, err; !reflect.DeepEqual(e, a) { t.Errorf("expected error %v. got %v", e, a) } } }	1.54 KB	2025-05-25 12:27

Edit File:

Terminal