11996898886

Committed 24 Nov 2024 02:18PM UTC coverage: 74.491% (+1.7%) from 72.763%

Build # 11996898886

Build Type

push

github

Committed by

asticode

Commit Message

Fixed windows github action

Run Details

2634 of 3536 relevant lines covered (74.49%)

2903.53 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

68.37

/http.go

package astikit

import (
        "bytes"
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "io"
        "net"
        "net/http"
        "os"
        "path/filepath"
        "strconv"
        "strings"
        "sync"
        "time"
)

var ErrHTTPSenderUnmarshaledError = errors.New("astikit: unmarshaled error")

// ServeHTTPOptions represents serve options
type ServeHTTPOptions struct {
        Addr    string
        Handler http.Handler
}

// ServeHTTP spawns an HTTP server
func ServeHTTP(w *Worker, o ServeHTTPOptions) {
        // Create server
        s := &http.Server{Addr: o.Addr, Handler: o.Handler}

        // Execute in a task
        w.NewTask().Do(func() {
                // Log
                w.Logger().Infof("astikit: serving on %s", o.Addr)

                // Serve
                var done = make(chan error)
                go func() {
                        if err := s.ListenAndServe(); err != nil {
                                done <- err
                        }
                }()

                // Wait for context or done to be done
                select {
                case <-w.ctx.Done():
                        if w.ctx.Err() != context.Canceled {
                                w.Logger().Error(fmt.Errorf("astikit: context error: %w", w.ctx.Err()))
                        }
                case err := <-done:
                        if err != nil {
                                w.Logger().Error(fmt.Errorf("astikit: serving failed: %w", err))
                        }
                }

                // Shutdown
                w.Logger().Infof("astikit: shutting down server on %s", o.Addr)
                if err := s.Shutdown(context.Background()); err != nil {
                        w.Logger().Error(fmt.Errorf("astikit: shutting down server on %s failed: %w", o.Addr, err))
                }
        })
}

// HTTPClient represents an HTTP client
type HTTPClient interface {
        Do(req *http.Request) (*http.Response, error)
}

// HTTPSender represents an object capable of sending http requests
type HTTPSender struct {
        client     HTTPClient
        l          SeverityLogger
        retryFunc  HTTPSenderRetryFunc
        retryMax   int
        retrySleep time.Duration
        timeout    time.Duration
}

// HTTPSenderRetryFunc is a function that decides whether to retry an HTTP request
type HTTPSenderRetryFunc func(resp *http.Response) error

// HTTPSenderOptions represents HTTPSender options
type HTTPSenderOptions struct {
        Client     HTTPClient
        Logger     StdLogger
        RetryFunc  HTTPSenderRetryFunc
        RetryMax   int
        RetrySleep time.Duration
        Timeout    time.Duration
}

// NewHTTPSender creates a new HTTP sender
func NewHTTPSender(o HTTPSenderOptions) (s *HTTPSender) {
        s = &HTTPSender{
                client:     o.Client,
                l:          AdaptStdLogger(o.Logger),
                retryFunc:  o.RetryFunc,
                retryMax:   o.RetryMax,
                retrySleep: o.RetrySleep,
                timeout:    o.Timeout,
        }
        if s.client == nil {
                s.client = &http.Client{}
        }
        if s.retryFunc == nil {
                s.retryFunc = s.defaultHTTPRetryFunc
        }
        return
}

func (s *HTTPSender) defaultHTTPRetryFunc(resp *http.Response) error {
        if resp.StatusCode >= http.StatusInternalServerError {
                return fmt.Errorf("astikit: invalid status code %d", resp.StatusCode)
        }
        return nil
}

// Send sends a new *http.Request
func (s *HTTPSender) Send(req *http.Request) (*http.Response, error) {
        return s.SendWithTimeout(req, s.timeout)
}

// SendWithTimeout sends a new *http.Request with a timeout
func (s *HTTPSender) SendWithTimeout(req *http.Request, timeout time.Duration) (*http.Response, error) {
        // Timeout
        if timeout > 0 {
                // Create context
                ctx, cancel := context.WithTimeout(req.Context(), timeout)
                defer cancel()

                // Update request
                req = req.WithContext(ctx)
        }

        // Send
        return s.send(req, timeout)
}

func (s *HTTPSender) send(req *http.Request, timeout time.Duration) (resp *http.Response, err error) {
        // Set name
        name := req.Method + " request"
        if req.URL != nil {
                name += " to " + req.URL.String()
        }

        // Timeout
        if timeout > 0 {
                name += " with timeout " + timeout.String()
        }

        // Loop
        // We start at retryMax + 1 so that it runs at least once even if retryMax == 0
        tries := 0
        for retriesLeft := s.retryMax + 1; retriesLeft > 0; retriesLeft-- {
                // Get request name
                nr := name + " (" + strconv.Itoa(s.retryMax-retriesLeft+2) + "/" + strconv.Itoa(s.retryMax+1) + ")"
                tries++

                // Send request
                s.l.Debugf("astikit: sending %s", nr)
                if resp, err = s.client.Do(req); err != nil {
                        // Retry if error is temporary, stop here otherwise
                        if netError, ok := err.(net.Error); !ok || !netError.Timeout() {
                                err = fmt.Errorf("astikit: sending %s failed: %w", nr, err)
                                return
                        }
                } else if err = req.Context().Err(); err != nil {
                        err = fmt.Errorf("astikit: request context failed: %w", err)
                        return
                } else {
                        err = s.retryFunc(resp)
                }

                // Retry
                if err != nil {
                        if retriesLeft > 1 {
                                s.l.Errorf("astikit: sending %s failed, sleeping %s and retrying... (%d retries left): %w", nr, s.retrySleep, retriesLeft-1, err)
                                time.Sleep(s.retrySleep)
                        }
                        continue
                }

                // Return if conditions for retrying were not met
                return
        }

        // Max retries limit reached
        err = fmt.Errorf("astikit: sending %s failed after %d tries: %w", name, tries, err)
        return
}

