GaoreGo/goes
7
0
Fork 0
Code Pull Requests Releases 2 Activity

Merge pull request #6 from OwnLocal/split_core

Browse Source 
Split core, update testing
This commit is contained in:
Paul Bonser 2017-02-01 15:23:08 -06:00 committed by GitHub
commit c2f6a74d74
10 changed files with 508 additions and 459 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.travis.yml
View File

@ -3,14 +3,12 @@ language: go
go:
- 1.5.4
- 1.6.3
- 1.7.1
env:
global:
- GO15VENDOREXPERIMENT=1
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
@ -22,7 +20,6 @@ before_script:
- wget https://download.elastic.co/elasticsearch/elasticsearch/elasticsearch-${ES_VERSION}.tar.gz
- 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

9
Makefile
View File

@ -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 .

36
example_test.go
View File

@ -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)
}

10
glide.lock generated
View File

@ -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

6
glide.yaml
View File

@ -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

427
goes.go
View File

@ -19,118 +19,114 @@ 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
}
// 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",
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",
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",
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",
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",
Method: "POST",
API: "_optimize",
}
return r.Run()
return c.Do(&r)
}
// 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",
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",
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
@ -157,7 +153,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,
},
})
@ -200,256 +196,153 @@ func (c *Connection) BulkSend(documents []Document) (*Response, error) {
bulkData[len(bulkData)-1] = []byte(nil)
r := Request{
Conn: c,
method: "POST",
api: "_bulk",
bulkData: bytes.Join(bulkData, []byte("\n")),
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,
method: "POST",
api: "_search",
Method: "POST",
API: "_search",
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,
method: "POST",
api: "_count",
Method: "POST",
API: "_count",
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,
method: httpMethod,
api: "_query",
Method: httpMethod,
API: "_query",
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) {
func (c *Client) Scan(query interface{}, indexList []string, typeList []string, timeout string, size int) (*Response, error) {
v := url.Values{}
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,
method: "POST",
api: "_search",
Method: "POST",
API: "_search",
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) {
func (c *Client) Scroll(scrollID string, timeout string) (*Response, error) {
v := url.Values{}
v.Add("scroll", timeout)
r := Request{
Conn: c,
method: "POST",
api: "_search/scroll",
Method: "POST",
API: "_search/scroll",
ExtraArgs: v,
Body: []byte(scrollId),
Body: []byte(scrollID),
}
return r.Run()
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,
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},
ExtraArgs: extraArgs,
method: "POST",
Method: "POST",
}
if d.Id != nil {
r.method = "PUT"
r.id = d.Id.(string)
if d.ID != nil {
r.Method = "PUT"
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),
Method: "DELETE",
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,83 +371,79 @@ 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,
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, ","),
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",
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},
ExtraArgs: extraArgs,
method: "POST",
api: "_update",
Method: "POST",
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) {
r := Request{
Conn: c,
IndexList: indexes,
method: "DELETE",
api: "_mappings/" + typeName,
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),
}
@ -569,35 +458,93 @@ func (c *Connection) modifyAlias(action string, alias string, indexes []string)
}
r := Request{
Conn: c,
Query: command,
method: "POST",
api: "_aliases",
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,
Method: "HEAD",
API: "_alias/" + alias,
}
resp, err := r.Run()
resp, err := c.Do(&r)
return resp.Status == 200, err
}
// 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
}
req.URL.Scheme = "http"
req.URL.Host = fmt.Sprintf("%s:%s", c.Host, c.Port)
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
}

13
goes_suite_test.go Normal file
View File

@ -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")
}

279
goes_test.go
View File

