Complete retry initial implementation

retry/retry.go

@@ -0,0 +1,83 @@
+package retry
+
+import (
+	"net/http"
+)
+
+type Retry struct {
+	next        http.Handler
+	errHandler  http.Handler
+	maxAttempts int
+	predicate   hpredicate
+}
+
+func New(next http.Handler, predicate string, settings ...optSetter) (*Retry, error) {
+	p, err := parseExpression(predicate)
+	if err != nil {
+		return nil, err
+	}
+
+	r := &Retry{
+		next:      next,
+		predicate: p,
+	}
+	for _, s := range settings {
+		if err := s(r); err != nil {
+			return nil, err
+		}
+	}
+	if r.maxAttempts == 0 {
+		r.maxAttempts = DefaultMaxAttempts
+	}
+	return r, nil
+}
+
+func (r *Retry) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	c := &context{r: req}
+	for i := 0; i < r.maxAttempts; i++ {
+		pw := &proxyWriter{w: w}
+		r.next.ServeHTTP(pw, req)
+		c.attempt = i + 1
+		if !r.predicate(c) {
+			return
+		}
+	}
+	// if we ended up here, we have excceded all the max attempts what is a system error
+	r.errHandler.ServeHTTP(w, req)
+}
+
+func MaxAttempts(a int) optSetter {
+	return func(r *Retry) error {
+		r.maxAttempts = a
+		return nil
+	}
+}
+
+func ErrorHandler(h http.Handler) optSetter {
+	return func(r *Retry) error {
+		r.errHandler = h
+		return nil
+	}
+}
+
+type optSetter func(r *Retry) error
+
+const DefaultMaxAttempts = 10
+
+type proxyWriter struct {
+	w    http.ResponseWriter
+	code int
+}
+
+func (p *proxyWriter) Header() http.Header {
+	return p.w.Header()
+}
+
+func (p *proxyWriter) Write(buf []byte) (int, error) {
+	return p.w.Write(buf)
+}
+
+func (p *proxyWriter) WriteHeader(code int) {
+	p.code = code
+	p.WriteHeader(code)
+}

retry/threshold.go

@@ -0,0 +1,151 @@
+package retry
+
+import (
+	"fmt"
+	"net"
+	"net/http"
+
+	"github.com/mailgun/predicate"
+)
+
+type context struct {
+	r            *http.Request
+	attempt      int
+	responseCode int
+	err          error
+}
+
+type hpredicate func(*context) bool
+
+// Parses expression in the go language into Failover predicates
+func parseExpression(in string) (hpredicate, error) {
+	p, err := predicate.NewParser(predicate.Def{
+		Operators: predicate.Operators{
+			AND: and,
+			OR:  or,
+			EQ:  eq,
+			NEQ: neq,
+		},
+		Functions: map[string]interface{}{
+			"RequestMethod":  requestMethod,
+			"IsNetworkError": isNetworkError,
+			"Attempts":       attempts,
+			"ResponseCode":   responseCode,
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+	out, err := p.Parse(in)
+	if err != nil {
+		return nil, err
+	}
+	pr, ok := out.(hpredicate)
+	if !ok {
+		return nil, fmt.Errorf("expected predicate, got %T", out)
+	}
+	return pr, nil
+}
+
+type toString func(c *context) string
+type toInt func(c *context) int
+
+// RequestMethod returns mapper of the request to its method e.g. POST
+func requestMethod() toString {
+	return func(c *context) string {
+		return c.r.Method
+	}
+}
+
+// Attempts returns mapper of the request to the number of proxy attempts
+func attempts() toInt {
+	return func(c *context) int {
+		return c.attempt
+	}
+}
+
+// ResponseCode returns mapper of the request to the last response code, returns 0 if there was no response code.
+func responseCode() toInt {
+	return func(c *context) int {
+		return c.responseCode
+	}
+}
+
+// IsNetworkError returns a predicate that returns true if last attempt ended with network error.
+func isNetworkError() hpredicate {
+	return func(c *context) bool {
+		_, ok := c.err.(net.Error)
+		return ok
+	}
+}
+
+// and returns predicate by joining the passed predicates with logical 'and'
+func and(fns ...hpredicate) hpredicate {
+	return func(c *context) bool {
+		for _, fn := range fns {
+			if !fn(c) {
+				return false
+			}
+		}
+		return true
+	}
+}
+
+// or returns predicate by joining the passed predicates with logical 'or'
+func or(fns ...hpredicate) hpredicate {
+	return func(c *context) bool {
+		for _, fn := range fns {
+			if fn(c) {
+				return true
+			}
+		}
+		return false
+	}
+}
+
+// not creates negation of the passed predicate
+func not(p hpredicate) hpredicate {
+	return func(c *context) bool {
+		return !p(c)
+	}
+}
+
+// eq returns predicate that tests for equality of the value of the mapper and the constant
+func eq(m interface{}, value interface{}) (hpredicate, error) {
+	switch mapper := m.(type) {
+	case toString:
+		return stringEQ(mapper, value)
+	case toInt:
+		return intEQ(mapper, value)
+	}
+	return nil, fmt.Errorf("unsupported argument: %T", m)
+}
+
+// neq returns predicate that tests for inequality of the value of the mapper and the constant
+func neq(m interface{}, value interface{}) (hpredicate, error) {
+	p, err := eq(m, value)
+	if err != nil {
+		return nil, err
+	}
+	return not(p), nil
+}
+
+func stringEQ(m toString, val interface{}) (hpredicate, error) {
+	value, ok := val.(string)
+	if !ok {
+		return nil, fmt.Errorf("expected string, got %T", val)
+	}
+	return func(c *context) bool {
+		return m(c) == value
+	}, nil
+}
+
+func intEQ(m toInt, val interface{}) (hpredicate, error) {
+	value, ok := val.(int)
+	if !ok {
+		return nil, fmt.Errorf("expected int, got %T", val)
+	}
+	return func(c *context) bool {
+		return m(c) == value
+	}, nil
+}