Refactor SearchRequest, SearchSource and Reindex

This commit refactors the above structs and aligns them with the Java
source (see #955).

Furthermore, it fixes go vet errors in `example_test.go`.

Author: olivere

example_test.go

@@ -242,7 +242,7 @@ func Example() {
 	}
 }
 
-func ExampleClient_NewClient_default() {
+func ExampleNewClient_default() {
 	// Obtain a client to the Elasticsearch instance on http://127.0.0.1:9200.
 	client, err := elastic.NewClient()
 	if err != nil {
@@ -256,7 +256,7 @@ func ExampleClient_NewClient_default() {
 	// connected
 }
 
-func ExampleClient_NewClient_cluster() {
+func ExampleNewClient_cluster() {
 	// Obtain a client for an Elasticsearch cluster of two nodes,
 	// running on 10.0.1.1 and 10.0.1.2.
 	client, err := elastic.NewClient(elastic.SetURL("http://10.0.1.1:9200", "http://10.0.1.2:9200"))
@@ -267,7 +267,7 @@ func ExampleClient_NewClient_cluster() {
 	_ = client
 }
 
-func ExampleClient_NewClient_manyOptions() {
+func ExampleNewClient_manyOptions() {
 	// Obtain a client for an Elasticsearch cluster of two nodes,
 	// running on 10.0.1.1 and 10.0.1.2. Do not run the sniffer.
 	// Set the healthcheck interval to 10s. When requests fail,
@@ -287,7 +287,7 @@ func ExampleClient_NewClient_manyOptions() {
 	_ = client
 }
 
-func ExampleIndexExistsService() {
+func ExampleIndicesExistsService() {
 	// Get a client to the local Elasticsearch instance.
 	client, err := elastic.NewClient()
 	if err != nil {
@@ -305,7 +305,7 @@ func ExampleIndexExistsService() {
 	}
 }
 
-func ExampleCreateIndexService() {
+func ExampleIndicesCreateService() {
 	// Get a client to the local Elasticsearch instance.
 	client, err := elastic.NewClient()
 	if err != nil {
@@ -323,7 +323,7 @@ func ExampleCreateIndexService() {
 	}
 }
 
-func ExampleDeleteIndexService() {
+func ExampleIndicesDeleteService() {
 	// Get a client to the local Elasticsearch instance.
 	client, err := elastic.NewClient()
 	if err != nil {
@@ -508,7 +508,7 @@ func ExampleClusterHealthService() {
 	fmt.Printf("Cluster status is %q\n", res.Status)
 }
 
-func ExampleClusterHealthService_WaitForGreen() {
+func ExampleClusterHealthService_WaitForStatus() {
 	client, err := elastic.NewClient()
 	if err != nil {
 		panic(err)

reindex.go

@@ -6,6 +6,7 @@ package elastic
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/url"
 )
@@ -214,7 +215,7 @@ func (s *ReindexService) Validate() error {
 	if s.source == nil {
 		invalid = append(invalid, "Source")
 	} else {
-		if len(s.source.indices) == 0 {
+		if len(s.source.request.indices) == 0 {
 			invalid = append(invalid, "Source.Index")
 		}
 	}
@@ -353,85 +354,124 @@ func (s *ReindexService) DoAsync(ctx context.Context) (*StartTaskResult, error)
 
 // ReindexSource specifies the source of a Reindex process.
 type ReindexSource struct {
-	searchType   string // default in ES is "query_then_fetch"
-	indices      []string
-	types        []string
-	routing      *string
-	preference   *string
-	requestCache *bool
-	scroll       string
-	query        Query
-	sorts        []SortInfo
-	sorters      []Sorter
-	searchSource *SearchSource
-	remoteInfo   *ReindexRemoteInfo
+	searchType string // default in ES is "query_then_fetch"
+	request    *SearchRequest
+	/*
+		indices      []string
+		types        []string
+		routing      *string
+		preference   *string
+		requestCache *bool
+		scroll       string
+		query        Query
+		sorts        []SortInfo
+		sorters      []Sorter
+		searchSource *SearchSource
+	*/
+	remoteInfo *ReindexRemoteInfo
 }
 
 // NewReindexSource creates a new ReindexSource.
 func NewReindexSource() *ReindexSource {
-	return &ReindexSource{}
+	return &ReindexSource{
+		request: NewSearchRequest(),
+	}
+}
+
+// Request specifies the search request used for source.
+func (r *ReindexSource) Request(request *SearchRequest) *ReindexSource {
+	if request == nil {
+		r.request = NewSearchRequest()
+	} else {
+		r.request = request
+	}
+	return r
 }
 
 // SearchType is the search operation type. Possible values are
 // "query_then_fetch" and "dfs_query_then_fetch".
 func (r *ReindexSource) SearchType(searchType string) *ReindexSource {
-	r.searchType = searchType
+	r.request = r.request.SearchType(searchType)
 	return r
 }
 
 func (r *ReindexSource) SearchTypeDfsQueryThenFetch() *ReindexSource {
-	return r.SearchType("dfs_query_then_fetch")
+	r.request = r.request.SearchType("dfs_query_then_fetch")
+	return r
 }
 
 func (r *ReindexSource) SearchTypeQueryThenFetch() *ReindexSource {
-	return r.SearchType("query_then_fetch")
+	r.request = r.request.SearchType("query_then_fetch")
+	return r
 }
 
 func (r *ReindexSource) Index(indices ...string) *ReindexSource {
-	r.indices = append(r.indices, indices...)
+	r.request = r.request.Index(indices...)
 	return r
 }
 
 func (r *ReindexSource) Type(types ...string) *ReindexSource {
-	r.types = append(r.types, types...)
+	r.request = r.request.Type(types...)
 	return r
 }
 
 func (r *ReindexSource) Preference(preference string) *ReindexSource {
-	r.preference = &preference
+	r.request = r.request.Preference(preference)
 	return r
 }
 
 func (r *ReindexSource) RequestCache(requestCache bool) *ReindexSource {
-	r.requestCache = &requestCache
+	r.request = r.request.RequestCache(requestCache)
 	return r
 }
 
 func (r *ReindexSource) Scroll(scroll string) *ReindexSource {
-	r.scroll = scroll
+	r.request = r.request.Scroll(scroll)
 	return r
 }
 
 func (r *ReindexSource) Query(query Query) *ReindexSource {
-	r.query = query
+	r.request = r.request.Query(query)
 	return r
 }
 
 // Sort adds a sort order.
 func (r *ReindexSource) Sort(field string, ascending bool) *ReindexSource {
-	r.sorts = append(r.sorts, SortInfo{Field: field, Ascending: ascending})
+	r.request = r.request.Sort(field, ascending)
 	return r
 }
 
 // SortWithInfo adds a sort order.
 func (r *ReindexSource) SortWithInfo(info SortInfo) *ReindexSource {
-	r.sorts = append(r.sorts, info)
+	r.request = r.request.SortWithInfo(info)
 	return r
 }
 
 // SortBy adds a sort order.
 func (r *ReindexSource) SortBy(sorter ...Sorter) *ReindexSource {
-	r.sorters = append(r.sorters, sorter...)
+	r.request = r.request.SortBy(sorter...)
+	return r
+}
+
+// FetchSource indicates whether the response should contain the stored
+// _source for every hit.
+func (r *ReindexSource) FetchSource(fetchSource bool) *ReindexSource {
+	r.request = r.request.FetchSource(fetchSource)
+	return r
+}
+
+// FetchSourceIncludeExclude specifies that _source should be returned
+// with each hit, where "include" and "exclude" serve as a simple wildcard
+// matcher that gets applied to its fields
+// (e.g. include := []string{"obj1.*","obj2.*"}, exclude := []string{"description.*"}).
+func (r *ReindexSource) FetchSourceIncludeExclude(include, exclude []string) *ReindexSource {
+	r.request = r.request.FetchSourceIncludeExclude(include, exclude)
+	return r
+}
+
+// FetchSourceContext indicates how the _source should be fetched.
+func (r *ReindexSource) FetchSourceContext(fsc *FetchSourceContext) *ReindexSource {
+	r.request = r.request.FetchSourceContext(fsc)
 	return r
 }
 
@@ -443,84 +483,35 @@ func (r *ReindexSource) RemoteInfo(ri *ReindexRemoteInfo) *ReindexSource {
 
 // Source returns a serializable JSON request for the request.
 func (r *ReindexSource) Source() (interface{}, error) {
-	source := make(map[string]interface{})
-
-	if r.query != nil {
-		src, err := r.query.Source()
-		if err != nil {
-			return nil, err
-		}
-		source["query"] = src
-	} else if r.searchSource != nil {
-		src, err := r.searchSource.Source()
-		if err != nil {
-			return nil, err
-		}
-		source["source"] = src
+	src, err := r.request.sourceAsMap()
+	if err != nil {
+		return nil, err
 	}
-
-	if r.searchType != "" {
-		source["search_type"] = r.searchType
+	source, ok := src.(map[string]interface{})
+	if !ok {
+		return nil, errors.New("unable to use SearchRequest as map[string]interface{}")
 	}
 
-	switch len(r.indices) {
-	case 0:
+	switch len(r.request.indices) {
 	case 1:
-		source["index"] = r.indices[0]
+		source["index"] = r.request.indices[0]
 	default:
-		source["index"] = r.indices
+		source["index"] = r.request.indices
 	}
-
-	switch len(r.types) {
+	switch len(r.request.types) {
 	case 0:
 	case 1:
-		source["type"] = r.types[0]
+		source["type"] = r.request.types[0]
 	default:
-		source["type"] = r.types
-	}
-
-	if r.preference != nil && *r.preference != "" {
-		source["preference"] = *r.preference
+		source["type"] = r.request.types
 	}
-
-	if r.requestCache != nil {
-		source["request_cache"] = fmt.Sprintf("%v", *r.requestCache)
-	}
-
-	if r.scroll != "" {
-		source["scroll"] = r.scroll
-	}
-
 	if r.remoteInfo != nil {
 		src, err := r.remoteInfo.Source()
 		if err != nil {
 			return nil, err
 		}
 		source["remote"] = src
 	}
-
-	if len(r.sorters) > 0 {
-		var sortarr []interface{}
-		for _, sorter := range r.sorters {
-			src, err := sorter.Source()
-			if err != nil {
-				return nil, err
-			}
-			sortarr = append(sortarr, src)
-		}
-		source["sort"] = sortarr
-	} else if len(r.sorts) > 0 {
-		var sortarr []interface{}
-		for _, sort := range r.sorts {
-			src, err := sort.Source()
-			if err != nil {
-				return nil, err
-			}
-			sortarr = append(sortarr, src)
-		}
-		source["sort"] = sortarr
-	}
-
 	return source, nil
 }
 

reindex_test.go

@@ -260,15 +260,28 @@ func TestReindexSourceWithRouting(t *testing.T) {
 	}
 }
 
-func TestReindex(t *testing.T) {
-	client := setupTestClientAndCreateIndexAndAddDocs(t) // , SetTraceLog(log.New(os.Stdout, "", 0)))
-	esversion, err := client.ElasticsearchVersion(DefaultURL)
+func TestReindexSourceWithSourceFilter(t *testing.T) {
+	client := setupTestClient(t)
+	src := NewReindexSource().Index("twitter").
+		FetchSourceIncludeExclude([]string{"obj1.*", "obj2.*"}, []string{"*.description"})
+	dst := NewReindexDestination().Index("new_twitter")
+	out, err := client.Reindex().Source(src).Destination(dst).getBody()
 	if err != nil {
 		t.Fatal(err)
 	}
-	if esversion < "2.3.0" {
-		t.Skipf("Elasticsearch %v does not support Reindex API yet", esversion)
+	b, err := json.Marshal(out)
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := string(b)
+	want := `{"dest":{"index":"new_twitter"},"source":{"_source":{"excludes":["*.description"],"includes":["obj1.*","obj2.*"]},"index":"twitter"}}`
+	if got != want {
+		t.Fatalf("\ngot  %s\nwant %s", got, want)
 	}
+}
+
+func TestReindex(t *testing.T) {
+	client := setupTestClientAndCreateIndexAndAddDocs(t) // , SetTraceLog(log.New(os.Stdout, "", 0)))
 
 	sourceCount, err := client.Count(testIndexName).Do(context.TODO())
 	if err != nil {