17
vendor/github.com/elastic/go-elasticsearch/v5/estransport/doc.go
generated
vendored
Normal file
17
vendor/github.com/elastic/go-elasticsearch/v5/estransport/doc.go
generated
vendored
Normal file
@@ -0,0 +1,17 @@
|
||||
/*
|
||||
Package estransport provides the transport layer for the Elasticsearch client.
|
||||
|
||||
It is automatically included in the client provided by the github.com/elastic/go-elasticsearch package
|
||||
and is not intended for direct use: to configure the client, use the elasticsearch.Config struct.
|
||||
|
||||
The default HTTP transport of the client is http.Transport.
|
||||
|
||||
The package defines the "Selector" interface for getting a URL from the list. At the moment,
|
||||
the implementation is rather minimal: the client takes a slice of url.URL pointers,
|
||||
and round-robins across them when performing the request.
|
||||
|
||||
The package defines the "Logger" interface for logging information about request and response.
|
||||
It comes with several bundled loggers for logging in text and JSON.
|
||||
|
||||
*/
|
||||
package estransport
|
||||
198
vendor/github.com/elastic/go-elasticsearch/v5/estransport/estransport.go
generated
vendored
Normal file
198
vendor/github.com/elastic/go-elasticsearch/v5/estransport/estransport.go
generated
vendored
Normal file
@@ -0,0 +1,198 @@
|
||||
package estransport
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"regexp"
|
||||
"runtime"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/elastic/go-elasticsearch/v5/internal/version"
|
||||
)
|
||||
|
||||
// Version returns the package version as a string.
|
||||
//
|
||||
const Version = version.Client
|
||||
|
||||
var (
|
||||
userAgent string
|
||||
reGoVersion = regexp.MustCompile(`go(\d+\.\d+\..+)`)
|
||||
)
|
||||
|
||||
func init() {
|
||||
userAgent = initUserAgent()
|
||||
}
|
||||
|
||||
// Interface defines the interface for HTTP client.
|
||||
//
|
||||
type Interface interface {
|
||||
Perform(*http.Request) (*http.Response, error)
|
||||
}
|
||||
|
||||
// Config represents the configuration of HTTP client.
|
||||
//
|
||||
type Config struct {
|
||||
URLs []*url.URL
|
||||
Username string
|
||||
Password string
|
||||
|
||||
Transport http.RoundTripper
|
||||
Logger Logger
|
||||
}
|
||||
|
||||
// Client represents the HTTP client.
|
||||
//
|
||||
type Client struct {
|
||||
urls []*url.URL
|
||||
username string
|
||||
password string
|
||||
|
||||
transport http.RoundTripper
|
||||
selector Selector
|
||||
logger Logger
|
||||
}
|
||||
|
||||
// New creates new HTTP client.
|
||||
//
|
||||
// http.DefaultTransport will be used if no transport is passed in the configuration.
|
||||
//
|
||||
func New(cfg Config) *Client {
|
||||
if cfg.Transport == nil {
|
||||
cfg.Transport = http.DefaultTransport
|
||||
}
|
||||
|
||||
return &Client{
|
||||
urls: cfg.URLs,
|
||||
username: cfg.Username,
|
||||
password: cfg.Password,
|
||||
|
||||
transport: cfg.Transport,
|
||||
selector: NewRoundRobinSelector(cfg.URLs...),
|
||||
logger: cfg.Logger,
|
||||
}
|
||||
}
|
||||
|
||||
// Perform executes the request and returns a response or error.
|
||||
//
|
||||
func (c *Client) Perform(req *http.Request) (*http.Response, error) {
|
||||
u, err := c.getURL()
|
||||
if err != nil {
|
||||
// TODO(karmi): Log error
|
||||
return nil, fmt.Errorf("cannot get URL: %s", err)
|
||||
}
|
||||
|
||||
c.setURL(u, req)
|
||||
c.setUserAgent(req)
|
||||
|
||||
if _, ok := req.Header["Authorization"]; !ok {
|
||||
c.setBasicAuth(u, req)
|
||||
}
|
||||
|
||||
var dupReqBody *bytes.Buffer
|
||||
if c.logger != nil && c.logger.RequestBodyEnabled() {
|
||||
if req.Body != nil && req.Body != http.NoBody {
|
||||
dupReqBody = bytes.NewBuffer(make([]byte, 0, int(req.ContentLength)))
|
||||
dupReqBody.ReadFrom(req.Body)
|
||||
req.Body = ioutil.NopCloser(bytes.NewBuffer(dupReqBody.Bytes()))
|
||||
}
|
||||
}
|
||||
|
||||
start := time.Now().UTC()
|
||||
res, err := c.transport.RoundTrip(req)
|
||||
dur := time.Since(start)
|
||||
|
||||
if c.logger != nil {
|
||||
var dupRes http.Response
|
||||
if res != nil {
|
||||
dupRes = *res
|
||||
}
|
||||
if c.logger.RequestBodyEnabled() {
|
||||
if req.Body != nil && req.Body != http.NoBody {
|
||||
req.Body = ioutil.NopCloser(dupReqBody)
|
||||
}
|
||||
}
|
||||
if c.logger.ResponseBodyEnabled() {
|
||||
if res != nil && res.Body != nil && res.Body != http.NoBody {
|
||||
b1, b2, _ := duplicateBody(res.Body)
|
||||
dupRes.Body = b1
|
||||
res.Body = b2
|
||||
}
|
||||
}
|
||||
c.logger.LogRoundTrip(req, &dupRes, err, start, dur) // errcheck exclude
|
||||
}
|
||||
|
||||
// TODO(karmi): Wrap error
|
||||
return res, err
|
||||
}
|
||||
|
||||
// URLs returns a list of transport URLs.
|
||||
//
|
||||
func (c *Client) URLs() []*url.URL {
|
||||
return c.urls
|
||||
}
|
||||
|
||||
func (c *Client) getURL() (*url.URL, error) {
|
||||
return c.selector.Select()
|
||||
}
|
||||
|
||||
func (c *Client) setURL(u *url.URL, req *http.Request) *http.Request {
|
||||
req.URL.Scheme = u.Scheme
|
||||
req.URL.Host = u.Host
|
||||
|
||||
if u.Path != "" {
|
||||
var b strings.Builder
|
||||
b.Grow(len(u.Path) + len(req.URL.Path))
|
||||
b.WriteString(u.Path)
|
||||
b.WriteString(req.URL.Path)
|
||||
req.URL.Path = b.String()
|
||||
}
|
||||
|
||||
return req
|
||||
}
|
||||
|
||||
func (c *Client) setBasicAuth(u *url.URL, req *http.Request) *http.Request {
|
||||
if u.User != nil {
|
||||
password, _ := u.User.Password()
|
||||
req.SetBasicAuth(u.User.Username(), password)
|
||||
return req
|
||||
}
|
||||
|
||||
if c.username != "" && c.password != "" {
|
||||
req.SetBasicAuth(c.username, c.password)
|
||||
return req
|
||||
}
|
||||
|
||||
return req
|
||||
}
|
||||
|
||||
func (c *Client) setUserAgent(req *http.Request) *http.Request {
|
||||
req.Header.Set("User-Agent", userAgent)
|
||||
return req
|
||||
}
|
||||
|
||||
func initUserAgent() string {
|
||||
var b strings.Builder
|
||||
|
||||
b.WriteString("go-elasticsearch")
|
||||
b.WriteRune('/')
|
||||
b.WriteString(Version)
|
||||
b.WriteRune(' ')
|
||||
b.WriteRune('(')
|
||||
b.WriteString(runtime.GOOS)
|
||||
b.WriteRune(' ')
|
||||
b.WriteString(runtime.GOARCH)
|
||||
b.WriteString("; ")
|
||||
b.WriteString("Go ")
|
||||
if v := reGoVersion.ReplaceAllString(runtime.Version(), "$1"); v != "" {
|
||||
b.WriteString(v)
|
||||
} else {
|
||||
b.WriteString(runtime.Version())
|
||||
}
|
||||
b.WriteRune(')')
|
||||
|
||||
return b.String()
|
||||
}
|
||||
388
vendor/github.com/elastic/go-elasticsearch/v5/estransport/logger.go
generated
vendored
Normal file
388
vendor/github.com/elastic/go-elasticsearch/v5/estransport/logger.go
generated
vendored
Normal file
@@ -0,0 +1,388 @@
|
||||
package estransport
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Logger defines an interface for logging request and response.
|
||||
//
|
||||
type Logger interface {
|
||||
// LogRoundTrip should not modify the request or response, except for consuming and closing the body.
|
||||
// Implementations have to check for nil values in request and response.
|
||||
LogRoundTrip(*http.Request, *http.Response, error, time.Time, time.Duration) error
|
||||
// RequestBodyEnabled makes the client pass a copy of request body to the logger.
|
||||
RequestBodyEnabled() bool
|
||||
// ResponseBodyEnabled makes the client pass a copy of response body to the logger.
|
||||
ResponseBodyEnabled() bool
|
||||
}
|
||||
|
||||
// TextLogger prints the log message in plain text.
|
||||
//
|
||||
type TextLogger struct {
|
||||
Output io.Writer
|
||||
EnableRequestBody bool
|
||||
EnableResponseBody bool
|
||||
}
|
||||
|
||||
// ColorLogger prints the log message in a terminal-optimized plain text.
|
||||
//
|
||||
type ColorLogger struct {
|
||||
Output io.Writer
|
||||
EnableRequestBody bool
|
||||
EnableResponseBody bool
|
||||
}
|
||||
|
||||
// CurlLogger prints the log message as a runnable curl command.
|
||||
//
|
||||
type CurlLogger struct {
|
||||
Output io.Writer
|
||||
EnableRequestBody bool
|
||||
EnableResponseBody bool
|
||||
}
|
||||
|
||||
// JSONLogger prints the log message as JSON.
|
||||
//
|
||||
type JSONLogger struct {
|
||||
Output io.Writer
|
||||
EnableRequestBody bool
|
||||
EnableResponseBody bool
|
||||
}
|
||||
|
||||
// LogRoundTrip prints the information about request and response.
|
||||
//
|
||||
func (l *TextLogger) LogRoundTrip(req *http.Request, res *http.Response, err error, start time.Time, dur time.Duration) error {
|
||||
fmt.Fprintf(l.Output, "%s %s %s [status:%d request:%s]\n",
|
||||
start.Format(time.RFC3339),
|
||||
req.Method,
|
||||
req.URL.String(),
|
||||
resStatusCode(res),
|
||||
dur.Truncate(time.Millisecond),
|
||||
)
|
||||
if l.RequestBodyEnabled() && req != nil && req.Body != nil && req.Body != http.NoBody {
|
||||
var buf bytes.Buffer
|
||||
buf.ReadFrom(req.Body)
|
||||
logBodyAsText(l.Output, &buf, ">")
|
||||
}
|
||||
if l.ResponseBodyEnabled() && res != nil && res.Body != nil && res.Body != http.NoBody {
|
||||
defer res.Body.Close()
|
||||
var buf bytes.Buffer
|
||||
buf.ReadFrom(res.Body)
|
||||
logBodyAsText(l.Output, &buf, "<")
|
||||
}
|
||||
if err != nil {
|
||||
fmt.Fprintf(l.Output, "! ERROR: %v\n", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// RequestBodyEnabled returns true when the request body should be logged.
|
||||
func (l *TextLogger) RequestBodyEnabled() bool { return l.EnableRequestBody }
|
||||
|
||||
// ResponseBodyEnabled returns true when the response body should be logged.
|
||||
func (l *TextLogger) ResponseBodyEnabled() bool { return l.EnableResponseBody }
|
||||
|
||||
// LogRoundTrip prints the information about request and response.
|
||||
//
|
||||
func (l *ColorLogger) LogRoundTrip(req *http.Request, res *http.Response, err error, start time.Time, dur time.Duration) error {
|
||||
query, _ := url.QueryUnescape(req.URL.RawQuery)
|
||||
if query != "" {
|
||||
query = "?" + query
|
||||
}
|
||||
|
||||
var (
|
||||
status string
|
||||
color string
|
||||
)
|
||||
|
||||
status = res.Status
|
||||
switch {
|
||||
case res.StatusCode > 0 && res.StatusCode < 300:
|
||||
color = "\x1b[32m"
|
||||
case res.StatusCode > 299 && res.StatusCode < 500:
|
||||
color = "\x1b[33m"
|
||||
case res.StatusCode > 499:
|
||||
color = "\x1b[31m"
|
||||
default:
|
||||
status = "ERROR"
|
||||
color = "\x1b[31;4m"
|
||||
}
|
||||
|
||||
fmt.Fprintf(l.Output, "%6s \x1b[1;4m%s://%s%s\x1b[0m%s %s%s\x1b[0m \x1b[2m%s\x1b[0m\n",
|
||||
req.Method,
|
||||
req.URL.Scheme,
|
||||
req.URL.Host,
|
||||
req.URL.Path,
|
||||
query,
|
||||
color,
|
||||
status,
|
||||
dur.Truncate(time.Millisecond),
|
||||
)
|
||||
|
||||
if l.RequestBodyEnabled() && req != nil && req.Body != nil && req.Body != http.NoBody {
|
||||
var buf bytes.Buffer
|
||||
buf.ReadFrom(req.Body)
|
||||
fmt.Fprint(l.Output, "\x1b[2m")
|
||||
logBodyAsText(l.Output, &buf, " »")
|
||||
fmt.Fprint(l.Output, "\x1b[0m")
|
||||
}
|
||||
|
||||
if l.ResponseBodyEnabled() && res != nil && res.Body != nil && res.Body != http.NoBody {
|
||||
defer res.Body.Close()
|
||||
var buf bytes.Buffer
|
||||
buf.ReadFrom(res.Body)
|
||||
fmt.Fprint(l.Output, "\x1b[2m")
|
||||
logBodyAsText(l.Output, &buf, " «")
|
||||
fmt.Fprint(l.Output, "\x1b[0m")
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
fmt.Fprintf(l.Output, "\x1b[31;1m» ERROR \x1b[31m%v\x1b[0m\n", err)
|
||||
}
|
||||
|
||||
if l.RequestBodyEnabled() || l.ResponseBodyEnabled() {
|
||||
fmt.Fprintf(l.Output, "\x1b[2m%s\x1b[0m\n", strings.Repeat("─", 80))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// RequestBodyEnabled returns true when the request body should be logged.
|
||||
func (l *ColorLogger) RequestBodyEnabled() bool { return l.EnableRequestBody }
|
||||
|
||||
// ResponseBodyEnabled returns true when the response body should be logged.
|
||||
func (l *ColorLogger) ResponseBodyEnabled() bool { return l.EnableResponseBody }
|
||||
|
||||
// LogRoundTrip prints the information about request and response.
|
||||
//
|
||||
func (l *CurlLogger) LogRoundTrip(req *http.Request, res *http.Response, err error, start time.Time, dur time.Duration) error {
|
||||
var b bytes.Buffer
|
||||
|
||||
var query string
|
||||
qvalues := url.Values{}
|
||||
for k, v := range req.URL.Query() {
|
||||
if k == "pretty" {
|
||||
continue
|
||||
}
|
||||
for _, qv := range v {
|
||||
qvalues.Add(k, qv)
|
||||
}
|
||||
}
|
||||
if len(qvalues) > 0 {
|
||||
query = qvalues.Encode()
|
||||
}
|
||||
|
||||
b.WriteString(`curl`)
|
||||
if req.Method == "HEAD" {
|
||||
b.WriteString(" --head")
|
||||
} else {
|
||||
fmt.Fprintf(&b, " -X %s", req.Method)
|
||||
}
|
||||
|
||||
if len(req.Header) > 0 {
|
||||
for k, vv := range req.Header {
|
||||
if k == "Authorization" || k == "User-Agent" {
|
||||
continue
|
||||
}
|
||||
v := strings.Join(vv, ",")
|
||||
b.WriteString(fmt.Sprintf(" -H '%s: %s'", k, v))
|
||||
}
|
||||
}
|
||||
|
||||
b.WriteString(" 'http://localhost:9200")
|
||||
b.WriteString(req.URL.Path)
|
||||
b.WriteString("?pretty")
|
||||
if query != "" {
|
||||
fmt.Fprintf(&b, "&%s", query)
|
||||
}
|
||||
b.WriteString("'")
|
||||
|
||||
if req != nil && req.Body != nil && req.Body != http.NoBody {
|
||||
var buf bytes.Buffer
|
||||
buf.ReadFrom(req.Body)
|
||||
|
||||
b.Grow(buf.Len())
|
||||
b.WriteString(" -d \\\n'")
|
||||
json.Indent(&b, buf.Bytes(), "", " ")
|
||||
b.WriteString("'")
|
||||
}
|
||||
|
||||
b.WriteRune('\n')
|
||||
|
||||
var status string
|
||||
status = res.Status
|
||||
|
||||
fmt.Fprintf(&b, "# => %s [%s] %s\n", start.UTC().Format(time.RFC3339), status, dur.Truncate(time.Millisecond))
|
||||
if l.ResponseBodyEnabled() && res != nil && res.Body != nil && res.Body != http.NoBody {
|
||||
var buf bytes.Buffer
|
||||
buf.ReadFrom(res.Body)
|
||||
|
||||
b.Grow(buf.Len())
|
||||
b.WriteString("# ")
|
||||
json.Indent(&b, buf.Bytes(), "# ", " ")
|
||||
}
|
||||
|
||||
b.WriteString("\n")
|
||||
if l.ResponseBodyEnabled() && res != nil && res.Body != nil && res.Body != http.NoBody {
|
||||
b.WriteString("\n")
|
||||
}
|
||||
|
||||
b.WriteTo(l.Output)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// RequestBodyEnabled returns true when the request body should be logged.
|
||||
func (l *CurlLogger) RequestBodyEnabled() bool { return l.EnableRequestBody }
|
||||
|
||||
// ResponseBodyEnabled returns true when the response body should be logged.
|
||||
func (l *CurlLogger) ResponseBodyEnabled() bool { return l.EnableResponseBody }
|
||||
|
||||
// LogRoundTrip prints the information about request and response.
|
||||
//
|
||||
func (l *JSONLogger) LogRoundTrip(req *http.Request, res *http.Response, err error, start time.Time, dur time.Duration) error {
|
||||
// https://github.com/elastic/ecs/blob/master/schemas/http.yml
|
||||
//
|
||||
// TODO(karmi): Research performance optimization of using sync.Pool
|
||||
|
||||
bsize := 200
|
||||
var b = bytes.NewBuffer(make([]byte, 0, bsize))
|
||||
var v = make([]byte, 0, bsize)
|
||||
|
||||
appendTime := func(t time.Time) {
|
||||
v = v[:0]
|
||||
v = t.AppendFormat(v, time.RFC3339)
|
||||
b.Write(v)
|
||||
}
|
||||
|
||||
appendQuote := func(s string) {
|
||||
v = v[:0]
|
||||
v = strconv.AppendQuote(v, s)
|
||||
b.Write(v)
|
||||
}
|
||||
|
||||
appendInt := func(i int64) {
|
||||
v = v[:0]
|
||||
v = strconv.AppendInt(v, i, 10)
|
||||
b.Write(v)
|
||||
}
|
||||
|
||||
port := req.URL.Port()
|
||||
|
||||
b.WriteRune('{')
|
||||
// -- Timestamp
|
||||
b.WriteString(`"@timestamp":"`)
|
||||
appendTime(start.UTC())
|
||||
b.WriteRune('"')
|
||||
// -- Event
|
||||
b.WriteString(`,"event":{`)
|
||||
b.WriteString(`"duration":`)
|
||||
appendInt(dur.Nanoseconds())
|
||||
b.WriteRune('}')
|
||||
// -- URL
|
||||
b.WriteString(`,"url":{`)
|
||||
b.WriteString(`"scheme":`)
|
||||
appendQuote(req.URL.Scheme)
|
||||
b.WriteString(`,"domain":`)
|
||||
appendQuote(req.URL.Hostname())
|
||||
if port != "" {
|
||||
b.WriteString(`,"port":`)
|
||||
b.WriteString(port)
|
||||
}
|
||||
b.WriteString(`,"path":`)
|
||||
appendQuote(req.URL.Path)
|
||||
b.WriteString(`,"query":`)
|
||||
appendQuote(req.URL.RawQuery)
|
||||
b.WriteRune('}') // Close "url"
|
||||
// -- HTTP
|
||||
b.WriteString(`,"http":`)
|
||||
// ---- Request
|
||||
b.WriteString(`{"request":{`)
|
||||
b.WriteString(`"method":`)
|
||||
appendQuote(req.Method)
|
||||
if l.RequestBodyEnabled() && req != nil && req.Body != nil && req.Body != http.NoBody {
|
||||
var buf bytes.Buffer
|
||||
buf.ReadFrom(req.Body)
|
||||
|
||||
b.Grow(buf.Len() + 8)
|
||||
b.WriteString(`,"body":`)
|
||||
appendQuote(buf.String())
|
||||
}
|
||||
b.WriteRune('}') // Close "http.request"
|
||||
// ---- Response
|
||||
b.WriteString(`,"response":{`)
|
||||
b.WriteString(`"status_code":`)
|
||||
appendInt(int64(resStatusCode(res)))
|
||||
if l.ResponseBodyEnabled() && res != nil && res.Body != nil && res.Body != http.NoBody {
|
||||
defer res.Body.Close()
|
||||
var buf bytes.Buffer
|
||||
buf.ReadFrom(res.Body)
|
||||
|
||||
b.Grow(buf.Len() + 8)
|
||||
b.WriteString(`,"body":`)
|
||||
appendQuote(buf.String())
|
||||
}
|
||||
b.WriteRune('}') // Close "http.response"
|
||||
b.WriteRune('}') // Close "http"
|
||||
// -- Error
|
||||
if err != nil {
|
||||
b.WriteString(`,"error":{"message":`)
|
||||
appendQuote(err.Error())
|
||||
b.WriteRune('}') // Close "error"
|
||||
}
|
||||
b.WriteRune('}')
|
||||
b.WriteRune('\n')
|
||||
b.WriteTo(l.Output)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// RequestBodyEnabled returns true when the request body should be logged.
|
||||
func (l *JSONLogger) RequestBodyEnabled() bool { return l.EnableRequestBody }
|
||||
|
||||
// ResponseBodyEnabled returns true when the response body should be logged.
|
||||
func (l *JSONLogger) ResponseBodyEnabled() bool { return l.EnableResponseBody }
|
||||
|
||||
func logBodyAsText(dst io.Writer, body io.Reader, prefix string) {
|
||||
scanner := bufio.NewScanner(body)
|
||||
for scanner.Scan() {
|
||||
s := scanner.Text()
|
||||
if s != "" {
|
||||
fmt.Fprintf(dst, "%s %s\n", prefix, s)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func duplicateBody(body io.ReadCloser) (io.ReadCloser, io.ReadCloser, error) {
|
||||
var (
|
||||
b1 bytes.Buffer
|
||||
b2 bytes.Buffer
|
||||
tr = io.TeeReader(body, &b2)
|
||||
)
|
||||
_, err := b1.ReadFrom(tr)
|
||||
if err != nil {
|
||||
return ioutil.NopCloser(io.MultiReader(&b1, errorReader{err: err})), ioutil.NopCloser(io.MultiReader(&b2, errorReader{err: err})), err
|
||||
}
|
||||
defer func() { body.Close() }()
|
||||
|
||||
return ioutil.NopCloser(&b1), ioutil.NopCloser(&b2), nil
|
||||
}
|
||||
|
||||
func resStatusCode(res *http.Response) int {
|
||||
if res == nil {
|
||||
return -1
|
||||
}
|
||||
return res.StatusCode
|
||||
}
|
||||
|
||||
type errorReader struct{ err error }
|
||||
|
||||
func (r errorReader) Read(p []byte) (int, error) { return 0, r.err }
|
||||
54
vendor/github.com/elastic/go-elasticsearch/v5/estransport/selector.go
generated
vendored
Normal file
54
vendor/github.com/elastic/go-elasticsearch/v5/estransport/selector.go
generated
vendored
Normal file
@@ -0,0 +1,54 @@
|
||||
package estransport
|
||||
|
||||
import (
|
||||
"container/ring"
|
||||
"errors"
|
||||
"net/url"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// Selector defines the interface for selecting URLs for performing request.
|
||||
//
|
||||
type Selector interface {
|
||||
Select() (*url.URL, error)
|
||||
}
|
||||
|
||||
// RoundRobinSelector implements a round-robin selection strategy.
|
||||
//
|
||||
type RoundRobinSelector struct {
|
||||
sync.Mutex
|
||||
ring *ring.Ring
|
||||
}
|
||||
|
||||
// Select returns a URL or error from the list of URLs in a round-robin fashion.
|
||||
//
|
||||
func (r *RoundRobinSelector) Select() (*url.URL, error) {
|
||||
r.Lock()
|
||||
defer r.Unlock()
|
||||
|
||||
if r.ring.Len() < 1 {
|
||||
return nil, errors.New("No URL available")
|
||||
}
|
||||
|
||||
v := r.ring.Value
|
||||
if ov, ok := v.(*url.URL); !ok || ov == nil {
|
||||
return nil, errors.New("No URL available")
|
||||
}
|
||||
|
||||
r.ring = r.ring.Next()
|
||||
return v.(*url.URL), nil
|
||||
}
|
||||
|
||||
// NewRoundRobinSelector creates a new RoundRobinSelector.
|
||||
//
|
||||
func NewRoundRobinSelector(urls ...*url.URL) *RoundRobinSelector {
|
||||
r := RoundRobinSelector{}
|
||||
|
||||
r.ring = ring.New(len(urls))
|
||||
for _, u := range urls {
|
||||
r.ring.Value = u
|
||||
r.ring = r.ring.Next()
|
||||
}
|
||||
|
||||
return &r
|
||||
}
|
||||
Reference in New Issue
Block a user