更新测试用例

ES 2 and 5

.travis.yml

@@ -1,30 +1,31 @@
 language: go
 
+addons:
+  apt:
+    packages:
+      - oracle-java8-set-default
+
 go:
-  - 1.5.4
-  - 1.6.3
+  - 1.6.4
+  - 1.7.5
 
 env:
   global:
-    - GO15VENDOREXPERIMENT=1
+    - JAVA_HOME=/usr/lib/jvm/java-8-oracle
   matrix:
-    - ES_VERSION=1.0.3 GROOVY_VER=2.0.0
-    - ES_VERSION=1.1.2 GROOVY_VER=2.0.0
-    - ES_VERSION=1.2.1 GROOVY_VER=2.2.0
-    - ES_VERSION=1.3.4
-    - ES_VERSION=1.4.4
-    - ES_VERSION=1.5.2
-    - ES_VERSION=1.6.0
-    - ES_VERSION=1.7.0
+    - ES_VERSION=1.7.5 ES_URL=https://download.elastic.co/elasticsearch/elasticsearch/elasticsearch-1.7.5.tar.gz
+    - ES_VERSION=2.4.4 ES_URL=https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.4.4/elasticsearch-2.4.4.tar.gz
+    - ES_VERSION=5.2.0 ES_URL=https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-5.2.0.tar.gz
 
 before_script:
+  - java -version
+  - echo $JAVA_HOME
   - mkdir ${HOME}/elasticsearch
-  - wget https://download.elastic.co/elasticsearch/elasticsearch/elasticsearch-${ES_VERSION}.tar.gz
+  - wget $ES_URL
   - tar -xzf elasticsearch-${ES_VERSION}.tar.gz -C ${HOME}/elasticsearch
-  - "echo 'script.groovy.sandbox.enabled: true' >> ${HOME}/elasticsearch/elasticsearch-${ES_VERSION}/config/elasticsearch.yml"
-  - 'if [[ "${ES_VERSION}" < "1.3" ]]; then ${HOME}/elasticsearch/elasticsearch-${ES_VERSION}/bin/plugin --install elasticsearch/elasticsearch-lang-groovy/${GROOVY_VER}; fi'
-  - ${HOME}/elasticsearch/elasticsearch-${ES_VERSION}/bin/elasticsearch >/dev/null &
-  - sleep 10 # Wait for ES to start up
+  - "echo 'script.inline: true' >> ${HOME}/elasticsearch/elasticsearch-${ES_VERSION}/config/elasticsearch.yml"
+  - ${HOME}/elasticsearch/elasticsearch-${ES_VERSION}/bin/elasticsearch &
+  - wget --retry-connrefused http://127.0.0.1:9200/ # Wait for ES to start up
 
 install:
   - go get github.com/Masterminds/glide

Makefile

@@ -2,9 +2,14 @@ help:
 	@echo "Available targets:"
 	@echo "- test: run tests"
 	@echo "- deps: installs dependencies with glide"
+	@echo "- watch: watch for changes and re-run tests"
 
 deps:
-	glide up
+	glide install	&& mkdir -p vendor/bin && go build -o vendor/bin/ginkgo ./vendor/github.com/onsi/ginkgo/ginkgo
+
 
 test: deps
-	go test -i && go test
+	vendor/bin/ginkgo -race -randomizeAllSpecs -r -skipPackage vendor -progress .
+
+watch: deps
+	vendor/bin/ginkgo watch -race -randomizeAllSpecs -r -skipPackage vendor -progress -notify .

example_test.go

@@ -19,7 +19,7 @@ var (
 	ES_PORT = "9200"
 )
 
