func (c *Client) updateConns(conns []*conn) {
	c.connsMu.Lock()
	var newConns []*conn
	for _, conn := range conns {
		var found bool
		for _, oldConn := range c.conns {
			if oldConn.NodeID() == conn.NodeID() {
				newConns = append(newConns, oldConn)
				found = true
				break
			}
		}
		if !found {
			c.infof("elastic: %s joined the cluster", conn.URL())
			newConns = append(newConns, conn)
		}
	}
	c.conns = newConns
	c.cindex = -1
	c.connsMu.Unlock()
}
updateConns updates the clients' connections with new information
gather by a sniff operation.
package elastic
import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"net/http/httputil"
	"net/url"
	"os"
	"regexp"
	"strings"
	"sync"
	"time"
	"github.com/pkg/errors"
	"github.com/olivere/elastic/config"
)
const (
	Version = "6.1.5"
	DefaultURL = "http://127.0.0.1:9200"
	DefaultScheme = "http"
	DefaultHealthcheckEnabled = true
	DefaultHealthcheckTimeoutStartup = 5 * time.Second
	DefaultHealthcheckTimeout = 1 * time.Second
	DefaultHealthcheckInterval = 60 * time.Second
	DefaultSnifferEnabled = true
	DefaultSnifferInterval = 15 * time.Minute
	DefaultSnifferTimeoutStartup = 5 * time.Second
	DefaultSnifferTimeout = 2 * time.Second
	DefaultSendGetBodyAs = "GET"
	off = -1 * time.Second
)
var (
	ErrNoClient = errors.New("no Elasticsearch node available")
	ErrRetry = errors.New("cannot connect after several retries")
	ErrTimeout = errors.New("timeout")
	noRetries = NewStopRetrier()
)
type ClientOptionFunc func(*Client) error
type Client struct {
	c *http.Client  
	connsMu sync.RWMutex  
	conns   []*conn       
	cindex  int           
	mu                        sync.RWMutex     
	urls                      []string         
	running                   bool             
	errorlog                  Logger           
	infolog                   Logger           
	tracelog                  Logger           
	scheme                    string           
	healthcheckEnabled        bool             
	healthcheckTimeoutStartup time.Duration    
	healthcheckTimeout        time.Duration    
	healthcheckInterval       time.Duration    
	healthcheckStop           chan bool        
	snifferEnabled            bool             
	snifferTimeoutStartup     time.Duration    
	snifferTimeout            time.Duration    
	snifferInterval           time.Duration    
	snifferCallback           SnifferCallback  
	snifferStop               chan bool        
	decoder                   Decoder          
	basicAuth                 bool             
	basicAuthUsername         string           
	basicAuthPassword         string           
	sendGetBodyAs             string           
	requiredPlugins           []string         
	retrier                   Retrier          
}
func NewClient(options ...ClientOptionFunc) (*Client, error) {
	c := &Client{
		c:                         http.DefaultClient,
		conns:                     make([]*conn, 0),
		cindex:                    -1,
		scheme:                    DefaultScheme,
		decoder:                   &DefaultDecoder{},
		healthcheckEnabled:        DefaultHealthcheckEnabled,
		healthcheckTimeoutStartup: DefaultHealthcheckTimeoutStartup,
		healthcheckTimeout:        DefaultHealthcheckTimeout,
		healthcheckInterval:       DefaultHealthcheckInterval,
		healthcheckStop:           make(chan bool),
		snifferEnabled:            DefaultSnifferEnabled,
		snifferTimeoutStartup:     DefaultSnifferTimeoutStartup,
		snifferTimeout:            DefaultSnifferTimeout,
		snifferInterval:           DefaultSnifferInterval,
		snifferCallback:           nopSnifferCallback,
		snifferStop:               make(chan bool),
		sendGetBodyAs:             DefaultSendGetBodyAs,
		retrier:                   noRetries,  
	}
	for _, option := range options {
		if err := option(c); err != nil {
			return nil, err
		}
	}
	if len(c.urls) == 0 {
		c.urls = []string{DefaultURL}
	}
	c.urls = canonicalize(c.urls...)
	if !c.basicAuth {
		for _, urlStr := range c.urls {
			u, err := url.Parse(urlStr)
			if err == nil && u.User != nil {
				c.basicAuth = true
				c.basicAuthUsername = u.User.Username()
				c.basicAuthPassword, _ = u.User.Password()
				break
			}
		}
	}
	if c.healthcheckEnabled {
		if err := c.startupHealthcheck(c.healthcheckTimeoutStartup); err != nil {
			return nil, err
		}
	}
	if c.snifferEnabled {
		if err := c.sniff(c.snifferTimeoutStartup); err != nil {
			return nil, err
		}
	} else {
		for _, url := range c.urls {
			c.conns = append(c.conns, newConn(url, url))
		}
	}
	if c.healthcheckEnabled {
		c.healthcheck(c.healthcheckTimeoutStartup, true)
	}
	if err := c.mustActiveConn(); err != nil {
		return nil, err
	}
	for _, plugin := range c.requiredPlugins {
		found, err := c.HasPlugin(plugin)
		if err != nil {
			return nil, err
		}
		if !found {
			return nil, fmt.Errorf("elastic: plugin %s not found", plugin)
		}
	}
	if c.snifferEnabled {
		go c.sniffer()  
	}
	if c.healthcheckEnabled {
		go c.healthchecker()  
	}
	c.mu.Lock()
	c.running = true
	c.mu.Unlock()
	return c, nil
}
func NewClientFromConfig(cfg *config.Config) (*Client, error) {
	var options []ClientOptionFunc
	if cfg != nil {
		if cfg.URL != "" {
			options = append(options, SetURL(cfg.URL))
		}
		if cfg.Errorlog != "" {
			f, err := os.OpenFile(cfg.Errorlog, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
			if err != nil {
				return nil, errors.Wrap(err, "unable to initialize error log")
			}
			l := log.New(f, "", 0)
			options = append(options, SetErrorLog(l))
		}
		if cfg.Tracelog != "" {
			f, err := os.OpenFile(cfg.Tracelog, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
			if err != nil {
				return nil, errors.Wrap(err, "unable to initialize trace log")
			}
			l := log.New(f, "", 0)
			options = append(options, SetTraceLog(l))
		}
		if cfg.Infolog != "" {
			f, err := os.OpenFile(cfg.Infolog, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
			if err != nil {
				return nil, errors.Wrap(err, "unable to initialize info log")
			}
			l := log.New(f, "", 0)
			options = append(options, SetInfoLog(l))
		}
		if cfg.Username != "" || cfg.Password != "" {
			options = append(options, SetBasicAuth(cfg.Username, cfg.Password))
		}
		if cfg.Sniff != nil {
			options = append(options, SetSniff(*cfg.Sniff))
		}
	}
	return NewClient(options...)
}
func NewSimpleClient(options ...ClientOptionFunc) (*Client, error) {
	c := &Client{
		c:                         http.DefaultClient,
		conns:                     make([]*conn, 0),
		cindex:                    -1,
		scheme:                    DefaultScheme,
		decoder:                   &DefaultDecoder{},
		healthcheckEnabled:        false,
		healthcheckTimeoutStartup: off,
		healthcheckTimeout:        off,
		healthcheckInterval:       off,
		healthcheckStop:           make(chan bool),
		snifferEnabled:            false,
		snifferTimeoutStartup:     off,
		snifferTimeout:            off,
		snifferInterval:           off,
		snifferCallback:           nopSnifferCallback,
		snifferStop:               make(chan bool),
		sendGetBodyAs:             DefaultSendGetBodyAs,
		retrier:                   noRetries,  
	}
	for _, option := range options {
		if err := option(c); err != nil {
			return nil, err
		}
	}
	if len(c.urls) == 0 {
		c.urls = []string{DefaultURL}
	}
	c.urls = canonicalize(c.urls...)
	if !c.basicAuth {
		for _, urlStr := range c.urls {
			u, err := url.Parse(urlStr)
			if err == nil && u.User != nil {
				c.basicAuth = true
				c.basicAuthUsername = u.User.Username()
				c.basicAuthPassword, _ = u.User.Password()
				break
			}
		}
	}
	for _, url := range c.urls {
		c.conns = append(c.conns, newConn(url, url))
	}
	if err := c.mustActiveConn(); err != nil {
		return nil, err
	}
	for _, plugin := range c.requiredPlugins {
		found, err := c.HasPlugin(plugin)
		if err != nil {
			return nil, err
		}
		if !found {
			return nil, fmt.Errorf("elastic: plugin %s not found", plugin)
		}
	}
	c.mu.Lock()
	c.running = true
	c.mu.Unlock()
	return c, nil
}
func SetHttpClient(httpClient *http.Client) ClientOptionFunc {
	return func(c *Client) error {
		if httpClient != nil {
			c.c = httpClient
		} else {
			c.c = http.DefaultClient
		}
		return nil
	}
}
func SetBasicAuth(username, password string) ClientOptionFunc {
	return func(c *Client) error {
		c.basicAuthUsername = username
		c.basicAuthPassword = password
		c.basicAuth = c.basicAuthUsername != "" || c.basicAuthPassword != ""
		return nil
	}
}
func SetURL(urls ...string) ClientOptionFunc {
	return func(c *Client) error {
		switch len(urls) {
		case 0:
			c.urls = []string{DefaultURL}
		default:
			c.urls = urls
		}
		return nil
	}
}
func SetScheme(scheme string) ClientOptionFunc {
	return func(c *Client) error {
		c.scheme = scheme
		return nil
	}
}
func SetSniff(enabled bool) ClientOptionFunc {
	return func(c *Client) error {
		c.snifferEnabled = enabled
		return nil
	}
}
func SetSnifferTimeoutStartup(timeout time.Duration) ClientOptionFunc {
	return func(c *Client) error {
		c.snifferTimeoutStartup = timeout
		return nil
	}
}
func SetSnifferTimeout(timeout time.Duration) ClientOptionFunc {
	return func(c *Client) error {
		c.snifferTimeout = timeout
		return nil
	}
}
func SetSnifferInterval(interval time.Duration) ClientOptionFunc {
	return func(c *Client) error {
		c.snifferInterval = interval
		return nil
	}
}
type SnifferCallback func(*NodesInfoNode) bool
var nopSnifferCallback = func(*NodesInfoNode) bool { return true }
func SetSnifferCallback(f SnifferCallback) ClientOptionFunc {
	return func(c *Client) error {
		if f != nil {
			c.snifferCallback = f
		}
		return nil
	}
}
func SetHealthcheck(enabled bool) ClientOptionFunc {
	return func(c *Client) error {
		c.healthcheckEnabled = enabled
		return nil
	}
}
func SetHealthcheckTimeoutStartup(timeout time.Duration) ClientOptionFunc {
	return func(c *Client) error {
		c.healthcheckTimeoutStartup = timeout
		return nil
	}
}
func SetHealthcheckTimeout(timeout time.Duration) ClientOptionFunc {
	return func(c *Client) error {
		c.healthcheckTimeout = timeout
		return nil
	}
}
func SetHealthcheckInterval(interval time.Duration) ClientOptionFunc {
	return func(c *Client) error {
		c.healthcheckInterval = interval
		return nil
	}
}
func SetMaxRetries(maxRetries int) ClientOptionFunc {
	return func(c *Client) error {
		if maxRetries < 0 {
			return errors.New("MaxRetries must be greater than or equal to 0")
		} else if maxRetries == 0 {
			c.retrier = noRetries
		} else {
			ticks := make([]int, maxRetries)
			for i := 0; i < len(ticks); i++ {
				ticks[i] = 100
			}
			backoff := NewSimpleBackoff(ticks...)
			c.retrier = NewBackoffRetrier(backoff)
		}
		return nil
	}
}
func SetDecoder(decoder Decoder) ClientOptionFunc {
	return func(c *Client) error {
		if decoder != nil {
			c.decoder = decoder
		} else {
			c.decoder = &DefaultDecoder{}
		}
		return nil
	}
}
func SetRequiredPlugins(plugins ...string) ClientOptionFunc {
	return func(c *Client) error {
		if c.requiredPlugins == nil {
			c.requiredPlugins = make([]string, 0)
		}
		c.requiredPlugins = append(c.requiredPlugins, plugins...)
		return nil
	}
}
func SetErrorLog(logger Logger) ClientOptionFunc {
	return func(c *Client) error {
		c.errorlog = logger
		return nil
	}
}
func SetInfoLog(logger Logger) ClientOptionFunc {
	return func(c *Client) error {
		c.infolog = logger
		return nil
	}
}
func SetTraceLog(logger Logger) ClientOptionFunc {
	return func(c *Client) error {
		c.tracelog = logger
		return nil
	}
}
func SetSendGetBodyAs(httpMethod string) ClientOptionFunc {
	return func(c *Client) error {
		c.sendGetBodyAs = httpMethod
		return nil
	}
}
func SetRetrier(retrier Retrier) ClientOptionFunc {
	return func(c *Client) error {
		if retrier == nil {
			retrier = noRetries  
		}
		c.retrier = retrier
		return nil
	}
}
func (c *Client) String() string {
	c.connsMu.Lock()
	conns := c.conns
	c.connsMu.Unlock()
	var buf bytes.Buffer
	for i, conn := range conns {
		if i > 0 {
			buf.WriteString(", ")
		}
		buf.WriteString(conn.String())
	}
	return buf.String()
}
func (c *Client) IsRunning() bool {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.running
}
func (c *Client) Start() {
	c.mu.RLock()
	if c.running {
		c.mu.RUnlock()
		return
	}
	c.mu.RUnlock()
	if c.snifferEnabled {
		go c.sniffer()
	}
	if c.healthcheckEnabled {
		go c.healthchecker()
	}
	c.mu.Lock()
	c.running = true
	c.mu.Unlock()
	c.infof("elastic: client started")
}
func (c *Client) Stop() {
	c.mu.RLock()
	if !c.running {
		c.mu.RUnlock()
		return
	}
	c.mu.RUnlock()
	if c.healthcheckEnabled {
		c.healthcheckStop <- true
		<-c.healthcheckStop
	}
	if c.snifferEnabled {
		c.snifferStop <- true
		<-c.snifferStop
	}
	c.mu.Lock()
	c.running = false
	c.mu.Unlock()
	c.infof("elastic: client stopped")
}
func (c *Client) errorf(format string, args ...interface{}) {
	if c.errorlog != nil {
		c.errorlog.Printf(format, args...)
	}
}
func (c *Client) infof(format string, args ...interface{}) {
	if c.infolog != nil {
		c.infolog.Printf(format, args...)
	}
}
func (c *Client) tracef(format string, args ...interface{}) {
	if c.tracelog != nil {
		c.tracelog.Printf(format, args...)
	}
}
func (c *Client) dumpRequest(r *http.Request) {
	if c.tracelog != nil {
		out, err := httputil.DumpRequestOut(r, true)
		if err == nil {
			c.tracef("%s\n", string(out))
		}
	}
}
func (c *Client) dumpResponse(resp *http.Response) {
	if c.tracelog != nil {
		out, err := httputil.DumpResponse(resp, true)
		if err == nil {
			c.tracef("%s\n", string(out))
		}
	}
}
func (c *Client) sniffer() {
	c.mu.RLock()
	timeout := c.snifferTimeout
	interval := c.snifferInterval
	c.mu.RUnlock()
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-c.snifferStop:
			c.snifferStop <- true
			return
		case <-ticker.C:
			c.sniff(timeout)
		}
	}
}
func (c *Client) sniff(timeout time.Duration) error {
	c.mu.RLock()
	if !c.snifferEnabled {
		c.mu.RUnlock()
		return nil
	}
	var urls []string
	urlsMap := make(map[string]bool)
	for _, url := range c.urls {
		urlsMap[url] = true
		urls = append(urls, url)
	}
	c.mu.RUnlock()
	c.connsMu.RLock()
	for _, conn := range c.conns {
		if !conn.IsDead() {
			url := conn.URL()
			if _, found := urlsMap[url]; !found {
				urls = append(urls, url)
			}
		}
	}
	c.connsMu.RUnlock()
	if len(urls) == 0 {
		return errors.Wrap(ErrNoClient, "no URLs found")
	}
	ch := make(chan []*conn, len(urls))
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	for _, url := range urls {
		go func(url string) { ch <- c.sniffNode(ctx, url) }(url)
	}
	for {
		select {
		case conns := <-ch:
			if len(conns) > 0 {
				c.updateConns(conns)
				return nil
			}
		case <-ctx.Done():
			return errors.Wrap(ErrNoClient, "sniff timeout")
		}
	}
}
func (c *Client) sniffNode(ctx context.Context, url string) []*conn {
	var nodes []*conn
	req, err := NewRequest("GET", url+"/_nodes/http")
	if err != nil {
		return nodes
	}
	c.mu.RLock()
	if c.basicAuth {
		req.SetBasicAuth(c.basicAuthUsername, c.basicAuthPassword)
	}
	c.mu.RUnlock()
	res, err := c.c.Do((*http.Request)(req).WithContext(ctx))
	if err != nil {
		return nodes
	}
	if res == nil {
		return nodes
	}
	if res.Body != nil {
		defer res.Body.Close()
	}
	var info NodesInfoResponse
	if err := json.NewDecoder(res.Body).Decode(&info); err == nil {
		if len(info.Nodes) > 0 {
			for nodeID, node := range info.Nodes {
				if c.snifferCallback(node) {
					if node.HTTP != nil && len(node.HTTP.PublishAddress) > 0 {
						url := c.extractHostname(c.scheme, node.HTTP.PublishAddress)
						if url != "" {
							nodes = append(nodes, newConn(nodeID, url))
						}
					}
				}
			}
		}
	}
	return nodes
}
var reSniffHostAndPort = regexp.MustCompile(`\/([^:]*):([0-9]+)\]`)
func (c *Client) extractHostname(scheme, address string) string {
	if strings.HasPrefix(address, "inet") {
		m := reSniffHostAndPort.FindStringSubmatch(address)
		if len(m) == 3 {
			return fmt.Sprintf("%s://%s:%s", scheme, m[1], m[2])
		}
	}
	s := address
	if idx := strings.Index(s, "/"); idx >= 0 {
		s = s[idx+1:]
	}
	if strings.Index(s, ":") < 0 {
		return ""
	}
	return fmt.Sprintf("%s://%s", scheme, s)
}