type HTTPSenderHeaderFunc func(h http.Header)

type HTTPSenderStatusCodeFunc func(code int) error

// HTTPSendJSONOptions represents SendJSON options
type HTTPSendJSONOptions struct {
        BodyError      interface{}
        BodyIn         interface{}
        BodyOut        interface{}
        HeadersIn      map[string]string
        HeadersOut     HTTPSenderHeaderFunc
        Host           string
        Method         string
        StatusCodeFunc HTTPSenderStatusCodeFunc
        Timeout        time.Duration
        URL            string
}

// SendJSON sends a new JSON HTTP request
func (s *HTTPSender) SendJSON(o HTTPSendJSONOptions) (err error) {
        // Marshal body in
        var bi io.Reader
        if o.BodyIn != nil {
                bb := &bytes.Buffer{}
                if err = json.NewEncoder(bb).Encode(o.BodyIn); err != nil {
                        err = fmt.Errorf("astikit: marshaling body in failed: %w", err)
                        return
                }
                bi = bb
        }

        // Create request
        var req *http.Request
        if req, err = http.NewRequest(o.Method, o.URL, bi); err != nil {
                err = fmt.Errorf("astikit: creating request failed: %w", err)
                return
        }

        // Update host
        if o.Host != "" {
                req.Host = o.Host
        }

        // Add headers
        for k, v := range o.HeadersIn {
                req.Header.Set(k, v)
        }

        // Get timeout
        timeout := s.timeout
        if o.Timeout > 0 {
                timeout = o.Timeout
        }

        // Handle timeout outside the .send() method otherwise context may be cancelled before
        // reading all response body
        if timeout > 0 {
                // Create context
                ctx, cancel := context.WithTimeout(req.Context(), timeout)
                defer cancel()

                // Update request
                req = req.WithContext(ctx)
        }

        // Send request
        var resp *http.Response
        if resp, err = s.send(req, timeout); err != nil {
                err = fmt.Errorf("astikit: sending request failed: %w", err)
                return
        }
        defer resp.Body.Close()

        // Process headers
        if o.HeadersOut != nil {
                o.HeadersOut(resp.Header)
        }

        // Process status code
        fn := HTTPSenderDefaultStatusCodeFunc
        if o.StatusCodeFunc != nil {
                fn = o.StatusCodeFunc
        }
        if err = fn(resp.StatusCode); err != nil {
                // Try unmarshaling error
                if o.BodyError != nil {
                        if err2 := json.NewDecoder(resp.Body).Decode(o.BodyError); err2 == nil {
                                err = ErrHTTPSenderUnmarshaledError
                                return
                        }
                }

                // Default error
                err = fmt.Errorf("astikit: validating status code %d failed: %w", resp.StatusCode, err)
                return
        }

        // Unmarshal body out
        if o.BodyOut != nil {
                // Read all
                var b []byte
                if b, err = io.ReadAll(resp.Body); err != nil {
                        err = fmt.Errorf("astikit: reading all failed: %w", err)
                        return
                }

                // Unmarshal
                if err = json.Unmarshal(b, o.BodyOut); err != nil {
                        err = fmt.Errorf("astikit: unmarshaling failed: %w (json: %s)", err, b)
                        return
                }
        }
        return
}

func HTTPSenderDefaultStatusCodeFunc(code int) error {
        if code < 200 || code > 299 {
                return errors.New("astikit: status code should be between 200 and 299")
        }
        return nil
}

// HTTPResponseFunc is a func that can process an $http.Response
type HTTPResponseFunc func(resp *http.Response) error

func defaultHTTPResponseFunc(resp *http.Response) (err error) {
        if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
                err = fmt.Errorf("astikit: invalid status code %d", resp.StatusCode)
                return
        }
        return
}

// HTTPDownloader represents an object capable of downloading several HTTP srcs simultaneously
// and doing stuff to the results
type HTTPDownloader struct {
        bp           *BufferPool
        l            *GoroutineLimiter
        responseFunc HTTPResponseFunc
        s            *HTTPSender
}

// HTTPDownloaderOptions represents HTTPDownloader options
type HTTPDownloaderOptions struct {
        Limiter      GoroutineLimiterOptions
        ResponseFunc HTTPResponseFunc
        Sender       HTTPSenderOptions
}

// NewHTTPDownloader creates a new HTTPDownloader
func NewHTTPDownloader(o HTTPDownloaderOptions) (d *HTTPDownloader) {
        d = &HTTPDownloader{
                bp:           NewBufferPool(),
                l:            NewGoroutineLimiter(o.Limiter),
                responseFunc: o.ResponseFunc,
                s:            NewHTTPSender(o.Sender),
        }
        if d.responseFunc == nil {
                d.responseFunc = defaultHTTPResponseFunc
        }
        return
}

// Close closes the downloader properly
func (d *HTTPDownloader) Close() error {
        return d.l.Close()
}

type HTTPDownloaderSrc struct {
        Body   io.Reader
        Header http.Header
        Method string
        URL    string
}

// It is the responsibility of the caller to call i.Close()
type httpDownloaderFunc func(ctx context.Context, idx int, i *BufferPoolItem) error

