limit number of redirect attempts
@@ -29,8 +29,12 @@ var (
ErrMalformed = errors.New("scheduler: malformed request")
// ErrMediaType is returned by Do calls that are sent with an unsupported media type.
ErrMediaType = errors.New("scheduler: unsupported media type")
- // ErrRateLimit is returned by Do calls that are rate limited.
+ // ErrRateLimit is returned by Do calls that are rate limited. This is a temporary condition
+ // that should clear.
ErrRateLimit = errors.New("scheduler: rate limited")
+ // ErrUnavailable is returned by Do calls that are sent to a master that's in recovery, or
+ // does not yet realize that it's the leader. This is a temporary condition that should clear.
+ ErrUnavailable = errors.New("scheduler: master unavailable")
// codeErrors maps HTTP response codes to their respective errors.
codeErrors = map[int]error{
@@ -42,7 +46,8 @@ var (
http.StatusForbidden: ErrUnsubscribed,
http.StatusUnauthorized: ErrAuth,
http.StatusNotAcceptable: ErrMediaType,
- 429: ErrRateLimit,
+ http.StatusServiceUnavailable: ErrUnavailable,
+ http.StatusTooManyRequests: ErrRateLimit,
}
)
@@ -116,6 +121,8 @@ func (c *Client) With(opts ...Opt) Opt {
return last
}
+const maxRedirectAttempts = 9
+
// Do sends a Call and returns a streaming Decoder from which callers can read
// Events from, an io.Closer to close the event stream on graceful termination
// and an error in case of failure. Callers are expected to *always* close a
@@ -123,6 +130,8 @@ func (c *Client) With(opts ...Opt) Opt {
// but also for which there is no expected object stream as a result the
// returned Decoder will be nil.
func (c *Client) Do(m encoding.Marshaler, opt ...RequestOpt) (*Response, error) {
+ attempt := 0
+ for {
var body bytes.Buffer
if err := c.codec.NewEncoder(&body).Invoke(m); err != nil {
return nil, err
@@ -172,21 +181,26 @@ func (c *Client) Do(m encoding.Marshaler, opt ...RequestOpt) (*Response, error)
if debug {
log.Println("master changed!")
}
+ if attempt < maxRedirectAttempts {
+ attempt++
newMaster := res.Header.Get("Location")
if newMaster != "" {
// current format appears to be //x.y.z.w:port
hostport, parseErr := url.Parse(newMaster)
- if parseErr == nil && hostport.Host != "" {
+ if parseErr != nil || hostport.Host == "" {
+ break
+ }
current, parseErr := url.Parse(c.url)
- if parseErr == nil {
+ if parseErr != nil {
+ break
+ }
current.Host = hostport.Host
c.url = current.String()
if debug {
log.Println("master changed, redirecting to " + c.url)
}
res.Body.Close()
- return c.Do(m, opt...)
- }
+ continue
}
}
default:
@@ -203,6 +217,7 @@ func (c *Client) Do(m encoding.Marshaler, opt ...RequestOpt) (*Response, error)
Closer: res.Body,
}, codeErrors[res.StatusCode]
}
+}
// URL returns an Opt that sets a Client's URL.
func URL(rawurl string) Opt {
