func (p *BulkProcessor) Start(ctx context.Context) error {
	p.startedMu.Lock()
	defer p.startedMu.Unlock()
	if p.started {
		return nil
	}
	if p.numWorkers < 1 {
		p.numWorkers = 1
	}
	p.requestsC = make(chan BulkableRequest)
	p.executionId = 0
	p.stats = newBulkProcessorStats(p.numWorkers)
	p.workers = make([]*bulkWorker, p.numWorkers)
	for i := 0; i < p.numWorkers; i++ {
		p.workerWg.Add(1)
		p.workers[i] = newBulkWorker(p, i)
		go p.workers[i].work(ctx)
	}
	if int64(p.flushInterval) > 0 {
		p.flusherStopC = make(chan struct{})
		go p.flusher(p.flushInterval)
	}
	p.started = true
	return nil
}
Start starts the bulk processor. If the processor is already started,
nil is returned.
package elastic
import (
	"context"
	"sync"
	"sync/atomic"
	"time"
)
type BulkProcessorService struct {
	c             *Client
	beforeFn      BulkBeforeFunc
	afterFn       BulkAfterFunc
	name          string         
	numWorkers    int            
	bulkActions   int            
	bulkSize      int            
	flushInterval time.Duration  
	wantStats     bool           
	backoff       Backoff        
}
func NewBulkProcessorService(client *Client) *BulkProcessorService {
	return &BulkProcessorService{
		c:           client,
		numWorkers:  1,
		bulkActions: 1000,
		bulkSize:    5 << 20,  
		backoff: NewExponentialBackoff(
			time.Duration(200)*time.Millisecond,
			time.Duration(10000)*time.Millisecond,
		),
	}
}
type BulkBeforeFunc func(executionId int64, requests []BulkableRequest)
type BulkAfterFunc func(executionId int64, requests []BulkableRequest, response *BulkResponse, err error)
func (s *BulkProcessorService) Before(fn BulkBeforeFunc) *BulkProcessorService {
	s.beforeFn = fn
	return s
}
func (s *BulkProcessorService) After(fn BulkAfterFunc) *BulkProcessorService {
	s.afterFn = fn
	return s
}
func (s *BulkProcessorService) Name(name string) *BulkProcessorService {
	s.name = name
	return s
}
func (s *BulkProcessorService) Workers(num int) *BulkProcessorService {
	s.numWorkers = num
	return s
}
func (s *BulkProcessorService) BulkActions(bulkActions int) *BulkProcessorService {
	s.bulkActions = bulkActions
	return s
}
func (s *BulkProcessorService) BulkSize(bulkSize int) *BulkProcessorService {
	s.bulkSize = bulkSize
	return s
}
func (s *BulkProcessorService) FlushInterval(interval time.Duration) *BulkProcessorService {
	s.flushInterval = interval
	return s
}
func (s *BulkProcessorService) Stats(wantStats bool) *BulkProcessorService {
	s.wantStats = wantStats
	return s
}
func (s *BulkProcessorService) Backoff(backoff Backoff) *BulkProcessorService {
	s.backoff = backoff
	return s
}
func (s *BulkProcessorService) Do(ctx context.Context) (*BulkProcessor, error) {
	p := newBulkProcessor(
		s.c,
		s.beforeFn,
		s.afterFn,
		s.name,
		s.numWorkers,
		s.bulkActions,
		s.bulkSize,
		s.flushInterval,
		s.wantStats,
		s.backoff)
	err := p.Start(ctx)
	if err != nil {
		return nil, err
	}
	return p, nil
}
type BulkProcessorStats struct {
	Flushed   int64  
	Committed int64  
	Indexed   int64  
	Created   int64  
	Updated   int64  
	Deleted   int64  
	Succeeded int64  
	Failed    int64  
	Workers []*BulkProcessorWorkerStats  
}
type BulkProcessorWorkerStats struct {
	Queued       int64          
	LastDuration time.Duration  
}
func newBulkProcessorStats(workers int) *BulkProcessorStats {
	stats := &BulkProcessorStats{
		Workers: make([]*BulkProcessorWorkerStats, workers),
	}
	for i := 0; i < workers; i++ {
		stats.Workers[i] = &BulkProcessorWorkerStats{}
	}
	return stats
}
func (st *BulkProcessorStats) dup() *BulkProcessorStats {
	dst := new(BulkProcessorStats)
	dst.Flushed = st.Flushed
	dst.Committed = st.Committed
	dst.Indexed = st.Indexed
	dst.Created = st.Created
	dst.Updated = st.Updated
	dst.Deleted = st.Deleted
	dst.Succeeded = st.Succeeded
	dst.Failed = st.Failed
	for _, src := range st.Workers {
		dst.Workers = append(dst.Workers, src.dup())
	}
	return dst
}
func (st *BulkProcessorWorkerStats) dup() *BulkProcessorWorkerStats {
	dst := new(BulkProcessorWorkerStats)
	dst.Queued = st.Queued
	dst.LastDuration = st.LastDuration
	return dst
}
type BulkProcessor struct {
	c             *Client
	beforeFn      BulkBeforeFunc
	afterFn       BulkAfterFunc
	name          string
	bulkActions   int
	bulkSize      int
	numWorkers    int
	executionId   int64
	requestsC     chan BulkableRequest
	workerWg      sync.WaitGroup
	workers       []*bulkWorker
	flushInterval time.Duration
	flusherStopC  chan struct{}
	wantStats     bool
	backoff       Backoff
	startedMu sync.Mutex  
	started   bool
	statsMu sync.Mutex  
	stats   *BulkProcessorStats
}
func newBulkProcessor(
	client *Client,
	beforeFn BulkBeforeFunc,
	afterFn BulkAfterFunc,
	name string,
	numWorkers int,
	bulkActions int,
	bulkSize int,
	flushInterval time.Duration,
	wantStats bool,
	backoff Backoff) *BulkProcessor {
	return &BulkProcessor{
		c:             client,
		beforeFn:      beforeFn,
		afterFn:       afterFn,
		name:          name,
		numWorkers:    numWorkers,
		bulkActions:   bulkActions,
		bulkSize:      bulkSize,
		flushInterval: flushInterval,
		wantStats:     wantStats,
		backoff:       backoff,
	}
}