-func getConnection() (conn *goes.Connection) {
+func getClient() (conn *goes.Client) {
 	h := os.Getenv("TEST_ELASTICSEARCH_HOST")
 	if h == "" {
 		h = ES_HOST
@@ -30,13 +30,13 @@ func getConnection() (conn *goes.Connection) {
 		p = ES_PORT
 	}
 
-	conn = goes.NewConnection(h, p)
+	conn = goes.NewClient(h, p)
 
 	return
 }
 
-func ExampleConnection_CreateIndex() {
-	conn := getConnection()
+func ExampleClient_CreateIndex() {
+	conn := getClient()
 
 	mapping := map[string]interface{}{
 		"settings": map[string]interface{}{
@@ -64,8 +64,8 @@ func ExampleConnection_CreateIndex() {
 	fmt.Printf("%s", resp)
 }
 
-func ExampleConnection_DeleteIndex() {
-	conn := getConnection()
+func ExampleClient_DeleteIndex() {
+	conn := getClient()
 	resp, err := conn.DeleteIndex("yourinde")
 
 	if err != nil {
@@ -75,8 +75,8 @@ func ExampleConnection_DeleteIndex() {
 	fmt.Printf("%s", resp)
 }
 
-func ExampleConnection_RefreshIndex() {
-	conn := getConnection()
+func ExampleClient_RefreshIndex() {
+	conn := getClient()
 	resp, err := conn.RefreshIndex("yourindex")
 
 	if err != nil {
@@ -86,8 +86,8 @@ func ExampleConnection_RefreshIndex() {
 	fmt.Printf("%s", resp)
 }
 
-func ExampleConnection_Search() {
-	conn := getConnection()
+func ExampleClient_Search() {
+	conn := getClient()
 
 	var query = map[string]interface{}{
 		"query": map[string]interface{}{
@@ -123,8 +123,8 @@ func ExampleConnection_Search() {
 	fmt.Printf("%s", searchResults)
 }
 
-func ExampleConnection_Index() {
-	conn := getConnection()
+func ExampleClient_Index() {
+	conn := getClient()
 
 	d := goes.Document{
 		Index: "twitter",
@@ -147,15 +147,15 @@ func ExampleConnection_Index() {
 	fmt.Printf("%s", response)
 }
 
-func ExampleConnection_Delete() {
-	conn := getConnection()
+func ExampleClient_Delete() {
+	conn := getClient()
 
 	//[create index, index document ...]
 
 	d := goes.Document{
 		Index: "twitter",
 		Type:  "tweet",
-		Id:    "1",
+		ID:    "1",
 		Fields: map[string]interface{}{
 			"user": "foo",
 		},
@@ -169,15 +169,15 @@ func ExampleConnection_Delete() {
 	fmt.Printf("%s", response)
 }
 
-func ExampleConnectionOverrideHttpClient() {
+func ExampleClient_WithHTTPClient() {
 	tr := &http.Transport{
 		ResponseHeaderTimeout: 1 * time.Second,
 	}
 	cl := &http.Client{
 		Transport: tr,
 	}
-	conn := getConnection()
-	conn.WithClient(cl)
+	conn := getClient()
+	conn.WithHTTPClient(cl)
 
 	fmt.Printf("%v\n", conn.Client)
 }

glide.lock

@@ -1,6 +1,12 @@
-hash: 7e39b5bd354c3f0770ca3a9d28a74a0523695d18effa1e540f006b325defcfe7
-updated: 2016-07-26T14:50:32.597335795-05:00
+hash: 02db47097959405b1a7e0e1e583c6fbb11c7236c450264909a4e9ac690ef4d47
+updated: 2016-09-27T11:54:53.427061181-05:00
 imports: []
 testImports:
 - name: github.com/go-check/check
   version: 4f90aeace3a26ad7021961c297b22c42160c7b25
+- name: github.com/onsi/ginkgo
+  version: 462326b1628e124b23f42e87a8f2750e3c4e2d24
+  subpackages:
+  - ginkgo
+- name: github.com/onsi/gomega
+  version: a78ae492d53aad5a7a232d0d0462c14c400e3ee7

glide.yaml

@@ -3,3 +3,9 @@ import: []
 testImport:
 - package: github.com/go-check/check
   version: v1
+- package: github.com/onsi/ginkgo
+  version: ^1.2.0
+  subpackages:
+  - ginkgo
+- package: github.com/onsi/gomega
+  version: ^1.0.0

goes.go

@@ -19,118 +19,144 @@ import (
 )
 
 const (
-	BULK_COMMAND_INDEX  = "index"
-	BULK_COMMAND_DELETE = "delete"
+	// BulkCommandIndex specifies a bulk doc should be indexed
+	BulkCommandIndex = "index"
+	// BulkCommandDelete specifies a bulk doc should be deleted
+	BulkCommandDelete = "delete"
 )
 
 func (err *SearchError) Error() string {
 	return fmt.Sprintf("[%d] %s", err.StatusCode, err.Msg)
 }
 
-// NewConnection initiates a new Connection to an elasticsearch server
+// NewClient initiates a new client for an elasticsearch server
 //
 // This function is pretty useless for now but might be useful in a near future
 // if wee need more features like connection pooling or load balancing.
-func NewConnection(host string, port string) *Connection {
-	return &Connection{host, port, http.DefaultClient}
+func NewClient(host string, port string) *Client {
+	return &Client{host, port, http.DefaultClient, "", "", ""}
 }
 
-func (c *Connection) WithClient(cl *http.Client) *Connection {
+// WithHTTPClient sets the http.Client to be used with the connection. Returns the original client.
+func (c *Client) WithHTTPClient(cl *http.Client) *Client {
 	c.Client = cl
 	return c
 }
 
+// Version returns the detected version of the connected ES server
+func (c *Client) Version() (string, error) {
+	// Use cached version if it was already fetched
+	if c.version != "" {
+		return c.version, nil
+	}
+
+	// Get the version if it was not cached
+	r := Request{Method: "GET"}
+	res, err := c.Do(&r)
+	if err != nil {
+		return "", err
+	}
+	if version, ok := res.Raw["version"].(map[string]interface{}); ok {
+		if number, ok := version["number"].(string); ok {
+			c.version = number
+			return number, nil
+		}
+	}
+	return "", errors.New("No version returned by ElasticSearch Server")
+}
+
 // CreateIndex creates a new index represented by a name and a mapping
-func (c *Connection) CreateIndex(name string, mapping interface{}) (*Response, error) {
+func (c *Client) CreateIndex(name string, mapping interface{}) (*Response, error) {
 	r := Request{
-		Conn:      c,
 		Query:     mapping,
 		IndexList: []string{name},
 		Method:    "PUT",
 	}
 
-	return r.Run()
+	return c.Do(&r)
 }
 
 // DeleteIndex deletes an index represented by a name
-func (c *Connection) DeleteIndex(name string) (*Response, error) {
+func (c *Client) DeleteIndex(name string) (*Response, error) {
 	r := Request{
-		Conn:      c,
 		IndexList: []string{name},
 		Method:    "DELETE",
 	}
 
-	return r.Run()
+	return c.Do(&r)
 }
 
 // RefreshIndex refreshes an index represented by a name
-func (c *Connection) RefreshIndex(name string) (*Response, error) {
+func (c *Client) RefreshIndex(name string) (*Response, error) {
 	r := Request{
-		Conn:      c,
 		IndexList: []string{name},
 		Method:    "POST",
 		API:       "_refresh",
 	}
 
-	return r.Run()
+	return c.Do(&r)
 }
 
 // UpdateIndexSettings updates settings for existing index represented by a name and a settings
 // as described here: https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-update-settings.html
-func (c *Connection) UpdateIndexSettings(name string, settings interface{}) (*Response, error) {
+func (c *Client) UpdateIndexSettings(name string, settings interface{}) (*Response, error) {
 	r := Request{
-		Conn:      c,
 		Query:     settings,
 		IndexList: []string{name},
 		Method:    "PUT",
 		API:       "_settings",
 	}
 
-	return r.Run()
+	return c.Do(&r)
 }
 
 // Optimize an index represented by a name, extra args are also allowed please check:
 // http://www.elasticsearch.org/guide/en/elasticsearch/reference/current/indices-optimize.html#indices-optimize
-func (c *Connection) Optimize(indexList []string, extraArgs url.Values) (*Response, error) {
+func (c *Client) Optimize(indexList []string, extraArgs url.Values) (*Response, error) {
 	r := Request{
-		Conn:      c,
 		IndexList: indexList,
 		ExtraArgs: extraArgs,
 		Method:    "POST",
 		API:       "_optimize",
 	}
+	if version, _ := c.Version(); version > "2.1" {
+		r.API = "_forcemerge"
+	}
 
-	return r.Run()
+	return c.Do(&r)
+}
+
+// ForceMerge is the same as Optimize, but matches the naming of the endpoint as of ES 2.1.0
+func (c *Client) ForceMerge(indexList []string, extraArgs url.Values) (*Response, error) {
+	return c.Optimize(indexList, extraArgs)
 }
 
 // Stats fetches statistics (_stats) for the current elasticsearch server
-func (c *Connection) Stats(indexList []string, extraArgs url.Values) (*Response, error) {
+func (c *Client) Stats(indexList []string, extraArgs url.Values) (*Response, error) {
 	r := Request{
-		Conn:      c,
 		IndexList: indexList,
 		ExtraArgs: extraArgs,
 		Method:    "GET",
 		API:       "_stats",
 	}
 
-	return r.Run()
+	return c.Do(&r)
 }
 
 // IndexStatus fetches the status (_status) for the indices defined in
 // indexList. Use _all in indexList to get stats for all indices
-func (c *Connection) IndexStatus(indexList []string) (*Response, error) {
+func (c *Client) IndexStatus(indexList []string) (*Response, error) {
 	r := Request{
-		Conn:      c,
 		IndexList: indexList,
 		Method:    "GET",
 		API:       "_status",
 	}
 
-	return r.Run()
+	return c.Do(&r)
 }
 
-// Bulk adds multiple documents in bulk mode
-func (c *Connection) BulkSend(documents []Document) (*Response, error) {
+// BulkSend bulk adds multiple documents in bulk mode
+func (c *Client) BulkSend(documents []Document) (*Response, error) {
 	// We do not generate a traditional JSON here (often a one liner)
 	// Elasticsearch expects one line of JSON per line (EOL = \n)
 	// plus an extra \n at the very end of the document
@@ -149,7 +175,7 @@ func (c *Connection) BulkSend(documents []Document) (*Response, error) {
 
 	// len(documents) * 2 : action + optional_sources
 	// + 1 : room for the trailing \n
-	bulkData := make([][]byte, len(documents)*2+1)
+	bulkData := make([][]byte, 0, len(documents)*2+1)
 	i := 0
 
 	for _, doc := range documents {
@@ -157,7 +183,7 @@ func (c *Connection) BulkSend(documents []Document) (*Response, error) {
 			doc.BulkCommand: map[string]interface{}{
 				"_index": doc.Index,
 				"_type":  doc.Type,
-				"_id":    doc.Id,
+				"_id":    doc.ID,
 			},
 		})
 
@@ -165,7 +191,7 @@ func (c *Connection) BulkSend(documents []Document) (*Response, error) {
 			return &Response{}, err
 		}
 
-		bulkData[i] = action
+		bulkData = append(bulkData, action)
 		i++
 
 		if doc.Fields != nil {
@@ -191,28 +217,42 @@ func (c *Connection) BulkSend(documents []Document) (*Response, error) {
 				return &Response{}, err
 			}
 
-			bulkData[i] = sources
+			bulkData = append(bulkData, sources)
 			i++
 		}
 	}
 
 	// forces an extra trailing \n absolutely necessary for elasticsearch
-	bulkData[len(bulkData)-1] = []byte(nil)
+	bulkData = append(bulkData, []byte(nil))
 
 	r := Request{
-		Conn:     c,
 		Method:   "POST",
 		API:      "_bulk",
 		BulkData: bytes.Join(bulkData, []byte("\n")),
 	}
 
-	return r.Run()
+	resp, err := c.Do(&r)
+	if err != nil {
+		return resp, err
+	}
+
+	if resp.Errors {
+		for _, item := range resp.Items {
+			for _, i := range item {
+				if i.Error != "" {
+					return resp, &SearchError{i.Error, i.Status}
+				}
+			}
+		}
+		return resp, &SearchError{Msg: "Unknown error while bulk indexing"}
+	}
+
+	return resp, err
 }
 
 // Search executes a search query against an index
-func (c *Connection) Search(query interface{}, indexList []string, typeList []string, extraArgs url.Values) (*Response, error) {
+func (c *Client) Search(query interface{}, indexList []string, typeList []string, extraArgs url.Values) (*Response, error) {
 	r := Request{
-		Conn:      c,
 		Query:     query,
 		IndexList: indexList,
 		TypeList:  typeList,
@@ -221,13 +261,12 @@ func (c *Connection) Search(query interface{}, indexList []string, typeList []st
 		ExtraArgs: extraArgs,
 	}
 
-	return r.Run()
+	return c.Do(&r)
 }
 
 // Count executes a count query against an index, use the Count field in the response for the result
-func (c *Connection) Count(query interface{}, indexList []string, typeList []string, extraArgs url.Values) (*Response, error) {
+func (c *Client) Count(query interface{}, indexList []string, typeList []string, extraArgs url.Values) (*Response, error) {
 	r := Request{
-		Conn:      c,
 		Query:     query,
 		IndexList: indexList,
 		TypeList:  typeList,
@@ -236,15 +275,14 @@ func (c *Connection) Count(query interface{}, indexList []string, typeList []str
 		ExtraArgs: extraArgs,
 	}
 
-	return r.Run()
+	return c.Do(&r)
 }
 
 //Query runs a query against an index using the provided http method.
 //This method can be used to execute a delete by query, just pass in "DELETE"
 //for the HTTP method.
-func (c *Connection) Query(query interface{}, indexList []string, typeList []string, httpMethod string, extraArgs url.Values) (*Response, error) {
+func (c *Client) Query(query interface{}, indexList []string, typeList []string, httpMethod string, extraArgs url.Values) (*Response, error) {
 	r := Request{
-		Conn:      c,
 		Query:     query,
 		IndexList: indexList,
 		TypeList:  typeList,
@@ -253,18 +291,56 @@ func (c *Connection) Query(query interface{}, indexList []string, typeList []str
 		ExtraArgs: extraArgs,
 	}
 
-	return r.Run()
+	return c.Do(&r)
 }
 
-// Scan starts scroll over an index
-func (c *Connection) Scan(query interface{}, indexList []string, typeList []string, timeout string, size int) (*Response, error) {
+// DeleteByQuery deletes documents matching the specified query. It will return an error for ES 2.x,
+// because delete by query support was removed in those versions.
+func (c *Client) DeleteByQuery(query interface{}, indexList []string, typeList []string, extraArgs url.Values) (*Response, error) {
+	version, err := c.Version()
+	if err != nil {
+		return nil, err
+	}
+	if version > "2" && version < "5" {
+		return nil, errors.New("ElasticSearch 2.x does not support delete by query")
+	}
+
+	r := Request{
+		Query:     query,
+		IndexList: indexList,
+		TypeList:  typeList,
+		Method:    "DELETE",
+		API:       "_query",
+		ExtraArgs: extraArgs,
+	}
+
+	if version > "5" {
+		r.API = "_delete_by_query"
+		r.Method = "POST"
+	}
+
+	return c.Do(&r)
+}
+
+// Scan starts scroll over an index.
+// For ES versions < 5.x, it uses search_type=scan; for 5.x it uses sort=_doc. This means that data
+// will  be returned in the initial response for 5.x versions, but not for older versions. Code
+// wishing to be compatible with both should be written to handle either case.
+func (c *Client) Scan(query interface{}, indexList []string, typeList []string, timeout string, size int) (*Response, error) {
 	v := url.Values{}
+	version, err := c.Version()
+	if err != nil {
+		return nil, err
+	}
+	if version > "5" {
+		v.Add("sort", "_doc")
+	} else {
 		v.Add("search_type", "scan")
+	}
 	v.Add("scroll", timeout)
 	v.Add("size", strconv.Itoa(size))
 
 	r := Request{
-		Conn:      c,
 		Query:     query,
 		IndexList: indexList,
 		TypeList:  typeList,
@@ -273,44 +349,51 @@ func (c *Connection) Scan(query interface{}, indexList []string, typeList []stri
 		ExtraArgs: v,
 	}
 
-	return r.Run()
+	return c.Do(&r)
 }
 
 // Scroll fetches data by scroll id
-func (c *Connection) Scroll(scrollId string, timeout string) (*Response, error) {
-	v := url.Values{}
-	v.Add("scroll", timeout)
-
+func (c *Client) Scroll(scrollID string, timeout string) (*Response, error) {
 	r := Request{
-		Conn:      c,
 		Method: "POST",
 		API:    "_search/scroll",
-		ExtraArgs: v,
-		Body:      []byte(scrollId),
 	}
 
-	return r.Run()
+	if version, err := c.Version(); err != nil {
+		return nil, err
+	} else if version > "2" {
+		r.Body, err = json.Marshal(map[string]string{"scroll": timeout, "scroll_id": scrollID})
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		v := url.Values{}
+		v.Add("scroll", timeout)
+		v.Add("scroll_id", scrollID)
+
+		r.ExtraArgs = v
+	}
+
+	return c.Do(&r)
 }
 
 // Get a typed document by its id
-func (c *Connection) Get(index string, documentType string, id string, extraArgs url.Values) (*Response, error) {
+func (c *Client) Get(index string, documentType string, id string, extraArgs url.Values) (*Response, error) {
 	r := Request{
-		Conn:      c,
 		IndexList: []string{index},
 		Method:    "GET",
 		API:       documentType + "/" + id,
 		ExtraArgs: extraArgs,
 	}
 
-	return r.Run()
+	return c.Do(&r)
 }
 
 // Index indexes a Document
 // The extraArgs is a list of url.Values that you can send to elasticsearch as
 // URL arguments, for example, to control routing, ttl, version, op_type, etc.
-func (c *Connection) Index(d Document, extraArgs url.Values) (*Response, error) {
+func (c *Client) Index(d Document, extraArgs url.Values) (*Response, error) {
 	r := Request{
-		Conn:      c,
 		Query:     d.Fields,
 		IndexList: []string{d.Index.(string)},
 		TypeList:  []string{d.Type},
@@ -318,138 +401,27 @@ func (c *Connection) Index(d Document, extraArgs url.Values) (*Response, error)
 		Method:    "POST",
 	}
 
-	if d.Id != nil {
+	if d.ID != nil {
 		r.Method = "PUT"
-		r.ID = d.Id.(string)
+		r.ID = d.ID.(string)
 	}
 
-	return r.Run()
+	return c.Do(&r)
 }
 
 // Delete deletes a Document d
 // The extraArgs is a list of url.Values that you can send to elasticsearch as
 // URL arguments, for example, to control routing.
-func (c *Connection) Delete(d Document, extraArgs url.Values) (*Response, error) {
+func (c *Client) Delete(d Document, extraArgs url.Values) (*Response, error) {
 	r := Request{
-		Conn:      c,
 		IndexList: []string{d.Index.(string)},
 		TypeList:  []string{d.Type},
 		ExtraArgs: extraArgs,
 		Method:    "DELETE",
-		ID:        d.Id.(string),
+		ID:        d.ID.(string),
 	}
 
-	return r.Run()
-}
-
-// Run executes an elasticsearch Request. It converts data to Json, sends the
-// request and returns the Response obtained
-func (req *Request) Run() (*Response, error) {
-	body, statusCode, err := req.run()
-	esResp := &Response{Status: statusCode}
-
-	if err != nil {
-		return esResp, err
-	}
-
-	if req.Method != "HEAD" {
-		err = json.Unmarshal(body, &esResp)
-		if err != nil {
-			return esResp, err
-		}
-		err = json.Unmarshal(body, &esResp.Raw)
-		if err != nil {
-			return esResp, err
-		}
-	}
-
-	if req.API == "_bulk" && esResp.Errors {
-		for _, item := range esResp.Items {
-			for _, i := range item {
-				if i.Error != "" {
-					return esResp, &SearchError{i.Error, i.Status}
-				}
-			}
-		}
-		return esResp, &SearchError{Msg: "Unknown error while bulk indexing"}
-	}
-
-	if esResp.Error != "" {
-		return esResp, &SearchError{esResp.Error, esResp.Status}
-	}
-
-	return esResp, nil
-}
-
-func (req *Request) run() ([]byte, uint64, error) {
-	postData := []byte{}
-
-	// XXX : refactor this
-	if len(req.Body) > 0 {
-		postData = req.Body
-	} else if req.API == "_bulk" {
-		postData = req.BulkData
-	} else {
-		b, err := json.Marshal(req.Query)
-		if err != nil {
-			return nil, 0, err
-		}
-		postData = b
-	}
-
-	reader := bytes.NewReader(postData)
-
-	newReq, err := http.NewRequest(req.Method, req.Url(), reader)
-	if err != nil {
-		return nil, 0, err
-	}
-
-	if req.Method == "POST" || req.Method == "PUT" {
-		newReq.Header.Set("Content-Type", "application/json")
-	}
-
-	resp, err := req.Conn.Client.Do(newReq)
-	if err != nil {
-		return nil, 0, err
-	}
-
-	defer resp.Body.Close()
-
-	body, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		return nil, uint64(resp.StatusCode), err
-	}
-
-	if resp.StatusCode > 201 && resp.StatusCode < 400 {
-		return nil, uint64(resp.StatusCode), errors.New(string(body))
-	}
-
-	return body, uint64(resp.StatusCode), nil
-}
-
-// Url builds a Request for a URL
-func (r *Request) Url() string {
-	path := "/" + strings.Join(r.IndexList, ",")
-
-	if len(r.TypeList) > 0 {
-		path += "/" + strings.Join(r.TypeList, ",")
-	}
-
-	// XXX : for indexing documents using the normal (non bulk) API
-	if len(r.ID) > 0 {
-		path += "/" + r.ID
-	}
-
-	path += "/" + r.API
-
-	u := url.URL{
-		Scheme:   "http",
-		Host:     fmt.Sprintf("%s:%s", r.Conn.Host, r.Conn.Port),
-		Path:     path,
-		RawQuery: r.ExtraArgs.Encode(),
-	}
-
-	return u.String()
+	return c.Do(&r)
 }
 
 // Buckets returns list of buckets in aggregation
@@ -478,54 +450,51 @@ func (b Bucket) DocCount() uint64 {
 func (b Bucket) Aggregation(name string) Aggregation {
 	if agg, ok := b[name]; ok {
 		return agg.(map[string]interface{})
-	} else {
-		return Aggregation{}
 	}
+	return Aggregation{}
 }
 
 // PutMapping registers a specific mapping for one or more types in one or more indexes
-func (c *Connection) PutMapping(typeName string, mapping interface{}, indexes []string) (*Response, error) {
+func (c *Client) PutMapping(typeName string, mapping interface{}, indexes []string) (*Response, error) {
 
 	r := Request{
-		Conn:      c,
 		Query:     mapping,
 		IndexList: indexes,
 		Method:    "PUT",
 		API:       "_mappings/" + typeName,
 	}
 
-	return r.Run()
+	return c.Do(&r)
 }
 
-func (c *Connection) GetMapping(types []string, indexes []string) (*Response, error) {
+// GetMapping returns the mappings for the specified types
+func (c *Client) GetMapping(types []string, indexes []string) (*Response, error) {
 
 	r := Request{
-		Conn:      c,
 		IndexList: indexes,
 		Method:    "GET",
 		API:       "_mapping/" + strings.Join(types, ","),
 	}
 
-	return r.Run()
+	return c.Do(&r)
 }
 
 // IndicesExist checks whether index (or indices) exist on the server
-func (c *Connection) IndicesExist(indexes []string) (bool, error) {
+func (c *Client) IndicesExist(indexes []string) (bool, error) {
 
 	r := Request{
-		Conn:      c,
 		IndexList: indexes,
 		Method:    "HEAD",
 	}
 
-	resp, err := r.Run()
+	resp, err := c.Do(&r)
 
 	return resp.Status == 200, err
 }
 
-func (c *Connection) Update(d Document, query interface{}, extraArgs url.Values) (*Response, error) {
+// Update updates the specified document using the _update endpoint
+func (c *Client) Update(d Document, query interface{}, extraArgs url.Values) (*Response, error) {
 	r := Request{
-		Conn:      c,
 		Query:     query,
 		IndexList: []string{d.Index.(string)},
 		TypeList:  []string{d.Type},
@@ -534,29 +503,33 @@ func (c *Connection) Update(d Document, query interface{}, extraArgs url.Values)
 		API:       "_update",
 	}
 
-	if d.Id != nil {
-		r.ID = d.Id.(string)
+	if d.ID != nil {
+		r.ID = d.ID.(string)
 	}
 
-	return r.Run()
+	return c.Do(&r)
 }
 
 // DeleteMapping deletes a mapping along with all data in the type
-func (c *Connection) DeleteMapping(typeName string, indexes []string) (*Response, error) {
+func (c *Client) DeleteMapping(typeName string, indexes []string) (*Response, error) {
+	if version, err := c.Version(); err != nil {
+		return nil, err
+	} else if version > "2" {
+		return nil, errors.New("Deletion of mappings is not supported in ES 2.x and above.")
+	}
 
 	r := Request{
-		Conn:      c,
 		IndexList: indexes,
 		Method:    "DELETE",
 		API:       "_mappings/" + typeName,
 	}
 
-	return r.Run()
+	return c.Do(&r)
 }
 
-func (c *Connection) modifyAlias(action string, alias string, indexes []string) (*Response, error) {
+func (c *Client) modifyAlias(action string, alias string, indexes []string) (*Response, error) {
 	command := map[string]interface{}{
-		"actions": make([]map[string]interface{}, 1),
+		"actions": make([]map[string]interface{}, 0, 1),
 	}
 
 	for _, index := range indexes {
@@ -569,35 +542,116 @@ func (c *Connection) modifyAlias(action string, alias string, indexes []string)
 	}
 
 	r := Request{
-		Conn:   c,
 		Query:  command,
 		Method: "POST",
 		API:    "_aliases",
 	}
 
-	return r.Run()
+	return c.Do(&r)
 }
 
 // AddAlias creates an alias to one or more indexes
-func (c *Connection) AddAlias(alias string, indexes []string) (*Response, error) {
+func (c *Client) AddAlias(alias string, indexes []string) (*Response, error) {
 	return c.modifyAlias("add", alias, indexes)
 }
 
 // RemoveAlias removes an alias to one or more indexes
-func (c *Connection) RemoveAlias(alias string, indexes []string) (*Response, error) {
+func (c *Client) RemoveAlias(alias string, indexes []string) (*Response, error) {
 	return c.modifyAlias("remove", alias, indexes)
 }
 
 // AliasExists checks whether alias is defined on the server
-func (c *Connection) AliasExists(alias string) (bool, error) {
+func (c *Client) AliasExists(alias string) (bool, error) {
 
 	r := Request{
-		Conn:   c,
 		Method: "HEAD",
 		API:    "_alias/" + alias,
 	}
 
-	resp, err := r.Run()
+	resp, err := c.Do(&r)
 
 	return resp.Status == 200, err
 }
+
+func (c *Client) replaceHost(req *http.Request) {
+	req.URL.Scheme = "http"
+	req.URL.Host = fmt.Sprintf("%s:%s", c.Host, c.Port)
+}
+
+// DoRaw Does the provided requeset and returns the raw bytes and the status code of the response
+func (c *Client) DoRaw(r Requester) ([]byte, uint64, error) {
+	req, err := r.Request()
+	if err != nil {
+		return nil, 0, err
+	}
+	c.replaceHost(req)
+
+	if c.AuthUsername != "" {
+		req.SetBasicAuth(c.AuthUsername, c.AuthPassword)
+	}
+
+	return c.doRequest(req)
+}
+
+// Do runs the request returned by the requestor and returns the parsed response
+func (c *Client) Do(r Requester) (*Response, error) {
+	req, err := r.Request()
+	if err != nil {
+		return &Response{}, err
+	}
+	c.replaceHost(req)
+
+	if c.AuthUsername != "" {
+		req.SetBasicAuth(c.AuthUsername, c.AuthPassword)
+	}
+
+	body, statusCode, err := c.doRequest(req)
+	esResp := &Response{Status: statusCode}
+
+	if err != nil {
+		return esResp, err
+	}
+
+	if req.Method != "HEAD" {
+		err = json.Unmarshal(body, &esResp)
+		if err != nil {
+			return esResp, err
+		}
+		err = json.Unmarshal(body, &esResp.Raw)
+		if err != nil {
+			return esResp, err
+		}
+	}
+
+	if len(esResp.RawError) > 0 && esResp.RawError[0] == '"' {
+		json.Unmarshal(esResp.RawError, &esResp.Error)
+	} else {
+		esResp.Error = string(esResp.RawError)
+	}
+	esResp.RawError = nil
+
+	if esResp.Error != "" {
+		return esResp, &SearchError{esResp.Error, esResp.Status}
+	}
+
+	return esResp, nil
+}
+
+func (c *Client) doRequest(req *http.Request) ([]byte, uint64, error) {
+	resp, err := c.Client.Do(req)
+	if err != nil {
+		return nil, 0, err
+	}
+	defer resp.Body.Close()
+
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, uint64(resp.StatusCode), err
+	}
+
+	if resp.StatusCode > 201 && resp.StatusCode < 400 {
+		return nil, uint64(resp.StatusCode), errors.New(string(body))
+	}
+
+	return body, uint64(resp.StatusCode), nil
+}

goes_suite_test.go

@@ -0,0 +1,13 @@
+package goes_test
+
+import (
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"testing"
+)
+
+func TestGoes(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Goes Suite")
+}

request.go

@@ -0,0 +1,113 @@
+package goes
+
+import (
+	"bytes"
+	"encoding/json"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"strings"
+)
+
+// Requester implements Request which builds an HTTP request for Elasticsearch
+type Requester interface {
+	// Request should set the URL and Body (if needed). The host of the URL will be overwritten by the client.
+	Request() (*http.Request, error)
+}
+
+// Request holds a single request to elasticsearch
+type Request struct {
+	// A search query
+	Query interface{}
+
+	// Which index to search into
+	IndexList []string
+
+	// Which type to search into
+	TypeList []string
+
+	// HTTP Method to user (GET, POST ...)
+	Method string
+
+	// Which api keyword (_search, _bulk, etc) to use
+	API string
+
+	// Bulk data
+	BulkData []byte
+
+	// Request body
+	Body []byte
+
+	// A list of extra URL arguments
+	ExtraArgs url.Values
+
+	// Used for the id field when indexing a document
+	ID string
+
+	// Auth username
+	AuthUsername string
+
+	// Auth password
+	AuthPassword string
+}
+
+// URL builds a URL for a Request
+func (req *Request) URL() *url.URL {
+	var path string
+	if len(req.IndexList) > 0 {
+		path = "/" + strings.Join(req.IndexList, ",")
+	}
+
+	if len(req.TypeList) > 0 {
+		path += "/" + strings.Join(req.TypeList, ",")
+	}
+
+	// XXX : for indexing documents using the normal (non bulk) API
+	if len(req.ID) > 0 {
+		path += "/" + req.ID
+	}
+
+	path += "/" + req.API
+
+	u := url.URL{
+		//Scheme:   "http",
+		//Host:     fmt.Sprintf("%s:%s", req.Conn.Host, req.Conn.Port),
+		Path:     path,
+		RawQuery: req.ExtraArgs.Encode(),
+	}
+
+	return &u
+}
+
+// Request generates an http.Request based on the contents of the Request struct
+func (req *Request) Request() (*http.Request, error) {
+	postData := []byte{}
+
+	// XXX : refactor this
+	if len(req.Body) > 0 {
+		postData = req.Body
+	} else if req.API == "_bulk" {
+		postData = req.BulkData
+	} else if req.Query != nil {
+		b, err := json.Marshal(req.Query)
+		if err != nil {
+			return nil, err
+		}
+		postData = b
+	}
+
+	newReq, err := http.NewRequest(req.Method, "", nil)
+	if err != nil {
+		return nil, err
+	}
+	newReq.URL = req.URL()
+	newReq.Body = ioutil.NopCloser(bytes.NewReader(postData))
+	newReq.ContentLength = int64(len(postData))
+
+	if req.Method == "POST" || req.Method == "PUT" {
+		newReq.Header.Set("Content-Type", "application/json")
+	}
+	return newReq, nil
+}
+
+var _ Requester = (*Request)(nil)

structs.go

@@ -5,12 +5,12 @@
 package goes
 
 import (
+	"encoding/json"
 	"net/http"
-	"net/url"
 )
 
-// Represents a Connection object to elasticsearch
-type Connection struct {
+// Client represents a connection to elasticsearch
+type Client struct {
 	// The host to connect to
 	Host string
 
@@ -20,45 +20,22 @@ type Connection struct {
 	// Client is the http client used to make requests, allowing settings things
 	// such as timeouts etc
 	Client *http.Client
+
+	// Detected version of ES
+	version string
+
+	// user name for http basic auth
+	AuthUsername string `json:"username"`
+
+	// pass word  for http basic auth
+	AuthPassword string `json:"password"`
 }
 
-// Represents a Request to elasticsearch
-type Request struct {
-	// Which connection will be used
-	Conn *Connection
-
-	// A search query
-	Query interface{}
-
-	// Which index to search into
-	IndexList []string
-
-	// Which type to search into
-	TypeList []string
-
-	// HTTP Method to user (GET, POST ...)
-	Method string
-
-	// Which api keyword (_search, _bulk, etc) to use
-	API string
-
-	// Bulk data
-	BulkData []byte
-
-	// Request body
-	Body []byte
-
-	// A list of extra URL arguments
-	ExtraArgs url.Values
-
-	// Used for the id field when indexing a document
-	ID string
-}
-
-// Represents a Response from elasticsearch
+// Response holds an elasticsearch response
 type Response struct {
 	Acknowledged bool
 	Error        string
+	RawError     json.RawMessage `json:"error"`
 	Errors       bool
 	Status       uint64
 	Took         uint64
@@ -66,7 +43,7 @@ type Response struct {
 	Shards       Shard `json:"_shards"`
 	Hits         Hits
 	Index        string `json:"_index"`
-	Id           string `json:"_id"`
+	ID           string `json:"_id"`
 	Type         string `json:"_type"`
 	Version      int    `json:"_version"`
 	Found        bool
@@ -86,75 +63,77 @@ type Response struct {
 	Indices map[string]IndexStatus
 
 	// Scroll id for iteration
-	ScrollId string `json:"_scroll_id"`
+	ScrollID string `json:"_scroll_id"`
 
 	Aggregations map[string]Aggregation `json:"aggregations,omitempty"`
 
 	Raw map[string]interface{}
 }
 
-// Represents an aggregation from response
+// Aggregation holds the aggregation portion of an ES response
 type Aggregation map[string]interface{}
 
-// Represents a bucket for aggregation
+// Bucket represents a bucket for aggregation
 type Bucket map[string]interface{}
 
-// Represents a document to send to elasticsearch
+// Document holds a document to send to elasticsearch
 type Document struct {
 	// XXX : interface as we can support nil values
 	Index       interface{}
 	Type        string
-	Id          interface{}
+	ID          interface{}
 	BulkCommand string
 	Fields      interface{}
 }
 
-// Represents the "items" field in a _bulk response
+// Item holds an item from the "items" field in a _bulk response
 type Item struct {
 	Type    string `json:"_type"`
-	Id      string `json:"_id"`
+	ID      string `json:"_id"`
 	Index   string `json:"_index"`
 	Version int    `json:"_version"`
 	Error   string `json:"error"`
 	Status  uint64 `json:"status"`
 }
 
-// Represents the "_all" field when calling the _stats API
+// All represents the "_all" field when calling the _stats API
 // This is minimal but this is what I only need
 type All struct {
 	Indices   map[string]StatIndex   `json:"indices"`
 	Primaries map[string]StatPrimary `json:"primaries"`
 }
 
+// StatIndex contains stats for a specific index
 type StatIndex struct {
 	Primaries map[string]StatPrimary `json:"primaries"`
 }
 
+// StatPrimary contains stats for a primary index
 type StatPrimary struct {
 	// primary/docs:
 	Count   int
 	Deleted int
 }
 
-// Represents the "shard" struct as returned by elasticsearch
+// Shard holds the "shard" struct as returned by elasticsearch
 type Shard struct {
 	Total      uint64
 	Successful uint64
 	Failed     uint64
 }
 
-// Represent a hit returned by a search
+// Hit holds a hit returned by a search
 type Hit struct {
 	Index     string                 `json:"_index"`
 	Type      string                 `json:"_type"`
-	Id        string                 `json:"_id"`
+	ID        string                 `json:"_id"`
 	Score     float64                `json:"_score"`
 	Source    map[string]interface{} `json:"_source"`
 	Highlight map[string]interface{} `json:"highlight"`
 	Fields    map[string]interface{} `json:"fields"`
 }
 
-// Represent the hits structure as returned by elasticsearch
+// Hits holds the hits structure as returned by elasticsearch
 type Hits struct {
 	Total uint64
 	// max_score may contain the "null" value
@@ -162,12 +141,13 @@ type Hits struct {
 	Hits     []Hit
 }
 
+// SearchError holds errors returned from an ES search
 type SearchError struct {
 	Msg        string
 	StatusCode uint64
 }
 
-// Represent the status for a given index for the _status command
+// IndexStatus holds the status for a given index for the _status command
 type IndexStatus struct {
 	// XXX : problem, int will be marshaled to a float64 which seems logical
 	// XXX : is it better to use strings even for int values or to keep