add assert of response code, and fix the bug of parent test

elastic/client_test.go

@@ -53,6 +53,7 @@ func (s *elasticTestSuite) TestSimple(c *C) {
 
 	r, err := s.c.Get(index, docType, "1")
 	c.Assert(err, IsNil)
+	c.Assert(r.Code, Equals, 200)
 	c.Assert(r.ID, Equals, "1")
 
 	err = s.c.Delete(index, docType, "1")
@@ -78,6 +79,7 @@ func (s *elasticTestSuite) TestSimple(c *C) {
 
 	resp, err := s.c.IndexTypeBulk(index, docType, items)
 	c.Assert(err, IsNil)
+	c.Assert(resp.Code, Equals, 200)
 	c.Assert(resp.Errors, Equals, false)
 
 	for i := 0; i < 10; i++ {
@@ -90,16 +92,26 @@ func (s *elasticTestSuite) TestSimple(c *C) {
 
 	resp, err = s.c.IndexTypeBulk(index, docType, items)
 	c.Assert(err, IsNil)
+	c.Assert(resp.Code, Equals, 200)
 	c.Assert(resp.Errors, Equals, false)
 }
 
 // this requires a parent setting in _mapping
 func (s *elasticTestSuite) TestParent(c *C) {
 	index := "dummy"
 	docType := "comment"
+	ParentType := "parent"
 
+	mapping := map[string]interface{}{
+			docType: map[string]interface{}{
+				"_parent": map[string]string{"type": ParentType},
+		},
+	}
+	err := s.c.CreateMapping(index, docType,  mapping)
+	c.Assert(err, IsNil)
+
 	items := make([]*BulkRequest, 10)
-
+	
 	for i := 0; i < 10; i++ {
 		id := fmt.Sprintf("%d", i)
 		req := new(BulkRequest)
@@ -112,16 +124,20 @@ func (s *elasticTestSuite) TestParent(c *C) {
 
 	resp, err := s.c.IndexTypeBulk(index, docType, items)
 	c.Assert(err, IsNil)
+	c.Assert(resp.Code, Equals, 200)
 	c.Assert(resp.Errors, Equals, false)
-
 	for i := 0; i < 10; i++ {
 		id := fmt.Sprintf("%d", i)
 		req := new(BulkRequest)
+		req.Index = index
+		req.Type = docType
 		req.Action = ActionDelete
 		req.ID = id
+		req.Parent = "1"
 		items[i] = req
 	}
-	resp, err = s.c.IndexTypeBulk(index, docType, items)
+	resp, err = s.c.Bulk(items)
 	c.Assert(err, IsNil)
+	c.Assert(resp.Code, Equals, 200)
 	c.Assert(resp.Errors, Equals, false)
 }

etc/river.toml

@@ -3,6 +3,7 @@
 my_addr = "127.0.0.1:3306"
 my_user = "root"
 my_pass = ""
+my_charset = "utf8"
 
 # Elasticsearch address
 es_addr = "127.0.0.1:9200"

glide.yaml

@@ -7,7 +7,7 @@ import:
 - package: github.com/satori/go.uuid
   version: ^1.1.0
 - package: github.com/siddontang/go-mysql
-  version: 3fef0652795ca3a9d47e4af1a2f6d885e574e9bb
+  version: ead11cac47bd127a8c667efa07f171a9143d8a25
   subpackages:
   - canal
   - client

river/config.go

@@ -17,6 +17,7 @@ type Config struct {
 	MyAddr     string `toml:"my_addr"`
 	MyUser     string `toml:"my_user"`
 	MyPassword string `toml:"my_pass"`
+	MyCharset  string `toml:"my_charset"`
 
 	ESAddr string `toml:"es_addr"`
 

river/river.go

@@ -79,6 +79,7 @@ func (r *River) newCanal() error {
 	cfg.Addr = r.c.MyAddr
 	cfg.User = r.c.MyUser
 	cfg.Password = r.c.MyPassword
+	cfg.Charset = r.c.MyCharset
 	cfg.Flavor = r.c.Flavor
 
 	cfg.ServerID = r.c.ServerID
@@ -234,6 +235,7 @@ func (r *River) prepareRule() error {
 		if len(rule.TableInfo.PKColumns) == 0 {
 			return errors.Errorf("%s.%s must have a PK for a column", rule.Schema, rule.Table)
 		}
+
 	}
 
 	return nil

river/river_test.go

@@ -59,6 +59,7 @@ func (s *riverTestSuite) SetUpSuite(c *C) {
 	cfg.MyAddr = *my_addr
 	cfg.MyUser = "root"
 	cfg.MyPassword = ""
+	cfg.MyCharset = "utf8"
 	cfg.ESAddr = *es_addr
 
 	cfg.ServerID = 1001
@@ -121,7 +122,7 @@ func (s *riverTestSuite) TestConfig(c *C) {
 my_addr = "127.0.0.1:3306"
 my_user = "root"
 my_pass = ""
-
+my_charset = "utf8"
 es_addr = "127.0.0.1:9200"
 
 data_dir = "./var"