func (d *HTTPDownloader) do(ctx context.Context, fn httpDownloaderFunc, idx int, src HTTPDownloaderSrc) (err error) {
        // Defaults
        if src.Method == "" {
                src.Method = http.MethodGet
        }

        // Create request
        var r *http.Request
        if r, err = http.NewRequestWithContext(ctx, src.Method, src.URL, src.Body); err != nil {
                err = fmt.Errorf("astikit: creating request to %s failed: %w", src.URL, err)
                return
        }

        // Copy header
        for k := range src.Header {
                r.Header.Set(k, src.Header.Get(k))
        }

        // Send request
        var resp *http.Response
        if resp, err = d.s.Send(r); err != nil {
                err = fmt.Errorf("astikit: sending request to %s failed: %w", src.URL, err)
                return
        }
        defer resp.Body.Close()

        // Create buffer pool item
        buf := d.bp.New()

        // Process response
        if err = d.responseFunc(resp); err != nil {
                err = fmt.Errorf("astikit: response for request to %s is invalid: %w", src.URL, err)
                return
        }

        // Copy body
        if _, err = Copy(ctx, buf, resp.Body); err != nil {
                err = fmt.Errorf("astikit: copying body of %s failed: %w", src.URL, err)
                return
        }

        // Custom
        if err = fn(ctx, idx, buf); err != nil {
                err = fmt.Errorf("astikit: custom callback on %s failed: %w", src.URL, err)
                return
        }
        return
}

func (d *HTTPDownloader) download(ctx context.Context, srcs []HTTPDownloaderSrc, fn httpDownloaderFunc) (err error) {
        // Nothing to download
        if len(srcs) == 0 {
                return nil
        }

        // Loop through srcs
        wg := &sync.WaitGroup{}
        wg.Add(len(srcs))
        for idx, src := range srcs {
                func(idx int, src HTTPDownloaderSrc) {
                        // Update error with ctx
                        if ctx.Err() != nil {
                                err = ctx.Err()
                        }

                        // Do nothing if error
                        if err != nil {
                                wg.Done()
                                return
                        }

                        // Do
                        //nolint:errcheck
                        d.l.Do(func() {
                                // Task is done
                                defer wg.Done()

                                // Do
                                if errD := d.do(ctx, fn, idx, src); errD != nil && err == nil {
                                        err = errD
                                        return
                                }
                        })
                }(idx, src)
        }

        // Wait
        wg.Wait()
        return
}

// DownloadInDirectory downloads in parallel a set of srcs and saves them in a dst directory
func (d *HTTPDownloader) DownloadInDirectory(ctx context.Context, dst string, srcs ...HTTPDownloaderSrc) error {
        return d.download(ctx, srcs, func(ctx context.Context, idx int, buf *BufferPoolItem) (err error) {
                // Make sure to close buffer
                defer buf.Close()

                // Make sure destination directory exists
                if err = os.MkdirAll(dst, DefaultDirMode); err != nil {
                        err = fmt.Errorf("astikit: mkdirall %s failed: %w", dst, err)
                        return
                }

                // Create destination file
                var f *os.File
                dst := filepath.Join(dst, filepath.Base(srcs[idx].URL))
                if f, err = os.Create(dst); err != nil {
                        err = fmt.Errorf("astikit: creating %s failed: %w", dst, err)
                        return
                }
                defer f.Close()

                // Copy buffer
                if _, err = Copy(ctx, f, buf); err != nil {
                        err = fmt.Errorf("astikit: copying content to %s failed: %w", dst, err)
                        return
                }
                return
        })
}

// DownloadInWriter downloads in parallel a set of srcs and concatenates them in a writer while
// maintaining the initial order
func (d *HTTPDownloader) DownloadInWriter(ctx context.Context, dst io.Writer, srcs ...HTTPDownloaderSrc) error {
        // Init
        type chunk struct {
                buf *BufferPoolItem
                idx int
        }
        var cs []chunk
        var m sync.Mutex // Locks cs
        var requiredIdx int

        // Make sure to close all buffers
        defer func() {
                for _, c := range cs {
                        c.buf.Close()
                }
        }()

        // Download
        return d.download(ctx, srcs, func(ctx context.Context, idx int, buf *BufferPoolItem) (err error) {
                // Lock
                m.Lock()
                defer m.Unlock()

                // Check where to insert chunk
                var idxInsert = -1
                for idxChunk := 0; idxChunk < len(cs); idxChunk++ {
                        if idx < cs[idxChunk].idx {
                                idxInsert = idxChunk
                                break
                        }
                }

                // Create chunk
                c := chunk{
                        buf: buf,
                        idx: idx,
                }

                // Add chunk
                if idxInsert > -1 {
                        cs = append(cs[:idxInsert], append([]chunk{c}, cs[idxInsert:]...)...)
                } else {
                        cs = append(cs, c)
                }

                // Loop through chunks
                for idxChunk := 0; idxChunk < len(cs); idxChunk++ {
                        // Get chunk
                        c := cs[idxChunk]

                        // The chunk should be copied
                        if c.idx == requiredIdx {
                                // Copy chunk content
                                // Do not check error right away since we still want to close the buffer
                                // and remove the chunk
                                _, err = Copy(ctx, dst, c.buf)

                                // Close buffer
                                c.buf.Close()

                                // Remove chunk
                                requiredIdx++
                                cs = append(cs[:idxChunk], cs[idxChunk+1:]...)
                                idxChunk--

                                // Check error
                                if err != nil {
                                        err = fmt.Errorf("astikit: copying chunk #%d to dst failed: %w", c.idx, err)
                                        return
                                }
                        }
                }
                return
        })
}

// DownloadInFile downloads in parallel a set of srcs and concatenates them in a dst file while
// maintaining the initial order
func (d *HTTPDownloader) DownloadInFile(ctx context.Context, dst string, srcs ...HTTPDownloaderSrc) (err error) {
        // Make sure destination directory exists
        if err = os.MkdirAll(filepath.Dir(dst), DefaultDirMode); err != nil {
                err = fmt.Errorf("astikit: mkdirall %s failed: %w", filepath.Dir(dst), err)
                return
        }

        // Create destination file
        var f *os.File
        if f, err = os.Create(dst); err != nil {
                err = fmt.Errorf("astikit: creating %s failed: %w", dst, err)
                return
        }
        defer f.Close()

        // Download in writer
        return d.DownloadInWriter(ctx, f, srcs...)
}

// HTTPMiddleware represents an HTTP middleware
type HTTPMiddleware func(http.Handler) http.Handler