@ -16,8 +16,8 @@ import (
)
var (
ES_HOST = "localhost"
ES_PORT = "9200"
ESHost = "localhost"
ESPort = "9200"
)
// Hook up gocheck into the gotest runner.
@ -30,21 +30,21 @@ var _ = Suite(&GoesTestSuite{})
func (s *GoesTestSuite) SetUpTest(c *C) {
h := os.Getenv("TEST_ELASTICSEARCH_HOST")
if h != "" {
ES_HOST = h
ESHost = h
}
p := os.Getenv("TEST_ELASTICSEARCH_PORT")
if p != "" {
ES_PORT = p
ESPort = p
}
}
func (s *GoesTestSuite) TestNewConnection(c *C) {
conn := NewConnection(ES_HOST, ES_PORT)
c.Assert(conn, DeepEquals, &Connection{ES_HOST, ES_PORT, http.DefaultClient})
func (s *GoesTestSuite) TestNewClient(c *C) {
conn := NewClient(ESHost, ESPort)
c.Assert(conn, DeepEquals, &Client{ESHost, ESPort, http.DefaultClient})
}
func (s *GoesTestSuite) TestWithClient(c *C) {
func (s *GoesTestSuite) TestWithHTTPClient(c *C) {
tr := &http.Transport{
DisableCompression: true,
ResponseHeaderTimeout: 1 * time.Second,
@ -52,72 +52,67 @@ func (s *GoesTestSuite) TestWithClient(c *C) {
cl := &http.Client{
Transport: tr,
}
conn := NewConnection(ES_HOST, ES_PORT).WithClient(cl)
conn := NewClient(ESHost, ESPort).WithHTTPClient(cl)
c.Assert(conn, DeepEquals, &Connection{ES_HOST, ES_PORT, cl})
c.Assert(conn, DeepEquals, &Client{ESHost, ESPort, cl})
c.Assert(conn.Client.Transport.(*http.Transport).DisableCompression, Equals, true)
c.Assert(conn.Client.Transport.(*http.Transport).ResponseHeaderTimeout, Equals, 1*time.Second)
}
func (s *GoesTestSuite) TestUrl(c *C) {
conn := NewConnection(ES_HOST, ES_PORT)
r := Request{
Conn: conn,
Query: "q",
IndexList: []string{"i"},
TypeList: []string{},
method: "GET",
api: "_search",
Method: "GET",
API: "_search",
}
c.Assert(r.Url(), Equals, "http://"+ES_HOST+":"+ES_PORT+"/i/_search")
c.Assert(r.URL().String(), Equals, "/i/_search")
r.IndexList = []string{"a", "b"}
c.Assert(r.Url(), Equals, "http://"+ES_HOST+":"+ES_PORT+"/a,b/_search")
c.Assert(r.URL().String(), Equals, "/a,b/_search")
r.TypeList = []string{"c", "d"}
c.Assert(r.Url(), Equals, "http://"+ES_HOST+":"+ES_PORT+"/a,b/c,d/_search")
c.Assert(r.URL().String(), Equals, "/a,b/c,d/_search")
r.ExtraArgs = make(url.Values, 1)
r.ExtraArgs.Set("version", "1")
c.Assert(r.Url(), Equals, "http://"+ES_HOST+":"+ES_PORT+"/a,b/c,d/_search?version=1")
c.Assert(r.URL().String(), Equals, "/a,b/c,d/_search?version=1")
r.id = "1234"
r.api = ""
c.Assert(r.Url(), Equals, "http://"+ES_HOST+":"+ES_PORT+"/a,b/c,d/1234/?version=1")
r.ID = "1234"
r.API = ""
c.Assert(r.URL().String(), Equals, "/a,b/c,d/1234/?version=1")
}
func (s *GoesTestSuite) TestEsDown(c *C) {
conn := NewConnection("a.b.c.d", "1234")
conn := NewClient("a.b.c.d", "1234")
var query = map[string]interface{}{"query": "foo"}
r := Request{
Conn: conn,
Query: query,
IndexList: []string{"i"},
method: "GET",
api: "_search",
Method: "GET",
API: "_search",
}
_, err := r.Run()
_, err := conn.Do(&r)
c.Assert(err, ErrorMatches, ".* no such host")
}
func (s *GoesTestSuite) TestRunMissingIndex(c *C) {
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
var query = map[string]interface{}{"query": "foo"}
r := Request{
Conn: conn,
Query: query,
IndexList: []string{"i"},
method: "GET",
api: "_search",
Method: "GET",
API: "_search",
}
_, err := r.Run()
_, err := conn.Do(&r)
c.Assert(err.Error(), Equals, "[404] IndexMissingException[[i] missing]")
}
@ -125,7 +120,7 @@ func (s *GoesTestSuite) TestRunMissingIndex(c *C) {
func (s *GoesTestSuite) TestCreateIndex(c *C) {
indexName := "testcreateindexgoes"
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
defer conn.DeleteIndex(indexName)
mapping := map[string]interface{}{
@ -152,7 +147,7 @@ func (s *GoesTestSuite) TestCreateIndex(c *C) {
}
func (s *GoesTestSuite) TestDeleteIndexInexistantIndex(c *C) {
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
resp, err := conn.DeleteIndex("foobar")
c.Assert(err.Error(), Equals, "[404] IndexMissingException[[foobar] missing]")
@ -161,7 +156,7 @@ func (s *GoesTestSuite) TestDeleteIndexInexistantIndex(c *C) {
}
func (s *GoesTestSuite) TestDeleteIndexExistingIndex(c *C) {
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
indexName := "testdeleteindexexistingindex"
@ -181,7 +176,7 @@ func (s *GoesTestSuite) TestDeleteIndexExistingIndex(c *C) {
}
func (s *GoesTestSuite) TestUpdateIndexSettings(c *C) {
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
indexName := "testupdateindex"
_, err := conn.CreateIndex(indexName, map[string]interface{}{})
@ -199,7 +194,7 @@ func (s *GoesTestSuite) TestUpdateIndexSettings(c *C) {
}
func (s *GoesTestSuite) TestRefreshIndex(c *C) {
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
indexName := "testrefreshindex"
_, err := conn.CreateIndex(indexName, map[string]interface{}{})
@ -213,7 +208,7 @@ func (s *GoesTestSuite) TestRefreshIndex(c *C) {
}
func (s *GoesTestSuite) TestOptimize(c *C) {
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
indexName := "testoptimize"
conn.DeleteIndex(indexName)
@ -238,10 +233,10 @@ func (s *GoesTestSuite) TestBulkSend(c *C) {
tweets := []Document{
{
Id: "123",
ID: "123",
Index: indexName,
Type: docType,
BulkCommand: BULK_COMMAND_INDEX,
BulkCommand: BulkCommandIndex,
Fields: map[string]interface{}{
"user": "foo",
"message": "some foo message",
@ -249,10 +244,10 @@ func (s *GoesTestSuite) TestBulkSend(c *C) {
},
{
Id: nil,
ID: nil,
Index: indexName,
Type: docType,
BulkCommand: BULK_COMMAND_INDEX,
BulkCommand: BulkCommandIndex,
Fields: map[string]interface{}{
"user": "bar",
"message": "some bar message",
@ -260,7 +255,7 @@ func (s *GoesTestSuite) TestBulkSend(c *C) {
},
}
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
conn.DeleteIndex(indexName)
_, err := conn.CreateIndex(indexName, nil)
@ -268,13 +263,13 @@ func (s *GoesTestSuite) TestBulkSend(c *C) {
response, err := conn.BulkSend(tweets)
i := Item{
Id: "123",
ID: "123",
Type: docType,
Version: 1,
Index: indexName,
Status: 201, //201 for indexing ( https://issues.apache.org/jira/browse/CONNECTORS-634 )
}
c.Assert(response.Items[0][BULK_COMMAND_INDEX], Equals, i)
c.Assert(response.Items[0][BulkCommandIndex], Equals, i)
c.Assert(err, IsNil)
_, err = conn.RefreshIndex(indexName)
@ -292,17 +287,17 @@ func (s *GoesTestSuite) TestBulkSend(c *C) {
var expectedTotal uint64 = 2
c.Assert(searchResults.Hits.Total, Equals, expectedTotal)
extraDocId := ""
extraDocID := ""
checked := 0
for _, hit := range searchResults.Hits.Hits {
if hit.Source["user"] == "foo" {
c.Assert(hit.Id, Equals, "123")
c.Assert(hit.ID, Equals, "123")
checked++
}
if hit.Source["user"] == "bar" {
c.Assert(len(hit.Id) > 0, Equals, true)
extraDocId = hit.Id
c.Assert(len(hit.ID) > 0, Equals, true)
extraDocID = hit.ID
checked++
}
}
@ -310,28 +305,28 @@ func (s *GoesTestSuite) TestBulkSend(c *C) {
docToDelete := []Document{
{
Id: "123",
ID: "123",
Index: indexName,
Type: docType,
BulkCommand: BULK_COMMAND_DELETE,
BulkCommand: BulkCommandDelete,
},
{
Id: extraDocId,
ID: extraDocID,
Index: indexName,
Type: docType,
BulkCommand: BULK_COMMAND_DELETE,
BulkCommand: BulkCommandDelete,
},
}
response, err = conn.BulkSend(docToDelete)
i = Item{
Id: "123",
ID: "123",
Type: docType,
Version: 2,
Index: indexName,
Status: 200, //200 for updates
}
c.Assert(response.Items[0][BULK_COMMAND_DELETE], Equals, i)
c.Assert(response.Items[0][BulkCommandDelete], Equals, i)
c.Assert(err, IsNil)
@ -349,7 +344,7 @@ func (s *GoesTestSuite) TestBulkSend(c *C) {
}
func (s *GoesTestSuite) TestStats(c *C) {
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
indexName := "teststats"
conn.DeleteIndex(indexName)
@ -371,9 +366,9 @@ func (s *GoesTestSuite) TestStats(c *C) {
func (s *GoesTestSuite) TestIndexWithFieldsInStruct(c *C) {
indexName := "testindexwithfieldsinstruct"
docType := "tweet"
docId := "1234"
docID := "1234"
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
// just in case
conn.DeleteIndex(indexName)
@ -384,7 +379,7 @@ func (s *GoesTestSuite) TestIndexWithFieldsInStruct(c *C) {
d := Document{
Index: indexName,
Type: docType,
Id: docId,
ID: docID,
Fields: struct {
user string
message string
@ -402,7 +397,7 @@ func (s *GoesTestSuite) TestIndexWithFieldsInStruct(c *C) {
expectedResponse := &Response{
Status: 201,
Index: indexName,
Id: docId,
ID: docID,
Type: docType,
Version: 1,
}
@ -414,9 +409,9 @@ func (s *GoesTestSuite) TestIndexWithFieldsInStruct(c *C) {
func (s *GoesTestSuite) TestIndexWithFieldsNotInMapOrStruct(c *C) {
indexName := "testindexwithfieldsnotinmaporstruct"
docType := "tweet"
docId := "1234"
docID := "1234"
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
// just in case
conn.DeleteIndex(indexName)
@ -427,7 +422,7 @@ func (s *GoesTestSuite) TestIndexWithFieldsNotInMapOrStruct(c *C) {
d := Document{
Index: indexName,
Type: docType,
Id: docId,
ID: docID,
Fields: "test",
}
@ -440,9 +435,9 @@ func (s *GoesTestSuite) TestIndexWithFieldsNotInMapOrStruct(c *C) {
func (s *GoesTestSuite) TestIndexIdDefined(c *C) {
indexName := "testindexiddefined"
docType := "tweet"
docId := "1234"
docID := "1234"
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
// just in case
conn.DeleteIndex(indexName)
@ -453,7 +448,7 @@ func (s *GoesTestSuite) TestIndexIdDefined(c *C) {
d := Document{
Index: indexName,
Type: docType,
Id: docId,
ID: docID,
Fields: map[string]interface{}{
"user": "foo",
"message": "bar",
@ -468,7 +463,7 @@ func (s *GoesTestSuite) TestIndexIdDefined(c *C) {
expectedResponse := &Response{
Status: 201,
Index: indexName,
Id: docId,
ID: docID,
Type: docType,
Version: 1,
}
@ -481,7 +476,7 @@ func (s *GoesTestSuite) TestIndexIdNotDefined(c *C) {
indexName := "testindexidnotdefined"
docType := "tweet"
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
// just in case
conn.DeleteIndex(indexName)
@ -504,15 +499,15 @@ func (s *GoesTestSuite) TestIndexIdNotDefined(c *C) {
c.Assert(response.Index, Equals, indexName)
c.Assert(response.Type, Equals, docType)
c.Assert(response.Version, Equals, 1)
c.Assert(response.Id != "", Equals, true)
c.Assert(response.ID != "", Equals, true)
}
func (s *GoesTestSuite) TestDelete(c *C) {
indexName := "testdelete"
docType := "tweet"
docId := "1234"
docID := "1234"
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
// just in case
conn.DeleteIndex(indexName)
@ -523,7 +518,7 @@ func (s *GoesTestSuite) TestDelete(c *C) {
d := Document{
Index: indexName,
Type: docType,
Id: docId,
ID: docID,
Fields: map[string]interface{}{
"user": "foo",
},
@ -540,7 +535,7 @@ func (s *GoesTestSuite) TestDelete(c *C) {
Found: true,
Index: indexName,
Type: docType,
Id: docId,
ID: docID,
// XXX : even after a DELETE the version number seems to be incremented
Version: 2,
}
@ -555,7 +550,7 @@ func (s *GoesTestSuite) TestDelete(c *C) {
Found: false,
Index: indexName,
Type: docType,
Id: docId,
ID: docID,
// XXX : even after a DELETE the version number seems to be incremented
Version: 3,
}
@ -566,9 +561,9 @@ func (s *GoesTestSuite) TestDelete(c *C) {
func (s *GoesTestSuite) TestDeleteByQuery(c *C) {
indexName := "testdeletebyquery"
docType := "tweet"
docId := "1234"
docID := "1234"
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
// just in case
conn.DeleteIndex(indexName)
@ -579,7 +574,7 @@ func (s *GoesTestSuite) TestDeleteByQuery(c *C) {
d := Document{
Index: indexName,
Type: docType,
Id: docId,
ID: docID,
Fields: map[string]interface{}{
"user": "foo",
},
@ -617,7 +612,7 @@ func (s *GoesTestSuite) TestDeleteByQuery(c *C) {
Found: false,
Index: "",
Type: "",
Id: "",
ID: "",
Version: 0,
}
response.Raw = nil
@ -632,13 +627,13 @@ func (s *GoesTestSuite) TestDeleteByQuery(c *C) {
func (s *GoesTestSuite) TestGet(c *C) {
indexName := "testget"
docType := "tweet"
docId := "111"
docID := "111"
source := map[string]interface{}{
"f1": "foo",
"f2": "foo",
}
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
conn.DeleteIndex(indexName)
_, err := conn.CreateIndex(indexName, map[string]interface{}{})
@ -648,21 +643,21 @@ func (s *GoesTestSuite) TestGet(c *C) {
d := Document{
Index: indexName,
Type: docType,
Id: docId,
ID: docID,
Fields: source,
}
_, err = conn.Index(d, url.Values{})
c.Assert(err, IsNil)
response, err := conn.Get(indexName, docType, docId, url.Values{})
response, err := conn.Get(indexName, docType, docID, url.Values{})
c.Assert(err, IsNil)
expectedResponse := &Response{
Status: 200,
Index: indexName,
Type: docType,
Id: docId,
ID: docID,
Version: 1,
Found: true,
Source: source,
@ -673,14 +668,14 @@ func (s *GoesTestSuite) TestGet(c *C) {
fields := make(url.Values, 1)
fields.Set("fields", "f1")
response, err = conn.Get(indexName, docType, docId, fields)
response, err = conn.Get(indexName, docType, docID, fields)
c.Assert(err, IsNil)
expectedResponse = &Response{
Status: 200,
Index: indexName,
Type: docType,
Id: docId,
ID: docID,
Version: 1,
Found: true,
Fields: map[string]interface{}{
@ -695,13 +690,13 @@ func (s *GoesTestSuite) TestGet(c *C) {
func (s *GoesTestSuite) TestSearch(c *C) {
indexName := "testsearch"
docType := "tweet"
docId := "1234"
docID := "1234"
source := map[string]interface{}{
"user": "foo",
"message": "bar",
}
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
conn.DeleteIndex(indexName)
_, err := conn.CreateIndex(indexName, map[string]interface{}{})
@ -711,7 +706,7 @@ func (s *GoesTestSuite) TestSearch(c *C) {
d := Document{
Index: indexName,
Type: docType,
Id: docId,
ID: docID,
Fields: source,
}
@ -733,7 +728,7 @@ func (s *GoesTestSuite) TestSearch(c *C) {
},
},
}
response, err := conn.Search(query, []string{indexName}, []string{docType}, url.Values{})
response, _ := conn.Search(query, []string{indexName}, []string{docType}, url.Values{})
expectedHits := Hits{
Total: 1,
@ -742,7 +737,7 @@ func (s *GoesTestSuite) TestSearch(c *C) {
{
Index: indexName,
Type: docType,
Id: docId,
ID: docID,
Score: 1.0,
Source: source,
},
@ -755,13 +750,13 @@ func (s *GoesTestSuite) TestSearch(c *C) {
func (s *GoesTestSuite) TestCount(c *C) {
indexName := "testcount"
docType := "tweet"
docId := "1234"
docID := "1234"
source := map[string]interface{}{
"user": "foo",
"message": "bar",
}
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
conn.DeleteIndex(indexName)
_, err := conn.CreateIndex(indexName, map[string]interface{}{})
@ -771,7 +766,7 @@ func (s *GoesTestSuite) TestCount(c *C) {
d := Document{
Index: indexName,
Type: docType,
Id: docId,
ID: docID,
Fields: source,
}
@ -793,14 +788,14 @@ func (s *GoesTestSuite) TestCount(c *C) {
},
},
}
response, err := conn.Count(query, []string{indexName}, []string{docType}, url.Values{})
response, _ := conn.Count(query, []string{indexName}, []string{docType}, url.Values{})
c.Assert(response.Count, Equals, 1)
}
func (s *GoesTestSuite) TestIndexStatus(c *C) {
indexName := "testindexstatus"
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
conn.DeleteIndex(indexName)
mapping := map[string]interface{}{
@ -876,10 +871,10 @@ func (s *GoesTestSuite) TestScroll(c *C) {
tweets := []Document{
{
Id: nil,
ID: nil,
Index: indexName,
Type: docType,
BulkCommand: BULK_COMMAND_INDEX,
BulkCommand: BulkCommandIndex,
Fields: map[string]interface{}{
"user": "foo",
"message": "some foo message",
@ -887,10 +882,10 @@ func (s *GoesTestSuite) TestScroll(c *C) {
},
{
Id: nil,
ID: nil,
Index: indexName,
Type: docType,
BulkCommand: BULK_COMMAND_INDEX,
BulkCommand: BulkCommandIndex,
Fields: map[string]interface{}{
"user": "bar",
"message": "some bar message",
@ -898,10 +893,10 @@ func (s *GoesTestSuite) TestScroll(c *C) {
},
{
Id: nil,
ID: nil,
Index: indexName,
Type: docType,
BulkCommand: BULK_COMMAND_INDEX,
BulkCommand: BulkCommandIndex,
Fields: map[string]interface{}{
"user": "foo",
"message": "another foo message",
@ -909,7 +904,7 @@ func (s *GoesTestSuite) TestScroll(c *C) {
},
}
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
mapping := map[string]interface{}{
"settings": map[string]interface{}{
@ -943,30 +938,30 @@ func (s *GoesTestSuite) TestScroll(c *C) {
scan, err := conn.Scan(query, []string{indexName}, []string{docType}, "1m", 1)
c.Assert(err, IsNil)
c.Assert(len(scan.ScrollId) > 0, Equals, true)
c.Assert(len(scan.ScrollID) > 0, Equals, true)
searchResults, err := conn.Scroll(scan.ScrollId, "1m")
searchResults, err := conn.Scroll(scan.ScrollID, "1m")
c.Assert(err, IsNil)
// some data in first chunk
c.Assert(searchResults.Hits.Total, Equals, uint64(2))
c.Assert(len(searchResults.ScrollId) > 0, Equals, true)
c.Assert(len(searchResults.ScrollID) > 0, Equals, true)
c.Assert(len(searchResults.Hits.Hits), Equals, 1)
searchResults, err = conn.Scroll(searchResults.ScrollId, "1m")
searchResults, err = conn.Scroll(searchResults.ScrollID, "1m")
c.Assert(err, IsNil)
// more data in second chunk
c.Assert(searchResults.Hits.Total, Equals, uint64(2))
c.Assert(len(searchResults.ScrollId) > 0, Equals, true)
c.Assert(len(searchResults.ScrollID) > 0, Equals, true)
c.Assert(len(searchResults.Hits.Hits), Equals, 1)
searchResults, err = conn.Scroll(searchResults.ScrollId, "1m")
searchResults, err = conn.Scroll(searchResults.ScrollID, "1m")
c.Assert(err, IsNil)
// nothing in third chunk
c.Assert(searchResults.Hits.Total, Equals, uint64(2))
c.Assert(len(searchResults.ScrollId) > 0, Equals, true)
c.Assert(len(searchResults.ScrollID) > 0, Equals, true)
c.Assert(len(searchResults.Hits.Hits), Equals, 0)
}
@ -976,10 +971,10 @@ func (s *GoesTestSuite) TestAggregations(c *C) {
tweets := []Document{
{
Id: nil,
ID: nil,
Index: indexName,
Type: docType,
BulkCommand: BULK_COMMAND_INDEX,
BulkCommand: BulkCommandIndex,
Fields: map[string]interface{}{
"user": "foo",
"message": "some foo message",
@ -988,10 +983,10 @@ func (s *GoesTestSuite) TestAggregations(c *C) {
},
{
Id: nil,
ID: nil,
Index: indexName,
Type: docType,
BulkCommand: BULK_COMMAND_INDEX,
BulkCommand: BulkCommandIndex,
Fields: map[string]interface{}{
"user": "bar",
"message": "some bar message",
@ -1000,10 +995,10 @@ func (s *GoesTestSuite) TestAggregations(c *C) {
},
{
Id: nil,
ID: nil,
Index: indexName,
Type: docType,
BulkCommand: BULK_COMMAND_INDEX,
BulkCommand: BulkCommandIndex,
Fields: map[string]interface{}{
"user": "foo",
"message": "another foo message",
@ -1011,7 +1006,7 @@ func (s *GoesTestSuite) TestAggregations(c *C) {
},
}
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
mapping := map[string]interface{}{
"settings": map[string]interface{}{
@ -1056,7 +1051,7 @@ func (s *GoesTestSuite) TestAggregations(c *C) {
},
}
resp, err := conn.Search(query, []string{indexName}, []string{docType}, url.Values{})
resp, _ := conn.Search(query, []string{indexName}, []string{docType}, url.Values{})
user, ok := resp.Aggregations["user"]
c.Assert(ok, Equals, true)
@ -1084,7 +1079,7 @@ func (s *GoesTestSuite) TestPutMapping(c *C) {
indexName := "testputmapping"
docType := "tweet"
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
// just in case
conn.DeleteIndex(indexName)
@ -1125,7 +1120,7 @@ func (s *GoesTestSuite) TestPutMapping(c *C) {
func (s *GoesTestSuite) TestIndicesExist(c *C) {
indices := []string{"testindicesexist"}
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
// just in case
conn.DeleteIndex(indices[0])
@ -1137,20 +1132,20 @@ func (s *GoesTestSuite) TestIndicesExist(c *C) {
defer conn.DeleteIndex(indices[0])
time.Sleep(200 * time.Millisecond)
exists, err = conn.IndicesExist(indices)
exists, _ = conn.IndicesExist(indices)
c.Assert(exists, Equals, true)
indices = append(indices, "nonexistent")
exists, err = conn.IndicesExist(indices)
exists, _ = conn.IndicesExist(indices)
c.Assert(exists, Equals, false)
}
func (s *GoesTestSuite) TestUpdate(c *C) {
indexName := "testupdate"
docType := "tweet"
docId := "1234"
docID := "1234"
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
// just in case
conn.DeleteIndex(indexName)
@ -1161,7 +1156,7 @@ func (s *GoesTestSuite) TestUpdate(c *C) {
d := Document{
Index: indexName,
Type: docType,
Id: docId,
ID: docID,
Fields: map[string]interface{}{
"user": "foo",
"message": "bar",
@ -1177,7 +1172,7 @@ func (s *GoesTestSuite) TestUpdate(c *C) {
expectedResponse := &Response{
Status: 201,
Index: indexName,
Id: docId,
ID: docID,
Type: docType,
Version: 1,
}
@ -1208,20 +1203,20 @@ func (s *GoesTestSuite) TestUpdate(c *C) {
c.Assert(err, Equals, nil)
response, err = conn.Get(indexName, docType, docId, url.Values{})
response, err = conn.Get(indexName, docType, docID, url.Values{})
c.Assert(err, Equals, nil)
c.Assert(response.Source["counter"], Equals, float64(6))
c.Assert(response.Source["user"], Equals, "foo")
c.Assert(response.Source["message"], Equals, "bar")
// Test another document, non-existent
docId = "555"
d.Id = docId
docID = "555"
d.ID = docID
response, err = conn.Update(d, query, extraArgs)
c.Assert(err, Equals, nil)
time.Sleep(200 * time.Millisecond)
response, err = conn.Get(indexName, docType, docId, url.Values{})
response, err = conn.Get(indexName, docType, docID, url.Values{})
c.Assert(err, Equals, nil)
c.Assert(response.Source["user"], Equals, "admin")
c.Assert(response.Source["message"], Equals, "candybar")
@ -1231,7 +1226,7 @@ func (s *GoesTestSuite) TestGetMapping(c *C) {
indexName := "testmapping"
docType := "tweet"
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
// just in case
conn.DeleteIndex(indexName)
@ -1267,7 +1262,7 @@ func (s *GoesTestSuite) TestDeleteMapping(c *C) {
indexName := "testdeletemapping"
docType := "tweet"
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
// just in case
conn.DeleteIndex(indexName)
@ -1313,13 +1308,13 @@ func (s *GoesTestSuite) TestAddAlias(c *C) {
aliasName := "testAlias"
indexName := "testalias_1"
docType := "testDoc"
docId := "1234"
docID := "1234"
source := map[string]interface{}{
"user": "foo",
"message": "bar",
}
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
defer conn.DeleteIndex(indexName)
_, err := conn.CreateIndex(indexName, map[string]interface{}{})
@ -1329,7 +1324,7 @@ func (s *GoesTestSuite) TestAddAlias(c *C) {
d := Document{
Index: indexName,
Type: docType,
Id: docId,
ID: docID,
Fields: source,
}
@ -1342,14 +1337,14 @@ func (s *GoesTestSuite) TestAddAlias(c *C) {
c.Assert(err, IsNil)
// Get document via alias
response, err := conn.Get(aliasName, docType, docId, url.Values{})
response, err := conn.Get(aliasName, docType, docID, url.Values{})
c.Assert(err, IsNil)
expectedResponse := &Response{
Status: 200,
Index: indexName,
Type: docType,
Id: docId,
ID: docID,
Version: 1,
Found: true,
Source: source,
@ -1363,13 +1358,13 @@ func (s *GoesTestSuite) TestRemoveAlias(c *C) {
aliasName := "testAlias"
indexName := "testalias_1"
docType := "testDoc"
docId := "1234"
docID := "1234"
source := map[string]interface{}{
"user": "foo",
"message": "bar",
}
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
defer conn.DeleteIndex(indexName)
_, err := conn.CreateIndex(indexName, map[string]interface{}{})
@ -1379,7 +1374,7 @@ func (s *GoesTestSuite) TestRemoveAlias(c *C) {
d := Document{
Index: indexName,
Type: docType,
Id: docId,
ID: docID,
Fields: source,
}
@ -1396,7 +1391,7 @@ func (s *GoesTestSuite) TestRemoveAlias(c *C) {
c.Assert(err, IsNil)
// Get document via alias
_, err = conn.Get(aliasName, docType, docId, url.Values{})
_, err = conn.Get(aliasName, docType, docID, url.Values{})
c.Assert(err.Error(), Equals, "[404] IndexMissingException[["+aliasName+"] missing]")
}
@ -1404,7 +1399,7 @@ func (s *GoesTestSuite) TestAliasExists(c *C) {
index := "testaliasexist_1"
alias := "testaliasexists"
conn := NewConnection(ES_HOST, ES_PORT)
conn := NewClient(ESHost, ESPort)
// just in case
conn.DeleteIndex(index)
@ -1421,6 +1416,6 @@ func (s *GoesTestSuite) TestAliasExists(c *C) {
time.Sleep(200 * time.Millisecond)
defer conn.RemoveAlias(alias, []string{index})
exists, err = conn.AliasExists(alias)
exists, _ = conn.AliasExists(alias)
c.Assert(exists, Equals, true)
}

109
request.go Normal file
View File

@ -0,0 +1,109 @@
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
}
// 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 {
b, err := json.Marshal(req.Query)
if err != nil {
return nil, err
}
postData = b
}
reader := ioutil.NopCloser(bytes.NewReader(postData))
newReq, err := http.NewRequest(req.Method, "", nil)
if err != nil {
return nil, err
}
newReq.URL = req.URL()
newReq.Body = reader
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)

73
structs.go
View File

@ -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
@ -22,43 +22,11 @@ type Connection struct {
Client *http.Client
}
// 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 +34,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 +54,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 +132,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