func (w *Service) pointToEncode(point *write.Point) lp.Metric {
	var m lp.Metric
	if len(w.writeOptions.DefaultTags()) > 0 {
		m = &pointWithDefaultTags{
			point:       point,
			defaultTags: w.writeOptions.DefaultTags(),
		}
	} else {
		m = point
	}
	return m
}
pointToEncode determines whether default tags should be applied
and returns point with default tags instead of point
package write
import (
	"bytes"
	"context"
	"fmt"
	"io"
	"math/rand"
	"net/http"
	"net/url"
	"sort"
	"strings"
	"sync"
	"time"
	http2 "github.com/influxdata/influxdb-client-go/v2/api/http"
	"github.com/influxdata/influxdb-client-go/v2/api/write"
	"github.com/influxdata/influxdb-client-go/v2/internal/gzip"
	"github.com/influxdata/influxdb-client-go/v2/internal/log"
	ilog "github.com/influxdata/influxdb-client-go/v2/log"
	lp "github.com/influxdata/line-protocol"
)
type Batch struct {
	Batch string
	RetryDelay uint
	RetryAttempts uint
	Evicted bool
	Expires time.Time
}
func NewBatch(data string, retryDelay uint, expireDelayMs uint) *Batch {
	return &Batch{
		Batch:      data,
		RetryDelay: retryDelay,
		Expires:    time.Now().Add(time.Duration(expireDelayMs) * time.Millisecond),
	}
}
type BatchErrorCallback func(batch *Batch, error2 http2.Error) bool
type Service struct {
	org                  string
	bucket               string
	httpService          http2.Service
	url                  string
	lastWriteAttempt     time.Time
	retryQueue           *queue
	lock                 sync.Mutex
	writeOptions         *write.Options
	retryExponentialBase uint
	errorCb              BatchErrorCallback
}
func NewService(org string, bucket string, httpService http2.Service, options *write.Options) *Service {
	retryBufferLimit := options.RetryBufferLimit() / options.BatchSize()
	if retryBufferLimit == 0 {
		retryBufferLimit = 1
	}
	u, _ := url.Parse(httpService.ServerAPIURL())
	u, _ = u.Parse("write")
	params := u.Query()
	params.Set("org", org)
	params.Set("bucket", bucket)
	params.Set("precision", precisionToString(options.Precision()))
	u.RawQuery = params.Encode()
	writeURL := u.String()
	return &Service{
		org:                  org,
		bucket:               bucket,
		httpService:          httpService,
		url:                  writeURL,
		writeOptions:         options,
		retryQueue:           newQueue(int(retryBufferLimit)),
		retryExponentialBase: 2,
	}
}
func (w *Service) SetBatchErrorCallback(cb BatchErrorCallback) {
	w.errorCb = cb
}
func (w *Service) HandleWrite(ctx context.Context, batch *Batch) error {
	log.Debug("Write proc: received write request")
	batchToWrite := batch
	retrying := false
	for {
		select {
		case <-ctx.Done():
			log.Debug("Write proc: ctx cancelled req")
			return ctx.Err()
		default:
		}
		if !w.retryQueue.isEmpty() {
			log.Debug("Write proc: taking batch from retry queue")
			if !retrying {
				b := w.retryQueue.first()
				if w.lastWriteAttempt.IsZero() || time.Now().After(w.lastWriteAttempt.Add(time.Millisecond*time.Duration(b.RetryDelay))) {
					retrying = true
				} else {
					log.Warn("Write proc: cannot write yet, storing batch to queue")
					if w.retryQueue.push(batch) {
						log.Warn("Write proc: Retry buffer full, discarding oldest batch")
					}
					batchToWrite = nil
				}
			}
			if retrying {
				batchToWrite = w.retryQueue.first()
				if batch != nil {  
					if w.retryQueue.push(batch) {
						log.Warn("Write proc: Retry buffer full, discarding oldest batch")
					}
					batch = nil
				}
			}
		}
		if batchToWrite != nil {
			if time.Now().After(batchToWrite.Expires) {
				if !batchToWrite.Evicted {
					w.retryQueue.pop()
				}
				return fmt.Errorf("write failed (attempts %d): max retry time exceeded", batchToWrite.RetryAttempts)
			}
			perror := w.WriteBatch(ctx, batchToWrite)
			if perror != nil {
				if w.writeOptions.MaxRetries() != 0 && (perror.StatusCode == 0 || perror.StatusCode >= http.StatusTooManyRequests) {
					log.Errorf("Write error: %s\nBatch kept for retrying\n", perror.Error())
					if perror.RetryAfter > 0 {
						batchToWrite.RetryDelay = perror.RetryAfter * 1000
					} else {
						batchToWrite.RetryDelay = w.computeRetryDelay(batchToWrite.RetryAttempts)
					}
					if w.errorCb != nil && !w.errorCb(batchToWrite, *perror) {
						log.Warn("Callback rejected batch, discarding")
						if !batchToWrite.Evicted {
							w.retryQueue.pop()
						}
						return perror
					}
					if !batchToWrite.Evicted && batchToWrite != w.retryQueue.first() {
						if w.retryQueue.push(batch) {
							log.Warn("Retry buffer full, discarding oldest batch")
						}
					} else if batchToWrite.RetryAttempts == w.writeOptions.MaxRetries() {
						log.Warn("Reached maximum number of retries, discarding batch")
						if !batchToWrite.Evicted {
							w.retryQueue.pop()
						}
					}
					batchToWrite.RetryAttempts++
					log.Debugf("Write proc: next wait for write is %dms\n", batchToWrite.RetryDelay)
				} else {
					log.Errorf("Write error: %s\n", perror.Error())
				}
				return fmt.Errorf("write failed (attempts %d): %w", batchToWrite.RetryAttempts, perror)
			}
			if retrying && !batchToWrite.Evicted {
				w.retryQueue.pop()
			}
			batchToWrite = nil
		} else {
			break
		}
	}
	return nil
}
func (w *Service) computeRetryDelay(attempts uint) uint {
	minDelay := int(w.writeOptions.RetryInterval() * pow(w.writeOptions.ExponentialBase(), attempts))
	maxDelay := int(w.writeOptions.RetryInterval() * pow(w.writeOptions.ExponentialBase(), attempts+1))
	retryDelay := uint(rand.Intn(maxDelay-minDelay) + minDelay)
	if retryDelay > w.writeOptions.MaxRetryInterval() {
		retryDelay = w.writeOptions.MaxRetryInterval()
	}
	return retryDelay
}
func pow(x, y uint) uint {
	p := uint(1)
	if y == 0 {
		return 1
	}
	for i := uint(1); i <= y; i++ {
		p = p * x
	}
	return p
}
func (w *Service) WriteBatch(ctx context.Context, batch *Batch) *http2.Error {
	var body io.Reader
	var err error
	body = strings.NewReader(batch.Batch)
	if log.Level() >= ilog.DebugLevel {
		log.Debugf("Writing batch: %s", batch.Batch)
	}
	if w.writeOptions.UseGZip() {
		body, err = gzip.CompressWithGzip(body)
		if err != nil {
			return http2.NewError(err)
		}
	}
	w.lock.Lock()
	w.lastWriteAttempt = time.Now()
	w.lock.Unlock()
	perror := w.httpService.DoPostRequest(ctx, w.url, body, func(req *http.Request) {
		if w.writeOptions.UseGZip() {
			req.Header.Set("Content-Encoding", "gzip")
		}
	}, func(r *http.Response) error {
		return nil
	})
	return perror
}
type pointWithDefaultTags struct {
	point       *write.Point
	defaultTags map[string]string
}
func (p *pointWithDefaultTags) Name() string {
	return p.point.Name()
}
func (p *pointWithDefaultTags) Time() time.Time {
	return p.point.Time()
}
func (p *pointWithDefaultTags) FieldList() []*lp.Field {
	return p.point.FieldList()
}
func (p *pointWithDefaultTags) TagList() []*lp.Tag {
	tags := make([]*lp.Tag, 0, len(p.point.TagList())+len(p.defaultTags))
	tags = append(tags, p.point.TagList()...)
	for k, v := range p.defaultTags {
		if !existTag(p.point.TagList(), k) {
			tags = append(tags, &lp.Tag{
				Key:   k,
				Value: v,
			})
		}
	}
	sort.Slice(tags, func(i, j int) bool { return tags[i].Key < tags[j].Key })
	return tags
}
func existTag(tags []*lp.Tag, key string) bool {
	for _, tag := range tags {
		if key == tag.Key {
			return true
		}
	}
	return false
}
func (w *Service) EncodePoints(points ...*write.Point) (string, error) {
	var buffer bytes.Buffer
	e := lp.NewEncoder(&buffer)
	e.SetFieldTypeSupport(lp.UintSupport)
	e.FailOnFieldErr(true)
	e.SetPrecision(w.writeOptions.Precision())
	for _, point := range points {
		_, err := e.Encode(w.pointToEncode(point))
		if err != nil {
			return "", err
		}
	}
	return buffer.String(), nil
}