// ChainHTTPMiddlewares chains HTTP middlewares
func ChainHTTPMiddlewares(h http.Handler, ms ...HTTPMiddleware) http.Handler {
        return ChainHTTPMiddlewaresWithPrefix(h, []string{}, ms...)
}

// ChainHTTPMiddlewaresWithPrefix chains HTTP middlewares if one of prefixes is present
func ChainHTTPMiddlewaresWithPrefix(h http.Handler, prefixes []string, ms ...HTTPMiddleware) http.Handler {
        for _, m := range ms {
                if m == nil {
                        continue
                }
                if len(prefixes) == 0 {
                        h = m(h)
                } else {
                        t := h
                        h = http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
                                for _, prefix := range prefixes {
                                        if strings.HasPrefix(r.URL.EscapedPath(), prefix) {
                                                m(t).ServeHTTP(rw, r)
                                                return
                                        }
                                }
                                t.ServeHTTP(rw, r)
                        })
                }
        }
        return h
}

func handleHTTPBasicAuth(username, password string, rw http.ResponseWriter, r *http.Request) bool {
        if u, p, ok := r.BasicAuth(); !ok || u != username || p != password {
                rw.Header().Set("WWW-Authenticate", "Basic Realm=Please enter your credentials")
                rw.WriteHeader(http.StatusUnauthorized)
                return true
        }
        return false
}

// HTTPMiddlewareBasicAuth adds basic HTTP auth to an HTTP handler
func HTTPMiddlewareBasicAuth(username, password string) HTTPMiddleware {
        if username == "" && password == "" {
                return nil
        }
        return func(h http.Handler) http.Handler {
                return http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
                        // Handle basic auth
                        if handleHTTPBasicAuth(username, password, rw, r) {
                                return
                        }

                        // Next handler
                        h.ServeHTTP(rw, r)
                })
        }
}

func setHTTPContentType(contentType string, rw http.ResponseWriter) {
        rw.Header().Set("Content-Type", contentType)
}

// HTTPMiddlewareContentType adds a content type to an HTTP handler
func HTTPMiddlewareContentType(contentType string) HTTPMiddleware {
        return func(h http.Handler) http.Handler {
                return http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
                        // Set content type
                        setHTTPContentType(contentType, rw)

                        // Next handler
                        h.ServeHTTP(rw, r)
                })
        }
}

func setHTTPHeaders(vs map[string]string, rw http.ResponseWriter) {
        for k, v := range vs {
                rw.Header().Set(k, v)
        }
}

// HTTPMiddlewareHeaders adds headers to an HTTP handler
func HTTPMiddlewareHeaders(vs map[string]string) HTTPMiddleware {
        return func(h http.Handler) http.Handler {
                return http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
                        // Set headers
                        setHTTPHeaders(vs, rw)

                        // Next handler
                        h.ServeHTTP(rw, r)
                })
        }
}

// HTTPMiddlewareCORSHeaders adds CORS headers to an HTTP handler
func HTTPMiddlewareCORSHeaders() HTTPMiddleware {
        return HTTPMiddlewareHeaders(map[string]string{
                "Access-Control-Allow-Headers": "*",
                "Access-Control-Allow-Methods": "*",
                "Access-Control-Allow-Origin":  "*",
        })
}

