func (e *Elasticsearch) Stop() {
	e.Client.Stop()
}
Stop stops the background processes that the client is running,
i.e. sniffing the cluster periodically and running health checks
on the nodes.
package v1
import (
	"context"
	"crypto/tls"
	"encoding/json"
	"fmt"
	"net/http"
	"net/url"
	"time"
	"github.com/Sirupsen/logrus"
	"github.com/google/uuid"
	"gopkg.in/olivere/elastic.v2"
	"gopkg.in/olivere/elastic.v2/backoff"
)
const version = 1
type Elasticsearch struct {
	*elastic.Client
	*elastic.BulkProcessor
	*elastic.BulkProcessorService
}
func NewClient(address, username, password string, timeout time.Duration, sniff bool, insecure bool) (*Elasticsearch, error) {
	url, _ := url.Parse(address)
	tr := new(http.Transport)
	if url.Scheme == "https" {
		tr = &http.Transport{
			TLSClientConfig: &tls.Config{InsecureSkipVerify: insecure},
		}
	}
	client := &http.Client{Transport: tr}
	c, err := elastic.NewClient(
		elastic.SetURL(address),
		elastic.SetScheme(url.Scheme),
		elastic.SetBasicAuth(username, password),
		elastic.SetHttpClient(client),
		elastic.SetSniff(sniff),
		elastic.SetRetrier(NewMyRetrier(timeout)),
	)
	if err != nil {
		return nil, fmt.Errorf("elasticsearch: cannot connect to the endpoint: %s\n%v", url, err)
	}
	return &Elasticsearch{
		Client:               c,
		BulkProcessorService: c.BulkProcessor(),
	}, nil
}
func (e *Elasticsearch) Log(_ context.Context, index, tzpe string, msg interface{}) error {
	if _, err := e.Client.Index().Index(index).Type(tzpe).BodyJson(msg).Do(); err != nil {
		return err
	}
	return nil
}
func (e *Elasticsearch) NewBulkProcessorService(_ context.Context, workers, actions, size int, flushInterval, timeout time.Duration, stats bool, log *logrus.Entry) error {
	afterFunc := func(executionId int64, bulkableRequests []elastic.BulkableRequest, response *elastic.BulkResponse, err error) {
		if response != nil && response.Errors {
			requests, perr := parseRequest(bulkableRequests)
			if perr != nil {
				log.WithError(err).Error("could not parse request")
			}
			for _, result := range response.Failed() {
				if result.Error == "" {
					continue
				}
				log.WithFields(logrus.Fields{
					"workerId":  executionId,
					"requestId": result.Id,
					"request":   requests.ById(result.Id),
					"reason":    result.Error,
					"status":    result.Status,
				}).Error("response error message and status code")
			}
		}
		if err != nil {
			log.WithError(err).WithFields(logrus.Fields{
				"workerId": executionId,
				"requests": bulkableRequests,
				"response": response,
			}).Error("after func")
		}
	}
	p, err := e.BulkProcessorService.
		Workers(workers).
		BulkActions(actions).          
		BulkSize(size).                
		FlushInterval(flushInterval).  
		Stats(stats).                  
		After(afterFunc).
		Do()
	if err != nil {
		return err
	}
	e.BulkProcessor = p
	return nil
}
func (e *Elasticsearch) Add(index, tzpe string, msg interface{}) error {
	id := uuid.New().String()
	r := elastic.NewBulkIndexRequest().Index(index).Type(tzpe).Doc(msg).Id(id)
	e.BulkProcessor.Add(r)
	return nil
}
func (e *Elasticsearch) Close() error {
	return e.BulkProcessor.Close()
}
func (e *Elasticsearch) Flush() error {
	return e.BulkProcessor.Flush()
}