Merge pull request #2 from randlabs/httpclient

Added a load-balanced http client library code.

Author: mxmauro

README.md

@@ -2,7 +2,7 @@
 
 A round-robin server selection (aka load balancer) library.
 
-**IMPORTANT NOTE**: This library DOES NOT DO any kind of networking. It aims to automatically select an upstream handler in a set of primary and backup configured servers.
+The base code of this library, the balancer, *DOES NOT DO* any kind of network access. It's goal is to automatically select an upstream handler in a set of primary and backup configured servers.
 
 ## Usage with example
 
@@ -81,5 +81,61 @@ func main() {
 }
 ```
 
+## httpclient
+
+The `httpclient` module, implements an alternative to `http.Client` that allows to use a set of servers and balance
+requests among them.
+
+> Most load-balanced http client libraries makes use of the `RoundTripper` interface, but we don't.
+>
+> The major reason for this is we want to allow the dev, to be able to mark a server (temporarily) offline or retry
+> the operation, not only if connection to the server is established, but also depending on the response.
+>
+> For e.g., let's say your backend correctly answers a request but the output indicates the internal processing is not
+> up-to-date, then you can decide to stop using that server until it is.
+
+### Usage:
+
+```golang
+import (
+    "fmt"
+
+    balancer "github.com/randlabs/go-loadbalancer"
+)
+
+func main() {
+    hc := httpclient.Create()
+    _ = hc.AddSource("https://server1.test-network", httpclient.SourceOptions{
+        ServerOptions: httpclient.ServerOptions{
+            Weight:      1,
+            MaxFails:    1,
+            FailTimeout: 10 * time.Second,
+        },
+    })
+    _ = hc.AddSource("https://server2.test-network", httpclient.SourceOptions{
+        ServerOptions: httpclient.ServerOptions{
+            Weight:      1,
+            MaxFails:    1,
+            FailTimeout: 10 * time.Second,
+        },
+    })
+
+    req := hc.NewRequest("GET", "/api-test")
+    err := req.Exec(context.Background(), func (ctx context.Context, res httpclient.Response) error {
+        if res.Err() != nil || res.StatusCode != 200 {
+            // Retry on the next available server on failed request
+            res.RetryOnNextServer()
+            return nil
+        }
+
+        // Process response
+        // ...
+
+        // Done
+        return nil
+    })
+}
+```
+
 ## License
 See `LICENSE` file for details.

httpclient/error.go

@@ -0,0 +1,84 @@
+package httpclient
+
+import (
+	"errors"
+	"fmt"
+	"net"
+)
+
+// -----------------------------------------------------------------------------
+
+const (
+	errorTypeIsTimeout = 1
+	errorTypeIsCanceled = 2
+)
+
+// -----------------------------------------------------------------------------
+
+// Error is the error type usually returned by us.
+type Error struct {
+	message    string
+	url        string
+	statusCode int
+	errType    int
+	// err is the underlying error that occurred during the operation.
+	err        error
+}
+
+// -----------------------------------------------------------------------------
+
+func (c *HttpClient) newError(wrappedErr error, message string, url string, statusCode int) *Error {
+	err := Error{
+		message:    message,
+		url:        url,
+		statusCode: statusCode,
+		err:        wrappedErr,
+	}
+	return &err
+}
+
+// -----------------------------------------------------------------------------
+
+func (e *Error) URL() string {
+	return e.url
+}
+
+func (e *Error) StatusCode() int {
+	return e.statusCode
+}
+
+func (e *Error) Unwrap() error {
+	return e.err
+}
+
+func (e *Error) Error() string {
+	if e == nil {
+		return "<nil>"
+	}
+	s := e.message + fmt.Sprintf(" [URL=%v]", e.url)
+	if e.err != nil {
+		s += " [err=" + e.err.Error() + "]"
+	}
+	return s
+}
+
+func (e *Error) IsTimeout() bool {
+	return e.errType == errorTypeIsTimeout
+}
+
+func (e *Error) IsCanceled() bool {
+	return e.errType == errorTypeIsCanceled
+}
+
+func (e *Error) IsNetworkError() bool {
+	if e.err != nil {
+		var netErr net.Error
+		var netOpErr *net.OpError
+		var netDnsErr *net.DNSError
+
+		if errors.As(e.err, &netErr) || errors.As(e.err, &netOpErr)  || errors.As(e.err, &netDnsErr) {
+			return true
+		}
+	}
+	return false
+}

httpclient/exec.go

@@ -0,0 +1,208 @@
+package httpclient
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"net"
+	"net/http"
+	"strings"
+	"time"
+)
+
+// -----------------------------------------------------------------------------
+
+const (
+	defaultTimeout = 20 * time.Second
+
+	errUnableToExecuteRequest = "failed to execute http request"
+	errNoAvailableServer      = "no available upstream server"
+)
+
+// -----------------------------------------------------------------------------
+
+func (c *HttpClient) exec(parentCtx context.Context, cb ExecCallback, req *Request) error {
+	var httpreq *http.Request
+	var getBody func() io.ReadCloser
+	var err error
+
+	// Define a body getter to return multiple copies of the reader to be used in retries.
+	if req.Body == nil {
+		// If no body, getter will return nil
+		getBody = func() io.ReadCloser {
+			return nil
+		}
+	} else {
+		// Convert to a ReadCloser if just a reader
+		rc, ok := req.Body.(io.ReadCloser)
+		if !ok {
+			rc = io.NopCloser(req.Body)
+		}
+
+		// Defer close of the original body
+		defer func() {
+			_ = rc.Close()
+		}()
+
+		// Set up a body reader cloning function
+		switch v := req.Body.(type) {
+		case *bytes.Buffer:
+			buf := v.Bytes()
+			getBody = func() io.ReadCloser {
+				r := bytes.NewReader(buf)
+				return io.NopCloser(r)
+			}
+
+		case *bytes.Reader:
+			snapshot := *v
+			getBody = func() io.ReadCloser {
+				r := snapshot
+				return io.NopCloser(&r)
+			}
+
+		case *strings.Reader:
+			snapshot := *v
+			getBody = func() io.ReadCloser {
+				r := snapshot
+				return io.NopCloser(&r)
+			}
+
+		default:
+			return errors.New("unsupported body reader")
+		}
+	}
+
+	// Initialize retry counter
+	retryCounter := 0
+
+	// Loop
+	for {
+		// Get next available server
+		srv := c.lb.Next()
+		if srv == nil {
+			return c.newError(nil, errNoAvailableServer, req.Resource, 0)
+		}
+
+		src := srv.UserData().(*Source)
+
+		// Create the final url
+		url := src.baseURL + req.Resource
+
+		// Create a new http request
+		httpreq, err = http.NewRequest(req.Method, url, getBody())
+		if err != nil {
+			err = c.newError(err, errUnableToExecuteRequest, url, 0)
+			src.setLastError(err)
+			return err
+		}
+
+		// Add load balancer source headers
+		httpreq.Header = src.header.Clone()
+
+		// Add request headers
+		for k, v := range req.Header {
+			vLen := len(v)
+			if vLen > 0 {
+				httpreq.Header.Set(k, v[0])
+				for vIdx := 1; vIdx < vLen; vIdx++ {
+					httpreq.Header.Add(k, v[vIdx])
+				}
+			}
+		}
+
+		// Create http client requester
+		client := http.Client{
+			Transport: c.transport,
+		}
+
+		// Establish a new context with a default timeout if a deadline is not present
+		ctx := parentCtx
+		var cancelCtx context.CancelFunc
+		if _, hasDeadline := parentCtx.Deadline(); !hasDeadline {
+			ctx, cancelCtx = context.WithTimeout(parentCtx, defaultTimeout)
+		}
+
+		// Build callback info
+		upstreamOffline := false
+		retry := false
+		execResult := Response{
+			fullUrl:         url,
+			source:          src,
+			retryCount:      retryCounter,
+			upstreamOffline: &upstreamOffline,
+			retry:           &retry,
+		}
+
+		// Execute real request
+		execResult.Response, err = client.Do(httpreq.WithContext(ctx))
+		if err != nil {
+			var netErr net.Error
+
+			if errors.Is(err, context.DeadlineExceeded) {
+				// Deadline exceeded?
+				if cancelCtx != nil {
+					cancelCtx() // To avoid defer calling inside a for loop and warnings, we call it here
+				}
+				return ErrTimeout
+			}
+			if errors.As(err, &netErr); netErr.Timeout() {
+				// Network timeout?
+				srv.SetOffline()
+
+				err = ErrTimeout
+			} else if errors.Is(err, context.Canceled) {
+				// Canceled?
+				if cancelCtx != nil {
+					cancelCtx() // To avoid defer calling inside a for loop and warnings, we call it here
+				}
+				return ErrCanceled
+			} else {
+				// Other type of error
+				srv.SetOffline()
+
+				err = c.newError(err, errUnableToExecuteRequest, url, 0)
+			}
+		}
+
+		// Set error in callback
+		execResult.err = err
+
+		// Call the callback
+		err = cb(parentCtx, execResult)
+
+		// To avoid defer calling inside a for loop and warnings, we call it here
+		if cancelCtx != nil {
+			cancelCtx()
+		}
+
+		// Close the response body if one exist
+		if execResult.Response != nil {
+			_ = execResult.Response.Body.Close()
+		}
+
+		// Set the last error (even success)
+		src.setLastError(err)
+
+		// Raise callbak
+		c.raiseRequestEvent(srv, err)
+
+		// Set server online/offline based on the callback response
+		if !upstreamOffline {
+			srv.SetOnline()
+		} else {
+			srv.SetOffline()
+		}
+
+		// Should we retry on next server?
+		if !retry {
+			break
+		}
+
+		// Increment retry counter
+		retryCounter += 1
+	}
+
+	// Done
+	return err
+}