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
📄 connection_reset_error.go package request import ( "strings" ) func isErrConnectionReset(err error) bool { if strings.Contains(err.Error(), "read: connection reset") { return false } if strings.Contains(err.Error(), "use of closed network connection") \|\| strings.Contains(err.Error(), "connection reset") \|\| strings.Contains(err.Error(), "broken pipe") { return true } return false }	376 B	2025-05-25 12:27
📄 connection_reset_error_test.go //go:build go1.7 // +build go1.7 package request_test import ( "net/http" "reflect" "testing" "time" "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/client/metadata" "github.com/aws/aws-sdk-go/aws/corehandlers" "github.com/aws/aws-sdk-go/aws/request" v4 "github.com/aws/aws-sdk-go/aws/signer/v4" "github.com/aws/aws-sdk-go/awstesting/unit" "github.com/aws/aws-sdk-go/private/protocol/jsonrpc" ) type connResetCloser struct { Err error } func (rc connResetCloser) Read(b []byte) (int, error) { return 0, rc.Err } func (rc connResetCloser) Close() error { return nil } func TestSerializationErrConnectionReset_accept(t testing.T) { cases := map[string]struct { Err error ExpectAttempts int }{ "accept with temporary": { Err: errAcceptConnectionResetStub, ExpectAttempts: 6, }, "read not temporary": { Err: errReadConnectionResetStub, ExpectAttempts: 1, }, "write with temporary": { Err: errWriteConnectionResetStub, ExpectAttempts: 6, }, "write broken pipe with temporary": { Err: errWriteBrokenPipeStub, ExpectAttempts: 6, }, "generic connection reset": { Err: errConnectionResetStub, ExpectAttempts: 6, }, "use of closed network connection": { Err: errUseOfClosedConnectionStub, ExpectAttempts: 6, }, } for name, c := range cases { t.Run(name, func(t testing.T) { count := 0 handlers := request.Handlers{} handlers.Send.PushBack(func(r request.Request) { count++ r.HTTPResponse = &http.Response{} r.HTTPResponse.Body = &connResetCloser{ Err: c.Err, } }) handlers.Sign.PushBackNamed(v4.SignRequestHandler) handlers.Build.PushBackNamed(jsonrpc.BuildHandler) handlers.Unmarshal.PushBackNamed(jsonrpc.UnmarshalHandler) handlers.UnmarshalMeta.PushBackNamed(jsonrpc.UnmarshalMetaHandler) handlers.UnmarshalError.PushBackNamed(jsonrpc.UnmarshalErrorHandler) handlers.AfterRetry.PushBackNamed(corehandlers.AfterRetryHandler) op := &request.Operation{ Name: "op", HTTPMethod: "POST", HTTPPath: "/", } meta := metadata.ClientInfo{ ServiceName: "fooService", SigningName: "foo", SigningRegion: "foo", Endpoint: "localhost", APIVersion: "2001-01-01", JSONVersion: "1.1", TargetPrefix: "Foo", } cfg := unit.Session.Config.Copy() cfg.MaxRetries = aws.Int(5) cfg.SleepDelay = func(time.Duration) {} req := request.New( cfg, meta, handlers, client.DefaultRetryer{NumMaxRetries: 5}, op, &struct{}{}, &struct{}{}, ) osErr := c.Err req.ApplyOptions(request.WithResponseReadTimeout(time.Second)) err := req.Send() if err == nil { t.Error("Expected error 'SerializationError', but received nil") } if aerr, ok := err.(awserr.Error); ok && aerr.Code() != request.ErrCodeSerialization { t.Errorf("Expected 'SerializationError', but received %q", aerr.Code()) } else if !ok { t.Errorf("Expected 'awserr.Error', but received %v", reflect.TypeOf(err)) } else if aerr.OrigErr().Error() != osErr.Error() { t.Errorf("Expected %q, but received %q", osErr.Error(), aerr.OrigErr().Error()) } if e, a := c.ExpectAttempts, count; e != a { t.Errorf("Expected %v, but received %v", e, a) } }) } }	3.38 KB	2025-05-25 12:27
📄 handlers.go package request import ( "fmt" "strings" ) // A Handlers provides a collection of request handlers for various // stages of handling requests. type Handlers struct { Validate HandlerList Build HandlerList BuildStream HandlerList Sign HandlerList Send HandlerList ValidateResponse HandlerList Unmarshal HandlerList UnmarshalStream HandlerList UnmarshalMeta HandlerList UnmarshalError HandlerList Retry HandlerList AfterRetry HandlerList CompleteAttempt HandlerList Complete HandlerList } // Copy returns a copy of this handler's lists. func (h Handlers) Copy() Handlers { return Handlers{ Validate: h.Validate.copy(), Build: h.Build.copy(), BuildStream: h.BuildStream.copy(), Sign: h.Sign.copy(), Send: h.Send.copy(), ValidateResponse: h.ValidateResponse.copy(), Unmarshal: h.Unmarshal.copy(), UnmarshalStream: h.UnmarshalStream.copy(), UnmarshalError: h.UnmarshalError.copy(), UnmarshalMeta: h.UnmarshalMeta.copy(), Retry: h.Retry.copy(), AfterRetry: h.AfterRetry.copy(), CompleteAttempt: h.CompleteAttempt.copy(), Complete: h.Complete.copy(), } } // Clear removes callback functions for all handlers. func (h Handlers) Clear() { h.Validate.Clear() h.Build.Clear() h.BuildStream.Clear() h.Send.Clear() h.Sign.Clear() h.Unmarshal.Clear() h.UnmarshalStream.Clear() h.UnmarshalMeta.Clear() h.UnmarshalError.Clear() h.ValidateResponse.Clear() h.Retry.Clear() h.AfterRetry.Clear() h.CompleteAttempt.Clear() h.Complete.Clear() } // IsEmpty returns if there are no handlers in any of the handlerlists. func (h Handlers) IsEmpty() bool { if h.Validate.Len() != 0 { return false } if h.Build.Len() != 0 { return false } if h.BuildStream.Len() != 0 { return false } if h.Send.Len() != 0 { return false } if h.Sign.Len() != 0 { return false } if h.Unmarshal.Len() != 0 { return false } if h.UnmarshalStream.Len() != 0 { return false } if h.UnmarshalMeta.Len() != 0 { return false } if h.UnmarshalError.Len() != 0 { return false } if h.ValidateResponse.Len() != 0 { return false } if h.Retry.Len() != 0 { return false } if h.AfterRetry.Len() != 0 { return false } if h.CompleteAttempt.Len() != 0 { return false } if h.Complete.Len() != 0 { return false } return true } // A HandlerListRunItem represents an entry in the HandlerList which // is being run. type HandlerListRunItem struct { Index int Handler NamedHandler Request Request } // A HandlerList manages zero or more handlers in a list. type HandlerList struct { list []NamedHandler // Called after each request handler in the list is called. If set // and the func returns true the HandlerList will continue to iterate // over the request handlers. If false is returned the HandlerList // will stop iterating. // // Should be used if extra logic to be performed between each handler // in the list. This can be used to terminate a list's iteration // based on a condition such as error like, HandlerListStopOnError. // Or for logging like HandlerListLogItem. AfterEachFn func(item HandlerListRunItem) bool } // A NamedHandler is a struct that contains a name and function callback. type NamedHandler struct { Name string Fn func(Request) } // copy creates a copy of the handler list. func (l HandlerList) copy() HandlerList { n := HandlerList{ AfterEachFn: l.AfterEachFn, } if len(l.list) == 0 { return n } n.list = append(make([]NamedHandler, 0, len(l.list)), l.list...) return n } // Clear clears the handler list. func (l HandlerList) Clear() { l.list = l.list[0:0] } // Len returns the number of handlers in the list. func (l HandlerList) Len() int { return len(l.list) } // PushBack pushes handler f to the back of the handler list. func (l HandlerList) PushBack(f func(Request)) { l.PushBackNamed(NamedHandler{"__anonymous", f}) } // PushBackNamed pushes named handler f to the back of the handler list. func (l HandlerList) PushBackNamed(n NamedHandler) { if cap(l.list) == 0 { l.list = make([]NamedHandler, 0, 5) } l.list = append(l.list, n) } // PushFront pushes handler f to the front of the handler list. func (l HandlerList) PushFront(f func(Request)) { l.PushFrontNamed(NamedHandler{"__anonymous", f}) } // PushFrontNamed pushes named handler f to the front of the handler list. func (l HandlerList) PushFrontNamed(n NamedHandler) { if cap(l.list) == len(l.list) { // Allocating new list required l.list = append([]NamedHandler{n}, l.list...) } else { // Enough room to prepend into list. l.list = append(l.list, NamedHandler{}) copy(l.list[1:], l.list) l.list[0] = n } } // Remove removes a NamedHandler n func (l HandlerList) Remove(n NamedHandler) { l.RemoveByName(n.Name) } // RemoveByName removes a NamedHandler by name. func (l HandlerList) RemoveByName(name string) { for i := 0; i < len(l.list); i++ { m := l.list[i] if m.Name == name { // Shift array preventing creating new arrays copy(l.list[i:], l.list[i+1:]) l.list[len(l.list)-1] = NamedHandler{} l.list = l.list[:len(l.list)-1] // decrement list so next check to length is correct i-- } } } // SwapNamed will swap out any existing handlers with the same name as the // passed in NamedHandler returning true if handlers were swapped. False is // returned otherwise. func (l HandlerList) SwapNamed(n NamedHandler) (swapped bool) { for i := 0; i < len(l.list); i++ { if l.list[i].Name == n.Name { l.list[i].Fn = n.Fn swapped = true } } return swapped } // Swap will swap out all handlers matching the name passed in. The matched // handlers will be swapped in. True is returned if the handlers were swapped. func (l HandlerList) Swap(name string, replace NamedHandler) bool { var swapped bool for i := 0; i < len(l.list); i++ { if l.list[i].Name == name { l.list[i] = replace swapped = true } } return swapped } // SetBackNamed will replace the named handler if it exists in the handler list. // If the handler does not exist the handler will be added to the end of the list. func (l HandlerList) SetBackNamed(n NamedHandler) { if !l.SwapNamed(n) { l.PushBackNamed(n) } } // SetFrontNamed will replace the named handler if it exists in the handler list. // If the handler does not exist the handler will be added to the beginning of // the list. func (l HandlerList) SetFrontNamed(n NamedHandler) { if !l.SwapNamed(n) { l.PushFrontNamed(n) } } // Run executes all handlers in the list with a given request object. func (l HandlerList) Run(r Request) { for i, h := range l.list { h.Fn(r) item := HandlerListRunItem{ Index: i, Handler: h, Request: r, } if l.AfterEachFn != nil && !l.AfterEachFn(item) { return } } } // HandlerListLogItem logs the request handler and the state of the // request's Error value. Always returns true to continue iterating // request handlers in a HandlerList. func HandlerListLogItem(item HandlerListRunItem) bool { if item.Request.Config.Logger == nil { return true } item.Request.Config.Logger.Log("DEBUG: RequestHandler", item.Index, item.Handler.Name, item.Request.Error) return true } // HandlerListStopOnError returns false to stop the HandlerList iterating // over request handlers if Request.Error is not nil. True otherwise // to continue iterating. func HandlerListStopOnError(item HandlerListRunItem) bool { return item.Request.Error == nil } // WithAppendUserAgent will add a string to the user agent prefixed with a // single white space. func WithAppendUserAgent(s string) Option { return func(r Request) { r.Handlers.Build.PushBack(func(r2 Request) { AddToUserAgent(r, s) }) } } // MakeAddToUserAgentHandler will add the name/version pair to the User-Agent request // header. If the extra parameters are provided they will be added as metadata to the // name/version pair resulting in the following format. // "name/version (extra0; extra1; ...)" // The user agent part will be concatenated with this current request's user agent string. func MakeAddToUserAgentHandler(name, version string, extra ...string) func(Request) { ua := fmt.Sprintf("%s/%s", name, version) if len(extra) > 0 { ua += fmt.Sprintf(" (%s)", strings.Join(extra, "; ")) } return func(r Request) { AddToUserAgent(r, ua) } } // MakeAddToUserAgentFreeFormHandler adds the input to the User-Agent request header. // The input string will be concatenated with the current request's user agent string. func MakeAddToUserAgentFreeFormHandler(s string) func(Request) { return func(r Request) { AddToUserAgent(r, s) } } // WithSetRequestHeaders updates the operation request's HTTP header to contain // the header key value pairs provided. If the header key already exists in the // request's HTTP header set, the existing value(s) will be replaced. // // Header keys added will be added as canonical format with title casing // applied via http.Header.Set method. func WithSetRequestHeaders(h map[string]string) Option { return withRequestHeader(h).SetRequestHeaders } type withRequestHeader map[string]string func (h withRequestHeader) SetRequestHeaders(r *Request) { for k, v := range h { r.HTTPRequest.Header.Set(k, v) } }	9.16 KB	2025-05-25 12:27
📄 handlers_test.go package request_test import ( "net/http" "reflect" "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/service/s3" ) func TestHandlerList(t testing.T) { s := "" r := &request.Request{} l := request.HandlerList{} l.PushBack(func(r request.Request) { s += "a" r.Data = s }) l.Run(r) if e, a := "a", s; e != a { t.Errorf("expect %q update got %q", e, a) } if e, a := "a", r.Data.(string); e != a { t.Errorf("expect %q data update got %q", e, a) } } func TestMultipleHandlers(t testing.T) { r := &request.Request{} l := request.HandlerList{} l.PushBack(func(r request.Request) { r.Data = nil }) l.PushFront(func(r request.Request) { r.Data = aws.Bool(true) }) l.Run(r) if r.Data != nil { t.Error("Expected handler to execute") } } func TestNamedHandlers(t testing.T) { l := request.HandlerList{} named := request.NamedHandler{Name: "Name", Fn: func(r request.Request) {}} named2 := request.NamedHandler{Name: "NotName", Fn: func(r request.Request) {}} l.PushBackNamed(named) l.PushBackNamed(named) l.PushBackNamed(named2) l.PushBack(func(r request.Request) {}) if e, a := 4, l.Len(); e != a { t.Errorf("expect %d list length, got %d", e, a) } l.Remove(named) if e, a := 2, l.Len(); e != a { t.Errorf("expect %d list length, got %d", e, a) } } func TestSwapHandlers(t testing.T) { firstHandlerCalled := 0 swappedOutHandlerCalled := 0 swappedInHandlerCalled := 0 l := request.HandlerList{} named := request.NamedHandler{Name: "Name", Fn: func(r request.Request) { firstHandlerCalled++ }} named2 := request.NamedHandler{Name: "SwapOutName", Fn: func(r request.Request) { swappedOutHandlerCalled++ }} l.PushBackNamed(named) l.PushBackNamed(named2) l.PushBackNamed(named) l.SwapNamed(request.NamedHandler{Name: "SwapOutName", Fn: func(r request.Request) { swappedInHandlerCalled++ }}) l.Run(&request.Request{}) if e, a := 2, firstHandlerCalled; e != a { t.Errorf("expect first handler to be called %d, was called %d times", e, a) } if n := swappedOutHandlerCalled; n != 0 { t.Errorf("expect swapped out handler to not be called, was called %d times", n) } if e, a := 1, swappedInHandlerCalled; e != a { t.Errorf("expect swapped in handler to be called %d, was called %d times", e, a) } } func TestSetBackNamed_Exists(t testing.T) { firstHandlerCalled := 0 swappedOutHandlerCalled := 0 swappedInHandlerCalled := 0 l := request.HandlerList{} named := request.NamedHandler{Name: "Name", Fn: func(r request.Request) { firstHandlerCalled++ }} named2 := request.NamedHandler{Name: "SwapOutName", Fn: func(r request.Request) { swappedOutHandlerCalled++ }} l.PushBackNamed(named) l.PushBackNamed(named2) l.SetBackNamed(request.NamedHandler{Name: "SwapOutName", Fn: func(r request.Request) { swappedInHandlerCalled++ }}) l.Run(&request.Request{}) if e, a := 1, firstHandlerCalled; e != a { t.Errorf("expect first handler to be called %d, was called %d times", e, a) } if n := swappedOutHandlerCalled; n != 0 { t.Errorf("expect swapped out handler to not be called, was called %d times", n) } if e, a := 1, swappedInHandlerCalled; e != a { t.Errorf("expect swapped in handler to be called %d, was called %d times", e, a) } } func TestSetBackNamed_NotExists(t testing.T) { firstHandlerCalled := 0 secondHandlerCalled := 0 swappedInHandlerCalled := 0 l := request.HandlerList{} named := request.NamedHandler{Name: "Name", Fn: func(r request.Request) { firstHandlerCalled++ }} named2 := request.NamedHandler{Name: "OtherName", Fn: func(r request.Request) { secondHandlerCalled++ }} l.PushBackNamed(named) l.PushBackNamed(named2) l.SetBackNamed(request.NamedHandler{Name: "SwapOutName", Fn: func(r request.Request) { swappedInHandlerCalled++ }}) l.Run(&request.Request{}) if e, a := 1, firstHandlerCalled; e != a { t.Errorf("expect first handler to be called %d, was called %d times", e, a) } if e, a := 1, secondHandlerCalled; e != a { t.Errorf("expect second handler to be called %d, was called %d times", e, a) } if e, a := 1, swappedInHandlerCalled; e != a { t.Errorf("expect swapped in handler to be called %d, was called %d times", e, a) } } func TestLoggedHandlers(t testing.T) { expectedHandlers := []string{"name1", "name2"} l := request.HandlerList{} loggedHandlers := []string{} l.AfterEachFn = request.HandlerListLogItem cfg := aws.Config{Logger: aws.LoggerFunc(func(args ...interface{}) { loggedHandlers = append(loggedHandlers, args[2].(string)) })} named1 := request.NamedHandler{Name: "name1", Fn: func(r request.Request) {}} named2 := request.NamedHandler{Name: "name2", Fn: func(r request.Request) {}} l.PushBackNamed(named1) l.PushBackNamed(named2) l.Run(&request.Request{Config: cfg}) if !reflect.DeepEqual(expectedHandlers, loggedHandlers) { t.Errorf("expect handlers executed %v to match logged handlers, %v", expectedHandlers, loggedHandlers) } } func TestStopHandlers(t testing.T) { l := request.HandlerList{} stopAt := 1 l.AfterEachFn = func(item request.HandlerListRunItem) bool { return item.Index != stopAt } called := 0 l.PushBackNamed(request.NamedHandler{Name: "name1", Fn: func(r request.Request) { called++ }}) l.PushBackNamed(request.NamedHandler{Name: "name2", Fn: func(r request.Request) { called++ }}) l.PushBackNamed(request.NamedHandler{Name: "name3", Fn: func(r request.Request) { t.Fatalf("third handler should not be called") }}) l.Run(&request.Request{}) if e, a := 2, called; e != a { t.Errorf("expect %d handlers called, got %d", e, a) } } func TestWithSetRequestHeaders(t testing.T) { fn := request.WithSetRequestHeaders(map[string]string{ "x-foo-bar": "abc123", "X-Bar-foo": "efg456", }) req := &request.Request{HTTPRequest: &http.Request{Header: http.Header{}}} fn(req) expect := map[string][]string{ "X-Foo-Bar": {"abc123"}, "X-Bar-Foo": {"efg456"}, } if e, a := len(req.HTTPRequest.Header), len(expect); e != a { t.Fatalf("expect %v headers, got %v", e, a) } for k, expectVs := range expect { actualVs, ok := req.HTTPRequest.Header[k] if !ok { t.Errorf("expect %v header", k) } if e, a := len(expectVs), len(actualVs); e != a { t.Fatalf("expect %v values for %v, got %v", e, k, a) } for i, expectV := range expectVs { if e, a := expectV, actualVs[i]; e != a { t.Errorf("expect %v[%d] to be %v, got %v", k, i, e, a) } } } } func BenchmarkNewRequest(b testing.B) { svc := s3.New(unit.Session) for i := 0; i < b.N; i++ { r, _ := svc.GetObjectRequest(nil) if r == nil { b.Fatal("r should not be nil") } } } func BenchmarkHandlersCopy(b testing.B) { handlers := request.Handlers{} handlers.Validate.PushBack(func(r request.Request) {}) handlers.Validate.PushBack(func(r request.Request) {}) handlers.Build.PushBack(func(r request.Request) {}) handlers.Build.PushBack(func(r request.Request) {}) handlers.Send.PushBack(func(r request.Request) {}) handlers.Send.PushBack(func(r request.Request) {}) handlers.Unmarshal.PushBack(func(r request.Request) {}) handlers.Unmarshal.PushBack(func(r request.Request) {}) for i := 0; i < b.N; i++ { h := handlers.Copy() if e, a := handlers.Validate.Len(), h.Validate.Len(); e != a { b.Fatalf("expected %d handlers got %d", e, a) } } } func BenchmarkHandlersPushBack(b testing.B) { handlers := request.Handlers{} for i := 0; i < b.N; i++ { h := handlers.Copy() h.Validate.PushBack(func(r request.Request) {}) h.Validate.PushBack(func(r request.Request) {}) h.Validate.PushBack(func(r request.Request) {}) h.Validate.PushBack(func(r request.Request) {}) } } func BenchmarkHandlersPushFront(b testing.B) { handlers := request.Handlers{} for i := 0; i < b.N; i++ { h := handlers.Copy() h.Validate.PushFront(func(r request.Request) {}) h.Validate.PushFront(func(r request.Request) {}) h.Validate.PushFront(func(r request.Request) {}) h.Validate.PushFront(func(r request.Request) {}) } } func BenchmarkHandlersClear(b testing.B) { handlers := request.Handlers{} for i := 0; i < b.N; i++ { h := handlers.Copy() h.Validate.PushFront(func(r request.Request) {}) h.Validate.PushFront(func(r request.Request) {}) h.Validate.PushFront(func(r request.Request) {}) h.Validate.PushFront(func(r request.Request) {}) h.Clear() } }	8.28 KB	2025-05-25 12:27
📄 http_request.go package request import ( "io" "net/http" "net/url" ) func copyHTTPRequest(r http.Request, body io.ReadCloser) http.Request { req := new(http.Request) req = r req.URL = &url.URL{} req.URL = r.URL req.Body = body req.Header = http.Header{} for k, v := range r.Header { for _, vv := range v { req.Header.Add(k, vv) } } return req }	358 B	2025-05-25 12:27
📄 http_request_copy_test.go package request import ( "bytes" "io/ioutil" "net/http" "net/url" "sync" "testing" ) func TestRequestCopyRace(t *testing.T) { origReq := &http.Request{URL: &url.URL{}, Header: http.Header{}} origReq.Header.Set("Header", "OrigValue") var wg sync.WaitGroup for i := 0; i < 100; i++ { wg.Add(1) go func() { req := copyHTTPRequest(origReq, ioutil.NopCloser(&bytes.Buffer{})) req.Header.Set("Header", "Value") go func() { req2 := copyHTTPRequest(req, ioutil.NopCloser(&bytes.Buffer{})) req2.Header.Add("Header", "Value2") }() _ = req.Header.Get("Header") wg.Done() }() _ = origReq.Header.Get("Header") } origReq.Header.Get("Header") wg.Wait() }	691 B	2025-05-25 12:27
📄 http_request_retry_test.go package request_test import ( "strings" "testing" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/awstesting/mock" ) func TestRequestCancelRetry(t testing.T) { c := make(chan struct{}) reqNum := 0 s := mock.NewMockClient(&aws.Config{ MaxRetries: aws.Int(1), }) s.Handlers.Validate.Clear() s.Handlers.Unmarshal.Clear() s.Handlers.UnmarshalMeta.Clear() s.Handlers.UnmarshalError.Clear() s.Handlers.Send.PushFront(func(r request.Request) { reqNum++ }) out := &testData{} r := s.NewRequest(&request.Operation{Name: "Operation"}, nil, out) r.HTTPRequest.Cancel = c close(c) err := r.Send() if !strings.Contains(err.Error(), "canceled") { t.Errorf("expect canceled in error, %v", err) } if e, a := 1, reqNum; e != a { t.Errorf("expect %v, got %v", e, a) } }	841 B	2025-05-25 12:27
📄 offset_reader.go package request import ( "io" "sync" "github.com/aws/aws-sdk-go/internal/sdkio" ) // offsetReader is a thread-safe io.ReadCloser to prevent racing // with retrying requests type offsetReader struct { buf io.ReadSeeker lock sync.Mutex closed bool } func newOffsetReader(buf io.ReadSeeker, offset int64) (offsetReader, error) { reader := &offsetReader{} _, err := buf.Seek(offset, sdkio.SeekStart) if err != nil { return nil, err } reader.buf = buf return reader, nil } // Close will close the instance of the offset reader's access to // the underlying io.ReadSeeker. func (o offsetReader) Close() error { o.lock.Lock() defer o.lock.Unlock() o.closed = true return nil } // Read is a thread-safe read of the underlying io.ReadSeeker func (o offsetReader) Read(p []byte) (int, error) { o.lock.Lock() defer o.lock.Unlock() if o.closed { return 0, io.EOF } return o.buf.Read(p) } // Seek is a thread-safe seeking operation. func (o offsetReader) Seek(offset int64, whence int) (int64, error) { o.lock.Lock() defer o.lock.Unlock() return o.buf.Seek(offset, whence) } // CloseAndCopy will return a new offsetReader with a copy of the old buffer // and close the old buffer. func (o offsetReader) CloseAndCopy(offset int64) (offsetReader, error) { if err := o.Close(); err != nil { return nil, err } return newOffsetReader(o.buf, offset) }	1.36 KB	2025-05-25 12:27
📄 offset_reader_test.go package request import ( "bytes" "io" "math/rand" "sync" "testing" "time" "github.com/aws/aws-sdk-go/internal/sdkio" ) func TestOffsetReaderRead(t testing.T) { buf := []byte("testData") reader := &offsetReader{buf: bytes.NewReader(buf)} tempBuf := make([]byte, len(buf)) n, err := reader.Read(tempBuf) if e, a := n, len(buf); e != a { t.Errorf("expect %v, got %v", e, a) } if err != nil { t.Fatalf("expect no error, got %v", err) } if e, a := buf, tempBuf; !bytes.Equal(e, a) { t.Errorf("expect %v, got %v", e, a) } } func TestOffsetReaderSeek(t testing.T) { buf := []byte("testData") reader, err := newOffsetReader(bytes.NewReader(buf), 0) if err != nil { t.Fatalf("expect no error, got %v", err) } orig, err := reader.Seek(0, sdkio.SeekCurrent) if err != nil { t.Fatalf("expect no error, got %v", err) } if e, a := int64(0), orig; e != a { t.Errorf("expect %v, got %v", e, a) } n, err := reader.Seek(0, sdkio.SeekEnd) if err != nil { t.Fatalf("expect no error, got %v", err) } if e, a := int64(len(buf)), n; e != a { t.Errorf("expect %v, got %v", e, a) } n, err = reader.Seek(orig, sdkio.SeekStart) if err != nil { t.Fatalf("expect no error, got %v", err) } if e, a := int64(0), n; e != a { t.Errorf("expect %v, got %v", e, a) } } func TestOffsetReaderClose(t testing.T) { buf := []byte("testData") reader := &offsetReader{buf: bytes.NewReader(buf)} err := reader.Close() if err != nil { t.Fatalf("expect no error, got %v", err) } tempBuf := make([]byte, len(buf)) n, err := reader.Read(tempBuf) if e, a := n, 0; e != a { t.Errorf("expect %v, got %v", e, a) } if e, a := err, io.EOF; e != a { t.Errorf("expect %v, got %v", e, a) } } func TestOffsetReaderCloseAndCopy(t testing.T) { buf := []byte("testData") tempBuf := make([]byte, len(buf)) reader := &offsetReader{buf: bytes.NewReader(buf)} newReader, err := reader.CloseAndCopy(0) if err != nil { t.Fatalf("expect no error, got %v", err) } n, err := reader.Read(tempBuf) if e, a := n, 0; e != a { t.Errorf("expect %v, got %v", e, a) } if e, a := err, io.EOF; e != a { t.Errorf("expect %v, got %v", e, a) } n, err = newReader.Read(tempBuf) if e, a := n, len(buf); e != a { t.Errorf("expect %v, got %v", e, a) } if err != nil { t.Fatalf("expect no error, got %v", err) } if e, a := buf, tempBuf; !bytes.Equal(e, a) { t.Errorf("expect %v, got %v", e, a) } } func TestOffsetReaderCloseAndCopyOffset(t testing.T) { buf := []byte("testData") tempBuf := make([]byte, len(buf)) reader := &offsetReader{buf: bytes.NewReader(buf)} newReader, err := reader.CloseAndCopy(4) if err != nil { t.Fatalf("expect no error, got %v", err) } n, err := newReader.Read(tempBuf) if e, a := n, len(buf)-4; e != a { t.Errorf("expect %v, got %v", e, a) } if err != nil { t.Fatalf("expect no error, got %v", err) } expected := []byte{'D', 'a', 't', 'a', 0, 0, 0, 0} if e, a := expected, tempBuf; !bytes.Equal(e, a) { t.Errorf("expect %v, got %v", e, a) } } func TestOffsetReaderRace(t testing.T) { wg := sync.WaitGroup{} f := func(reader offsetReader) { defer wg.Done() var err error buf := make([]byte, 1) _, err = reader.Read(buf) for err != io.EOF { _, err = reader.Read(buf) } } closeFn := func(reader offsetReader) { defer wg.Done() time.Sleep(time.Duration(rand.Intn(20)+1) * time.Millisecond) reader.Close() } for i := 0; i < 50; i++ { reader := &offsetReader{buf: bytes.NewReader(make([]byte, 10241024))} wg.Add(1) go f(reader) wg.Add(1) go closeFn(reader) } wg.Wait() } func BenchmarkOffsetReader(b testing.B) { bufSize := 1024 * 1024 * 100 buf := make([]byte, bufSize) reader := &offsetReader{buf: bytes.NewReader(buf)} tempBuf := make([]byte, 1024) for i := 0; i < b.N; i++ { reader.Read(tempBuf) } } func BenchmarkBytesReader(b testing.B) { bufSize := 1024 1024 * 100 buf := make([]byte, bufSize) reader := bytes.NewReader(buf) tempBuf := make([]byte, 1024) for i := 0; i < b.N; i++ { reader.Read(tempBuf) } }	3.98 KB	2025-05-25 12:27
📄 request.go package request import ( "bytes" "fmt" "io" "io/ioutil" "net/http" "net/url" "reflect" "strings" "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/aws/client/metadata" "github.com/aws/aws-sdk-go/internal/sdkio" ) const ( // ErrCodeSerialization is the serialization error code that is received // during protocol unmarshaling. ErrCodeSerialization = "SerializationError" // ErrCodeRead is an error that is returned during HTTP reads. ErrCodeRead = "ReadError" // ErrCodeResponseTimeout is the connection timeout error that is received // during body reads. ErrCodeResponseTimeout = "ResponseTimeout" // ErrCodeInvalidPresignExpire is returned when the expire time provided to // presign is invalid ErrCodeInvalidPresignExpire = "InvalidPresignExpireError" // CanceledErrorCode is the error code that will be returned by an // API request that was canceled. Requests given a aws.Context may // return this error when canceled. CanceledErrorCode = "RequestCanceled" // ErrCodeRequestError is an error preventing the SDK from continuing to // process the request. ErrCodeRequestError = "RequestError" ) // A Request is the service request to be made. type Request struct { Config aws.Config ClientInfo metadata.ClientInfo Handlers Handlers Retryer AttemptTime time.Time Time time.Time Operation Operation HTTPRequest http.Request HTTPResponse http.Response Body io.ReadSeeker streamingBody io.ReadCloser BodyStart int64 // offset from beginning of Body that the request body starts Params interface{} Error error Data interface{} RequestID string RetryCount int Retryable bool RetryDelay time.Duration NotHoist bool SignedHeaderVals http.Header LastSignedAt time.Time DisableFollowRedirects bool // Additional API error codes that should be retried. IsErrorRetryable // will consider these codes in addition to its built in cases. RetryErrorCodes []string // Additional API error codes that should be retried with throttle backoff // delay. IsErrorThrottle will consider these codes in addition to its // built in cases. ThrottleErrorCodes []string // A value greater than 0 instructs the request to be signed as Presigned URL // You should not set this field directly. Instead use Request's // Presign or PresignRequest methods. ExpireTime time.Duration context aws.Context built bool // Need to persist an intermediate body between the input Body and HTTP // request body because the HTTP Client's transport can maintain a reference // to the HTTP request's body after the client has returned. This value is // safe to use concurrently and wrap the input Body for each HTTP request. safeBody offsetReader } // An Operation is the service API operation to be made. type Operation struct { Name string HTTPMethod string HTTPPath string Paginator BeforePresignFn func(r Request) error } // New returns a new Request pointer for the service API operation and // parameters. // // A Retryer should be provided to direct how the request is retried. If // Retryer is nil, a default no retry value will be used. You can use // NoOpRetryer in the Client package to disable retry behavior directly. // // Params is any value of input parameters to be the request payload. // Data is pointer value to an object which the request's response // payload will be deserialized to. func New(cfg aws.Config, clientInfo metadata.ClientInfo, handlers Handlers, retryer Retryer, operation Operation, params interface{}, data interface{}) Request { if retryer == nil { retryer = noOpRetryer{} } method := operation.HTTPMethod if method == "" { method = "POST" } httpReq, _ := http.NewRequest(method, "", nil) var err error httpReq.URL, err = url.Parse(clientInfo.Endpoint) if err != nil { httpReq.URL = &url.URL{} err = awserr.New("InvalidEndpointURL", "invalid endpoint uri", err) } if len(operation.HTTPPath) != 0 { opHTTPPath := operation.HTTPPath var opQueryString string if idx := strings.Index(opHTTPPath, "?"); idx >= 0 { opQueryString = opHTTPPath[idx+1:] opHTTPPath = opHTTPPath[:idx] } if strings.HasSuffix(httpReq.URL.Path, "/") && strings.HasPrefix(opHTTPPath, "/") { opHTTPPath = opHTTPPath[1:] } httpReq.URL.Path += opHTTPPath httpReq.URL.RawQuery = opQueryString } r := &Request{ Config: cfg, ClientInfo: clientInfo, Handlers: handlers.Copy(), Retryer: retryer, Time: time.Now(), ExpireTime: 0, Operation: operation, HTTPRequest: httpReq, Body: nil, Params: params, Error: err, Data: data, } r.SetBufferBody([]byte{}) return r } // A Option is a functional option that can augment or modify a request when // using a WithContext API operation method. type Option func(Request) // WithGetResponseHeader builds a request Option which will retrieve a single // header value from the HTTP Response. If there are multiple values for the // header key use WithGetResponseHeaders instead to access the http.Header // map directly. The passed in val pointer must be non-nil. // // This Option can be used multiple times with a single API operation. // // var id2, versionID string // svc.PutObjectWithContext(ctx, params, // request.WithGetResponseHeader("x-amz-id-2", &id2), // request.WithGetResponseHeader("x-amz-version-id", &versionID), // ) func WithGetResponseHeader(key string, val string) Option { return func(r Request) { r.Handlers.Complete.PushBack(func(req Request) { val = req.HTTPResponse.Header.Get(key) }) } } // WithGetResponseHeaders builds a request Option which will retrieve the // headers from the HTTP response and assign them to the passed in headers // variable. The passed in headers pointer must be non-nil. // // var headers http.Header // svc.PutObjectWithContext(ctx, params, request.WithGetResponseHeaders(&headers)) func WithGetResponseHeaders(headers http.Header) Option { return func(r Request) { r.Handlers.Complete.PushBack(func(req Request) { headers = req.HTTPResponse.Header }) } } // WithLogLevel is a request option that will set the request to use a specific // log level when the request is made. // // svc.PutObjectWithContext(ctx, params, request.WithLogLevel(aws.LogDebugWithHTTPBody) func WithLogLevel(l aws.LogLevelType) Option { return func(r Request) { r.Config.LogLevel = aws.LogLevel(l) } } // ApplyOptions will apply each option to the request calling them in the order // the were provided. func (r Request) ApplyOptions(opts ...Option) { for _, opt := range opts { opt(r) } } // Context will always returns a non-nil context. If Request does not have a // context aws.BackgroundContext will be returned. func (r Request) Context() aws.Context { if r.context != nil { return r.context } return aws.BackgroundContext() } // SetContext adds a Context to the current request that can be used to cancel // a in-flight request. The Context value must not be nil, or this method will // panic. // // Unlike http.Request.WithContext, SetContext does not return a copy of the // Request. It is not safe to use use a single Request value for multiple // requests. A new Request should be created for each API operation request. // // Go 1.6 and below: // The http.Request's Cancel field will be set to the Done() value of // the context. This will overwrite the Cancel field's value. // // Go 1.7 and above: // The http.Request.WithContext will be used to set the context on the underlying // http.Request. This will create a shallow copy of the http.Request. The SDK // may create sub contexts in the future for nested requests such as retries. func (r Request) SetContext(ctx aws.Context) { if ctx == nil { panic("context cannot be nil") } setRequestContext(r, ctx) } // WillRetry returns if the request's can be retried. func (r Request) WillRetry() bool { if !aws.IsReaderSeekable(r.Body) && r.HTTPRequest.Body != NoBody { return false } return r.Error != nil && aws.BoolValue(r.Retryable) && r.RetryCount < r.MaxRetries() } func fmtAttemptCount(retryCount, maxRetries int) string { return fmt.Sprintf("attempt %v/%v", retryCount, maxRetries) } // ParamsFilled returns if the request's parameters have been populated // and the parameters are valid. False is returned if no parameters are // provided or invalid. func (r Request) ParamsFilled() bool { return r.Params != nil && reflect.ValueOf(r.Params).Elem().IsValid() } // DataFilled returns true if the request's data for response deserialization // target has been set and is a valid. False is returned if data is not // set, or is invalid. func (r Request) DataFilled() bool { return r.Data != nil && reflect.ValueOf(r.Data).Elem().IsValid() } // SetBufferBody will set the request's body bytes that will be sent to // the service API. func (r Request) SetBufferBody(buf []byte) { r.SetReaderBody(bytes.NewReader(buf)) } // SetStringBody sets the body of the request to be backed by a string. func (r Request) SetStringBody(s string) { r.SetReaderBody(strings.NewReader(s)) } // SetReaderBody will set the request's body reader. func (r Request) SetReaderBody(reader io.ReadSeeker) { r.Body = reader if aws.IsReaderSeekable(reader) { var err error // Get the Bodies current offset so retries will start from the same // initial position. r.BodyStart, err = reader.Seek(0, sdkio.SeekCurrent) if err != nil { r.Error = awserr.New(ErrCodeSerialization, "failed to determine start of request body", err) return } } r.ResetBody() } // SetStreamingBody set the reader to be used for the request that will stream // bytes to the server. Request's Body must not be set to any reader. func (r Request) SetStreamingBody(reader io.ReadCloser) { r.streamingBody = reader r.SetReaderBody(aws.ReadSeekCloser(reader)) } // Presign returns the request's signed URL. Error will be returned // if the signing fails. The expire parameter is only used for presigned Amazon // S3 API requests. All other AWS services will use a fixed expiration // time of 15 minutes. // // It is invalid to create a presigned URL with a expire duration 0 or less. An // error is returned if expire duration is 0 or less. func (r Request) Presign(expire time.Duration) (string, error) { r = r.copy() // Presign requires all headers be hoisted. There is no way to retrieve // the signed headers not hoisted without this. Making the presigned URL // useless. r.NotHoist = false u, _, err := getPresignedURL(r, expire) return u, err } // PresignRequest behaves just like presign, with the addition of returning a // set of headers that were signed. The expire parameter is only used for // presigned Amazon S3 API requests. All other AWS services will use a fixed // expiration time of 15 minutes. // // It is invalid to create a presigned URL with a expire duration 0 or less. An // error is returned if expire duration is 0 or less. // // Returns the URL string for the API operation with signature in the query string, // and the HTTP headers that were included in the signature. These headers must // be included in any HTTP request made with the presigned URL. // // To prevent hoisting any headers to the query string set NotHoist to true on // this Request value prior to calling PresignRequest. func (r Request) PresignRequest(expire time.Duration) (string, http.Header, error) { r = r.copy() return getPresignedURL(r, expire) } // IsPresigned returns true if the request represents a presigned API url. func (r Request) IsPresigned() bool { return r.ExpireTime != 0 } func getPresignedURL(r Request, expire time.Duration) (string, http.Header, error) { if expire <= 0 { return "", nil, awserr.New( ErrCodeInvalidPresignExpire, "presigned URL requires an expire duration greater than 0", nil, ) } r.ExpireTime = expire if r.Operation.BeforePresignFn != nil { if err := r.Operation.BeforePresignFn(r); err != nil { return "", nil, err } } if err := r.Sign(); err != nil { return "", nil, err } return r.HTTPRequest.URL.String(), r.SignedHeaderVals, nil } const ( notRetrying = "not retrying" ) func debugLogReqError(r Request, stage, retryStr string, err error) { if !r.Config.LogLevel.Matches(aws.LogDebugWithRequestErrors) { return } r.Config.Logger.Log(fmt.Sprintf("DEBUG: %s %s/%s failed, %s, error %v", stage, r.ClientInfo.ServiceName, r.Operation.Name, retryStr, err)) } // Build will build the request's object so it can be signed and sent // to the service. Build will also validate all the request's parameters. // Any additional build Handlers set on this request will be run // in the order they were set. // // The request will only be built once. Multiple calls to build will have // no effect. // // If any Validate or Build errors occur the build will stop and the error // which occurred will be returned. func (r Request) Build() error { if !r.built { r.Handlers.Validate.Run(r) if r.Error != nil { debugLogReqError(r, "Validate Request", notRetrying, r.Error) return r.Error } r.Handlers.Build.Run(r) if r.Error != nil { debugLogReqError(r, "Build Request", notRetrying, r.Error) return r.Error } r.built = true } return r.Error } // Sign will sign the request, returning error if errors are encountered. // // Sign will build the request prior to signing. All Sign Handlers will // be executed in the order they were set. func (r Request) Sign() error { r.Build() if r.Error != nil { debugLogReqError(r, "Build Request", notRetrying, r.Error) return r.Error } SanitizeHostForHeader(r.HTTPRequest) r.Handlers.Sign.Run(r) return r.Error } func (r Request) getNextRequestBody() (body io.ReadCloser, err error) { if r.streamingBody != nil { return r.streamingBody, nil } if r.safeBody != nil { r.safeBody.Close() } r.safeBody, err = newOffsetReader(r.Body, r.BodyStart) if err != nil { return nil, awserr.New(ErrCodeSerialization, "failed to get next request body reader", err) } // Go 1.8 tightened and clarified the rules code needs to use when building // requests with the http package. Go 1.8 removed the automatic detection // of if the Request.Body was empty, or actually had bytes in it. The SDK // always sets the Request.Body even if it is empty and should not actually // be sent. This is incorrect. // // Go 1.8 did add a http.NoBody value that the SDK can use to tell the http // client that the request really should be sent without a body. The // Request.Body cannot be set to nil, which is preferable, because the // field is exported and could introduce nil pointer dereferences for users // of the SDK if they used that field. // // Related golang/go#18257 l, err := aws.SeekerLen(r.Body) if err != nil { return nil, awserr.New(ErrCodeSerialization, "failed to compute request body size", err) } if l == 0 { body = NoBody } else if l > 0 { body = r.safeBody } else { // Hack to prevent sending bodies for methods where the body // should be ignored by the server. Sending bodies on these // methods without an associated ContentLength will cause the // request to socket timeout because the server does not handle // Transfer-Encoding: chunked bodies for these methods. // // This would only happen if a aws.ReaderSeekerCloser was used with // a io.Reader that was not also an io.Seeker, or did not implement // Len() method. switch r.Operation.HTTPMethod { case "GET", "HEAD", "DELETE": body = NoBody default: body = r.safeBody } } return body, nil } // GetBody will return an io.ReadSeeker of the Request's underlying // input body with a concurrency safe wrapper. func (r Request) GetBody() io.ReadSeeker { return r.safeBody } // Send will send the request, returning error if errors are encountered. // // Send will sign the request prior to sending. All Send Handlers will // be executed in the order they were set. // // Canceling a request is non-deterministic. If a request has been canceled, // then the transport will choose, randomly, one of the state channels during // reads or getting the connection. // // readLoop() and getConn(req Request, cm connectMethod) // https://github.com/golang/go/blob/master/src/net/http/transport.go // // Send will not close the request.Request's body. func (r Request) Send() error { defer func() { // Ensure a non-nil HTTPResponse parameter is set to ensure handlers // checking for HTTPResponse values, don't fail. if r.HTTPResponse == nil { r.HTTPResponse = &http.Response{ Header: http.Header{}, Body: ioutil.NopCloser(&bytes.Buffer{}), } } // Regardless of success or failure of the request trigger the Complete // request handlers. r.Handlers.Complete.Run(r) }() if err := r.Error; err != nil { return err } for { r.Error = nil r.AttemptTime = time.Now() if err := r.Sign(); err != nil { debugLogReqError(r, "Sign Request", notRetrying, err) return err } if err := r.sendRequest(); err == nil { return nil } r.Handlers.Retry.Run(r) r.Handlers.AfterRetry.Run(r) if r.Error != nil \|\| !aws.BoolValue(r.Retryable) { return r.Error } if err := r.prepareRetry(); err != nil { r.Error = err return err } } } func (r Request) prepareRetry() error { if r.Config.LogLevel.Matches(aws.LogDebugWithRequestRetries) { r.Config.Logger.Log(fmt.Sprintf("DEBUG: Retrying Request %s/%s, attempt %d", r.ClientInfo.ServiceName, r.Operation.Name, r.RetryCount)) } // The previous http.Request will have a reference to the r.Body // and the HTTP Client's Transport may still be reading from // the request's body even though the Client's Do returned. r.HTTPRequest = copyHTTPRequest(r.HTTPRequest, nil) r.ResetBody() if err := r.Error; err != nil { return awserr.New(ErrCodeSerialization, "failed to prepare body for retry", err) } // Closing response body to ensure that no response body is leaked // between retry attempts. if r.HTTPResponse != nil && r.HTTPResponse.Body != nil { r.HTTPResponse.Body.Close() } return nil } func (r Request) sendRequest() (sendErr error) { defer r.Handlers.CompleteAttempt.Run(r) r.Retryable = nil r.Handlers.Send.Run(r) if r.Error != nil { debugLogReqError(r, "Send Request", fmtAttemptCount(r.RetryCount, r.MaxRetries()), r.Error) return r.Error } r.Handlers.UnmarshalMeta.Run(r) r.Handlers.ValidateResponse.Run(r) if r.Error != nil { r.Handlers.UnmarshalError.Run(r) debugLogReqError(r, "Validate Response", fmtAttemptCount(r.RetryCount, r.MaxRetries()), r.Error) return r.Error } r.Handlers.Unmarshal.Run(r) if r.Error != nil { debugLogReqError(r, "Unmarshal Response", fmtAttemptCount(r.RetryCount, r.MaxRetries()), r.Error) return r.Error } return nil } // copy will copy a request which will allow for local manipulation of the // request. func (r Request) copy() Request { req := &Request{} req = r req.Handlers = r.Handlers.Copy() op := r.Operation req.Operation = &op return req } // AddToUserAgent adds the string to the end of the request's current user agent. func AddToUserAgent(r Request, s string) { curUA := r.HTTPRequest.Header.Get("User-Agent") if len(curUA) > 0 { s = curUA + " " + s } r.HTTPRequest.Header.Set("User-Agent", s) } // SanitizeHostForHeader removes default port from host and updates request.Host func SanitizeHostForHeader(r http.Request) { host := getHost(r) port := portOnly(host) if port != "" && isDefaultPort(r.URL.Scheme, port) { r.Host = stripPort(host) } } // Returns host from request func getHost(r http.Request) string { if r.Host != "" { return r.Host } if r.URL == nil { return "" } return r.URL.Host } // Hostname returns u.Host, without any port number. // // If Host is an IPv6 literal with a port number, Hostname returns the // IPv6 literal without the square brackets. IPv6 literals may include // a zone identifier. // // Copied from the Go 1.8 standard library (net/url) func stripPort(hostport string) string { colon := strings.IndexByte(hostport, ':') if colon == -1 { return hostport } if i := strings.IndexByte(hostport, ']'); i != -1 { return strings.TrimPrefix(hostport[:i], "[") } return hostport[:colon] } // Port returns the port part of u.Host, without the leading colon. // If u.Host doesn't contain a port, Port returns an empty string. // // Copied from the Go 1.8 standard library (net/url) func portOnly(hostport string) string { colon := strings.IndexByte(hostport, ':') if colon == -1 { return "" } if i := strings.Index(hostport, "]:"); i != -1 { return hostport[i+len("]:"):] } if strings.Contains(hostport, "]") { return "" } return hostport[colon+len(":"):] } // Returns true if the specified URI is using the standard port // (i.e. port 80 for HTTP URIs or 443 for HTTPS URIs) func isDefaultPort(scheme, port string) bool { if port == "" { return true } lowerCaseScheme := strings.ToLower(scheme) if (lowerCaseScheme == "http" && port == "80") \|\| (lowerCaseScheme == "https" && port == "443") { return true } return false }	20.98 KB	2025-05-25 12:27
📄 request_1_5_test.go //go:build !go1.6 // +build !go1.6 package request_test import ( "errors" "github.com/aws/aws-sdk-go/aws/awserr" ) var errTimeout = awserr.New("foo", "bar", errors.New("net/http: request canceled Timeout"))	213 B	2025-05-25 12:27
📄 request_1_6_test.go //go:build go1.6 // +build go1.6 package request_test import ( "errors" "testing" "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/client/metadata" "github.com/aws/aws-sdk-go/aws/defaults" "github.com/aws/aws-sdk-go/aws/request" ) // go version 1.4 and 1.5 do not return an error. Version 1.5 will url encode // the uri while 1.4 will not func TestRequestInvalidEndpoint(t testing.T) { endpoint := "http://localhost:90 " r := request.New( aws.Config{}, metadata.ClientInfo{Endpoint: endpoint}, defaults.Handlers(), client.DefaultRetryer{}, &request.Operation{}, nil, nil, ) if r.Error == nil { t.Errorf("expect error") } } type timeoutErr struct { error } var errTimeout = awserr.New("foo", "bar", &timeoutErr{ errors.New("net/http: request canceled"), }) func (e timeoutErr) Timeout() bool { return true } func (e *timeoutErr) Temporary() bool { return true }	998 B	2025-05-25 12:27
📄 request_1_7.go //go:build !go1.8 // +build !go1.8 package request import "io" // NoBody is an io.ReadCloser with no bytes. Read always returns EOF // and Close always returns nil. It can be used in an outgoing client // request to explicitly signal that a request has zero bytes. // An alternative, however, is to simply set Request.Body to nil. // // Copy of Go 1.8 NoBody type from net/http/http.go type noBody struct{} func (noBody) Read([]byte) (int, error) { return 0, io.EOF } func (noBody) Close() error { return nil } func (noBody) WriteTo(io.Writer) (int64, error) { return 0, nil } // NoBody is an empty reader that will trigger the Go HTTP client to not include // and body in the HTTP request. var NoBody = noBody{} // ResetBody rewinds the request body back to its starting position, and // sets the HTTP Request body reference. When the body is read prior // to being sent in the HTTP request it will need to be rewound. // // ResetBody will automatically be called by the SDK's build handler, but if // the request is being used directly ResetBody must be called before the request // is Sent. SetStringBody, SetBufferBody, and SetReaderBody will automatically // call ResetBody. func (r *Request) ResetBody() { body, err := r.getNextRequestBody() if err != nil { r.Error = err return } r.HTTPRequest.Body = body }	1.33 KB	2025-05-25 12:27
📄 request_1_7_test.go //go:build !go1.8 // +build !go1.8 package request import ( "net/http" "strings" "testing" ) func TestResetBody_WithEmptyBody(t *testing.T) { r := Request{ HTTPRequest: &http.Request{}, } reader := strings.NewReader("") r.Body = reader r.ResetBody() if a, e := r.HTTPRequest.Body, (noBody{}); a != e { t.Errorf("expected request body to be set to reader, got %#v", r.HTTPRequest.Body) } }	409 B	2025-05-25 12:27
📄 request_1_8.go //go:build go1.8 // +build go1.8 package request import ( "net/http" "github.com/aws/aws-sdk-go/aws/awserr" ) // NoBody is a http.NoBody reader instructing Go HTTP client to not include // and body in the HTTP request. var NoBody = http.NoBody // ResetBody rewinds the request body back to its starting position, and // sets the HTTP Request body reference. When the body is read prior // to being sent in the HTTP request it will need to be rewound. // // ResetBody will automatically be called by the SDK's build handler, but if // the request is being used directly ResetBody must be called before the request // is Sent. SetStringBody, SetBufferBody, and SetReaderBody will automatically // call ResetBody. // // Will also set the Go 1.8's http.Request.GetBody member to allow retrying // PUT/POST redirects. func (r *Request) ResetBody() { body, err := r.getNextRequestBody() if err != nil { r.Error = awserr.New(ErrCodeSerialization, "failed to reset request body", err) return } r.HTTPRequest.Body = body r.HTTPRequest.GetBody = r.getNextRequestBody }	1.05 KB	2025-05-25 12:27
📄 request_1_8_test.go //go:build go1.8 // +build go1.8 package request_test import ( "bytes" "io" "net/http" "net/http/httptest" "strings" "testing" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/awstesting" "github.com/aws/aws-sdk-go/awstesting/unit" ) func TestResetBody_WithEmptyBody(t testing.T) { r := request.Request{ HTTPRequest: &http.Request{}, } reader := strings.NewReader("") r.Body = reader r.ResetBody() if a, e := r.HTTPRequest.Body, http.NoBody; a != e { t.Errorf("expected request body to be set to reader, got %#v", r.HTTPRequest.Body) } } func TestRequest_FollowPUTRedirects(t testing.T) { const bodySize = 1024 redirectHit := 0 endpointHit := 0 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r http.Request) { switch r.URL.Path { case "/redirect-me": u := r.URL u.Path = "/endpoint" w.Header().Set("Location", u.String()) w.WriteHeader(307) redirectHit++ case "/endpoint": b := bytes.Buffer{} io.Copy(&b, r.Body) r.Body.Close() if e, a := bodySize, b.Len(); e != a { t.Fatalf("expect %d body size, got %d", e, a) } endpointHit++ default: t.Fatalf("unexpected endpoint used, %q", r.URL.String()) } })) defer server.Close() svc := awstesting.NewClient(&aws.Config{ Region: unit.Session.Config.Region, DisableSSL: aws.Bool(true), Endpoint: aws.String(server.URL), }) req := svc.NewRequest(&request.Operation{ Name: "Operation", HTTPMethod: "PUT", HTTPPath: "/redirect-me", }, &struct{}{}, &struct{}{}) req.SetReaderBody(bytes.NewReader(make([]byte, bodySize))) err := req.Send() if err != nil { t.Errorf("expect no error, got %v", err) } if e, a := 1, redirectHit; e != a { t.Errorf("expect %d redirect hits, got %d", e, a) } if e, a := 1, endpointHit; e != a { t.Errorf("expect %d endpoint hits, got %d", e, a) } } func TestNewRequest_JoinEndpointWithOperationPathQuery(t testing.T) { cases := map[string]struct { HTTPPath string Endpoint string ExpectQuery string ExpectPath string }{ "no op HTTP Path": { HTTPPath: "", Endpoint: aws.String("https://foo.bar.aws/foo?bar=Baz"), ExpectPath: "/foo", ExpectQuery: "bar=Baz", }, "no trailing slash": { HTTPPath: "/", Endpoint: aws.String("https://foo.bar.aws"), ExpectPath: "/", ExpectQuery: "", }, "set query": { HTTPPath: "/?Foo=bar", Endpoint: aws.String("https://foo.bar.aws"), ExpectPath: "/", ExpectQuery: "Foo=bar", }, "squash query": { HTTPPath: "/?Foo=bar", Endpoint: aws.String("https://foo.bar.aws/?bar=Foo"), ExpectPath: "/", ExpectQuery: "Foo=bar", }, "trailing slash": { HTTPPath: "/", Endpoint: aws.String("https://foo.bar.aws/"), ExpectPath: "/", ExpectQuery: "", }, "trailing slash set query": { HTTPPath: "/?Foo=bar", Endpoint: aws.String("https://foo.bar.aws/"), ExpectPath: "/", ExpectQuery: "Foo=bar", }, } for name, c := range cases { t.Run(name, func(t *testing.T) { client := awstesting.NewClient(&aws.Config{ Endpoint: c.Endpoint, }) client.Handlers.Clear() r := client.NewRequest(&request.Operation{ Name: "FooBar", HTTPMethod: "GET", HTTPPath: c.HTTPPath, }, nil, nil) if e, a := c.ExpectPath, r.HTTPRequest.URL.Path; e != a { t.Errorf("expect %v path, got %v", e, a) } if e, a := c.ExpectQuery, r.HTTPRequest.URL.RawQuery; e != a { t.Errorf("expect %v query, got %v", e, a) } }) } }	3.52 KB	2025-05-25 12:27
📄 request_context.go //go:build go1.7 // +build go1.7 package request import "github.com/aws/aws-sdk-go/aws" // setContext updates the Request to use the passed in context for cancellation. // Context will also be used for request retry delay. // // Creates shallow copy of the http.Request with the WithContext method. func setRequestContext(r *Request, ctx aws.Context) { r.context = ctx r.HTTPRequest = r.HTTPRequest.WithContext(ctx) }	423 B	2025-05-25 12:27
📄 request_context_1_6.go //go:build !go1.7 // +build !go1.7 package request import "github.com/aws/aws-sdk-go/aws" // setContext updates the Request to use the passed in context for cancellation. // Context will also be used for request retry delay. // // Creates shallow copy of the http.Request with the WithContext method. func setRequestContext(r *Request, ctx aws.Context) { r.context = ctx r.HTTPRequest.Cancel = ctx.Done() }	412 B	2025-05-25 12:27
📄 request_context_test.go package request_test import ( "fmt" "strings" "testing" "github.com/aws/aws-sdk-go/aws/corehandlers" "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/awstesting" ) func TestRequest_SetContext(t testing.T) { svc := awstesting.NewClient() svc.Handlers.Clear() svc.Handlers.Send.PushBackNamed(corehandlers.SendHandler) r := svc.NewRequest(&request.Operation{Name: "Operation"}, nil, nil) ctx := &awstesting.FakeContext{DoneCh: make(chan struct{})} r.SetContext(ctx) ctx.Error = fmt.Errorf("context canceled") close(ctx.DoneCh) err := r.Send() if err == nil { t.Fatalf("expected error, got none") } // Only check against canceled because go 1.6 will not use the context's // Err(). if e, a := "canceled", err.Error(); !strings.Contains(a, e) { t.Errorf("expect %q to be in %q, but was not", e, a) } } func TestRequest_SetContextPanic(t testing.T) { defer func() { if r := recover(); r == nil { t.Fatalf("expect SetContext to panic, did not") } }() r := &request.Request{} r.SetContext(nil) }	1.03 KB	2025-05-25 12:27
📄 request_internal_test.go package request import ( "testing" ) func TestCopy(t *testing.T) { handlers := Handlers{} op := &Operation{} op.HTTPMethod = "Foo" req := &Request{} req.Operation = op req.Handlers = handlers r := req.copy() if r == req { t.Fatal("expect request pointer copy to be different") } if r.Operation == req.Operation { t.Errorf("expect request operation pointer to be different") } if e, a := req.Operation.HTTPMethod, r.Operation.HTTPMethod; e != a { t.Errorf("expect %q http method, got %q", e, a) } }	523 B	2025-05-25 12:27
📄 request_pagination.go package request import ( "reflect" "sync/atomic" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awsutil" ) // A Pagination provides paginating of SDK API operations which are paginatable. // Generally you should not use this type directly, but use the "Pages" API // operations method to automatically perform pagination for you. Such as, // "S3.ListObjectsPages", and "S3.ListObjectsPagesWithContext" methods. // // Pagination differs from a Paginator type in that pagination is the type that // does the pagination between API operations, and Paginator defines the // configuration that will be used per page request. // // for p.Next() { // data := p.Page().(s3.ListObjectsOutput) // // process the page's data // // ... // // break out of loop to stop fetching additional pages // } // // return p.Err() // // See service client API operation Pages methods for examples how the SDK will // use the Pagination type. type Pagination struct { // Function to return a Request value for each pagination request. // Any configuration or handlers that need to be applied to the request // prior to getting the next page should be done here before the request // returned. // // NewRequest should always be built from the same API operations. It is // undefined if different API operations are returned on subsequent calls. NewRequest func() (Request, error) // EndPageOnSameToken, when enabled, will allow the paginator to stop on // token that are the same as its previous tokens. EndPageOnSameToken bool started bool prevTokens []interface{} nextTokens []interface{} err error curPage interface{} } // HasNextPage will return true if Pagination is able to determine that the API // operation has additional pages. False will be returned if there are no more // pages remaining. // // Will always return true if Next has not been called yet. func (p Pagination) HasNextPage() bool { if !p.started { return true } hasNextPage := len(p.nextTokens) != 0 if p.EndPageOnSameToken { return hasNextPage && !awsutil.DeepEqual(p.nextTokens, p.prevTokens) } return hasNextPage } // Err returns the error Pagination encountered when retrieving the next page. func (p Pagination) Err() error { return p.err } // Page returns the current page. Page should only be called after a successful // call to Next. It is undefined what Page will return if Page is called after // Next returns false. func (p Pagination) Page() interface{} { return p.curPage } // Next will attempt to retrieve the next page for the API operation. When a page // is retrieved true will be returned. If the page cannot be retrieved, or there // are no more pages false will be returned. // // Use the Page method to retrieve the current page data. The data will need // to be cast to the API operation's output type. // // Use the Err method to determine if an error occurred if Page returns false. func (p Pagination) Next() bool { if !p.HasNextPage() { return false } req, err := p.NewRequest() if err != nil { p.err = err return false } if p.started { for i, intok := range req.Operation.InputTokens { awsutil.SetValueAtPath(req.Params, intok, p.nextTokens[i]) } } p.started = true err = req.Send() if err != nil { p.err = err return false } p.prevTokens = p.nextTokens p.nextTokens = req.nextPageTokens() p.curPage = req.Data return true } // A Paginator is the configuration data that defines how an API operation // should be paginated. This type is used by the API service models to define // the generated pagination config for service APIs. // // The Pagination type is what provides iterating between pages of an API. It // is only used to store the token metadata the SDK should use for performing // pagination. type Paginator struct { InputTokens []string OutputTokens []string LimitToken string TruncationToken string } // nextPageTokens returns the tokens to use when asking for the next page of data. func (r Request) nextPageTokens() []interface{} { if r.Operation.Paginator == nil { return nil } if r.Operation.TruncationToken != "" { tr, _ := awsutil.ValuesAtPath(r.Data, r.Operation.TruncationToken) if len(tr) == 0 { return nil } switch v := tr[0].(type) { case bool: if !aws.BoolValue(v) { return nil } case bool: if !v { return nil } } } tokens := []interface{}{} tokenAdded := false for _, outToken := range r.Operation.OutputTokens { vs, _ := awsutil.ValuesAtPath(r.Data, outToken) if len(vs) == 0 { tokens = append(tokens, nil) continue } v := vs[0] switch tv := v.(type) { case string: if len(aws.StringValue(tv)) == 0 { tokens = append(tokens, nil) continue } case string: if len(tv) == 0 { tokens = append(tokens, nil) continue } } tokenAdded = true tokens = append(tokens, v) } if !tokenAdded { return nil } return tokens } // Ensure a deprecated item is only logged once instead of each time its used. func logDeprecatedf(logger aws.Logger, flag int32, msg string) { if logger == nil { return } if atomic.CompareAndSwapInt32(flag, 0, 1) { logger.Log(msg) } } var ( logDeprecatedHasNextPage int32 logDeprecatedNextPage int32 logDeprecatedEachPage int32 ) // HasNextPage returns true if this request has more pages of data available. // // Deprecated Use Pagination type for configurable pagination of API operations func (r Request) HasNextPage() bool { logDeprecatedf(r.Config.Logger, &logDeprecatedHasNextPage, "Request.HasNextPage deprecated. Use Pagination type for configurable pagination of API operations") return len(r.nextPageTokens()) > 0 } // NextPage returns a new Request that can be executed to return the next // page of result data. Call .Send() on this request to execute it. // // Deprecated Use Pagination type for configurable pagination of API operations func (r Request) NextPage() Request { logDeprecatedf(r.Config.Logger, &logDeprecatedNextPage, "Request.NextPage deprecated. Use Pagination type for configurable pagination of API operations") tokens := r.nextPageTokens() if len(tokens) == 0 { return nil } data := reflect.New(reflect.TypeOf(r.Data).Elem()).Interface() nr := New(r.Config, r.ClientInfo, r.Handlers, r.Retryer, r.Operation, awsutil.CopyOf(r.Params), data) for i, intok := range nr.Operation.InputTokens { awsutil.SetValueAtPath(nr.Params, intok, tokens[i]) } return nr } // EachPage iterates over each page of a paginated request object. The fn // parameter should be a function with the following sample signature: // // func(page T, lastPage bool) bool { // return true // return false to stop iterating // } // // Where "T" is the structure type matching the output structure of the given // operation. For example, a request object generated by // DynamoDB.ListTablesRequest() would expect to see dynamodb.ListTablesOutput // as the structure "T". The lastPage value represents whether the page is // the last page of data or not. The return value of this function should // return true to keep iterating or false to stop. // // Deprecated Use Pagination type for configurable pagination of API operations func (r *Request) EachPage(fn func(data interface{}, isLastPage bool) (shouldContinue bool)) error { logDeprecatedf(r.Config.Logger, &logDeprecatedEachPage, "Request.EachPage deprecated. Use Pagination type for configurable pagination of API operations") for page := r; page != nil; page = page.NextPage() { if err := page.Send(); err != nil { return err } if getNextPage := fn(page.Data, !page.HasNextPage()); !getNextPage { return page.Error } } return nil }	7.58 KB	2025-05-25 12:27
📄 request_pagination_test.go package request_test import ( "reflect" "testing" "github.com/aws/aws-sdk-go/aws" "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/dynamodb" "github.com/aws/aws-sdk-go/service/route53" "github.com/aws/aws-sdk-go/service/s3" ) // Use DynamoDB methods for simplicity func TestPaginationQueryPage(t testing.T) { db := dynamodb.New(unit.Session) tokens, pages, numPages, gotToEnd := []map[string]dynamodb.AttributeValue{}, []map[string]dynamodb.AttributeValue{}, 0, false reqNum := 0 resps := []dynamodb.QueryOutput{ { LastEvaluatedKey: map[string]dynamodb.AttributeValue{"key": {S: aws.String("key1")}}, Count: aws.Int64(1), Items: []map[string]dynamodb.AttributeValue{ { "key": {S: aws.String("key1")}, }, }, }, { LastEvaluatedKey: map[string]dynamodb.AttributeValue{"key": {S: aws.String("key2")}}, Count: aws.Int64(1), Items: []map[string]dynamodb.AttributeValue{ { "key": {S: aws.String("key2")}, }, }, }, { LastEvaluatedKey: map[string]dynamodb.AttributeValue{}, Count: aws.Int64(1), Items: []map[string]dynamodb.AttributeValue{ { "key": {S: aws.String("key3")}, }, }, }, } db.Handlers.Send.Clear() // mock sending db.Handlers.Unmarshal.Clear() db.Handlers.UnmarshalMeta.Clear() db.Handlers.ValidateResponse.Clear() db.Handlers.Build.PushBack(func(r request.Request) { in := r.Params.(dynamodb.QueryInput) if in == nil { tokens = append(tokens, nil) } else if len(in.ExclusiveStartKey) != 0 { tokens = append(tokens, in.ExclusiveStartKey) } }) db.Handlers.Unmarshal.PushBack(func(r request.Request) { r.Data = resps[reqNum] reqNum++ }) params := &dynamodb.QueryInput{ Limit: aws.Int64(2), TableName: aws.String("tablename"), } err := db.QueryPages(params, func(p dynamodb.QueryOutput, last bool) bool { numPages++ pages = append(pages, p.Items...) if last { if gotToEnd { t.Errorf("last=true happened twice") } gotToEnd = true } return true }) if err != nil { t.Errorf("expect nil, %v", err) } if e, a := []map[string]dynamodb.AttributeValue{ {"key": {S: aws.String("key1")}}, {"key": {S: aws.String("key2")}}, }, tokens; !reflect.DeepEqual(e, a) { t.Errorf("expect %v, got %v", e, a) } if e, a := []map[string]dynamodb.AttributeValue{ {"key": {S: aws.String("key1")}}, {"key": {S: aws.String("key2")}}, {"key": {S: aws.String("key3")}}, }, pages; !reflect.DeepEqual(e, a) { t.Errorf("expect %v, got %v", e, a) } if e, a := 3, numPages; e != a { t.Errorf("expect %v, got %v", e, a) } if !gotToEnd { t.Errorf("expect true") } if params.ExclusiveStartKey != nil { t.Errorf("expect nil, %v", err) } } // Use DynamoDB methods for simplicity func TestPagination(t testing.T) { db := dynamodb.New(unit.Session) tokens, pages, numPages, gotToEnd := []string{}, []string{}, 0, false reqNum := 0 resps := []dynamodb.ListTablesOutput{ {TableNames: []string{aws.String("Table1"), aws.String("Table2")}, LastEvaluatedTableName: aws.String("Table2")}, {TableNames: []string{aws.String("Table3"), aws.String("Table4")}, LastEvaluatedTableName: aws.String("Table4")}, {TableNames: []string{aws.String("Table5")}}, } db.Handlers.Send.Clear() // mock sending db.Handlers.Unmarshal.Clear() db.Handlers.UnmarshalMeta.Clear() db.Handlers.ValidateResponse.Clear() db.Handlers.Build.PushBack(func(r request.Request) { in := r.Params.(dynamodb.ListTablesInput) if in == nil { tokens = append(tokens, "") } else if in.ExclusiveStartTableName != nil { tokens = append(tokens, in.ExclusiveStartTableName) } }) db.Handlers.Unmarshal.PushBack(func(r request.Request) { r.Data = resps[reqNum] reqNum++ }) params := &dynamodb.ListTablesInput{Limit: aws.Int64(2)} err := db.ListTablesPages(params, func(p dynamodb.ListTablesOutput, last bool) bool { numPages++ for _, t := range p.TableNames { pages = append(pages, t) } if last { if gotToEnd { t.Errorf("last=true happened twice") } gotToEnd = true } return true }) if e, a := []string{"Table2", "Table4"}, tokens; !reflect.DeepEqual(e, a) { t.Errorf("expect %v, got %v", e, a) } if e, a := []string{"Table1", "Table2", "Table3", "Table4", "Table5"}, pages; !reflect.DeepEqual(e, a) { t.Errorf("expect %v, got %v", e, a) } if e, a := 3, numPages; e != a { t.Errorf("expect %v, got %v", e, a) } if !gotToEnd { t.Errorf("expect true") } if err != nil { t.Errorf("expect nil, %v", err) } if params.ExclusiveStartTableName != nil { t.Errorf("expect nil, %v", err) } } // Use DynamoDB methods for simplicity func TestPaginationEachPage(t testing.T) { db := dynamodb.New(unit.Session) tokens, pages, numPages, gotToEnd := []string{}, []string{}, 0, false reqNum := 0 resps := []dynamodb.ListTablesOutput{ {TableNames: []string{aws.String("Table1"), aws.String("Table2")}, LastEvaluatedTableName: aws.String("Table2")}, {TableNames: []string{aws.String("Table3"), aws.String("Table4")}, LastEvaluatedTableName: aws.String("Table4")}, {TableNames: []string{aws.String("Table5")}}, } db.Handlers.Send.Clear() // mock sending db.Handlers.Unmarshal.Clear() db.Handlers.UnmarshalMeta.Clear() db.Handlers.ValidateResponse.Clear() db.Handlers.Build.PushBack(func(r request.Request) { in := r.Params.(dynamodb.ListTablesInput) if in == nil { tokens = append(tokens, "") } else if in.ExclusiveStartTableName != nil { tokens = append(tokens, in.ExclusiveStartTableName) } }) db.Handlers.Unmarshal.PushBack(func(r request.Request) { r.Data = resps[reqNum] reqNum++ }) params := &dynamodb.ListTablesInput{Limit: aws.Int64(2)} req, _ := db.ListTablesRequest(params) err := req.EachPage(func(p interface{}, last bool) bool { numPages++ for _, t := range p.(dynamodb.ListTablesOutput).TableNames { pages = append(pages, t) } if last { if gotToEnd { t.Errorf("last=true happened twice") } gotToEnd = true } return true }) if e, a := []string{"Table2", "Table4"}, tokens; !reflect.DeepEqual(e, a) { t.Errorf("expect %v, got %v", e, a) } if e, a := []string{"Table1", "Table2", "Table3", "Table4", "Table5"}, pages; !reflect.DeepEqual(e, a) { t.Errorf("expect %v, got %v", e, a) } if e, a := 3, numPages; e != a { t.Errorf("expect %v, got %v", e, a) } if !gotToEnd { t.Errorf("expect true") } if err != nil { t.Errorf("expect nil, %v", err) } } // Use DynamoDB methods for simplicity func TestPaginationEarlyExit(t testing.T) { db := dynamodb.New(unit.Session) numPages, gotToEnd := 0, false reqNum := 0 resps := []dynamodb.ListTablesOutput{ {TableNames: []string{aws.String("Table1"), aws.String("Table2")}, LastEvaluatedTableName: aws.String("Table2")}, {TableNames: []string{aws.String("Table3"), aws.String("Table4")}, LastEvaluatedTableName: aws.String("Table4")}, {TableNames: []string{aws.String("Table5")}}, } db.Handlers.Send.Clear() // mock sending db.Handlers.Unmarshal.Clear() db.Handlers.UnmarshalMeta.Clear() db.Handlers.ValidateResponse.Clear() db.Handlers.Unmarshal.PushBack(func(r request.Request) { r.Data = resps[reqNum] reqNum++ }) params := &dynamodb.ListTablesInput{Limit: aws.Int64(2)} err := db.ListTablesPages(params, func(p dynamodb.ListTablesOutput, last bool) bool { numPages++ if numPages == 2 { return false } if last { if gotToEnd { t.Errorf("last=true happened twice") } gotToEnd = true } return true }) if e, a := 2, numPages; e != a { t.Errorf("expect %v, got %v", e, a) } if gotToEnd { t.Errorf("expect false") } if err != nil { t.Errorf("expect nil, %v", err) } } func TestSkipPagination(t testing.T) { client := s3.New(unit.Session) client.Handlers.Send.Clear() // mock sending client.Handlers.Unmarshal.Clear() client.Handlers.UnmarshalMeta.Clear() client.Handlers.ValidateResponse.Clear() client.Handlers.Unmarshal.PushBack(func(r request.Request) { r.Data = &s3.HeadBucketOutput{} }) req, _ := client.HeadBucketRequest(&s3.HeadBucketInput{Bucket: aws.String("bucket")}) numPages, gotToEnd := 0, false req.EachPage(func(p interface{}, last bool) bool { numPages++ if last { gotToEnd = true } return true }) if e, a := 1, numPages; e != a { t.Errorf("expect %v, got %v", e, a) } if !gotToEnd { t.Errorf("expect true") } } // Use S3 for simplicity func TestPaginationTruncation(t testing.T) { client := s3.New(unit.Session) reqNum := 0 resps := []s3.ListObjectsOutput{ {IsTruncated: aws.Bool(true), Contents: []s3.Object{{Key: aws.String("Key1")}}}, {IsTruncated: aws.Bool(true), Contents: []s3.Object{{Key: aws.String("Key2")}}}, {IsTruncated: aws.Bool(false), Contents: []s3.Object{{Key: aws.String("Key3")}}}, {IsTruncated: aws.Bool(true), Contents: []s3.Object{{Key: aws.String("Key4")}}}, } client.Handlers.Send.Clear() // mock sending client.Handlers.Unmarshal.Clear() client.Handlers.UnmarshalMeta.Clear() client.Handlers.ValidateResponse.Clear() client.Handlers.Unmarshal.PushBack(func(r request.Request) { r.Data = resps[reqNum] reqNum++ }) params := &s3.ListObjectsInput{Bucket: aws.String("bucket")} results := []string{} err := client.ListObjectsPages(params, func(p s3.ListObjectsOutput, last bool) bool { results = append(results, p.Contents[0].Key) return true }) if e, a := []string{"Key1", "Key2", "Key3"}, results; !reflect.DeepEqual(e, a) { t.Errorf("expect %v, got %v", e, a) } if err != nil { t.Errorf("expect nil, %v", err) } // Try again without truncation token at all reqNum = 0 resps[1].IsTruncated = nil resps[2].IsTruncated = aws.Bool(true) results = []string{} err = client.ListObjectsPages(params, func(p s3.ListObjectsOutput, last bool) bool { results = append(results, p.Contents[0].Key) return true }) if e, a := []string{"Key1", "Key2"}, results; !reflect.DeepEqual(e, a) { t.Errorf("expect %v, got %v", e, a) } if err != nil { t.Errorf("expect nil, %v", err) } } func TestPaginationNilToken(t testing.T) { client := route53.New(unit.Session) reqNum := 0 resps := []route53.ListResourceRecordSetsOutput{ { ResourceRecordSets: []route53.ResourceRecordSet{ {Name: aws.String("first.example.com.")}, }, IsTruncated: aws.Bool(true), NextRecordName: aws.String("second.example.com."), NextRecordType: aws.String("MX"), NextRecordIdentifier: aws.String("second"), MaxItems: aws.String("1"), }, { ResourceRecordSets: []route53.ResourceRecordSet{ {Name: aws.String("second.example.com.")}, }, IsTruncated: aws.Bool(true), NextRecordName: aws.String("third.example.com."), NextRecordType: aws.String("MX"), MaxItems: aws.String("1"), }, { ResourceRecordSets: []route53.ResourceRecordSet{ {Name: aws.String("third.example.com.")}, }, IsTruncated: aws.Bool(false), MaxItems: aws.String("1"), }, } client.Handlers.Send.Clear() // mock sending client.Handlers.Unmarshal.Clear() client.Handlers.UnmarshalMeta.Clear() client.Handlers.ValidateResponse.Clear() idents := []string{} client.Handlers.Build.PushBack(func(r request.Request) { p := r.Params.(route53.ListResourceRecordSetsInput) idents = append(idents, aws.StringValue(p.StartRecordIdentifier)) }) client.Handlers.Unmarshal.PushBack(func(r request.Request) { r.Data = resps[reqNum] reqNum++ }) params := &route53.ListResourceRecordSetsInput{ HostedZoneId: aws.String("id-zone"), } results := []string{} err := client.ListResourceRecordSetsPages(params, func(p route53.ListResourceRecordSetsOutput, last bool) bool { results = append(results, p.ResourceRecordSets[0].Name) return true }) if err != nil { t.Errorf("expect nil, %v", err) } if e, a := []string{"", "second", ""}, idents; !reflect.DeepEqual(e, a) { t.Errorf("expect %v, got %v", e, a) } if e, a := []string{"first.example.com.", "second.example.com.", "third.example.com."}, results; !reflect.DeepEqual(e, a) { t.Errorf("expect %v, got %v", e, a) } } func TestPaginationNilInput(t testing.T) { // Code generation doesn't have a great way to verify the code is correct // other than being run via unit tests in the SDK. This should be fixed // So code generation can be validated independently. client := s3.New(unit.Session) client.Handlers.Validate.Clear() client.Handlers.Send.Clear() // mock sending client.Handlers.Unmarshal.Clear() client.Handlers.UnmarshalMeta.Clear() client.Handlers.ValidateResponse.Clear() client.Handlers.Unmarshal.PushBack(func(r request.Request) { r.Data = &s3.ListObjectsOutput{} }) gotToEnd := false numPages := 0 err := client.ListObjectsPages(nil, func(p s3.ListObjectsOutput, last bool) bool { numPages++ if last { gotToEnd = true } return true }) if err != nil { t.Fatalf("expect no error, but got %v", err) } if e, a := 1, numPages; e != a { t.Errorf("expect %d number pages but got %d", e, a) } if !gotToEnd { t.Errorf("expect to of gotten to end, did not") } } func TestPaginationWithContextNilInput(t testing.T) { // Code generation doesn't have a great way to verify the code is correct // other than being run via unit tests in the SDK. This should be fixed // So code generation can be validated independently. client := s3.New(unit.Session) client.Handlers.Validate.Clear() client.Handlers.Send.Clear() // mock sending client.Handlers.Unmarshal.Clear() client.Handlers.UnmarshalMeta.Clear() client.Handlers.ValidateResponse.Clear() client.Handlers.Unmarshal.PushBack(func(r request.Request) { r.Data = &s3.ListObjectsOutput{} }) gotToEnd := false numPages := 0 ctx := &awstesting.FakeContext{DoneCh: make(chan struct{})} err := client.ListObjectsPagesWithContext(ctx, nil, func(p s3.ListObjectsOutput, last bool) bool { numPages++ if last { gotToEnd = true } return true }) if err != nil { t.Fatalf("expect no error, but got %v", err) } if e, a := 1, numPages; e != a { t.Errorf("expect %d number pages but got %d", e, a) } if !gotToEnd { t.Errorf("expect to of gotten to end, did not") } } func TestPagination_Standalone(t testing.T) { type testPageInput struct { NextToken string } type testPageOutput struct { Value string NextToken string } type testCase struct { Value, PrevToken, NextToken string } type testCaseList struct { StopOnSameToken bool Cases []testCase } cases := []testCaseList{ { Cases: []testCase{ {aws.String("FirstValue"), aws.String("InitalToken"), aws.String("FirstToken")}, {aws.String("SecondValue"), aws.String("FirstToken"), aws.String("SecondToken")}, {aws.String("ThirdValue"), aws.String("SecondToken"), nil}, }, StopOnSameToken: false, }, { Cases: []testCase{ {aws.String("FirstValue"), aws.String("InitalToken"), aws.String("FirstToken")}, {aws.String("SecondValue"), aws.String("FirstToken"), aws.String("SecondToken")}, {aws.String("ThirdValue"), aws.String("SecondToken"), aws.String("")}, }, StopOnSameToken: false, }, { Cases: []testCase{ {aws.String("FirstValue"), aws.String("InitalToken"), aws.String("FirstToken")}, {aws.String("SecondValue"), aws.String("FirstToken"), aws.String("SecondToken")}, {nil, aws.String("SecondToken"), aws.String("SecondToken")}, }, StopOnSameToken: true, }, { Cases: []testCase{ {aws.String("FirstValue"), aws.String("InitalToken"), aws.String("FirstToken")}, {aws.String("SecondValue"), aws.String("FirstToken"), aws.String("SecondToken")}, {aws.String("SecondValue"), aws.String("SecondToken"), aws.String("SecondToken")}, }, StopOnSameToken: true, }, } for _, testcase := range cases { c := testcase.Cases input := testPageInput{ NextToken: c[0].PrevToken, } svc := awstesting.NewClient() i := 0 p := request.Pagination{ EndPageOnSameToken: testcase.StopOnSameToken, NewRequest: func() (request.Request, error) { r := svc.NewRequest( &request.Operation{ Name: "Operation", Paginator: &request.Paginator{ InputTokens: []string{"NextToken"}, OutputTokens: []string{"NextToken"}, }, }, &input, &testPageOutput{}, ) // Setup handlers for testing r.Handlers.Clear() r.Handlers.Build.PushBack(func(req request.Request) { if e, a := len(c), i+1; a > e { t.Fatalf("expect no more than %d requests, got %d", e, a) } in := req.Params.(testPageInput) if e, a := aws.StringValue(c[i].PrevToken), aws.StringValue(in.NextToken); e != a { t.Errorf("%d, expect NextToken input %q, got %q", i, e, a) } }) r.Handlers.Unmarshal.PushBack(func(req request.Request) { out := &testPageOutput{ Value: c[i].Value, } if c[i].NextToken != nil { next := c[i].NextToken out.NextToken = aws.String(next) } req.Data = out }) return r, nil }, } for p.Next() { data := p.Page().(testPageOutput) if e, a := aws.StringValue(c[i].Value), aws.StringValue(data.Value); e != a { t.Errorf("%d, expect Value to be %q, got %q", i, e, a) } if e, a := aws.StringValue(c[i].NextToken), aws.StringValue(data.NextToken); e != a { t.Errorf("%d, expect NextToken to be %q, got %q", i, e, a) } i++ } if e, a := len(c), i; e != a { t.Errorf("expected to process %d pages, did %d", e, a) } if err := p.Err(); err != nil { t.Fatalf("%d, expected no error, got %v", i, err) } } } // Benchmarks var benchResps = []dynamodb.ListTablesOutput{ {TableNames: []string{aws.String("TABLE"), aws.String("NXT")}, LastEvaluatedTableName: aws.String("NXT")}, {TableNames: []string{aws.String("TABLE"), aws.String("NXT")}, LastEvaluatedTableName: aws.String("NXT")}, {TableNames: []string{aws.String("TABLE"), aws.String("NXT")}, LastEvaluatedTableName: aws.String("NXT")}, {TableNames: []string{aws.String("TABLE"), aws.String("NXT")}, LastEvaluatedTableName: aws.String("NXT")}, {TableNames: []string{aws.String("TABLE"), aws.String("NXT")}, LastEvaluatedTableName: aws.String("NXT")}, {TableNames: []string{aws.String("TABLE"), aws.String("NXT")}, LastEvaluatedTableName: aws.String("NXT")}, {TableNames: []string{aws.String("TABLE"), aws.String("NXT")}, LastEvaluatedTableName: aws.String("NXT")}, {TableNames: []string{aws.String("TABLE"), aws.String("NXT")}, LastEvaluatedTableName: aws.String("NXT")}, {TableNames: []string{aws.String("TABLE"), aws.String("NXT")}, LastEvaluatedTableName: aws.String("NXT")}, {TableNames: []string{aws.String("TABLE"), aws.String("NXT")}, LastEvaluatedTableName: aws.String("NXT")}, {TableNames: []string{aws.String("TABLE"), aws.String("NXT")}, LastEvaluatedTableName: aws.String("NXT")}, {TableNames: []string{aws.String("TABLE"), aws.String("NXT")}, LastEvaluatedTableName: aws.String("NXT")}, {TableNames: []string{aws.String("TABLE"), aws.String("NXT")}, LastEvaluatedTableName: aws.String("NXT")}, {TableNames: []string{aws.String("TABLE")}}, } var benchDb = func() dynamodb.DynamoDB { db := dynamodb.New(unit.Session) db.Handlers.Send.Clear() // mock sending db.Handlers.Unmarshal.Clear() db.Handlers.UnmarshalMeta.Clear() db.Handlers.ValidateResponse.Clear() return db } func BenchmarkCodegenIterator(b testing.B) { reqNum := 0 db := benchDb() db.Handlers.Unmarshal.PushBack(func(r request.Request) { r.Data = benchResps[reqNum] reqNum++ }) input := &dynamodb.ListTablesInput{Limit: aws.Int64(2)} iter := func(fn func(dynamodb.ListTablesOutput, bool) bool) error { page, _ := db.ListTablesRequest(input) for ; page != nil; page = page.NextPage() { page.Send() out := page.Data.(dynamodb.ListTablesOutput) if result := fn(out, !page.HasNextPage()); page.Error != nil \|\| !result { return page.Error } } return nil } for i := 0; i < b.N; i++ { reqNum = 0 iter(func(p dynamodb.ListTablesOutput, last bool) bool { return true }) } } func BenchmarkEachPageIterator(b testing.B) { reqNum := 0 db := benchDb() db.Handlers.Unmarshal.PushBack(func(r request.Request) { r.Data = benchResps[reqNum] reqNum++ }) input := &dynamodb.ListTablesInput{Limit: aws.Int64(2)} for i := 0; i < b.N; i++ { reqNum = 0 req, _ := db.ListTablesRequest(input) req.EachPage(func(p interface{}, last bool) bool { return true }) } }	20.15 KB	2025-05-25 12:27
📄 request_resetbody_test.go package request import ( "bytes" "io" "net/http" "strings" "testing" "github.com/aws/aws-sdk-go/aws" ) func TestResetBody_WithBodyContents(t testing.T) { r := Request{ HTTPRequest: &http.Request{}, } reader := strings.NewReader("abc") r.Body = reader r.ResetBody() if v, ok := r.HTTPRequest.Body.(offsetReader); !ok \|\| v == nil { t.Errorf("expected request body to be set to reader, got %#v", r.HTTPRequest.Body) } } type mockReader struct{} func (mockReader) Read([]byte) (int, error) { return 0, io.EOF } func TestResetBody_ExcludeEmptyUnseekableBodyByMethod(t *testing.T) { cases := []struct { Method string Body io.ReadSeeker IsNoBody bool }{ { Method: "GET", IsNoBody: true, Body: aws.ReadSeekCloser(mockReader{}), }, { Method: "HEAD", IsNoBody: true, Body: aws.ReadSeekCloser(mockReader{}), }, { Method: "DELETE", IsNoBody: true, Body: aws.ReadSeekCloser(mockReader{}), }, { Method: "PUT", IsNoBody: false, Body: aws.ReadSeekCloser(mockReader{}), }, { Method: "PATCH", IsNoBody: false, Body: aws.ReadSeekCloser(mockReader{}), }, { Method: "POST", IsNoBody: false, Body: aws.ReadSeekCloser(mockReader{}), }, { Method: "GET", IsNoBody: false, Body: aws.ReadSeekCloser(bytes.NewBuffer([]byte("abc"))), }, { Method: "GET", IsNoBody: true, Body: aws.ReadSeekCloser(bytes.NewBuffer(nil)), }, { Method: "POST", IsNoBody: false, Body: aws.ReadSeekCloser(bytes.NewBuffer([]byte("abc"))), }, { Method: "POST", IsNoBody: true, Body: aws.ReadSeekCloser(bytes.NewBuffer(nil)), }, } for i, c := range cases { r := Request{ HTTPRequest: &http.Request{}, Operation: &Operation{ HTTPMethod: c.Method, }, } r.SetReaderBody(c.Body) if a, e := r.HTTPRequest.Body == NoBody, c.IsNoBody; a != e { t.Errorf("%d, expect body to be set to noBody(%t), but was %t", i, e, a) } } }	1.98 KB	2025-05-25 12:27
📄 request_retry_test.go package request import ( "fmt" "net" "net/http" "net/http/httptest" "net/url" "os" "testing" "time" "github.com/aws/aws-sdk-go/aws/awserr" ) func newRequest(t testing.T, url string) http.Request { r, err := http.NewRequest("GET", url, nil) if err != nil { t.Fatalf("can't forge request: %v", err) } return r } func TestShouldRetryError_nil(t testing.T) { if shouldRetryError(nil) != true { t.Error("shouldRetryError(nil) should return true") } } func TestShouldRetryError_timeout(t testing.T) { tr := &http.Transport{} defer tr.CloseIdleConnections() client := http.Client{ Timeout: time.Nanosecond, Transport: tr, } resp, err := client.Do(newRequest(t, "https://179.179.179.179/no/such/host")) if resp != nil { resp.Body.Close() } if err == nil { t.Fatal("This should have failed.") } debugerr(t, err) if shouldRetryError(err) == false { t.Errorf("this request timed out and should be retried") } } func TestShouldRetryError_cancelled(t testing.T) { tr := &http.Transport{} defer tr.CloseIdleConnections() client := http.Client{ Transport: tr, } cancelWait := make(chan bool) srvrWait := make(chan bool) srvr := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req http.Request) { close(cancelWait) // Trigger the request cancel. time.Sleep(100 * time.Millisecond) fmt.Fprintf(w, "Hello") w.(http.Flusher).Flush() // send headers and some body <-srvrWait // block forever })) defer srvr.Close() defer close(srvrWait) r := newRequest(t, srvr.URL) ch := make(chan struct{}) r.Cancel = ch // Ensure the request has started, and client has started to receive bytes. // This ensures the test is stable and does not run into timing with the // request being canceled, before or after the http request is made. go func() { <-cancelWait close(ch) // request is cancelled before anything }() resp, err := client.Do(r) if resp != nil { resp.Body.Close() } if err == nil { t.Fatal("This should have failed.") } debugerr(t, err) if shouldRetryError(err) == true { t.Errorf("this request was cancelled and should not be retried") } } func TestShouldRetry(t testing.T) { syscallError := os.SyscallError{ Err: ErrInvalidParams{}, Syscall: "open", } opError := net.OpError{ Op: "dial", Net: "tcp", Source: net.Addr(nil), Err: &syscallError, } urlError := url.Error{ Op: "Post", URL: "https://localhost:52398", Err: &opError, } origError := awserr.New("ErrorTestShouldRetry", "Test should retry when error received", &urlError).OrigErr() if e, a := true, shouldRetryError(origError); e != a { t.Errorf("Expected to return %v to retry when error occurred, got %v instead", e, a) } } func debugerr(t testing.T, err error) { t.Logf("Error, %v", err) switch err := err.(type) { case temporary: t.Logf("%s is a temporary error: %t", err, err.Temporary()) return case url.Error: t.Logf("err: %s, nested err: %#v", err, err.Err) if operr, ok := err.Err.(net.OpError); ok { t.Logf("operr: %#v", operr) } debugerr(t, err.Err) return default: return } } func TestRequest_retryCustomCodes(t *testing.T) { cases := map[string]struct { Code string RetryErrorCodes []string ThrottleErrorCodes []string Retryable bool Throttle bool }{ "retry code": { Code: "RetryMePlease", RetryErrorCodes: []string{ "RetryMePlease", "SomeOtherError", }, Retryable: true, }, "throttle code": { Code: "AThrottleableError", RetryErrorCodes: []string{ "RetryMePlease", "SomeOtherError", }, ThrottleErrorCodes: []string{ "AThrottleableError", "SomeOtherError", }, Throttle: true, }, "unknown code": { Code: "UnknownCode", RetryErrorCodes: []string{ "RetryMePlease", "SomeOtherError", }, Retryable: false, }, } for name, c := range cases { req := Request{ HTTPRequest: &http.Request{}, HTTPResponse: &http.Response{}, Error: awserr.New(c.Code, "some error", nil), RetryErrorCodes: c.RetryErrorCodes, ThrottleErrorCodes: c.ThrottleErrorCodes, } retryable := req.IsErrorRetryable() if e, a := c.Retryable, retryable; e != a { t.Errorf("%s, expect %v retryable, got %v", name, e, a) } throttle := req.IsErrorThrottle() if e, a := c.Throttle, throttle; e != a { t.Errorf("%s, expect %v throttle, got %v", name, e, a) } } }	4.39 KB	2025-05-25 12:27
📄 request_test.go package request_test import ( "bytes" "encoding/json" "errors" "fmt" "io" "io/ioutil" "net" "net/http" "net/http/httptest" "net/url" "reflect" "runtime" "strconv" "strings" "testing" "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" "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/awstesting" "github.com/aws/aws-sdk-go/awstesting/unit" "github.com/aws/aws-sdk-go/private/protocol/rest" ) type tempNetworkError struct { op string msg string isTemp bool } func (e tempNetworkError) Temporary() bool { return e.isTemp } func (e tempNetworkError) Error() string { return fmt.Sprintf("%s: %s", e.op, e.msg) } var ( // net.OpError accept, are always temporary errAcceptConnectionResetStub = &tempNetworkError{ isTemp: true, op: "accept", msg: "connection reset", } // net.OpError read for ECONNRESET is not temporary. errReadConnectionResetStub = &tempNetworkError{ isTemp: false, op: "read", msg: "connection reset", } // net.OpError write for ECONNRESET may not be temporary, but is treaded as // temporary by the SDK. errWriteConnectionResetStub = &tempNetworkError{ isTemp: false, op: "write", msg: "connection reset", } // net.OpError write for broken pipe may not be temporary, but is treaded as // temporary by the SDK. errWriteBrokenPipeStub = &tempNetworkError{ isTemp: false, op: "write", msg: "broken pipe", } // Generic connection reset error errConnectionResetStub = errors.New("connection reset") // use of closed network connection error errUseOfClosedConnectionStub = errors.New("use of closed network connection") ) type testData struct { Data string } func body(str string) io.ReadCloser { return ioutil.NopCloser(bytes.NewReader([]byte(str))) } func unmarshal(req request.Request) { defer req.HTTPResponse.Body.Close() if req.Data != nil { json.NewDecoder(req.HTTPResponse.Body).Decode(req.Data) } } func unmarshalError(req request.Request) { bodyBytes, err := ioutil.ReadAll(req.HTTPResponse.Body) if err != nil { req.Error = awserr.New("UnmarshaleError", req.HTTPResponse.Status, err) return } if len(bodyBytes) == 0 { req.Error = awserr.NewRequestFailure( awserr.New("UnmarshaleError", req.HTTPResponse.Status, fmt.Errorf("empty body")), req.HTTPResponse.StatusCode, "", ) return } var jsonErr jsonErrorResponse if err := json.Unmarshal(bodyBytes, &jsonErr); err != nil { req.Error = awserr.New("UnmarshaleError", "JSON unmarshal", err) return } req.Error = awserr.NewRequestFailure( awserr.New(jsonErr.Code, jsonErr.Message, nil), req.HTTPResponse.StatusCode, "", ) } type jsonErrorResponse struct { Code string `json:"__type"` Message string `json:"message"` } // test that retries occur for 5xx status codes func TestRequestRecoverRetry5xx(t testing.T) { reqNum := 0 reqs := []http.Response{ {StatusCode: 500, Body: body(`{"__type":"UnknownError","message":"An error occurred."}`)}, {StatusCode: 502, Body: body(`{"__type":"UnknownError","message":"An error occurred."}`)}, {StatusCode: 200, Body: body(`{"data":"valid"}`)}, } s := awstesting.NewClient(&aws.Config{ MaxRetries: aws.Int(10), SleepDelay: func(time.Duration) {}, }) s.Handlers.Validate.Clear() s.Handlers.Unmarshal.PushBack(unmarshal) s.Handlers.UnmarshalError.PushBack(unmarshalError) s.Handlers.Send.Clear() // mock sending s.Handlers.Send.PushBack(func(r request.Request) { r.HTTPResponse = &reqs[reqNum] reqNum++ }) out := &testData{} r := s.NewRequest(&request.Operation{Name: "Operation"}, nil, out) err := r.Send() if err != nil { t.Fatalf("expect no error, but got %v", err) } if e, a := 2, r.RetryCount; e != a { t.Errorf("expect %d retry count, got %d", e, a) } if e, a := "valid", out.Data; e != a { t.Errorf("expect %q output got %q", e, a) } } // test that retries occur for 4xx status codes with a response type that can be retried - see `shouldRetry` func TestRequestRecoverRetry4xxRetryable(t testing.T) { reqNum := 0 reqs := []http.Response{ {StatusCode: 400, Body: body(`{"__type":"Throttling","message":"Rate exceeded."}`)}, {StatusCode: 400, Body: body(`{"__type":"ProvisionedThroughputExceededException","message":"Rate exceeded."}`)}, {StatusCode: 429, Body: body(`{"__type":"FooException","message":"Rate exceeded."}`)}, {StatusCode: 200, Body: body(`{"data":"valid"}`)}, } s := awstesting.NewClient(&aws.Config{ MaxRetries: aws.Int(10), SleepDelay: func(time.Duration) {}, }) s.Handlers.Validate.Clear() s.Handlers.Unmarshal.PushBack(unmarshal) s.Handlers.UnmarshalError.PushBack(unmarshalError) s.Handlers.Send.Clear() // mock sending s.Handlers.Send.PushBack(func(r request.Request) { r.HTTPResponse = &reqs[reqNum] reqNum++ }) out := &testData{} r := s.NewRequest(&request.Operation{Name: "Operation"}, nil, out) err := r.Send() if err != nil { t.Fatalf("expect no error, but got %v", err) } if e, a := 3, r.RetryCount; e != a { t.Errorf("expect %d retry count, got %d", e, a) } if e, a := "valid", out.Data; e != a { t.Errorf("expect %q output got %q", e, a) } } // test that retries don't occur for 4xx status codes with a response type that can't be retried func TestRequest4xxUnretryable(t testing.T) { s := awstesting.NewClient(&aws.Config{ MaxRetries: aws.Int(1), SleepDelay: func(time.Duration) {}, }) s.Handlers.Validate.Clear() s.Handlers.Unmarshal.PushBack(unmarshal) s.Handlers.UnmarshalError.PushBack(unmarshalError) s.Handlers.Send.Clear() // mock sending s.Handlers.Send.PushBack(func(r request.Request) { r.HTTPResponse = &http.Response{ StatusCode: 401, Body: body(`{"__type":"SignatureDoesNotMatch","message":"Signature does not match."}`), } }) out := &testData{} r := s.NewRequest(&request.Operation{Name: "Operation"}, nil, out) err := r.Send() if err == nil { t.Fatalf("expect error, but did not get one") } aerr := err.(awserr.RequestFailure) if e, a := 401, aerr.StatusCode(); e != a { t.Errorf("expect %d status code, got %d", e, a) } if e, a := "SignatureDoesNotMatch", aerr.Code(); e != a { t.Errorf("expect %q error code, got %q", e, a) } if e, a := "Signature does not match.", aerr.Message(); e != a { t.Errorf("expect %q error message, got %q", e, a) } if e, a := 0, r.RetryCount; e != a { t.Errorf("expect %d retry count, got %d", e, a) } } func TestRequestExhaustRetries(t testing.T) { delays := []time.Duration{} sleepDelay := func(delay time.Duration) { delays = append(delays, delay) } reqNum := 0 reqs := []http.Response{ {StatusCode: 500, Body: body(`{"__type":"UnknownError","message":"An error occurred."}`)}, {StatusCode: 500, Body: body(`{"__type":"UnknownError","message":"An error occurred."}`)}, {StatusCode: 500, Body: body(`{"__type":"UnknownError","message":"An error occurred."}`)}, {StatusCode: 500, Body: body(`{"__type":"UnknownError","message":"An error occurred."}`)}, } s := awstesting.NewClient(&aws.Config{ SleepDelay: sleepDelay, }) s.Handlers.Validate.Clear() s.Handlers.Unmarshal.PushBack(unmarshal) s.Handlers.UnmarshalError.PushBack(unmarshalError) s.Handlers.Send.Clear() // mock sending s.Handlers.Send.PushBack(func(r request.Request) { r.HTTPResponse = &reqs[reqNum] reqNum++ }) r := s.NewRequest(&request.Operation{Name: "Operation"}, nil, nil) err := r.Send() if err == nil { t.Fatalf("expect error, but did not get one") } aerr := err.(awserr.RequestFailure) if e, a := 500, aerr.StatusCode(); e != a { t.Errorf("expect %d status code, got %d", e, a) } if e, a := "UnknownError", aerr.Code(); e != a { t.Errorf("expect %q error code, got %q", e, a) } if e, a := "An error occurred.", aerr.Message(); e != a { t.Errorf("expect %q error message, got %q", e, a) } if e, a := 3, r.RetryCount; e != a { t.Errorf("expect %d retry count, got %d", e, a) } expectDelays := []struct{ min, max time.Duration }{{30, 60}, {60, 120}, {120, 240}} for i, v := range delays { min := expectDelays[i].min * time.Millisecond max := expectDelays[i].max * time.Millisecond if !(min <= v && v <= max) { t.Errorf("Expect delay to be within range, i:%d, v:%s, min:%s, max:%s", i, v, min, max) } } } // test that the request is retried after the credentials are expired. func TestRequestRecoverExpiredCreds(t testing.T) { reqNum := 0 reqs := []http.Response{ {StatusCode: 400, Body: body(`{"__type":"ExpiredTokenException","message":"expired token"}`)}, {StatusCode: 200, Body: body(`{"data":"valid"}`)}, } s := awstesting.NewClient(&aws.Config{ MaxRetries: aws.Int(10), Credentials: credentials.NewStaticCredentials("AKID", "SECRET", ""), SleepDelay: func(time.Duration) {}, }) s.Handlers.Validate.Clear() s.Handlers.Unmarshal.PushBack(unmarshal) s.Handlers.UnmarshalError.PushBack(unmarshalError) credExpiredBeforeRetry := false credExpiredAfterRetry := false s.Handlers.AfterRetry.PushBack(func(r request.Request) { credExpiredAfterRetry = r.Config.Credentials.IsExpired() }) s.Handlers.Sign.Clear() s.Handlers.Sign.PushBack(func(r request.Request) { r.Config.Credentials.Get() }) s.Handlers.Send.Clear() // mock sending s.Handlers.Send.PushBack(func(r request.Request) { r.HTTPResponse = &reqs[reqNum] reqNum++ }) out := &testData{} r := s.NewRequest(&request.Operation{Name: "Operation"}, nil, out) err := r.Send() if err != nil { t.Fatalf("expect no error, got %v", err) } if credExpiredBeforeRetry { t.Errorf("Expect valid creds before retry check") } if !credExpiredAfterRetry { t.Errorf("Expect expired creds after retry check") } if s.Config.Credentials.IsExpired() { t.Errorf("Expect valid creds after cred expired recovery") } if e, a := 1, r.RetryCount; e != a { t.Errorf("expect %d retry count, got %d", e, a) } if e, a := "valid", out.Data; e != a { t.Errorf("expect %q output got %q", e, a) } } func TestMakeAddtoUserAgentHandler(t testing.T) { fn := request.MakeAddToUserAgentHandler("name", "version", "extra1", "extra2") r := &request.Request{HTTPRequest: &http.Request{Header: http.Header{}}} r.HTTPRequest.Header.Set("User-Agent", "foo/bar") fn(r) if e, a := "foo/bar name/version (extra1; extra2)", r.HTTPRequest.Header.Get("User-Agent"); !strings.HasPrefix(a, e) { t.Errorf("expect %q user agent, got %q", e, a) } } func TestMakeAddtoUserAgentFreeFormHandler(t testing.T) { fn := request.MakeAddToUserAgentFreeFormHandler("name/version (extra1; extra2)") r := &request.Request{HTTPRequest: &http.Request{Header: http.Header{}}} r.HTTPRequest.Header.Set("User-Agent", "foo/bar") fn(r) if e, a := "foo/bar name/version (extra1; extra2)", r.HTTPRequest.Header.Get("User-Agent"); !strings.HasPrefix(a, e) { t.Errorf("expect %q user agent, got %q", e, a) } } func TestRequestUserAgent(t testing.T) { s := awstesting.NewClient(&aws.Config{ Region: aws.String("us-east-1"), }) req := s.NewRequest(&request.Operation{Name: "Operation"}, nil, &testData{}) req.HTTPRequest.Header.Set("User-Agent", "foo/bar") if err := req.Build(); err != nil { t.Fatalf("expect no error, got %v", err) } expectUA := fmt.Sprintf("foo/bar %s/%s (%s; %s; %s)", aws.SDKName, aws.SDKVersion, runtime.Version(), runtime.GOOS, runtime.GOARCH) if e, a := expectUA, req.HTTPRequest.Header.Get("User-Agent"); !strings.HasPrefix(a, e) { t.Errorf("expect %q user agent, got %q", e, a) } } func TestRequestThrottleRetries(t testing.T) { var delays []time.Duration sleepDelay := func(delay time.Duration) { delays = append(delays, delay) } reqNum := 0 reqs := []http.Response{ {StatusCode: 500, Body: body(`{"__type":"Throttling","message":"An error occurred."}`)}, {StatusCode: 500, Body: body(`{"__type":"Throttling","message":"An error occurred."}`)}, {StatusCode: 500, Body: body(`{"__type":"Throttling","message":"An error occurred."}`)}, {StatusCode: 500, Body: body(`{"__type":"Throttling","message":"An error occurred."}`)}, } s := awstesting.NewClient(&aws.Config{ SleepDelay: sleepDelay, }) s.Handlers.Validate.Clear() s.Handlers.Unmarshal.PushBack(unmarshal) s.Handlers.UnmarshalError.PushBack(unmarshalError) s.Handlers.Send.Clear() // mock sending s.Handlers.Send.PushBack(func(r request.Request) { r.HTTPResponse = &reqs[reqNum] reqNum++ }) r := s.NewRequest(&request.Operation{Name: "Operation"}, nil, nil) err := r.Send() if err == nil { t.Fatalf("expect error, but did not get one") } aerr := err.(awserr.RequestFailure) if e, a := 500, aerr.StatusCode(); e != a { t.Errorf("expect %d status code, got %d", e, a) } if e, a := "Throttling", aerr.Code(); e != a { t.Errorf("expect %q error code, got %q", e, a) } if e, a := "An error occurred.", aerr.Message(); e != a { t.Errorf("expect %q error message, got %q", e, a) } if e, a := 3, r.RetryCount; e != a { t.Errorf("expect %d retry count, got %d", e, a) } expectDelays := []struct{ min, max time.Duration }{{500, 1000}, {1000, 2000}, {2000, 4000}} for i, v := range delays { min := expectDelays[i].min time.Millisecond max := expectDelays[i].max * time.Millisecond if !(min <= v && v <= max) { t.Errorf("Expect delay to be within range, i:%d, v:%s, min:%s, max:%s", i, v, min, max) } } } // test that retries occur for request timeouts when response.Body can be nil func TestRequestRecoverTimeoutWithNilBody(t testing.T) { reqNum := 0 reqs := []http.Response{ {StatusCode: 0, Body: nil}, // body can be nil when requests time out {StatusCode: 200, Body: body(`{"data":"valid"}`)}, } errors := []error{ errTimeout, nil, } s := awstesting.NewClient(aws.NewConfig().WithMaxRetries(10)) s.Handlers.Validate.Clear() s.Handlers.Unmarshal.PushBack(unmarshal) s.Handlers.UnmarshalError.PushBack(unmarshalError) s.Handlers.AfterRetry.Clear() // force retry on all errors s.Handlers.AfterRetry.PushBack(func(r request.Request) { if r.Error != nil { r.Error = nil r.Retryable = aws.Bool(true) r.RetryCount++ } }) s.Handlers.Send.Clear() // mock sending s.Handlers.Send.PushBack(func(r request.Request) { r.HTTPResponse = reqs[reqNum] r.Error = errors[reqNum] reqNum++ }) out := &testData{} r := s.NewRequest(&request.Operation{Name: "Operation"}, nil, out) err := r.Send() if err != nil { t.Fatalf("expect no error, but got %v", err) } if e, a := 1, r.RetryCount; e != a { t.Errorf("expect %d retry count, got %d", e, a) } if e, a := "valid", out.Data; e != a { t.Errorf("expect %q output got %q", e, a) } } func TestRequestRecoverTimeoutWithNilResponse(t testing.T) { reqNum := 0 reqs := []http.Response{ nil, {StatusCode: 200, Body: body(`{"data":"valid"}`)}, } errors := []error{ errTimeout, nil, } s := awstesting.NewClient(aws.NewConfig().WithMaxRetries(10)) s.Handlers.Validate.Clear() s.Handlers.Unmarshal.PushBack(unmarshal) s.Handlers.UnmarshalError.PushBack(unmarshalError) s.Handlers.AfterRetry.Clear() // force retry on all errors s.Handlers.AfterRetry.PushBack(func(r request.Request) { if r.Error != nil { r.Error = nil r.Retryable = aws.Bool(true) r.RetryCount++ } }) s.Handlers.Send.Clear() // mock sending s.Handlers.Send.PushBack(func(r request.Request) { r.HTTPResponse = reqs[reqNum] r.Error = errors[reqNum] reqNum++ }) out := &testData{} r := s.NewRequest(&request.Operation{Name: "Operation"}, nil, out) err := r.Send() if err != nil { t.Fatalf("expect no error, but got %v", err) } if e, a := 1, r.RetryCount; e != a { t.Errorf("expect %d retry count, got %d", e, a) } if e, a := "valid", out.Data; e != a { t.Errorf("expect %q output got %q", e, a) } } func TestRequest_NoBody(t testing.T) { cases := []string{ "GET", "HEAD", "DELETE", "PUT", "POST", "PATCH", } for i, c := range cases { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r http.Request) { if v := r.TransferEncoding; len(v) > 0 { t.Errorf("%d, expect no body sent with Transfer-Encoding, %v", i, v) } outMsg := []byte(`{"Value": "abc"}`) if b, err := ioutil.ReadAll(r.Body); err != nil { t.Fatalf("%d, expect no error reading request body, got %v", i, err) } else if n := len(b); n > 0 { t.Errorf("%d, expect no request body, got %d bytes", i, n) } w.Header().Set("Content-Length", strconv.Itoa(len(outMsg))) if _, err := w.Write(outMsg); err != nil { t.Fatalf("%d, expect no error writing server response, got %v", i, err) } })) s := awstesting.NewClient(&aws.Config{ Region: aws.String("mock-region"), MaxRetries: aws.Int(0), Endpoint: aws.String(server.URL), DisableSSL: aws.Bool(true), }) s.Handlers.Build.PushBack(rest.Build) s.Handlers.Validate.Clear() s.Handlers.Unmarshal.PushBack(unmarshal) s.Handlers.UnmarshalError.PushBack(unmarshalError) in := struct { Bucket string `location:"uri" locationName:"bucket"` Key string `location:"uri" locationName:"key"` }{ Bucket: aws.String("mybucket"), Key: aws.String("myKey"), } out := struct { Value string }{} r := s.NewRequest(&request.Operation{ Name: "OpName", HTTPMethod: c, HTTPPath: "/{bucket}/{key+}", }, &in, &out) err := r.Send() server.Close() if err != nil { t.Fatalf("%d, expect no error sending request, got %v", i, err) } } } func TestIsSerializationErrorRetryable(t testing.T) { testCases := []struct { err error expected bool }{ { err: awserr.New(request.ErrCodeSerialization, "foo error", nil), expected: false, }, { err: awserr.New("ErrFoo", "foo error", nil), expected: false, }, { err: nil, expected: false, }, { err: awserr.New(request.ErrCodeSerialization, "foo error", errAcceptConnectionResetStub), expected: true, }, } for i, c := range testCases { r := &request.Request{ Error: c.err, } if r.IsErrorRetryable() != c.expected { t.Errorf("Case %d: Expected %v, but received %v", i, c.expected, !c.expected) } } } func TestWithLogLevel(t testing.T) { r := &request.Request{} opt := request.WithLogLevel(aws.LogDebugWithHTTPBody) r.ApplyOptions(opt) if !r.Config.LogLevel.Matches(aws.LogDebugWithHTTPBody) { t.Errorf("expect log level to be set, but was not, %v", r.Config.LogLevel.Value()) } } func TestWithGetResponseHeader(t testing.T) { r := &request.Request{} var val, val2 string r.ApplyOptions( request.WithGetResponseHeader("x-a-header", &val), request.WithGetResponseHeader("x-second-header", &val2), ) r.HTTPResponse = &http.Response{ Header: func() http.Header { h := http.Header{} h.Set("x-a-header", "first") h.Set("x-second-header", "second") return h }(), } r.Handlers.Complete.Run(r) if e, a := "first", val; e != a { t.Errorf("expect %q header value got %q", e, a) } if e, a := "second", val2; e != a { t.Errorf("expect %q header value got %q", e, a) } } func TestWithGetResponseHeaders(t testing.T) { r := &request.Request{} var headers http.Header opt := request.WithGetResponseHeaders(&headers) r.ApplyOptions(opt) r.HTTPResponse = &http.Response{ Header: func() http.Header { h := http.Header{} h.Set("x-a-header", "headerValue") return h }(), } r.Handlers.Complete.Run(r) if e, a := "headerValue", headers.Get("x-a-header"); e != a { t.Errorf("expect %q header value got %q", e, a) } } type testRetryer struct { shouldRetry bool maxRetries int } func (d testRetryer) MaxRetries() int { return d.maxRetries } // RetryRules returns the delay duration before retrying this request again func (d testRetryer) RetryRules(r request.Request) time.Duration { return 0 } func (d testRetryer) ShouldRetry(r request.Request) bool { return d.shouldRetry } func TestEnforceShouldRetryCheck(t testing.T) { retryer := &testRetryer{ shouldRetry: true, maxRetries: 3, } s := awstesting.NewClient(&aws.Config{ Region: aws.String("mock-region"), MaxRetries: aws.Int(0), Retryer: retryer, EnforceShouldRetryCheck: aws.Bool(true), SleepDelay: func(time.Duration) {}, }) s.Handlers.Validate.Clear() s.Handlers.Send.Swap(corehandlers.SendHandler.Name, request.NamedHandler{ Name: "TestEnforceShouldRetryCheck", Fn: func(r request.Request) { r.HTTPResponse = &http.Response{ Header: http.Header{}, Body: ioutil.NopCloser(bytes.NewBuffer(nil)), } r.Retryable = aws.Bool(false) }, }) s.Handlers.Unmarshal.PushBack(unmarshal) s.Handlers.UnmarshalError.PushBack(unmarshalError) out := &testData{} r := s.NewRequest(&request.Operation{Name: "Operation"}, nil, out) err := r.Send() if err == nil { t.Fatalf("expect error, but got nil") } if e, a := 3, r.RetryCount; e != a { t.Errorf("expect %d retry count, got %d", e, a) } if !retryer.shouldRetry { t.Errorf("expect 'true' for ShouldRetry, but got %v", retryer.shouldRetry) } } type errReader struct { err error } func (reader errReader) Read(b []byte) (int, error) { return 0, reader.err } func (reader errReader) Close() error { return nil } func TestIsNoBodyReader(t testing.T) { cases := []struct { reader io.ReadCloser expect bool }{ {ioutil.NopCloser(bytes.NewReader([]byte("abc"))), false}, {ioutil.NopCloser(bytes.NewReader(nil)), false}, {nil, false}, {request.NoBody, true}, } for i, c := range cases { if e, a := c.expect, request.NoBody == c.reader; e != a { t.Errorf("%d, expect %t match, but was %t", i, e, a) } } } func TestRequest_TemporaryRetry(t testing.T) { done := make(chan struct{}) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r http.Request) { w.Header().Set("Content-Length", "1024") w.WriteHeader(http.StatusOK) w.Write(make([]byte, 100)) f := w.(http.Flusher) f.Flush() <-done })) defer server.Close() client := &http.Client{ Timeout: 100 time.Millisecond, } svc := awstesting.NewClient(&aws.Config{ Region: unit.Session.Config.Region, MaxRetries: aws.Int(1), HTTPClient: client, DisableSSL: aws.Bool(true), Endpoint: aws.String(server.URL), }) req := svc.NewRequest(&request.Operation{ Name: "name", HTTPMethod: "GET", HTTPPath: "/path", }, &struct{}{}, &struct{}{}) req.Handlers.Unmarshal.PushBack(func(r request.Request) { defer req.HTTPResponse.Body.Close() _, err := io.Copy(ioutil.Discard, req.HTTPResponse.Body) r.Error = awserr.New(request.ErrCodeSerialization, "error", err) }) err := req.Send() if err == nil { t.Errorf("expect error, got none") } close(done) aerr := err.(awserr.Error) if e, a := request.ErrCodeSerialization, aerr.Code(); e != a { t.Errorf("expect %q error code, got %q", e, a) } if e, a := 1, req.RetryCount; e != a { t.Errorf("expect %d retries, got %d", e, a) } type temporary interface { Temporary() bool } terr := aerr.OrigErr().(temporary) if !terr.Temporary() { t.Errorf("expect temporary error, was not") } } func TestRequest_Presign(t testing.T) { presign := func(r request.Request, expire time.Duration) (string, http.Header, error) { u, err := r.Presign(expire) return u, nil, err } presignRequest := func(r request.Request, expire time.Duration) (string, http.Header, error) { return r.PresignRequest(expire) } mustParseURL := func(v string) url.URL { u, err := url.Parse(v) if err != nil { panic(err) } return u } cases := []struct { Expire time.Duration PresignFn func(request.Request, time.Duration) (string, http.Header, error) SignerFn func(request.Request) URL string Header http.Header Err string }{ { PresignFn: presign, Err: request.ErrCodeInvalidPresignExpire, }, { PresignFn: presignRequest, Err: request.ErrCodeInvalidPresignExpire, }, { Expire: -1, PresignFn: presign, Err: request.ErrCodeInvalidPresignExpire, }, { // Presign clear NotHoist Expire: 1 time.Minute, PresignFn: func(r request.Request, dur time.Duration) (string, http.Header, error) { r.NotHoist = true return presign(r, dur) }, SignerFn: func(r request.Request) { r.HTTPRequest.URL = mustParseURL("https://endpoint/presignedURL") if r.NotHoist { r.Error = fmt.Errorf("expect NotHoist to be cleared") } }, URL: "https://endpoint/presignedURL", }, { // PresignRequest does not clear NotHoist Expire: 1 * time.Minute, PresignFn: func(r request.Request, dur time.Duration) (string, http.Header, error) { r.NotHoist = true return presignRequest(r, dur) }, SignerFn: func(r request.Request) { r.HTTPRequest.URL = mustParseURL("https://endpoint/presignedURL") if !r.NotHoist { r.Error = fmt.Errorf("expect NotHoist not to be cleared") } }, URL: "https://endpoint/presignedURL", }, { // PresignRequest returns signed headers Expire: 1 * time.Minute, PresignFn: presignRequest, SignerFn: func(r request.Request) { r.HTTPRequest.URL = mustParseURL("https://endpoint/presignedURL") r.HTTPRequest.Header.Set("UnsigndHeader", "abc") r.SignedHeaderVals = http.Header{ "X-Amzn-Header": []string{"abc", "123"}, "X-Amzn-Header2": []string{"efg", "456"}, } }, URL: "https://endpoint/presignedURL", Header: http.Header{ "X-Amzn-Header": []string{"abc", "123"}, "X-Amzn-Header2": []string{"efg", "456"}, }, }, } svc := awstesting.NewClient() svc.Handlers.Clear() for i, c := range cases { req := svc.NewRequest(&request.Operation{ Name: "name", HTTPMethod: "GET", HTTPPath: "/path", }, &struct{}{}, &struct{}{}) req.Handlers.Sign.PushBack(c.SignerFn) u, h, err := c.PresignFn(req, c.Expire) if len(c.Err) != 0 { if e, a := c.Err, err.Error(); !strings.Contains(a, e) { t.Errorf("%d, expect %v to be in %v", i, e, a) } continue } else if err != nil { t.Errorf("%d, expect no error, got %v", i, err) continue } if e, a := c.URL, u; e != a { t.Errorf("%d, expect %v URL, got %v", i, e, a) } if e, a := c.Header, h; !reflect.DeepEqual(e, a) { t.Errorf("%d, expect %v header got %v", i, e, a) } } } func TestSanitizeHostForHeader(t testing.T) { cases := []struct { url string expectedRequestHost string }{ {"https://estest.us-east-1.es.amazonaws.com:443", "estest.us-east-1.es.amazonaws.com"}, {"https://estest.us-east-1.es.amazonaws.com", "estest.us-east-1.es.amazonaws.com"}, {"https://localhost:9200", "localhost:9200"}, {"http://localhost:80", "localhost"}, {"http://localhost:8080", "localhost:8080"}, } for _, c := range cases { r, _ := http.NewRequest("GET", c.url, nil) request.SanitizeHostForHeader(r) if h := r.Host; h != c.expectedRequestHost { t.Errorf("expect %v host, got %q", c.expectedRequestHost, h) } } } func TestRequestWillRetry_ByBody(t testing.T) { svc := awstesting.NewClient() cases := []struct { WillRetry bool HTTPMethod string Body io.ReadSeeker IsReqNoBody bool }{ { WillRetry: true, HTTPMethod: "GET", Body: bytes.NewReader([]byte{}), IsReqNoBody: true, }, { WillRetry: true, HTTPMethod: "GET", Body: bytes.NewReader(nil), IsReqNoBody: true, }, { WillRetry: true, HTTPMethod: "POST", Body: bytes.NewReader([]byte("abc123")), }, { WillRetry: true, HTTPMethod: "POST", Body: aws.ReadSeekCloser(bytes.NewReader([]byte("abc123"))), }, { WillRetry: true, HTTPMethod: "GET", Body: aws.ReadSeekCloser(bytes.NewBuffer(nil)), IsReqNoBody: true, }, { WillRetry: true, HTTPMethod: "POST", Body: aws.ReadSeekCloser(bytes.NewBuffer(nil)), IsReqNoBody: true, }, { WillRetry: false, HTTPMethod: "POST", Body: aws.ReadSeekCloser(bytes.NewBuffer([]byte("abc123"))), }, } for i, c := range cases { req := svc.NewRequest(&request.Operation{ Name: "Operation", HTTPMethod: c.HTTPMethod, HTTPPath: "/", }, nil, nil) req.SetReaderBody(c.Body) req.Build() req.Error = fmt.Errorf("some error") req.Retryable = aws.Bool(true) req.HTTPResponse = &http.Response{ StatusCode: 500, } if e, a := c.IsReqNoBody, request.NoBody == req.HTTPRequest.Body; e != a { t.Errorf("%d, expect request to be no body, %t, got %t, %T", i, e, a, req.HTTPRequest.Body) } if e, a := c.WillRetry, req.WillRetry(); e != a { t.Errorf("%d, expect %t willRetry, got %t", i, e, a) } if req.Error == nil { t.Fatalf("%d, expect error, got none", i) } if e, a := "some error", req.Error.Error(); !strings.Contains(a, e) { t.Errorf("%d, expect %q error in %q", i, e, a) } if e, a := 0, req.RetryCount; e != a { t.Errorf("%d, expect retry count to be %d, got %d", i, e, a) } } } func Test501NotRetrying(t testing.T) { reqNum := 0 reqs := []http.Response{ {StatusCode: 500, Body: body(`{"__type":"UnknownError","message":"An error occurred."}`)}, {StatusCode: 501, Body: body(`{"__type":"NotImplemented","message":"An error occurred."}`)}, {StatusCode: 200, Body: body(`{"data":"valid"}`)}, } s := awstesting.NewClient(aws.NewConfig().WithMaxRetries(10)) s.Handlers.Validate.Clear() s.Handlers.Unmarshal.PushBack(unmarshal) s.Handlers.UnmarshalError.PushBack(unmarshalError) s.Handlers.Send.Clear() // mock sending s.Handlers.Send.PushBack(func(r request.Request) { r.HTTPResponse = &reqs[reqNum] reqNum++ }) out := &testData{} r := s.NewRequest(&request.Operation{Name: "Operation"}, nil, out) err := r.Send() if err == nil { t.Fatal("expect error, but got none") } aerr := err.(awserr.Error) if e, a := "NotImplemented", aerr.Code(); e != a { t.Errorf("expected error code %q, but received %q", e, a) } if e, a := 1, r.RetryCount; e != a { t.Errorf("expect %d retry count, got %d", e, a) } } func TestRequestNoConnection(t testing.T) { port, err := getFreePort() if err != nil { t.Fatalf("failed to get free port for test") } s := awstesting.NewClient(aws.NewConfig(). WithMaxRetries(10). WithEndpoint("https://localhost:" + strconv.Itoa(port)). WithSleepDelay(func(time.Duration) {}), ) s.Handlers.Validate.Clear() s.Handlers.Unmarshal.PushBack(unmarshal) s.Handlers.UnmarshalError.PushBack(unmarshalError) out := &testData{} r := s.NewRequest(&request.Operation{Name: "Operation"}, nil, out) if err = r.Send(); err == nil { t.Fatal("expect error, but got none") } t.Logf("Error, %v", err) awsError := err.(awserr.Error) origError := awsError.OrigErr() t.Logf("Orig Error: %#v of type %T", origError, origError) if e, a := 10, r.RetryCount; e != a { t.Errorf("expect %v retry count, got %v", e, a) } } func TestRequestBodySeekFails(t testing.T) { s := awstesting.NewClient() s.Handlers.Validate.Clear() s.Handlers.Build.Clear() out := &testData{} r := s.NewRequest(&request.Operation{Name: "Operation"}, nil, out) r.SetReaderBody(&stubSeekFail{ Err: fmt.Errorf("failed to seek reader"), }) err := r.Send() if err == nil { t.Fatal("expect error, but got none") } aerr := err.(awserr.Error) if e, a := request.ErrCodeSerialization, aerr.Code(); e != a { t.Errorf("expect %v error code, got %v", e, a) } } func TestRequestEndpointWithDefaultPort(t testing.T) { s := awstesting.NewClient(&aws.Config{ Endpoint: aws.String("https://example.test:443"), }) r := s.NewRequest(&request.Operation{ Name: "FooBar", HTTPMethod: "GET", HTTPPath: "/", }, nil, nil) r.Handlers.Validate.Clear() r.Handlers.ValidateResponse.Clear() r.Handlers.Send.Clear() r.Handlers.Send.PushFront(func(r request.Request) { req := r.HTTPRequest if e, a := "example.test", req.Host; e != a { t.Errorf("expected %v, got %v", e, a) } if e, a := "https://example.test:443/", req.URL.String(); e != a { t.Errorf("expected %v, got %v", e, a) } }) err := r.Send() if err != nil { t.Fatalf("expected no error, got %v", err) } } func TestRequestEndpointWithNonDefaultPort(t testing.T) { s := awstesting.NewClient(&aws.Config{ Endpoint: aws.String("https://example.test:8443"), }) r := s.NewRequest(&request.Operation{ Name: "FooBar", HTTPMethod: "GET", HTTPPath: "/", }, nil, nil) r.Handlers.Validate.Clear() r.Handlers.ValidateResponse.Clear() r.Handlers.Send.Clear() r.Handlers.Send.PushFront(func(r request.Request) { req := r.HTTPRequest // http.Request.Host should not be set for non-default ports if e, a := "", req.Host; e != a { t.Errorf("expected %v, got %v", e, a) } if e, a := "https://example.test:8443/", req.URL.String(); e != a { t.Errorf("expected %v, got %v", e, a) } }) err := r.Send() if err != nil { t.Fatalf("expected no error, got %v", err) } } func TestRequestMarshaledEndpointWithDefaultPort(t testing.T) { s := awstesting.NewClient(&aws.Config{ Endpoint: aws.String("https://example.test:443"), }) r := s.NewRequest(&request.Operation{ Name: "FooBar", HTTPMethod: "GET", HTTPPath: "/", }, nil, nil) r.Handlers.Validate.Clear() r.Handlers.ValidateResponse.Clear() r.Handlers.Build.PushBack(func(r request.Request) { req := r.HTTPRequest req.URL.Host = "foo." + req.URL.Host }) r.Handlers.Send.Clear() r.Handlers.Send.PushFront(func(r request.Request) { req := r.HTTPRequest if e, a := "foo.example.test", req.Host; e != a { t.Errorf("expected %v, got %v", e, a) } if e, a := "https://foo.example.test:443/", req.URL.String(); e != a { t.Errorf("expected %v, got %v", e, a) } }) err := r.Send() if err != nil { t.Fatalf("expected no error, got %v", err) } } func TestRequestMarshaledEndpointWithNonDefaultPort(t testing.T) { s := awstesting.NewClient(&aws.Config{ Endpoint: aws.String("https://example.test:8443"), }) r := s.NewRequest(&request.Operation{ Name: "FooBar", HTTPMethod: "GET", HTTPPath: "/", }, nil, nil) r.Handlers.Validate.Clear() r.Handlers.ValidateResponse.Clear() r.Handlers.Build.PushBack(func(r request.Request) { req := r.HTTPRequest req.URL.Host = "foo." + req.URL.Host }) r.Handlers.Send.Clear() r.Handlers.Send.PushFront(func(r request.Request) { req := r.HTTPRequest // http.Request.Host should not be set for non-default ports if e, a := "", req.Host; e != a { t.Errorf("expected %v, got %v", e, a) } if e, a := "https://foo.example.test:8443/", req.URL.String(); e != a { t.Errorf("expected %v, got %v", e, a) } }) err := r.Send() if err != nil { t.Fatalf("expected no error, got %v", err) } } func TestRequestCompleteWithoutHTTPResponse(t testing.T) { s := awstesting.NewClient(aws.NewConfig().WithRegion("mock-region")) r := s.NewRequest(&request.Operation{ Name: "FooBar", HTTPMethod: "GET", HTTPPath: "/", }, nil, nil) r.Handlers.Sign.Clear() r.Handlers.Sign.PushFront(func(r request.Request) { r.Error = fmt.Errorf("failed") }) r.Handlers.Complete.PushBack(func(r request.Request) { if r.HTTPResponse == nil { t.Fatalf("expect HTTPResponse not to be nil") } if r.HTTPResponse.Header == nil { t.Fatalf("expect HTTPResponse.Header not to be nil") } if r.HTTPResponse.Body == nil { t.Fatalf("expect HTTPResponse.Body not to be nil") } }) err := r.Send() if err == nil { t.Fatalf("expect error, got none") } if e, a := "failed", err.Error(); !strings.Contains(a, e) { t.Errorf("expect %v error in %v", e, a) } } type stubSeekFail struct { Err error } func (f stubSeekFail) Read(b []byte) (int, error) { return len(b), nil } func (f stubSeekFail) ReadAt(b []byte, offset int64) (int, error) { return len(b), nil } func (f *stubSeekFail) Seek(offset int64, mode int) (int64, error) { return 0, f.Err } func getFreePort() (int, error) { l, err := net.Listen("tcp", ":0") if err != nil { return 0, err } defer l.Close() strAddr := l.Addr().String() parts := strings.Split(strAddr, ":") strPort := parts[len(parts)-1] port, err := strconv.ParseInt(strPort, 10, 32) if err != nil { return 0, err } return int(port), nil }	35.4 KB	2025-05-25 12:27
📄 retryer.go package request import ( "net" "net/url" "strings" "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" ) // Retryer provides the interface drive the SDK's request retry behavior. The // Retryer implementation is responsible for implementing exponential backoff, // and determine if a request API error should be retried. // // client.DefaultRetryer is the SDK's default implementation of the Retryer. It // uses the Request.IsErrorRetryable and Request.IsErrorThrottle methods to // determine if the request is retried. type Retryer interface { // RetryRules return the retry delay that should be used by the SDK before // making another request attempt for the failed request. RetryRules(Request) time.Duration // ShouldRetry returns if the failed request is retryable. // // Implementations may consider request attempt count when determining if a // request is retryable, but the SDK will use MaxRetries to limit the // number of attempts a request are made. ShouldRetry(Request) bool // MaxRetries is the number of times a request may be retried before // failing. MaxRetries() int } // WithRetryer sets a Retryer value to the given Config returning the Config // value for chaining. The value must not be nil. func WithRetryer(cfg aws.Config, retryer Retryer) aws.Config { if retryer == nil { if cfg.Logger != nil { cfg.Logger.Log("ERROR: Request.WithRetryer called with nil retryer. Replacing with retry disabled Retryer.") } retryer = noOpRetryer{} } cfg.Retryer = retryer return cfg } // noOpRetryer is a internal no op retryer used when a request is created // without a retryer. // // Provides a retryer that performs no retries. // It should be used when we do not want retries to be performed. type noOpRetryer struct{} // MaxRetries returns the number of maximum returns the service will use to make // an individual API; For NoOpRetryer the MaxRetries will always be zero. func (d noOpRetryer) MaxRetries() int { return 0 } // ShouldRetry will always return false for NoOpRetryer, as it should never retry. func (d noOpRetryer) ShouldRetry(_ Request) bool { return false } // RetryRules returns the delay duration before retrying this request again; // since NoOpRetryer does not retry, RetryRules always returns 0. func (d noOpRetryer) RetryRules(_ Request) time.Duration { return 0 } // retryableCodes is a collection of service response codes which are retry-able // without any further action. var retryableCodes = map[string]struct{}{ ErrCodeRequestError: {}, "RequestTimeout": {}, ErrCodeResponseTimeout: {}, "RequestTimeoutException": {}, // Glacier's flavor of RequestTimeout } var throttleCodes = map[string]struct{}{ "ProvisionedThroughputExceededException": {}, "ThrottledException": {}, // SNS, XRay, ResourceGroupsTagging API "Throttling": {}, "ThrottlingException": {}, "RequestLimitExceeded": {}, "RequestThrottled": {}, "RequestThrottledException": {}, "TooManyRequestsException": {}, // Lambda functions "PriorRequestNotComplete": {}, // Route53 "TransactionInProgressException": {}, "EC2ThrottledException": {}, // EC2 } // credsExpiredCodes is a collection of error codes which signify the credentials // need to be refreshed. Expired tokens require refreshing of credentials, and // resigning before the request can be retried. var credsExpiredCodes = map[string]struct{}{ "ExpiredToken": {}, "ExpiredTokenException": {}, "RequestExpired": {}, // EC2 Only } func isCodeThrottle(code string) bool { _, ok := throttleCodes[code] return ok } func isCodeRetryable(code string) bool { if _, ok := retryableCodes[code]; ok { return true } return isCodeExpiredCreds(code) } func isCodeExpiredCreds(code string) bool { _, ok := credsExpiredCodes[code] return ok } var validParentCodes = map[string]struct{}{ ErrCodeSerialization: {}, ErrCodeRead: {}, } func isNestedErrorRetryable(parentErr awserr.Error) bool { if parentErr == nil { return false } if _, ok := validParentCodes[parentErr.Code()]; !ok { return false } err := parentErr.OrigErr() if err == nil { return false } if aerr, ok := err.(awserr.Error); ok { return isCodeRetryable(aerr.Code()) } if t, ok := err.(temporary); ok { return t.Temporary() \|\| isErrConnectionReset(err) } return isErrConnectionReset(err) } // IsErrorRetryable returns whether the error is retryable, based on its Code. // Returns false if error is nil. func IsErrorRetryable(err error) bool { if err == nil { return false } return shouldRetryError(err) } type temporary interface { Temporary() bool } func shouldRetryError(origErr error) bool { switch err := origErr.(type) { case awserr.Error: if err.Code() == CanceledErrorCode { return false } if isNestedErrorRetryable(err) { return true } origErr := err.OrigErr() var shouldRetry bool if origErr != nil { shouldRetry = shouldRetryError(origErr) if err.Code() == ErrCodeRequestError && !shouldRetry { return false } } if isCodeRetryable(err.Code()) { return true } return shouldRetry case url.Error: if strings.Contains(err.Error(), "connection refused") { // Refused connections should be retried as the service may not yet // be running on the port. Go TCP dial considers refused // connections as not temporary. return true } // url.Error only implements Temporary after golang 1.6 but since // url.Error only wraps the error: return shouldRetryError(err.Err) case temporary: if netErr, ok := err.(net.OpError); ok && netErr.Op == "dial" { return true } // If the error is temporary, we want to allow continuation of the // retry process return err.Temporary() \|\| isErrConnectionReset(origErr) case nil: // `awserr.Error.OrigErr()` can be nil, meaning there was an error but // because we don't know the cause, it is marked as retryable. See // TestRequest4xxUnretryable for an example. return true default: switch err.Error() { case "net/http: request canceled", "net/http: request canceled while waiting for connection": // known 1.5 error case when an http request is cancelled return false } // here we don't know the error; so we allow a retry. return true } } // IsErrorThrottle returns whether the error is to be throttled based on its code. // Returns false if error is nil. func IsErrorThrottle(err error) bool { if aerr, ok := err.(awserr.Error); ok && aerr != nil { return isCodeThrottle(aerr.Code()) } return false } // IsErrorExpiredCreds returns whether the error code is a credential expiry // error. Returns false if error is nil. func IsErrorExpiredCreds(err error) bool { if aerr, ok := err.(awserr.Error); ok && aerr != nil { return isCodeExpiredCreds(aerr.Code()) } return false } // IsErrorRetryable returns whether the error is retryable, based on its Code. // Returns false if the request has no Error set. // // Alias for the utility function IsErrorRetryable func (r Request) IsErrorRetryable() bool { if isErrCode(r.Error, r.RetryErrorCodes) { return true } // HTTP response status code 501 should not be retried. // 501 represents Not Implemented which means the request method is not // supported by the server and cannot be handled. if r.HTTPResponse != nil { // HTTP response status code 500 represents internal server error and // should be retried without any throttle. if r.HTTPResponse.StatusCode == 500 { return true } } return IsErrorRetryable(r.Error) } // IsErrorThrottle returns whether the error is to be throttled based on its // code. Returns false if the request has no Error set. // // Alias for the utility function IsErrorThrottle func (r Request) IsErrorThrottle() bool { if isErrCode(r.Error, r.ThrottleErrorCodes) { return true } if r.HTTPResponse != nil { switch r.HTTPResponse.StatusCode { case 429, // error caused due to too many requests 502, // Bad Gateway error should be throttled 503, // caused when service is unavailable 504: // error occurred due to gateway timeout return true } } return IsErrorThrottle(r.Error) } func isErrCode(err error, codes []string) bool { if aerr, ok := err.(awserr.Error); ok && aerr != nil { for _, code := range codes { if code == aerr.Code() { return true } } } return false } // IsErrorExpired returns whether the error code is a credential expiry error. // Returns false if the request has no Error set. // // Alias for the utility function IsErrorExpiredCreds func (r Request) IsErrorExpired() bool { return IsErrorExpiredCreds(r.Error) }	8.62 KB	2025-05-25 12:27
📄 retryer_test.go package request import ( "errors" "fmt" "net/http" "testing" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/aws/client/metadata" ) func TestRequestIsErrorThrottle(t testing.T) { cases := []struct { Err error Throttle bool Req Request }{ { Err: awserr.New("ProvisionedThroughputExceededException", "", nil), Throttle: true, }, { Err: awserr.New("ThrottledException", "", nil), Throttle: true, }, { Err: awserr.New("Throttling", "", nil), Throttle: true, }, { Err: awserr.New("ThrottlingException", "", nil), Throttle: true, }, { Err: awserr.New("RequestLimitExceeded", "", nil), Throttle: true, }, { Err: awserr.New("RequestThrottled", "", nil), Throttle: true, }, { Err: awserr.New("TooManyRequestsException", "", nil), Throttle: true, }, { Err: awserr.New("PriorRequestNotComplete", "", nil), Throttle: true, }, { Err: awserr.New("TransactionInProgressException", "", nil), Throttle: true, }, { Err: awserr.New("EC2ThrottledException", "", nil), Throttle: true, }, { Err: awserr.NewRequestFailure( awserr.New(ErrCodeSerialization, "some error", awserr.NewUnmarshalError(nil, "blah", []byte{}), ), 503, "request-id", ), Req: Request{ HTTPResponse: &http.Response{ StatusCode: 503, Header: http.Header{}, }, }, Throttle: true, }, { Err: awserr.NewRequestFailure( awserr.New(ErrCodeSerialization, "some error", awserr.NewUnmarshalError(nil, "blah", []byte{}), ), 400, "request-id", ), Req: Request{ HTTPResponse: &http.Response{ StatusCode: 400, Header: http.Header{}, }, }, Throttle: false, }, } for i, c := range cases { req := c.Req req.Error = c.Err if e, a := c.Throttle, req.IsErrorThrottle(); e != a { t.Errorf("%d, expect %v to be throttled, was %t", i, c.Err, a) } } } type mockTempError bool func (e mockTempError) Error() string { return fmt.Sprintf("mock temporary error: %t", e.Temporary()) } func (e mockTempError) Temporary() bool { return bool(e) } func TestRequestIsErrorRetryable(t testing.T) { cases := []struct { Err error Req Request Retryable bool }{ { Err: awserr.New(ErrCodeSerialization, "temporary error", mockTempError(true)), Retryable: true, }, { Err: awserr.New(ErrCodeSerialization, "temporary error", mockTempError(false)), Retryable: false, }, { Err: awserr.New(ErrCodeSerialization, "some error", errors.New("blah")), Retryable: true, }, { Err: awserr.NewRequestFailure( awserr.New(ErrCodeSerialization, "some error", awserr.NewUnmarshalError(nil, "blah", []byte{}), ), 503, "request-id", ), Req: Request{ HTTPResponse: &http.Response{ StatusCode: 503, Header: http.Header{}, }, }, Retryable: false, // classified as throttled not retryable }, { Err: awserr.NewRequestFailure( awserr.New(ErrCodeSerialization, "some error", awserr.NewUnmarshalError(nil, "blah", []byte{}), ), 400, "request-id", ), Req: Request{ HTTPResponse: &http.Response{ StatusCode: 400, Header: http.Header{}, }, }, Retryable: false, }, { Err: awserr.New("SomeError", "some error", nil), Retryable: false, }, { Err: awserr.New(ErrCodeRequestError, "some error", nil), Retryable: true, }, { Err: nil, Retryable: false, }, } for i, c := range cases { req := c.Req req.Error = c.Err if e, a := c.Retryable, req.IsErrorRetryable(); e != a { t.Errorf("%d, expect %v to be retryable, was %t", i, c.Err, a) } } } func TestRequest_NilRetyer(t *testing.T) { clientInfo := metadata.ClientInfo{Endpoint: "https://mock.region.amazonaws.com"} req := New(aws.Config{}, clientInfo, Handlers{}, nil, &Operation{}, nil, nil) if req.Retryer == nil { t.Fatalf("expect retryer to be set") } if e, a := 0, req.MaxRetries(); e != a { t.Errorf("expect no retries, got %v", a) } }	4.11 KB	2025-05-25 12:27
📄 timeout_read_closer.go package request import ( "io" "time" "github.com/aws/aws-sdk-go/aws/awserr" ) var timeoutErr = awserr.New( ErrCodeResponseTimeout, "read on body has reached the timeout limit", nil, ) type readResult struct { n int err error } // timeoutReadCloser will handle body reads that take too long. // We will return a ErrReadTimeout error if a timeout occurs. type timeoutReadCloser struct { reader io.ReadCloser duration time.Duration } // Read will spin off a goroutine to call the reader's Read method. We will // select on the timer's channel or the read's channel. Whoever completes first // will be returned. func (r timeoutReadCloser) Read(b []byte) (int, error) { timer := time.NewTimer(r.duration) c := make(chan readResult, 1) go func() { n, err := r.reader.Read(b) timer.Stop() c <- readResult{n: n, err: err} }() select { case data := <-c: return data.n, data.err case <-timer.C: return 0, timeoutErr } } func (r timeoutReadCloser) Close() error { return r.reader.Close() } const ( // HandlerResponseTimeout is what we use to signify the name of the // response timeout handler. HandlerResponseTimeout = "ResponseTimeoutHandler" ) // adaptToResponseTimeoutError is a handler that will replace any top level error // to a ErrCodeResponseTimeout, if its child is that. func adaptToResponseTimeoutError(req Request) { if err, ok := req.Error.(awserr.Error); ok { aerr, ok := err.OrigErr().(awserr.Error) if ok && aerr.Code() == ErrCodeResponseTimeout { req.Error = aerr } } } // WithResponseReadTimeout is a request option that will wrap the body in a timeout read closer. // This will allow for per read timeouts. If a timeout occurred, we will return the // ErrCodeResponseTimeout. // // svc.PutObjectWithContext(ctx, params, request.WithTimeoutReadCloser(30 time.Second) func WithResponseReadTimeout(duration time.Duration) Option { return func(r Request) { var timeoutHandler = NamedHandler{ HandlerResponseTimeout, func(req Request) { req.HTTPResponse.Body = &timeoutReadCloser{ reader: req.HTTPResponse.Body, duration: duration, } }} // remove the handler so we are not stomping over any new durations. r.Handlers.Send.RemoveByName(HandlerResponseTimeout) r.Handlers.Send.PushBackNamed(timeoutHandler) r.Handlers.Unmarshal.PushBack(adaptToResponseTimeoutError) r.Handlers.UnmarshalError.PushBack(adaptToResponseTimeoutError) } }	2.4 KB	2025-05-25 12:27
📄 timeout_read_closer_benchmark_test.go package request_test import ( "bytes" "io/ioutil" "net/http" "testing" "time" "github.com/aws/aws-sdk-go/aws/client" "github.com/aws/aws-sdk-go/aws/client/metadata" "github.com/aws/aws-sdk-go/aws/request" "github.com/aws/aws-sdk-go/aws/signer/v4" "github.com/aws/aws-sdk-go/awstesting/unit" "github.com/aws/aws-sdk-go/private/protocol/jsonrpc" ) func BenchmarkTimeoutReadCloser(b testing.B) { resp := ` { "Bar": "qux" } ` handlers := request.Handlers{} handlers.Send.PushBack(func(r request.Request) { r.HTTPResponse = &http.Response{ StatusCode: http.StatusOK, Body: ioutil.NopCloser(bytes.NewBuffer([]byte(resp))), } }) handlers.Sign.PushBackNamed(v4.SignRequestHandler) handlers.Build.PushBackNamed(jsonrpc.BuildHandler) handlers.Unmarshal.PushBackNamed(jsonrpc.UnmarshalHandler) handlers.UnmarshalMeta.PushBackNamed(jsonrpc.UnmarshalMetaHandler) handlers.UnmarshalError.PushBackNamed(jsonrpc.UnmarshalErrorHandler) op := &request.Operation{ Name: "op", HTTPMethod: "POST", HTTPPath: "/", } meta := metadata.ClientInfo{ ServiceName: "fooService", SigningName: "foo", SigningRegion: "foo", Endpoint: "localhost", APIVersion: "2001-01-01", JSONVersion: "1.1", TargetPrefix: "Foo", } req := request.New( unit.Session.Config, meta, handlers, client.DefaultRetryer{NumMaxRetries: 5}, op, &struct { Foo string }{}, &struct { Bar string }{}, ) req.ApplyOptions(request.WithResponseReadTimeout(15 time.Second)) for i := 0; i < b.N; i++ { err := req.Send() if err != nil { b.Errorf("Expected no error, but received %v", err) } } }	1.63 KB	2025-05-25 12:27
📄 timeout_read_closer_test.go package request import ( "bytes" "io" "io/ioutil" "net/http" "net/url" "testing" "time" "github.com/aws/aws-sdk-go/aws/awserr" ) type testReader struct { duration time.Duration count int } func (r testReader) Read(b []byte) (int, error) { if r.count > 0 { r.count-- return len(b), nil } time.Sleep(r.duration) return 0, io.EOF } func (r testReader) Close() error { return nil } func TestTimeoutReadCloser(t testing.T) { reader := timeoutReadCloser{ reader: &testReader{ duration: time.Second, count: 5, }, duration: time.Millisecond, } b := make([]byte, 100) _, err := reader.Read(b) if err != nil { t.Log(err) } } func TestTimeoutReadCloserSameDuration(t testing.T) { reader := timeoutReadCloser{ reader: &testReader{ duration: time.Millisecond, count: 5, }, duration: time.Millisecond, } b := make([]byte, 100) _, err := reader.Read(b) if err != nil { t.Log(err) } } func TestWithResponseReadTimeout(t testing.T) { r := Request{ HTTPRequest: &http.Request{ URL: &url.URL{}, }, HTTPResponse: &http.Response{ Body: ioutil.NopCloser(bytes.NewReader(nil)), }, } r.ApplyOptions(WithResponseReadTimeout(time.Second)) err := r.Send() if err != nil { t.Error(err) } v, ok := r.HTTPResponse.Body.(timeoutReadCloser) if !ok { t.Error("Expected the body to be a timeoutReadCloser") } if v.duration != time.Second { t.Errorf("Expected %v, but receive %v\n", time.Second, v.duration) } } func TestAdaptToResponseTimeout(t *testing.T) { testCases := []struct { childErr error r Request expectedRootCode string }{ { childErr: awserr.New(ErrCodeResponseTimeout, "timeout!", nil), r: Request{ Error: awserr.New("ErrTest", "FooBar", awserr.New(ErrCodeResponseTimeout, "timeout!", nil)), }, expectedRootCode: ErrCodeResponseTimeout, }, { childErr: awserr.New(ErrCodeResponseTimeout+"1", "timeout!", nil), r: Request{ Error: awserr.New("ErrTest", "FooBar", awserr.New(ErrCodeResponseTimeout+"1", "timeout!", nil)), }, expectedRootCode: "ErrTest", }, { r: Request{ Error: awserr.New("ErrTest", "FooBar", nil), }, expectedRootCode: "ErrTest", }, } for i, c := range testCases { adaptToResponseTimeoutError(&c.r) if aerr, ok := c.r.Error.(awserr.Error); !ok { t.Errorf("Case %d: Expected 'awserr', but received %v", i+1, c.r.Error) } else if aerr.Code() != c.expectedRootCode { t.Errorf("Case %d: Expected %q, but received %s", i+1, c.expectedRootCode, aerr.Code()) } } }	2.51 KB	2025-05-25 12:27
📄 validation.go package request import ( "bytes" "fmt" "github.com/aws/aws-sdk-go/aws/awserr" ) const ( // InvalidParameterErrCode is the error code for invalid parameters errors InvalidParameterErrCode = "InvalidParameter" // ParamRequiredErrCode is the error code for required parameter errors ParamRequiredErrCode = "ParamRequiredError" // ParamMinValueErrCode is the error code for fields with too low of a // number value. ParamMinValueErrCode = "ParamMinValueError" // ParamMinLenErrCode is the error code for fields without enough elements. ParamMinLenErrCode = "ParamMinLenError" // ParamMaxLenErrCode is the error code for value being too long. ParamMaxLenErrCode = "ParamMaxLenError" // ParamFormatErrCode is the error code for a field with invalid // format or characters. ParamFormatErrCode = "ParamFormatInvalidError" ) // Validator provides a way for types to perform validation logic on their // input values that external code can use to determine if a type's values // are valid. type Validator interface { Validate() error } // An ErrInvalidParams provides wrapping of invalid parameter errors found when // validating API operation input parameters. type ErrInvalidParams struct { // Context is the base context of the invalid parameter group. Context string errs []ErrInvalidParam } // Add adds a new invalid parameter error to the collection of invalid // parameters. The context of the invalid parameter will be updated to reflect // this collection. func (e ErrInvalidParams) Add(err ErrInvalidParam) { err.SetContext(e.Context) e.errs = append(e.errs, err) } // AddNested adds the invalid parameter errors from another ErrInvalidParams // value into this collection. The nested errors will have their nested context // updated and base context to reflect the merging. // // Use for nested validations errors. func (e ErrInvalidParams) AddNested(nestedCtx string, nested ErrInvalidParams) { for _, err := range nested.errs { err.SetContext(e.Context) err.AddNestedContext(nestedCtx) e.errs = append(e.errs, err) } } // Len returns the number of invalid parameter errors func (e ErrInvalidParams) Len() int { return len(e.errs) } // Code returns the code of the error func (e ErrInvalidParams) Code() string { return InvalidParameterErrCode } // Message returns the message of the error func (e ErrInvalidParams) Message() string { return fmt.Sprintf("%d validation error(s) found.", len(e.errs)) } // Error returns the string formatted form of the invalid parameters. func (e ErrInvalidParams) Error() string { w := &bytes.Buffer{} fmt.Fprintf(w, "%s: %s\n", e.Code(), e.Message()) for _, err := range e.errs { fmt.Fprintf(w, "- %s\n", err.Message()) } return w.String() } // OrigErr returns the invalid parameters as a awserr.BatchedErrors value func (e ErrInvalidParams) OrigErr() error { return awserr.NewBatchError( InvalidParameterErrCode, e.Message(), e.OrigErrs()) } // OrigErrs returns a slice of the invalid parameters func (e ErrInvalidParams) OrigErrs() []error { errs := make([]error, len(e.errs)) for i := 0; i < len(errs); i++ { errs[i] = e.errs[i] } return errs } // An ErrInvalidParam represents an invalid parameter error type. type ErrInvalidParam interface { awserr.Error // Field name the error occurred on. Field() string // SetContext updates the context of the error. SetContext(string) // AddNestedContext updates the error's context to include a nested level. AddNestedContext(string) } type errInvalidParam struct { context string nestedContext string field string code string msg string } // Code returns the error code for the type of invalid parameter. func (e errInvalidParam) Code() string { return e.code } // Message returns the reason the parameter was invalid, and its context. func (e errInvalidParam) Message() string { return fmt.Sprintf("%s, %s.", e.msg, e.Field()) } // Error returns the string version of the invalid parameter error. func (e errInvalidParam) Error() string { return fmt.Sprintf("%s: %s", e.code, e.Message()) } // OrigErr returns nil, Implemented for awserr.Error interface. func (e errInvalidParam) OrigErr() error { return nil } // Field Returns the field and context the error occurred. func (e errInvalidParam) Field() string { field := e.context if len(field) > 0 { field += "." } if len(e.nestedContext) > 0 { field += fmt.Sprintf("%s.", e.nestedContext) } field += e.field return field } // SetContext updates the base context of the error. func (e errInvalidParam) SetContext(ctx string) { e.context = ctx } // AddNestedContext prepends a context to the field's path. func (e errInvalidParam) AddNestedContext(ctx string) { if len(e.nestedContext) == 0 { e.nestedContext = ctx } else { e.nestedContext = fmt.Sprintf("%s.%s", ctx, e.nestedContext) } } // An ErrParamRequired represents an required parameter error. type ErrParamRequired struct { errInvalidParam } // NewErrParamRequired creates a new required parameter error. func NewErrParamRequired(field string) ErrParamRequired { return &ErrParamRequired{ errInvalidParam{ code: ParamRequiredErrCode, field: field, msg: fmt.Sprintf("missing required field"), }, } } // An ErrParamMinValue represents a minimum value parameter error. type ErrParamMinValue struct { errInvalidParam min float64 } // NewErrParamMinValue creates a new minimum value parameter error. func NewErrParamMinValue(field string, min float64) ErrParamMinValue { return &ErrParamMinValue{ errInvalidParam: errInvalidParam{ code: ParamMinValueErrCode, field: field, msg: fmt.Sprintf("minimum field value of %v", min), }, min: min, } } // MinValue returns the field's require minimum value. // // float64 is returned for both int and float min values. func (e ErrParamMinValue) MinValue() float64 { return e.min } // An ErrParamMinLen represents a minimum length parameter error. type ErrParamMinLen struct { errInvalidParam min int } // NewErrParamMinLen creates a new minimum length parameter error. func NewErrParamMinLen(field string, min int) ErrParamMinLen { return &ErrParamMinLen{ errInvalidParam: errInvalidParam{ code: ParamMinLenErrCode, field: field, msg: fmt.Sprintf("minimum field size of %v", min), }, min: min, } } // MinLen returns the field's required minimum length. func (e ErrParamMinLen) MinLen() int { return e.min } // An ErrParamMaxLen represents a maximum length parameter error. type ErrParamMaxLen struct { errInvalidParam max int } // NewErrParamMaxLen creates a new maximum length parameter error. func NewErrParamMaxLen(field string, max int, value string) ErrParamMaxLen { return &ErrParamMaxLen{ errInvalidParam: errInvalidParam{ code: ParamMaxLenErrCode, field: field, msg: fmt.Sprintf("maximum size of %v, %v", max, value), }, max: max, } } // MaxLen returns the field's required minimum length. func (e ErrParamMaxLen) MaxLen() int { return e.max } // An ErrParamFormat represents a invalid format parameter error. type ErrParamFormat struct { errInvalidParam format string } // NewErrParamFormat creates a new invalid format parameter error. func NewErrParamFormat(field string, format, value string) ErrParamFormat { return &ErrParamFormat{ errInvalidParam: errInvalidParam{ code: ParamFormatErrCode, field: field, msg: fmt.Sprintf("format %v, %v", format, value), }, format: format, } } // Format returns the field's required format. func (e ErrParamFormat) Format() string { return e.format }	7.44 KB	2025-05-25 12:27
📄 waiter.go package request import ( "fmt" "time" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/aws/awsutil" ) // WaiterResourceNotReadyErrorCode is the error code returned by a waiter when // the waiter's max attempts have been exhausted. const WaiterResourceNotReadyErrorCode = "ResourceNotReady" // A WaiterOption is a function that will update the Waiter value's fields to // configure the waiter. type WaiterOption func(Waiter) // WithWaiterMaxAttempts returns the maximum number of times the waiter should // attempt to check the resource for the target state. func WithWaiterMaxAttempts(max int) WaiterOption { return func(w Waiter) { w.MaxAttempts = max } } // WaiterDelay will return a delay the waiter should pause between attempts to // check the resource state. The passed in attempt is the number of times the // Waiter has checked the resource state. // // Attempt is the number of attempts the Waiter has made checking the resource // state. type WaiterDelay func(attempt int) time.Duration // ConstantWaiterDelay returns a WaiterDelay that will always return a constant // delay the waiter should use between attempts. It ignores the number of // attempts made. func ConstantWaiterDelay(delay time.Duration) WaiterDelay { return func(attempt int) time.Duration { return delay } } // WithWaiterDelay will set the Waiter to use the WaiterDelay passed in. func WithWaiterDelay(delayer WaiterDelay) WaiterOption { return func(w Waiter) { w.Delay = delayer } } // WithWaiterLogger returns a waiter option to set the logger a waiter // should use to log warnings and errors to. func WithWaiterLogger(logger aws.Logger) WaiterOption { return func(w Waiter) { w.Logger = logger } } // WithWaiterRequestOptions returns a waiter option setting the request // options for each request the waiter makes. Appends to waiter's request // options already set. func WithWaiterRequestOptions(opts ...Option) WaiterOption { return func(w Waiter) { w.RequestOptions = append(w.RequestOptions, opts...) } } // A Waiter provides the functionality to perform a blocking call which will // wait for a resource state to be satisfied by a service. // // This type should not be used directly. The API operations provided in the // service packages prefixed with "WaitUntil" should be used instead. type Waiter struct { Name string Acceptors []WaiterAcceptor Logger aws.Logger MaxAttempts int Delay WaiterDelay RequestOptions []Option NewRequest func([]Option) (Request, error) SleepWithContext func(aws.Context, time.Duration) error } // ApplyOptions updates the waiter with the list of waiter options provided. func (w Waiter) ApplyOptions(opts ...WaiterOption) { for _, fn := range opts { fn(w) } } // WaiterState are states the waiter uses based on WaiterAcceptor definitions // to identify if the resource state the waiter is waiting on has occurred. type WaiterState int // String returns the string representation of the waiter state. func (s WaiterState) String() string { switch s { case SuccessWaiterState: return "success" case FailureWaiterState: return "failure" case RetryWaiterState: return "retry" default: return "unknown waiter state" } } // States the waiter acceptors will use to identify target resource states. const ( SuccessWaiterState WaiterState = iota // waiter successful FailureWaiterState // waiter failed RetryWaiterState // waiter needs to be retried ) // WaiterMatchMode is the mode that the waiter will use to match the WaiterAcceptor // definition's Expected attribute. type WaiterMatchMode int // Modes the waiter will use when inspecting API response to identify target // resource states. const ( PathAllWaiterMatch WaiterMatchMode = iota // match on all paths PathWaiterMatch // match on specific path PathAnyWaiterMatch // match on any path PathListWaiterMatch // match on list of paths StatusWaiterMatch // match on status code ErrorWaiterMatch // match on error ) // String returns the string representation of the waiter match mode. func (m WaiterMatchMode) String() string { switch m { case PathAllWaiterMatch: return "pathAll" case PathWaiterMatch: return "path" case PathAnyWaiterMatch: return "pathAny" case PathListWaiterMatch: return "pathList" case StatusWaiterMatch: return "status" case ErrorWaiterMatch: return "error" default: return "unknown waiter match mode" } } // WaitWithContext will make requests for the API operation using NewRequest to // build API requests. The request's response will be compared against the // Waiter's Acceptors to determine the successful state of the resource the // waiter is inspecting. // // The passed in context must not be nil. If it is nil a panic will occur. The // Context will be used to cancel the waiter's pending requests and retry delays. // Use aws.BackgroundContext if no context is available. // // The waiter will continue until the target state defined by the Acceptors, // or the max attempts expires. // // Will return the WaiterResourceNotReadyErrorCode error code if the waiter's // retryer ShouldRetry returns false. This normally will happen when the max // wait attempts expires. func (w Waiter) WaitWithContext(ctx aws.Context) error { for attempt := 1; ; attempt++ { req, err := w.NewRequest(w.RequestOptions) if err != nil { waiterLogf(w.Logger, "unable to create request %v", err) return err } req.Handlers.Build.PushBack(MakeAddToUserAgentFreeFormHandler("Waiter")) err = req.Send() // See if any of the acceptors match the request's response, or error for _, a := range w.Acceptors { if matched, matchErr := a.match(w.Name, w.Logger, req, err); matched { return matchErr } } // The Waiter should only check the resource state MaxAttempts times // This is here instead of in the for loop above to prevent delaying // unnecessary when the waiter will not retry. if attempt == w.MaxAttempts { break } // Delay to wait before inspecting the resource again delay := w.Delay(attempt) if sleepFn := req.Config.SleepDelay; sleepFn != nil { // Support SleepDelay for backwards compatibility and testing sleepFn(delay) } else { sleepCtxFn := w.SleepWithContext if sleepCtxFn == nil { sleepCtxFn = aws.SleepWithContext } if err := sleepCtxFn(ctx, delay); err != nil { return awserr.New(CanceledErrorCode, "waiter context canceled", err) } } } return awserr.New(WaiterResourceNotReadyErrorCode, "exceeded wait attempts", nil) } // A WaiterAcceptor provides the information needed to wait for an API operation // to complete. type WaiterAcceptor struct { State WaiterState Matcher WaiterMatchMode Argument string Expected interface{} } // match returns if the acceptor found a match with the passed in request // or error. True is returned if the acceptor made a match, error is returned // if there was an error attempting to perform the match. func (a WaiterAcceptor) match(name string, l aws.Logger, req *Request, err error) (bool, error) { result := false var vals []interface{} switch a.Matcher { case PathAllWaiterMatch, PathWaiterMatch: // Require all matches to be equal for result to match vals, _ = awsutil.ValuesAtPath(req.Data, a.Argument) if len(vals) == 0 { break } result = true for _, val := range vals { if !awsutil.DeepEqual(val, a.Expected) { result = false break } } case PathAnyWaiterMatch: // Only a single match needs to equal for the result to match vals, _ = awsutil.ValuesAtPath(req.Data, a.Argument) for _, val := range vals { if awsutil.DeepEqual(val, a.Expected) { result = true break } } case PathListWaiterMatch: // ignored matcher case StatusWaiterMatch: s := a.Expected.(int) result = s == req.HTTPResponse.StatusCode case ErrorWaiterMatch: switch ex := a.Expected.(type) { case string: if aerr, ok := err.(awserr.Error); ok { result = aerr.Code() == ex } case bool: if ex { result = err != nil } else { result = err == nil } } default: waiterLogf(l, "WARNING: Waiter %s encountered unexpected matcher: %s", name, a.Matcher) } if !result { // If there was no matching result found there is nothing more to do // for this response, retry the request. return false, nil } switch a.State { case SuccessWaiterState: // waiter completed return true, nil case FailureWaiterState: // Waiter failure state triggered return true, awserr.New(WaiterResourceNotReadyErrorCode, "failed waiting for successful resource state", err) case RetryWaiterState: // clear the error and retry the operation return false, nil default: waiterLogf(l, "WARNING: Waiter %s encountered unexpected state: %s", name, a.State) return false, nil } } func waiterLogf(logger aws.Logger, msg string, args ...interface{}) { if logger != nil { logger.Log(fmt.Sprintf(msg, args...)) } }	8.96 KB	2025-05-25 12:27
📄 waiter_test.go package request_test import ( "bytes" "fmt" "io/ioutil" "net/http" "testing" "time" "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/awstesting" "github.com/aws/aws-sdk-go/awstesting/unit" "github.com/aws/aws-sdk-go/service/s3" ) type mockClient struct { client.Client } type MockInput struct{} type MockOutput struct { States []MockState } type MockState struct { State string } func (c mockClient) MockRequest(input MockInput) (request.Request, MockOutput) { op := &request.Operation{ Name: "Mock", HTTPMethod: "POST", HTTPPath: "/", } if input == nil { input = &MockInput{} } output := &MockOutput{} req := c.NewRequest(op, input, output) req.Data = output return req, output } func BuildNewMockRequest(c mockClient, in MockInput) func([]request.Option) (request.Request, error) { return func(opts []request.Option) (request.Request, error) { req, _ := c.MockRequest(in) req.ApplyOptions(opts...) return req, nil } } func TestWaiterPathAll(t testing.T) { svc := &mockClient{Client: awstesting.NewClient(&aws.Config{ Region: aws.String("mock-region"), })} svc.Handlers.Send.Clear() // mock sending svc.Handlers.Unmarshal.Clear() svc.Handlers.UnmarshalMeta.Clear() svc.Handlers.ValidateResponse.Clear() reqNum := 0 resps := []MockOutput{ { // Request 1 States: []MockState{ {State: aws.String("pending")}, {State: aws.String("pending")}, }, }, { // Request 2 States: []MockState{ {State: aws.String("running")}, {State: aws.String("pending")}, }, }, { // Request 3 States: []MockState{ {State: aws.String("running")}, {State: aws.String("running")}, }, }, } numBuiltReq := 0 svc.Handlers.Build.PushBack(func(r request.Request) { numBuiltReq++ }) svc.Handlers.Unmarshal.PushBack(func(r request.Request) { if reqNum >= len(resps) { t.Errorf("too many polling requests made") return } r.Data = resps[reqNum] reqNum++ }) w := request.Waiter{ MaxAttempts: 10, Delay: request.ConstantWaiterDelay(0), SleepWithContext: aws.SleepWithContext, Acceptors: []request.WaiterAcceptor{ { State: request.SuccessWaiterState, Matcher: request.PathAllWaiterMatch, Argument: "States[].State", Expected: "running", }, }, NewRequest: BuildNewMockRequest(svc, &MockInput{}), } err := w.WaitWithContext(aws.BackgroundContext()) if err != nil { t.Errorf("expect nil, %v", err) } if e, a := 3, numBuiltReq; e != a { t.Errorf("expect %v, got %v", e, a) } if e, a := 3, reqNum; e != a { t.Errorf("expect %v, got %v", e, a) } } func TestWaiterPath(t testing.T) { svc := &mockClient{Client: awstesting.NewClient(&aws.Config{ Region: aws.String("mock-region"), })} svc.Handlers.Send.Clear() // mock sending svc.Handlers.Unmarshal.Clear() svc.Handlers.UnmarshalMeta.Clear() svc.Handlers.ValidateResponse.Clear() reqNum := 0 resps := []MockOutput{ { // Request 1 States: []MockState{ {State: aws.String("pending")}, {State: aws.String("pending")}, }, }, { // Request 2 States: []MockState{ {State: aws.String("running")}, {State: aws.String("pending")}, }, }, { // Request 3 States: []MockState{ {State: aws.String("running")}, {State: aws.String("running")}, }, }, } numBuiltReq := 0 svc.Handlers.Build.PushBack(func(r request.Request) { numBuiltReq++ }) svc.Handlers.Unmarshal.PushBack(func(r request.Request) { if reqNum >= len(resps) { t.Errorf("too many polling requests made") return } r.Data = resps[reqNum] reqNum++ }) w := request.Waiter{ MaxAttempts: 10, Delay: request.ConstantWaiterDelay(0), SleepWithContext: aws.SleepWithContext, Acceptors: []request.WaiterAcceptor{ { State: request.SuccessWaiterState, Matcher: request.PathWaiterMatch, Argument: "States[].State", Expected: "running", }, }, NewRequest: BuildNewMockRequest(svc, &MockInput{}), } err := w.WaitWithContext(aws.BackgroundContext()) if err != nil { t.Errorf("expect nil, %v", err) } if e, a := 3, numBuiltReq; e != a { t.Errorf("expect %v, got %v", e, a) } if e, a := 3, reqNum; e != a { t.Errorf("expect %v, got %v", e, a) } } func TestWaiterFailure(t testing.T) { svc := &mockClient{Client: awstesting.NewClient(&aws.Config{ Region: aws.String("mock-region"), })} svc.Handlers.Send.Clear() // mock sending svc.Handlers.Unmarshal.Clear() svc.Handlers.UnmarshalMeta.Clear() svc.Handlers.ValidateResponse.Clear() reqNum := 0 resps := []MockOutput{ { // Request 1 States: []MockState{ {State: aws.String("pending")}, {State: aws.String("pending")}, }, }, { // Request 2 States: []MockState{ {State: aws.String("running")}, {State: aws.String("pending")}, }, }, { // Request 3 States: []MockState{ {State: aws.String("running")}, {State: aws.String("stopping")}, }, }, } numBuiltReq := 0 svc.Handlers.Build.PushBack(func(r request.Request) { numBuiltReq++ }) svc.Handlers.Unmarshal.PushBack(func(r request.Request) { if reqNum >= len(resps) { t.Errorf("too many polling requests made") return } r.Data = resps[reqNum] reqNum++ }) w := request.Waiter{ MaxAttempts: 10, Delay: request.ConstantWaiterDelay(0), SleepWithContext: aws.SleepWithContext, Acceptors: []request.WaiterAcceptor{ { State: request.SuccessWaiterState, Matcher: request.PathAllWaiterMatch, Argument: "States[].State", Expected: "running", }, { State: request.FailureWaiterState, Matcher: request.PathAnyWaiterMatch, Argument: "States[].State", Expected: "stopping", }, }, NewRequest: BuildNewMockRequest(svc, &MockInput{}), } err := w.WaitWithContext(aws.BackgroundContext()).(awserr.Error) if err == nil { t.Errorf("expect error") } if e, a := request.WaiterResourceNotReadyErrorCode, err.Code(); e != a { t.Errorf("expect %v, got %v", e, a) } if e, a := "failed waiting for successful resource state", err.Message(); e != a { t.Errorf("expect %v, got %v", e, a) } if e, a := 3, numBuiltReq; e != a { t.Errorf("expect %v, got %v", e, a) } if e, a := 3, reqNum; e != a { t.Errorf("expect %v, got %v", e, a) } } func TestWaiterError(t testing.T) { svc := &mockClient{Client: awstesting.NewClient(&aws.Config{ Region: aws.String("mock-region"), })} svc.Handlers.Send.Clear() // mock sending svc.Handlers.Unmarshal.Clear() svc.Handlers.UnmarshalMeta.Clear() svc.Handlers.UnmarshalError.Clear() svc.Handlers.ValidateResponse.Clear() reqNum := 0 resps := []MockOutput{ { // Request 1 States: []MockState{ {State: aws.String("pending")}, {State: aws.String("pending")}, }, }, { // Request 1, error case retry }, { // Request 2, error case failure }, { // Request 3 States: []MockState{ {State: aws.String("running")}, {State: aws.String("running")}, }, }, } reqErrs := make([]error, len(resps)) reqErrs[1] = awserr.New("MockException", "mock exception message", nil) reqErrs[2] = awserr.New("FailureException", "mock failure exception message", nil) numBuiltReq := 0 svc.Handlers.Build.PushBack(func(r request.Request) { numBuiltReq++ }) svc.Handlers.Send.PushBack(func(r request.Request) { code := 200 if reqNum == 1 { code = 400 } r.HTTPResponse = &http.Response{ StatusCode: code, Status: http.StatusText(code), Body: ioutil.NopCloser(bytes.NewReader([]byte{})), } }) svc.Handlers.Unmarshal.PushBack(func(r request.Request) { if reqNum >= len(resps) { t.Errorf("too many polling requests made") return } r.Data = resps[reqNum] reqNum++ }) svc.Handlers.UnmarshalMeta.PushBack(func(r request.Request) { // If there was an error unmarshal error will be called instead of unmarshal // need to increment count here also if err := reqErrs[reqNum]; err != nil { r.Error = err reqNum++ } }) w := request.Waiter{ MaxAttempts: 10, Delay: request.ConstantWaiterDelay(0), SleepWithContext: aws.SleepWithContext, Acceptors: []request.WaiterAcceptor{ { State: request.SuccessWaiterState, Matcher: request.PathAllWaiterMatch, Argument: "States[].State", Expected: "running", }, { State: request.RetryWaiterState, Matcher: request.ErrorWaiterMatch, Argument: "", Expected: "MockException", }, { State: request.FailureWaiterState, Matcher: request.ErrorWaiterMatch, Argument: "", Expected: "FailureException", }, }, NewRequest: BuildNewMockRequest(svc, &MockInput{}), } err := w.WaitWithContext(aws.BackgroundContext()) if err == nil { t.Fatalf("expected error, but did not get one") } aerr := err.(awserr.Error) if e, a := request.WaiterResourceNotReadyErrorCode, aerr.Code(); e != a { t.Errorf("expect %q error code, got %q", e, a) } if e, a := 3, numBuiltReq; e != a { t.Errorf("expect %d built requests got %d", e, a) } if e, a := 3, reqNum; e != a { t.Errorf("expect %d reqNum got %d", e, a) } } func TestWaiterRetryAnyError(t testing.T) { svc := &mockClient{Client: awstesting.NewClient(&aws.Config{ Region: aws.String("mock-region"), })} svc.Handlers.Send.Clear() // mock sending svc.Handlers.Unmarshal.Clear() svc.Handlers.UnmarshalMeta.Clear() svc.Handlers.UnmarshalError.Clear() svc.Handlers.ValidateResponse.Clear() var reqNum int results := []struct { Out MockOutput Err error }{ { // retry Err: awserr.New( "MockException1", "mock exception message", nil, ), }, { // retry Err: awserr.New( "MockException2", "mock exception message", nil, ), }, { // success Out: &MockOutput{ States: []MockState{ {aws.String("running")}, {aws.String("running")}, }, }, }, { // shouldn't happen Out: &MockOutput{ States: []MockState{ {aws.String("running")}, {aws.String("running")}, }, }, }, } numBuiltReq := 0 svc.Handlers.Build.PushBack(func(r request.Request) { numBuiltReq++ }) svc.Handlers.Send.PushBack(func(r request.Request) { code := http.StatusOK r.HTTPResponse = &http.Response{ StatusCode: code, Status: http.StatusText(code), Body: ioutil.NopCloser(bytes.NewReader([]byte{})), } }) svc.Handlers.Unmarshal.PushBack(func(r request.Request) { if reqNum >= len(results) { t.Errorf("too many polling requests made") return } r.Data = results[reqNum].Out reqNum++ }) svc.Handlers.UnmarshalMeta.PushBack(func(r request.Request) { // If there was an error unmarshal error will be called instead of unmarshal // need to increment count here also if err := results[reqNum].Err; err != nil { r.Error = err reqNum++ } }) w := request.Waiter{ MaxAttempts: 10, Delay: request.ConstantWaiterDelay(0), SleepWithContext: aws.SleepWithContext, Acceptors: []request.WaiterAcceptor{ { State: request.SuccessWaiterState, Matcher: request.PathAllWaiterMatch, Argument: "States[].State", Expected: "running", }, { State: request.RetryWaiterState, Matcher: request.ErrorWaiterMatch, Argument: "", Expected: true, }, { State: request.FailureWaiterState, Matcher: request.ErrorWaiterMatch, Argument: "", Expected: "FailureException", }, }, NewRequest: BuildNewMockRequest(svc, &MockInput{}), } err := w.WaitWithContext(aws.BackgroundContext()) if err != nil { t.Fatalf("expected no error, but did get one: %v", err) } if e, a := 3, numBuiltReq; e != a { t.Errorf("expect %d built requests got %d", e, a) } if e, a := 3, reqNum; e != a { t.Errorf("expect %d reqNum got %d", e, a) } } func TestWaiterSuccessNoError(t testing.T) { svc := &mockClient{Client: awstesting.NewClient(&aws.Config{ Region: aws.String("mock-region"), })} svc.Handlers.Send.Clear() // mock sending svc.Handlers.Unmarshal.Clear() svc.Handlers.UnmarshalMeta.Clear() svc.Handlers.UnmarshalError.Clear() svc.Handlers.ValidateResponse.Clear() var reqNum int results := []struct { Out MockOutput Err error }{ { // success Out: &MockOutput{ States: []MockState{ {aws.String("pending")}, }, }, }, { // shouldn't happen Out: &MockOutput{ States: []MockState{ {aws.String("pending")}, {aws.String("pending")}, }, }, }, } numBuiltReq := 0 svc.Handlers.Build.PushBack(func(r request.Request) { numBuiltReq++ }) svc.Handlers.Send.PushBack(func(r request.Request) { code := http.StatusOK r.HTTPResponse = &http.Response{ StatusCode: code, Status: http.StatusText(code), Body: ioutil.NopCloser(bytes.NewReader([]byte{})), } }) svc.Handlers.Unmarshal.PushBack(func(r request.Request) { if reqNum >= len(results) { t.Errorf("too many polling requests made") return } r.Data = results[reqNum].Out reqNum++ }) svc.Handlers.UnmarshalMeta.PushBack(func(r request.Request) { // If there was an error unmarshal error will be called instead of unmarshal // need to increment count here also if err := results[reqNum].Err; err != nil { r.Error = err reqNum++ } }) w := request.Waiter{ MaxAttempts: 10, Delay: request.ConstantWaiterDelay(0), SleepWithContext: aws.SleepWithContext, Acceptors: []request.WaiterAcceptor{ { State: request.SuccessWaiterState, Matcher: request.PathAllWaiterMatch, Argument: "States[].State", Expected: "running", }, { State: request.SuccessWaiterState, Matcher: request.ErrorWaiterMatch, Argument: "", Expected: false, }, { State: request.FailureWaiterState, Matcher: request.ErrorWaiterMatch, Argument: "", Expected: "FailureException", }, }, NewRequest: BuildNewMockRequest(svc, &MockInput{}), } err := w.WaitWithContext(aws.BackgroundContext()) if err != nil { t.Fatalf("expected no error, but did get one") } if e, a := 1, numBuiltReq; e != a { t.Errorf("expect %d built requests got %d", e, a) } if e, a := 1, reqNum; e != a { t.Errorf("expect %d reqNum got %d", e, a) } } func TestWaiterStatus(t testing.T) { svc := &mockClient{Client: awstesting.NewClient(&aws.Config{ Region: aws.String("mock-region"), })} svc.Handlers.Send.Clear() // mock sending svc.Handlers.Unmarshal.Clear() svc.Handlers.UnmarshalMeta.Clear() svc.Handlers.ValidateResponse.Clear() reqNum := 0 svc.Handlers.Build.PushBack(func(r request.Request) { reqNum++ }) svc.Handlers.Send.PushBack(func(r request.Request) { code := 200 if reqNum == 3 { code = 404 r.Error = awserr.New("NotFound", "Not Found", nil) } r.HTTPResponse = &http.Response{ StatusCode: code, Status: http.StatusText(code), Body: ioutil.NopCloser(bytes.NewReader([]byte{})), } }) w := request.Waiter{ MaxAttempts: 10, Delay: request.ConstantWaiterDelay(0), SleepWithContext: aws.SleepWithContext, Acceptors: []request.WaiterAcceptor{ { State: request.SuccessWaiterState, Matcher: request.StatusWaiterMatch, Argument: "", Expected: 404, }, }, NewRequest: BuildNewMockRequest(svc, &MockInput{}), } err := w.WaitWithContext(aws.BackgroundContext()) if err != nil { t.Errorf("expect nil, %v", err) } if e, a := 3, reqNum; e != a { t.Errorf("expect %v, got %v", e, a) } } func TestWaiter_ApplyOptions(t testing.T) { w := request.Waiter{} logger := aws.NewDefaultLogger() w.ApplyOptions( request.WithWaiterLogger(logger), request.WithWaiterRequestOptions(request.WithLogLevel(aws.LogDebug)), request.WithWaiterMaxAttempts(2), request.WithWaiterDelay(request.ConstantWaiterDelay(5time.Second)), ) if e, a := logger, w.Logger; e != a { t.Errorf("expect logger to be set, and match, was not, %v, %v", e, a) } if len(w.RequestOptions) != 1 { t.Fatalf("expect request options to be set to only a single option, %v", w.RequestOptions) } r := request.Request{} r.ApplyOptions(w.RequestOptions...) if e, a := aws.LogDebug, r.Config.LogLevel.Value(); e != a { t.Errorf("expect %v loglevel got %v", e, a) } if e, a := 2, w.MaxAttempts; e != a { t.Errorf("expect %d retryer max attempts, got %d", e, a) } if e, a := 5time.Second, w.Delay(0); e != a { t.Errorf("expect %d retryer delay, got %d", e, a) } } func TestWaiter_WithContextCanceled(t testing.T) { c := awstesting.NewClient() ctx := &awstesting.FakeContext{DoneCh: make(chan struct{})} reqCount := 0 w := request.Waiter{ Name: "TestWaiter", MaxAttempts: 10, Delay: request.ConstantWaiterDelay(1 time.Millisecond), SleepWithContext: aws.SleepWithContext, Acceptors: []request.WaiterAcceptor{ { State: request.SuccessWaiterState, Matcher: request.StatusWaiterMatch, Expected: 200, }, }, Logger: aws.NewDefaultLogger(), NewRequest: func(opts []request.Option) (request.Request, error) { req := c.NewRequest(&request.Operation{Name: "Operation"}, nil, nil) req.HTTPResponse = &http.Response{StatusCode: http.StatusNotFound} req.Handlers.Clear() req.Data = struct{}{} req.Handlers.Send.PushBack(func(r request.Request) { if reqCount == 1 { ctx.Error = fmt.Errorf("context canceled") close(ctx.DoneCh) } reqCount++ }) return req, nil }, } w.SleepWithContext = func(c aws.Context, delay time.Duration) error { context := c.(awstesting.FakeContext) select { case <-context.DoneCh: return context.Err() default: return nil } } err := w.WaitWithContext(ctx) if err == nil { t.Fatalf("expect waiter to be canceled.") } aerr := err.(awserr.Error) if e, a := request.CanceledErrorCode, aerr.Code(); e != a { t.Errorf("expect %q error code, got %q", e, a) } if e, a := 2, reqCount; e != a { t.Errorf("expect %d requests, got %d", e, a) } } func TestWaiter_WithContext(t testing.T) { c := awstesting.NewClient() ctx := &awstesting.FakeContext{DoneCh: make(chan struct{})} reqCount := 0 statuses := []int{http.StatusNotFound, http.StatusOK} w := request.Waiter{ Name: "TestWaiter", MaxAttempts: 10, Delay: request.ConstantWaiterDelay(1 * time.Millisecond), SleepWithContext: aws.SleepWithContext, Acceptors: []request.WaiterAcceptor{ { State: request.SuccessWaiterState, Matcher: request.StatusWaiterMatch, Expected: 200, }, }, Logger: aws.NewDefaultLogger(), NewRequest: func(opts []request.Option) (request.Request, error) { req := c.NewRequest(&request.Operation{Name: "Operation"}, nil, nil) req.HTTPResponse = &http.Response{StatusCode: statuses[reqCount]} req.Handlers.Clear() req.Data = struct{}{} req.Handlers.Send.PushBack(func(r request.Request) { if reqCount == 1 { ctx.Error = fmt.Errorf("context canceled") close(ctx.DoneCh) } reqCount++ }) return req, nil }, } err := w.WaitWithContext(ctx) if err != nil { t.Fatalf("expect no error, got %v", err) } if e, a := 2, reqCount; e != a { t.Errorf("expect %d requests, got %d", e, a) } } func TestWaiter_AttemptsExpires(t testing.T) { c := awstesting.NewClient() ctx := &awstesting.FakeContext{DoneCh: make(chan struct{})} reqCount := 0 w := request.Waiter{ Name: "TestWaiter", MaxAttempts: 2, Delay: request.ConstantWaiterDelay(1 time.Millisecond), SleepWithContext: aws.SleepWithContext, Acceptors: []request.WaiterAcceptor{ { State: request.SuccessWaiterState, Matcher: request.StatusWaiterMatch, Expected: 200, }, }, Logger: aws.NewDefaultLogger(), NewRequest: func(opts []request.Option) (request.Request, error) { req := c.NewRequest(&request.Operation{Name: "Operation"}, nil, nil) req.HTTPResponse = &http.Response{StatusCode: http.StatusNotFound} req.Handlers.Clear() req.Data = struct{}{} req.Handlers.Send.PushBack(func(r request.Request) { reqCount++ }) return req, nil }, } err := w.WaitWithContext(ctx) if err == nil { t.Fatalf("expect error did not get one") } aerr := err.(awserr.Error) if e, a := request.WaiterResourceNotReadyErrorCode, aerr.Code(); e != a { t.Errorf("expect %q error code, got %q", e, a) } if e, a := 2, reqCount; e != a { t.Errorf("expect %d requests, got %d", e, a) } } func TestWaiterNilInput(t testing.T) { // Code generation doesn't have a great way to verify the code is correct // other than being run via unit tests in the SDK. This should be fixed // So code generation can be validated independently. client := s3.New(unit.Session) client.Handlers.Validate.Clear() client.Handlers.Send.Clear() // mock sending client.Handlers.Send.PushBack(func(r request.Request) { r.HTTPResponse = &http.Response{ StatusCode: http.StatusOK, } }) client.Handlers.Unmarshal.Clear() client.Handlers.UnmarshalMeta.Clear() client.Handlers.ValidateResponse.Clear() client.Config.SleepDelay = func(dur time.Duration) {} // Ensure waiters do not panic on nil input. It doesn't make sense to // call a waiter without an input, Validation will err := client.WaitUntilBucketExists(nil) if err != nil { t.Fatalf("expect no error, but got %v", err) } } func TestWaiterWithContextNilInput(t testing.T) { // Code generation doesn't have a great way to verify the code is correct // other than being run via unit tests in the SDK. This should be fixed // So code generation can be validated independently. client := s3.New(unit.Session) client.Handlers.Validate.Clear() client.Handlers.Send.Clear() // mock sending client.Handlers.Send.PushBack(func(r request.Request) { r.HTTPResponse = &http.Response{ StatusCode: http.StatusOK, } }) client.Handlers.Unmarshal.Clear() client.Handlers.UnmarshalMeta.Clear() client.Handlers.ValidateResponse.Clear() // Ensure waiters do not panic on nil input ctx := &awstesting.FakeContext{DoneCh: make(chan struct{})} err := client.WaitUntilBucketExistsWithContext(ctx, nil, request.WithWaiterDelay(request.ConstantWaiterDelay(0)), request.WithWaiterMaxAttempts(1), ) if err != nil { t.Fatalf("expect no error, but got %v", err) } }	21.95 KB	2025-05-25 12:27

Edit File:

Terminal