1	package astikit
2
3	import (
4	"bytes"
5	"context"
6	"encoding/json"
7	"errors"
8	"fmt"
9	"io"
10	"net"
11	"net/http"
12	"os"
13	"path/filepath"
14	"strconv"
15	"strings"
16	"sync"
17	"time"
18	)
19
20	var ErrHTTPSenderUnmarshaledError = errors.New("astikit: unmarshaled error")
21
22	// ServeHTTPOptions represents serve options
23	type ServeHTTPOptions struct {
24	Addr string
25	Handler http.Handler
26	}
27
28	// ServeHTTP spawns an HTTP server
29	func ServeHTTP(w *Worker, o ServeHTTPOptions) {	2✔
30	// Create server	2✔
31	s := &http.Server{Addr: o.Addr, Handler: o.Handler}	2✔
32		2✔
33	// Execute in a task	2✔
34	w.NewTask().Do(func() {	4✔
35	// Log	2✔
36	w.Logger().Infof("astikit: serving on %s", o.Addr)	2✔
37		2✔
38	// Serve	2✔
39	var done = make(chan error)	2✔
40	go func() {	4✔
41	if err := s.ListenAndServe(); err != nil {	4✔
42	done <- err	2✔
43	}	2✔
44	}()
45
46	// Wait for context or done to be done
47	select {	2✔
48	case <-w.ctx.Done():	2✔
49	if w.ctx.Err() != context.Canceled {	2✔
50	w.Logger().Error(fmt.Errorf("astikit: context error: %w", w.ctx.Err()))	×
51	}	×
52	case err := <-done:	×
53	if err != nil {	×
54	w.Logger().Error(fmt.Errorf("astikit: serving failed: %w", err))	×
55	}	×
56	}
57
58	// Shutdown
59	w.Logger().Infof("astikit: shutting down server on %s", o.Addr)	2✔
60	if err := s.Shutdown(context.Background()); err != nil {	2✔
61	w.Logger().Error(fmt.Errorf("astikit: shutting down server on %s failed: %w", o.Addr, err))	×
62	}	×
63	})
64	}
65
66	// HTTPClient represents an HTTP client
67	type HTTPClient interface {
68	Do(req http.Request) (http.Response, error)
69	}
70
71	// HTTPSender represents an object capable of sending http requests
72	type HTTPSender struct {
73	client HTTPClient
74	l SeverityLogger
75	retryFunc HTTPSenderRetryFunc
76	retryMax int
77	retrySleep time.Duration
78	timeout time.Duration
79	}
80
81	// HTTPSenderRetryFunc is a function that decides whether to retry an HTTP request
82	type HTTPSenderRetryFunc func(resp *http.Response) error
83
84	// HTTPSenderOptions represents HTTPSender options
85	type HTTPSenderOptions struct {
86	Client HTTPClient
87	Logger StdLogger
88	RetryFunc HTTPSenderRetryFunc
89	RetryMax int
90	RetrySleep time.Duration
91	Timeout time.Duration
92	}
93
94	// NewHTTPSender creates a new HTTP sender
95	func NewHTTPSender(o HTTPSenderOptions) (s *HTTPSender) {	10✔
96	s = &HTTPSender{	10✔
97	client: o.Client,	10✔
98	l: AdaptStdLogger(o.Logger),	10✔
99	retryFunc: o.RetryFunc,	10✔
100	retryMax: o.RetryMax,	10✔
101	retrySleep: o.RetrySleep,	10✔
102	timeout: o.Timeout,	10✔
103	}	10✔
104	if s.client == nil {	10✔
105	s.client = &http.Client{}	×
106	}	×
107	if s.retryFunc == nil {	20✔
108	s.retryFunc = s.defaultHTTPRetryFunc	10✔
109	}	10✔
110	return	10✔
111	}
112
113	func (s HTTPSender) defaultHTTPRetryFunc(resp http.Response) error {	38✔
114	if resp.StatusCode >= http.StatusInternalServerError {	48✔
115	return fmt.Errorf("astikit: invalid status code %d", resp.StatusCode)	10✔
116	}	10✔
117	return nil	28✔
118	}
119
120	// Send sends a new *http.Request
121	func (s HTTPSender) Send(req http.Request) (*http.Response, error) {	22✔
122	return s.SendWithTimeout(req, s.timeout)	22✔
123	}	22✔
124
125	// SendWithTimeout sends a new *http.Request with a timeout
126	func (s HTTPSender) SendWithTimeout(req http.Request, timeout time.Duration) (*http.Response, error) {	24✔
127	// Timeout	24✔
128	if timeout > 0 {	26✔
129	// Create context	2✔
130	ctx, cancel := context.WithTimeout(req.Context(), timeout)	2✔
131	defer cancel()	2✔
132		2✔
133	// Update request	2✔
134	req = req.WithContext(ctx)	2✔
135	}	2✔
136
137	// Send
138	return s.send(req, timeout)	24✔
139	}
140
141	func (s HTTPSender) send(req http.Request, timeout time.Duration) (resp *http.Response, err error) {	34✔
142	// Set name	34✔
143	name := req.Method + " request"	34✔
144	if req.URL != nil {	62✔
145	name += " to " + req.URL.String()	28✔
146	}	28✔
147
148	// Timeout
149	if timeout > 0 {	40✔
150	name += " with timeout " + timeout.String()	6✔
151	}	6✔
152
153	// Loop
154	// We start at retryMax + 1 so that it runs at least once even if retryMax == 0
155	tries := 0	34✔
156	for retriesLeft := s.retryMax + 1; retriesLeft > 0; retriesLeft-- {	78✔
157	// Get request name	44✔
158	nr := name + " (" + strconv.Itoa(s.retryMax-retriesLeft+2) + "/" + strconv.Itoa(s.retryMax+1) + ")"	44✔
159	tries++	44✔
160		44✔
161	// Send request	44✔
162	s.l.Debugf("astikit: sending %s", nr)	44✔
163	if resp, err = s.client.Do(req); err != nil {	46✔
164	// Retry if error is temporary, stop here otherwise	2✔
165	if netError, ok := err.(net.Error); !ok \|\| !netError.Timeout() {	2✔
166	err = fmt.Errorf("astikit: sending %s failed: %w", nr, err)	×
167	return	×
168	}	×
169	} else if err = req.Context().Err(); err != nil {	46✔
170	err = fmt.Errorf("astikit: request context failed: %w", err)	4✔
171	return	4✔
172	} else {	42✔
173	err = s.retryFunc(resp)	38✔
174	}	38✔
175
176	// Retry
177	if err != nil {	52✔
178	if retriesLeft > 1 {	22✔
179	s.l.Errorf("astikit: sending %s failed, sleeping %s and retrying... (%d retries left): %w", nr, s.retrySleep, retriesLeft-1, err)	10✔
180	time.Sleep(s.retrySleep)	10✔
181	}	10✔
182	continue	12✔
183	}
184
185	// Return if conditions for retrying were not met
186	return	28✔
187	}
188
189	// Max retries limit reached
190	err = fmt.Errorf("astikit: sending %s failed after %d tries: %w", name, tries, err)	2✔
191	return	2✔
192	}
193
194	type HTTPSenderHeaderFunc func(h http.Header)
195
196	type HTTPSenderStatusCodeFunc func(code int) error
197
198	// HTTPSendJSONOptions represents SendJSON options
199	type HTTPSendJSONOptions struct {
200	BodyError interface{}
201	BodyIn interface{}
202	BodyOut interface{}
203	HeadersIn map[string]string
204	HeadersOut HTTPSenderHeaderFunc
205	Host string
206	Method string
207	StatusCodeFunc HTTPSenderStatusCodeFunc
208	Timeout time.Duration
209	URL string
210	}
211
212	// SendJSON sends a new JSON HTTP request
213	func (s *HTTPSender) SendJSON(o HTTPSendJSONOptions) (err error) {	10✔
214	// Marshal body in	10✔
215	var bi io.Reader	10✔
216	if o.BodyIn != nil {	12✔
217	bb := &bytes.Buffer{}	2✔
218	if err = json.NewEncoder(bb).Encode(o.BodyIn); err != nil {	2✔
219	err = fmt.Errorf("astikit: marshaling body in failed: %w", err)	×
220	return	×
221	}	×
222	bi = bb	2✔
223	}
224
225	// Create request
226	var req *http.Request	10✔
227	if req, err = http.NewRequest(o.Method, o.URL, bi); err != nil {	10✔
228	err = fmt.Errorf("astikit: creating request failed: %w", err)	×
229	return	×
230	}	×
231
232	// Update host
233	if o.Host != "" {	10✔
234	req.Host = o.Host	×
235	}	×
236
237	// Add headers
238	for k, v := range o.HeadersIn {	14✔
239	req.Header.Set(k, v)	4✔
240	}	4✔
241
242	// Get timeout
243	timeout := s.timeout	10✔
244	if o.Timeout > 0 {	14✔
245	timeout = o.Timeout	4✔
246	}	4✔
247
248	// Handle timeout outside the .send() method otherwise context may be cancelled before
249	// reading all response body
250	if timeout > 0 {	14✔
251	// Create context	4✔
252	ctx, cancel := context.WithTimeout(req.Context(), timeout)	4✔
253	defer cancel()	4✔
254		4✔
255	// Update request	4✔
256	req = req.WithContext(ctx)	4✔
257	}	4✔
258
259	// Send request
260	var resp *http.Response	10✔
261	if resp, err = s.send(req, timeout); err != nil {	12✔
262	err = fmt.Errorf("astikit: sending request failed: %w", err)	2✔
263	return	2✔
264	}	2✔
265	defer resp.Body.Close()	8✔
266		8✔
267	// Process headers	8✔
268	if o.HeadersOut != nil {	10✔
269	o.HeadersOut(resp.Header)	2✔
270	}	2✔
271
272	// Process status code
273	fn := HTTPSenderDefaultStatusCodeFunc	8✔
274	if o.StatusCodeFunc != nil {	10✔
275	fn = o.StatusCodeFunc	2✔
276	}	2✔
277	if err = fn(resp.StatusCode); err != nil {	12✔
278	// Try unmarshaling error	4✔
279	if o.BodyError != nil {	6✔
280	if err2 := json.NewDecoder(resp.Body).Decode(o.BodyError); err2 == nil {	4✔
281	err = ErrHTTPSenderUnmarshaledError	2✔
282	return	2✔
283	}	2✔
284	}
285
286	// Default error
287	err = fmt.Errorf("astikit: validating status code %d failed: %w", resp.StatusCode, err)	2✔
288	return	2✔
289	}
290
291	// Unmarshal body out
292	if o.BodyOut != nil {	8✔
293	// Read all	4✔
294	var b []byte	4✔
295	if b, err = io.ReadAll(resp.Body); err != nil {	4✔
296	err = fmt.Errorf("astikit: reading all failed: %w", err)	×
297	return	×
298	}	×
299
300	// Unmarshal
301	if err = json.Unmarshal(b, o.BodyOut); err != nil {	4✔
302	err = fmt.Errorf("astikit: unmarshaling failed: %w (json: %s)", err, b)	×
303	return	×
304	}	×
305	}
306	return	4✔
307	}
308
309	func HTTPSenderDefaultStatusCodeFunc(code int) error {	6✔
310	if code < 200 \|\| code > 299 {	8✔
311	return errors.New("astikit: status code should be between 200 and 299")	2✔
312	}	2✔
313	return nil	4✔
314	}
315
316	// HTTPResponseFunc is a func that can process an $http.Response
317	type HTTPResponseFunc func(resp *http.Response) error
318
319	func defaultHTTPResponseFunc(resp *http.Response) (err error) {	18✔
320	if resp.StatusCode < http.StatusOK \|\| resp.StatusCode >= http.StatusMultipleChoices {	18✔
321	err = fmt.Errorf("astikit: invalid status code %d", resp.StatusCode)	×
322	return	×
323	}	×
324	return	18✔
325	}
326
327	// HTTPDownloader represents an object capable of downloading several HTTP srcs simultaneously
328	// and doing stuff to the results
329	type HTTPDownloader struct {
330	bp *BufferPool
331	l *GoroutineLimiter
332	responseFunc HTTPResponseFunc
333	s *HTTPSender
334	}
335
336	// HTTPDownloaderOptions represents HTTPDownloader options
337	type HTTPDownloaderOptions struct {
338	Limiter GoroutineLimiterOptions
339	ResponseFunc HTTPResponseFunc
340	Sender HTTPSenderOptions
341	}
342
343	// NewHTTPDownloader creates a new HTTPDownloader
344	func NewHTTPDownloader(o HTTPDownloaderOptions) (d *HTTPDownloader) {	2✔
345	d = &HTTPDownloader{	2✔
346	bp: NewBufferPool(),	2✔
347	l: NewGoroutineLimiter(o.Limiter),	2✔
348	responseFunc: o.ResponseFunc,	2✔
349	s: NewHTTPSender(o.Sender),	2✔
350	}	2✔
351	if d.responseFunc == nil {	4✔
352	d.responseFunc = defaultHTTPResponseFunc	2✔
353	}	2✔
354	return	2✔
355	}
356
357	// Close closes the downloader properly
358	func (d *HTTPDownloader) Close() error {	2✔
359	return d.l.Close()	2✔
360	}	2✔
361
362	type HTTPDownloaderSrc struct {
363	Body io.Reader
364	Header http.Header
365	Method string
366	URL string
367	}
368
369	// It is the responsibility of the caller to call i.Close()
370	type httpDownloaderFunc func(ctx context.Context, idx int, i *BufferPoolItem) error
371
372	func (d *HTTPDownloader) do(ctx context.Context, fn httpDownloaderFunc, idx int, src HTTPDownloaderSrc) (err error) {	18✔
373	// Defaults	18✔
374	if src.Method == "" {	36✔
375	src.Method = http.MethodGet	18✔
376	}	18✔
377
378	// Create request
379	var r *http.Request	18✔
380	if r, err = http.NewRequestWithContext(ctx, src.Method, src.URL, src.Body); err != nil {	18✔
381	err = fmt.Errorf("astikit: creating request to %s failed: %w", src.URL, err)	×
382	return	×
383	}	×
384
385	// Copy header
386	for k := range src.Header {	18✔
387	r.Header.Set(k, src.Header.Get(k))	×
388	}	×
389
390	// Send request
391	var resp *http.Response	18✔
392	if resp, err = d.s.Send(r); err != nil {	18✔
393	err = fmt.Errorf("astikit: sending request to %s failed: %w", src.URL, err)	×
394	return	×
395	}	×
396	defer resp.Body.Close()	18✔
397		18✔
398	// Create buffer pool item	18✔
399	buf := d.bp.New()	18✔
400		18✔
401	// Process response	18✔
402	if err = d.responseFunc(resp); err != nil {	18✔
403	err = fmt.Errorf("astikit: response for request to %s is invalid: %w", src.URL, err)	×
404	return	×
405	}	×
406
407	// Copy body
408	if _, err = Copy(ctx, buf, resp.Body); err != nil {	18✔
409	err = fmt.Errorf("astikit: copying body of %s failed: %w", src.URL, err)	×
410	return	×
411	}	×
412
413	// Custom
414	if err = fn(ctx, idx, buf); err != nil {	18✔
415	err = fmt.Errorf("astikit: custom callback on %s failed: %w", src.URL, err)	×
416	return	×
417	}	×
418	return	18✔
419	}
420
421	func (d *HTTPDownloader) download(ctx context.Context, srcs []HTTPDownloaderSrc, fn httpDownloaderFunc) (err error) {	6✔
422	// Nothing to download	6✔
423	if len(srcs) == 0 {	6✔
424	return nil	×
425	}	×
426
427	// Loop through srcs
428	wg := &sync.WaitGroup{}	6✔
429	wg.Add(len(srcs))	6✔
430	for idx, src := range srcs {	24✔
431	func(idx int, src HTTPDownloaderSrc) {	36✔
432	// Update error with ctx	18✔
433	if ctx.Err() != nil {	18✔
434	err = ctx.Err()	×
435	}	×
436
437	// Do nothing if error
438	if err != nil {	18✔
439	wg.Done()	×
440	return	×
441	}	×
442
443	// Do
444	//nolint:errcheck
445	d.l.Do(func() {	36✔
446	// Task is done	18✔
447	defer wg.Done()	18✔
448		18✔
449	// Do	18✔
450	if errD := d.do(ctx, fn, idx, src); errD != nil && err == nil {	18✔
451	err = errD	×
452	return	×
453	}	×
454	})
455	}(idx, src)
456	}
457
458	// Wait
459	wg.Wait()	6✔
460	return	6✔
461	}
462
463	// DownloadInDirectory downloads in parallel a set of srcs and saves them in a dst directory
464	func (d *HTTPDownloader) DownloadInDirectory(ctx context.Context, dst string, srcs ...HTTPDownloaderSrc) error {	2✔
465	return d.download(ctx, srcs, func(ctx context.Context, idx int, buf *BufferPoolItem) (err error) {	8✔
466	// Make sure to close buffer	6✔
467	defer buf.Close()	6✔
468		6✔
469	// Make sure destination directory exists	6✔
470	if err = os.MkdirAll(dst, DefaultDirMode); err != nil {	6✔
471	err = fmt.Errorf("astikit: mkdirall %s failed: %w", dst, err)	×
472	return	×
473	}	×
474
475	// Create destination file
476	var f *os.File	6✔
477	dst := filepath.Join(dst, filepath.Base(srcs[idx].URL))	6✔
478	if f, err = os.Create(dst); err != nil {	6✔
479	err = fmt.Errorf("astikit: creating %s failed: %w", dst, err)	×
480	return	×
481	}	×
482	defer f.Close()	6✔
483		6✔
484	// Copy buffer	6✔
485	if _, err = Copy(ctx, f, buf); err != nil {	6✔
486	err = fmt.Errorf("astikit: copying content to %s failed: %w", dst, err)	×
487	return	×
488	}	×
489	return	6✔
490	})
491	}
492
493	// DownloadInWriter downloads in parallel a set of srcs and concatenates them in a writer while
494	// maintaining the initial order
495	func (d *HTTPDownloader) DownloadInWriter(ctx context.Context, dst io.Writer, srcs ...HTTPDownloaderSrc) error {	4✔
496	// Init	4✔
497	type chunk struct {	4✔
498	buf *BufferPoolItem	4✔
499	idx int	4✔
500	}	4✔
501	var cs []chunk	4✔
502	var m sync.Mutex // Locks cs	4✔
503	var requiredIdx int	4✔
504		4✔
505	// Make sure to close all buffers	4✔
506	defer func() {	8✔
507	for _, c := range cs {	4✔
508	c.buf.Close()	×
509	}	×
510	}()
511
512	// Download
513	return d.download(ctx, srcs, func(ctx context.Context, idx int, buf *BufferPoolItem) (err error) {	16✔
514	// Lock	12✔
515	m.Lock()	12✔
516	defer m.Unlock()	12✔
517		12✔
518	// Check where to insert chunk	12✔
519	var idxInsert = -1	12✔
520	for idxChunk := 0; idxChunk < len(cs); idxChunk++ {	16✔
521	if idx < cs[idxChunk].idx {	8✔
522	idxInsert = idxChunk	4✔
523	break	4✔
524	}
525	}
526
527	// Create chunk
528	c := chunk{	12✔
529	buf: buf,	12✔
530	idx: idx,	12✔
531	}	12✔
532		12✔
533	// Add chunk	12✔
534	if idxInsert > -1 {	16✔
535	cs = append(cs[:idxInsert], append([]chunk{c}, cs[idxInsert:]...)...)	4✔
536	} else {	12✔
537	cs = append(cs, c)	8✔
538	}	8✔
539
540	// Loop through chunks
541	for idxChunk := 0; idxChunk < len(cs); idxChunk++ {	28✔
542	// Get chunk	16✔
543	c := cs[idxChunk]	16✔
544		16✔
545	// The chunk should be copied	16✔
546	if c.idx == requiredIdx {	28✔
547	// Copy chunk content	12✔
548	// Do not check error right away since we still want to close the buffer	12✔
549	// and remove the chunk	12✔
550	_, err = Copy(ctx, dst, c.buf)	12✔
551		12✔
552	// Close buffer	12✔
553	c.buf.Close()	12✔
554		12✔
555	// Remove chunk	12✔
556	requiredIdx++	12✔
557	cs = append(cs[:idxChunk], cs[idxChunk+1:]...)	12✔
558	idxChunk--	12✔
559		12✔
560	// Check error	12✔
561	if err != nil {	12✔
562	err = fmt.Errorf("astikit: copying chunk #%d to dst failed: %w", c.idx, err)	×
563	return	×
564	}	×
565	}
566	}
567	return	12✔
568	})
569	}
570
571	// DownloadInFile downloads in parallel a set of srcs and concatenates them in a dst file while
572	// maintaining the initial order
573	func (d *HTTPDownloader) DownloadInFile(ctx context.Context, dst string, srcs ...HTTPDownloaderSrc) (err error) {	2✔
574	// Make sure destination directory exists	2✔
575	if err = os.MkdirAll(filepath.Dir(dst), DefaultDirMode); err != nil {	2✔
576	err = fmt.Errorf("astikit: mkdirall %s failed: %w", filepath.Dir(dst), err)	×
577	return	×
578	}	×
579
580	// Create destination file
581	var f *os.File	2✔
582	if f, err = os.Create(dst); err != nil {	2✔
583	err = fmt.Errorf("astikit: creating %s failed: %w", dst, err)	×
584	return	×
585	}	×
586	defer f.Close()	2✔
587		2✔
588	// Download in writer	2✔
589	return d.DownloadInWriter(ctx, f, srcs...)	2✔
590	}
591
592	// HTTPMiddleware represents an HTTP middleware
593	type HTTPMiddleware func(http.Handler) http.Handler
594
595	// ChainHTTPMiddlewares chains HTTP middlewares
596	func ChainHTTPMiddlewares(h http.Handler, ms ...HTTPMiddleware) http.Handler {	2✔
597	return ChainHTTPMiddlewaresWithPrefix(h, []string{}, ms...)	2✔
598	}	2✔
599
600	// ChainHTTPMiddlewaresWithPrefix chains HTTP middlewares if one of prefixes is present
601	func ChainHTTPMiddlewaresWithPrefix(h http.Handler, prefixes []string, ms ...HTTPMiddleware) http.Handler {	2✔
602	for _, m := range ms {	4✔
603	if m == nil {	2✔
604	continue	×
605	}
606	if len(prefixes) == 0 {	4✔
607	h = m(h)	2✔
608	} else {	2✔
609	t := h	×
610	h = http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {	×
611	for _, prefix := range prefixes {	×
612	if strings.HasPrefix(r.URL.EscapedPath(), prefix) {	×
613	m(t).ServeHTTP(rw, r)	×
614	return	×
615	}	×
616	}
617	t.ServeHTTP(rw, r)	×
618	})
619	}
620	}
621	return h	2✔
622	}
623
624	func handleHTTPBasicAuth(username, password string, rw http.ResponseWriter, r *http.Request) bool {	×
625	if u, p, ok := r.BasicAuth(); !ok \|\| u != username \|\| p != password {	×
626	rw.Header().Set("WWW-Authenticate", "Basic Realm=Please enter your credentials")	×
627	rw.WriteHeader(http.StatusUnauthorized)	×
628	return true	×
629	}	×
630	return false	×
631	}
632
633	// HTTPMiddlewareBasicAuth adds basic HTTP auth to an HTTP handler
634	func HTTPMiddlewareBasicAuth(username, password string) HTTPMiddleware {	×
635	if username == "" && password == "" {	×
636	return nil	×
637	}	×
638	return func(h http.Handler) http.Handler {	×
639	return http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {	×
640	// Handle basic auth	×
641	if handleHTTPBasicAuth(username, password, rw, r) {	×
642	return	×
643	}	×
644
645	// Next handler
646	h.ServeHTTP(rw, r)	×
647	})
648	}
649	}
650
651	func setHTTPContentType(contentType string, rw http.ResponseWriter) {	×
652	rw.Header().Set("Content-Type", contentType)	×
653	}	×
654
655	// HTTPMiddlewareContentType adds a content type to an HTTP handler
656	func HTTPMiddlewareContentType(contentType string) HTTPMiddleware {	×
657	return func(h http.Handler) http.Handler {	×
658	return http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {	×
659	// Set content type	×
660	setHTTPContentType(contentType, rw)	×
661		×
662	// Next handler	×
663	h.ServeHTTP(rw, r)	×
664	})	×
665	}
666	}
667
668	func setHTTPHeaders(vs map[string]string, rw http.ResponseWriter) {	×
669	for k, v := range vs {	×
670	rw.Header().Set(k, v)	×
671	}	×
672	}
673
674	// HTTPMiddlewareHeaders adds headers to an HTTP handler
675	func HTTPMiddlewareHeaders(vs map[string]string) HTTPMiddleware {	×
676	return func(h http.Handler) http.Handler {	×
677	return http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {	×
678	// Set headers	×
679	setHTTPHeaders(vs, rw)	×
680		×
681	// Next handler	×
682	h.ServeHTTP(rw, r)	×
683	})	×
684	}
685	}
686
687	// HTTPMiddlewareCORSHeaders adds CORS headers to an HTTP handler
688	func HTTPMiddlewareCORSHeaders() HTTPMiddleware {	×
689	return HTTPMiddlewareHeaders(map[string]string{	×
690	"Access-Control-Allow-Headers": "*",	×
691	"Access-Control-Allow-Methods": "*",	×
692	"Access-Control-Allow-Origin": "*",	×
693	})	×
694	}	×

asticode / go-astikit / 11996898886

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous