func (ix *Reindexer) count() (int64, error) {
	service := ix.sourceClient.Count(ix.sourceIndex)
	if ix.query != nil {
		service = service.Query(ix.query)
	}
	return service.Do()
}
count returns the number of documents in the source index.
The query is taken into account, if specified.
package elastic
import (
	"encoding/json"
	"errors"
)
type Reindexer struct {
	sourceClient, targetClient *Client
	sourceIndex                string
	query                      Query
	scanFields                 []string
	bulkSize                   int
	size                       int
	scroll                     string
	reindexerFunc              ReindexerFunc
	progress                   ReindexerProgressFunc
	statsOnly                  bool
}
type ReindexerFunc func(hit *SearchHit, bulkService *BulkService) error
func CopyToTargetIndex(targetIndex string) ReindexerFunc {
	return func(hit *SearchHit, bulkService *BulkService) error {
		source := make(map[string]interface{})
		if err := json.Unmarshal(*hit.Source, &source); err != nil {
			return err
		}
		req := NewBulkIndexRequest().Index(targetIndex).Type(hit.Type).Id(hit.Id).Doc(source)
		if hit.Parent != "" {
			req = req.Parent(hit.Parent)
		}
		if hit.Routing != "" {
			req = req.Routing(hit.Routing)
		}
		bulkService.Add(req)
		return nil
	}
}
type ReindexerProgressFunc func(current, total int64)
type ReindexerResponse struct {
	Success int64
	Failed  int64
	Errors  []*BulkResponseItem
}
func NewReindexer(client *Client, source string, reindexerFunc ReindexerFunc) *Reindexer {
	return &Reindexer{
		sourceClient:  client,
		sourceIndex:   source,
		reindexerFunc: reindexerFunc,
		statsOnly:     true,
	}
}
func (ix *Reindexer) TargetClient(c *Client) *Reindexer {
	ix.targetClient = c
	return ix
}
func (ix *Reindexer) Query(q Query) *Reindexer {
	ix.query = q
	return ix
}
func (ix *Reindexer) ScanFields(scanFields ...string) *Reindexer {
	ix.scanFields = scanFields
	return ix
}
func (ix *Reindexer) BulkSize(bulkSize int) *Reindexer {
	ix.bulkSize = bulkSize
	return ix
}
func (ix *Reindexer) Size(size int) *Reindexer {
	ix.size = size
	return ix
}
func (ix *Reindexer) Scroll(timeout string) *Reindexer {
	ix.scroll = timeout
	return ix
}
func (ix *Reindexer) Progress(f ReindexerProgressFunc) *Reindexer {
	ix.progress = f
	return ix
}
func (ix *Reindexer) StatsOnly(statsOnly bool) *Reindexer {
	ix.statsOnly = statsOnly
	return ix
}
func (ix *Reindexer) Do() (*ReindexerResponse, error) {
	if ix.sourceClient == nil {
		return nil, errors.New("no source client")
	}
	if ix.sourceIndex == "" {
		return nil, errors.New("no source index")
	}
	if ix.targetClient == nil {
		ix.targetClient = ix.sourceClient
	}
	if ix.scanFields == nil {
		ix.scanFields = []string{"_source", "_parent", "_routing"}
	}
	if ix.bulkSize <= 0 {
		ix.bulkSize = 500
	}
	if ix.scroll == "" {
		ix.scroll = "5m"
	}
	var err error
	var current, total int64
	if ix.progress != nil {
		total, err = ix.count()
		if err != nil {
			return nil, err
		}
	}
	scanner := ix.sourceClient.Scan(ix.sourceIndex).Scroll(ix.scroll).Fields(ix.scanFields...)
	if ix.query != nil {
		scanner = scanner.Query(ix.query)
	}
	if ix.size > 0 {
		scanner = scanner.Size(ix.size)
	}
	cursor, err := scanner.Do()
	bulk := ix.targetClient.Bulk()
	ret := &ReindexerResponse{
		Errors: make([]*BulkResponseItem, 0),
	}
	for {
		docs, err := cursor.Next()
		if err == EOS {
			break
		}
		if err != nil {
			return ret, err
		}
		if docs.TotalHits() > 0 {
			for _, hit := range docs.Hits.Hits {
				if ix.progress != nil {
					current++
					ix.progress(current, total)
				}
				err := ix.reindexerFunc(hit, bulk)
				if err != nil {
					return ret, err
				}
				if bulk.NumberOfActions() >= ix.bulkSize {
					bulk, err = ix.commit(bulk, ret)
					if err != nil {
						return ret, err
					}
				}
			}
		}
	}
	if bulk.NumberOfActions() > 0 {
		bulk, err = ix.commit(bulk, ret)
		if err != nil {
			return ret, err
		}
		bulk = nil
	}
	return ret, nil
}