From 8719e13c820411fa52a20b9920c29ca03073eabe Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xinux@mail.ustc.edu.cn>
Date: Wed, 29 Mar 2017 14:50:18 +0800
Subject: [PATCH 01/65] Update river.go

it looks that you have already support for multiple Primary key, but forget remove the code.
---
 river/river.go | 6 ------
 1 file changed, 6 deletions(-)

diff --git a/river/river.go b/river/river.go
index d9c66801..b91cbb53 100644
--- a/river/river.go
+++ b/river/river.go
@@ -221,12 +221,6 @@ func (r *River) prepareRule() error {
 		if rule.TableInfo, err = r.canal.GetTable(rule.Schema, rule.Table); err != nil {
 			return errors.Trace(err)
 		}
-
-		// table must have a PK for one column, multi columns may be supported later.
-
-		if len(rule.TableInfo.PKColumns) != 1 {
-			return errors.Errorf("%s.%s must have a PK for a column", rule.Schema, rule.Table)
-		}
 	}
 
 	return nil

From e5ff5dc96dabaa19ca36e02c57a4e08e912a5924 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Fri, 31 Mar 2017 22:40:43 +0800
Subject: [PATCH 02/65] if have only one request, use bulk will waste resource
 of elasticsearch

---
 river/sync.go | 42 ++++++++++++++++++++++++++++++++----------
 1 file changed, 32 insertions(+), 10 deletions(-)

diff --git a/river/sync.go b/river/sync.go
index ef975f43..4551139f 100644
--- a/river/sync.go
+++ b/river/sync.go
@@ -325,19 +325,41 @@ func (r *River) getParentID(rule *Rule, row []interface{}, columnName string) (s
 }
 
 func (r *River) doBulk(reqs []*elastic.BulkRequest) error {
+	flag := true
+	var err error
 	if len(reqs) == 0 {
 		return nil
 	}
-
-	if resp, err := r.es.Bulk(reqs); err != nil {
-		log.Errorf("sync docs err %v after binlog %s", err, r.canal.SyncedPosition())
-		return errors.Trace(err)
-	} else if resp.Errors {
-		for i := 0; i < len(resp.Items); i++ {
-			for action, item := range resp.Items[i] {
-				if len(item.Error) > 0 {
-					log.Errorf("%s index: %s, type: %s, id: %s, status: %d, error: %s",
-						action, item.Index, item.Type, item.ID, item.Status, item.Error)
+	if len(reqs) == 1{
+		switch reqs[0].Action {
+		case "index":
+			err = r.es.Update(reqs[0].Index, reqs[0].Type, reqs[0].ID, reqs[0].Data)
+		case "delete":
+			err = r.es.Delete(reqs[0].Index, reqs[0].Type, reqs[0].ID)
+		case "update":
+			err = r.es.Update(reqs[0].Index, reqs[0].Type, reqs[0].ID, reqs[0].Data)
+		default:
+			flag = false
+			err = nil
+		}
+		if flag {
+			if err != nil {
+				log.Errorf("sync docs err %v after binlog %s", err, r.canal.SyncedPosition())
+				return errors.Trace(err)
+			}
+			return nil
+		}
+	} else if len(reqs) > 1 || (!flag){
+		if resp, err := r.es.Bulk(reqs); err != nil {
+			log.Errorf("sync docs err %v after binlog %s", err, r.canal.SyncedPosition())
+			return errors.Trace(err)
+		} else if resp.Errors {
+			for i := 0; i < len(resp.Items); i++ {
+				for action, item := range resp.Items[i] {
+					if len(item.Error) > 0 {
+						log.Errorf("%s index: %s, type: %s, id: %s, status: %d, error: %s",
+							action, item.Index, item.Type, item.ID, item.Status, item.Error)
+					}
 				}
 			}
 		}

From 74e0c36076b4b8fa3bd8db078fa9a04f2186c3f5 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Wed, 12 Apr 2017 17:52:08 +0800
Subject: [PATCH 03/65] Update river.go

---
 river/river.go | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/river/river.go b/river/river.go
index cac1aa0e..a9db22d3 100644
--- a/river/river.go
+++ b/river/river.go
@@ -221,13 +221,11 @@ func (r *River) prepareRule() error {
 		if rule.TableInfo, err = r.canal.GetTable(rule.Schema, rule.Table); err != nil {
 			return errors.Trace(err)
 		}
-<<<<<<< HEAD
-=======
 
 		if len(rule.TableInfo.PKColumns) == 0 {
 			return errors.Errorf("%s.%s must have a PK for a column", rule.Schema, rule.Table)
 		}
->>>>>>> 4f21bc6ec1c64547ee05c698941ae8158b785760
+
 	}
 
 	return nil

From 45c069ab9cbfae38f92f40bdf48e70c551923b2a Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Thu, 13 Apr 2017 20:42:44 +0800
Subject: [PATCH 04/65] Update README.md

---
 README.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/README.md b/README.md
index ffacd7bc..7c43a6f3 100644
--- a/README.md
+++ b/README.md
@@ -23,7 +23,7 @@ It uses `mysqldump` to fetch the origin data at first, then syncs data increment
 + binlog format must be **row**.
 + binlog row image must be **full** for MySQL, you may lost some field data if you update PK data in MySQL with minimal or noblob binlog row image. MariaDB only supports full row image.
 + Can not alter table format at runtime.
-+ MySQL table which will be synced must have a PK(primary key), multi columns PK is allowed now, e,g, if the PKs is (a, b), we will use "a:b" as the key. The PK data will be used as "id" in Elasticsearch.
++ MySQL table which will be synced should have a PK(primary key), multi columns PK is allowed now, e,g, if the PKs is (a, b), we will use "a:b" as the key. The PK data will be used as "id" in Elasticsearch. And you can also config the id's constituent part.
 + You should create the associated mappings in Elasticsearch first, I don't think using the default mapping is a wise decision, you must know how to search accurately.
 + `mysqldump` must exist in the same node with go-mysql-elasticsearch, if not, go-mysql-elasticsearch will try to sync binlog only.
 + Don't change too many rows at same time in one SQL.

From 02ec489e96424a6e6753265d8d95643076feebae Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Thu, 13 Apr 2017 20:44:02 +0800
Subject: [PATCH 05/65] Update README.md

---
 README.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/README.md b/README.md
index 7c43a6f3..78dbffe2 100644
--- a/README.md
+++ b/README.md
@@ -23,7 +23,7 @@ It uses `mysqldump` to fetch the origin data at first, then syncs data increment
 + binlog format must be **row**.
 + binlog row image must be **full** for MySQL, you may lost some field data if you update PK data in MySQL with minimal or noblob binlog row image. MariaDB only supports full row image.
 + Can not alter table format at runtime.
-+ MySQL table which will be synced should have a PK(primary key), multi columns PK is allowed now, e,g, if the PKs is (a, b), we will use "a:b" as the key. The PK data will be used as "id" in Elasticsearch. And you can also config the id's constituent part.
++ MySQL table which will be synced should have a PK(primary key), multi columns PK is allowed now, e,g, if the PKs is (a, b), we will use "a:b" as the key. The PK data will be used as "id" in Elasticsearch. And you can also config the id's constituent part with other column.
 + You should create the associated mappings in Elasticsearch first, I don't think using the default mapping is a wise decision, you must know how to search accurately.
 + `mysqldump` must exist in the same node with go-mysql-elasticsearch, if not, go-mysql-elasticsearch will try to sync binlog only.
 + Don't change too many rows at same time in one SQL.

From 6c32bdc81532bc4c00d840d17b19f5c0d02cae05 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Thu, 13 Apr 2017 23:35:34 +0800
Subject: [PATCH 06/65] update commit version of go-mysql

---
 glide.lock | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/glide.lock b/glide.lock
index 9c2cfbc1..5328ed51 100644
--- a/glide.lock
+++ b/glide.lock
@@ -16,7 +16,7 @@ imports:
   - ioutil2
   - sync2
 - name: github.com/siddontang/go-mysql
-  version: 3ca161ffa3b5844340e534d4b3cdc03b60f32610
+  version: dd8c4fc06b092a5563dcd68431c857dd21a36d0a
   subpackages:
   - canal
   - client

From 4b0ba49529c2745e76a180e096d19d73bd32209b Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Sat, 29 Apr 2017 01:01:27 +0800
Subject: [PATCH 07/65] Update glide.lock

---
 glide.lock | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/glide.lock b/glide.lock
index 5328ed51..73252e14 100644
--- a/glide.lock
+++ b/glide.lock
@@ -15,8 +15,8 @@ imports:
   - hack
   - ioutil2
   - sync2
-- name: github.com/siddontang/go-mysql
-  version: dd8c4fc06b092a5563dcd68431c857dd21a36d0a
+- name: github.com/WangXiangUSTC/go-mysql
+  version: fddc32296e000e7d07ca11c49225f8ae4c03f915
   subpackages:
   - canal
   - client

From 4ef3830c0eb48523c86bdeeb158110816d27cd88 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 30 Apr 2017 16:03:15 +0800
Subject: [PATCH 08/65] add set of charcter

---
 river/config.go | 1 +
 river/river.go  | 1 +
 2 files changed, 2 insertions(+)

diff --git a/river/config.go b/river/config.go
index c21687ca..c6f626f3 100644
--- a/river/config.go
+++ b/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"`
 
diff --git a/river/river.go b/river/river.go
index 1e056002..fc349b63 100644
--- a/river/river.go
+++ b/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

From e1e147213a91ed51ebf3a7e75a8d3cad129c6c31 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Fri, 5 May 2017 20:31:58 +0800
Subject: [PATCH 09/65] update sync.go

---
 river/sync.go | 44 +++++++++++---------------------------------
 1 file changed, 11 insertions(+), 33 deletions(-)

diff --git a/river/sync.go b/river/sync.go
index aea2fd07..4217f851 100644
--- a/river/sync.go
+++ b/river/sync.go
@@ -432,45 +432,23 @@ func (r *River) getParentID(rule *Rule, row []interface{}, columnName string) (s
 }
 
 func (r *River) doBulk(reqs []*elastic.BulkRequest) error {
-	flag := true
-	var err error
 	if len(reqs) == 0 {
 		return nil
 	}
-	if len(reqs) == 1{
-		switch reqs[0].Action {
-		case "index":
-			err = r.es.Update(reqs[0].Index, reqs[0].Type, reqs[0].ID, reqs[0].Data)
-		case "delete":
-			err = r.es.Delete(reqs[0].Index, reqs[0].Type, reqs[0].ID)
-		case "update":
-			err = r.es.Update(reqs[0].Index, reqs[0].Type, reqs[0].ID, reqs[0].Data)
-		default:
-			flag = false
-			err = nil
-		}
-		if flag {
-			if err != nil {
-				log.Errorf("sync docs err %v after binlog %s", err, r.canal.SyncedPosition())
-				return errors.Trace(err)
-			}
-			return nil
-		}
-	} else if len(reqs) > 1 || (!flag){
-		if resp, err := r.es.Bulk(reqs); err != nil {
-			log.Errorf("sync docs err %v after binlog %s", err, r.canal.SyncedPosition())
-			return errors.Trace(err)
-		} else if resp.Errors {
-			for i := 0; i < len(resp.Items); i++ {
-				for action, item := range resp.Items[i] {
-					if len(item.Error) > 0 {
-						log.Errorf("%s index: %s, type: %s, id: %s, status: %d, error: %s",
-							action, item.Index, item.Type, item.ID, item.Status, item.Error)
-					}
+
+	if resp, err := r.es.Bulk(reqs); err != nil {
+		log.Errorf("sync docs err %v after binlog %s", err, r.canal.SyncedPosition())
+		return errors.Trace(err)
+	} else if resp.Errors {
+		for i := 0; i < len(resp.Items); i++ {
+			for action, item := range resp.Items[i] {
+				if len(item.Error) > 0 {
+					log.Errorf("%s index: %s, type: %s, id: %s, status: %d, error: %s",
+						action, item.Index, item.Type, item.ID, item.Status, item.Error)
 				}
 			}
 		}
 	}
 
-	return nil
+	return nil 
 }

From 56cb5d7e279534f708dbcc78b1f1f1ec8d4a270c Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Fri, 5 May 2017 20:53:32 +0800
Subject: [PATCH 10/65] update gp-mysql version

---
 glide.lock | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/glide.lock b/glide.lock
index 3c6a5f9a..c1c7fe56 100644
--- a/glide.lock
+++ b/glide.lock
@@ -16,7 +16,7 @@ imports:
   - ioutil2
   - sync2
 - name: github.com/siddontang/go-mysql
-  version: 3fef0652795ca3a9d47e4af1a2f6d885e574e9bb
+  version: ead11cac47bd127a8c667efa07f171a9143d8a25
   subpackages:
   - canal
   - client

From e9659253b0295a334f91f0b2a69ddc1d66f439a6 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Fri, 5 May 2017 20:54:25 +0800
Subject: [PATCH 11/65] update go-mysql version

---
 glide.yaml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/glide.yaml b/glide.yaml
index 888c5acc..e84692f4 100644
--- a/glide.yaml
+++ b/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

From cdc4543150c0e123ef8cfc8816aa37cb15e62caf Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Fri, 5 May 2017 20:56:14 +0800
Subject: [PATCH 12/65] add config of mysql charset

---
 etc/river.toml | 1 +
 1 file changed, 1 insertion(+)

diff --git a/etc/river.toml b/etc/river.toml
index 9b02ac02..0b9d5d68 100644
--- a/etc/river.toml
+++ b/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"

From 570317cbd00db0fd6e863516502b6ae30bd91962 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Fri, 5 May 2017 20:59:08 +0800
Subject: [PATCH 13/65] add test of mysql charset

---
 river/river_test.go | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/river/river_test.go b/river/river_test.go
index 38184da9..0cc7e28d 100644
--- a/river/river_test.go
+++ b/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"

From 43d24a4419b3aabdcbd820bbbdd9d39cbd2e98d2 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 7 May 2017 13:16:39 +0800
Subject: [PATCH 14/65] update go-mysql vendor

---
 .../github.com/siddontang/go-mysql/canal/canal.go |  4 ++++
 .../siddontang/go-mysql/canal/config.go           |  6 ++++--
 .../github.com/siddontang/go-mysql/dump/dump.go   | 13 ++++++++++++-
 .../go-mysql/replication/binlogsyncer.go          | 15 ++++++++++++++-
 4 files changed, 34 insertions(+), 4 deletions(-)

diff --git a/vendor/github.com/siddontang/go-mysql/canal/canal.go b/vendor/github.com/siddontang/go-mysql/canal/canal.go
index 4d668776..aa5cd65d 100644
--- a/vendor/github.com/siddontang/go-mysql/canal/canal.go
+++ b/vendor/github.com/siddontang/go-mysql/canal/canal.go
@@ -101,6 +101,9 @@ func (c *Canal) prepareDumper() error {
 		c.dumper.AddTables(tableDB, tables...)
 	}
 
+        charset := c.cfg.Charset
+        c.dumper.SetCharset(charset)
+
 	for _, ignoreTable := range c.cfg.Dump.IgnoreTables {
 		if seps := strings.Split(ignoreTable, ","); len(seps) == 2 {
 			c.dumper.AddIgnoreTables(seps[0], seps[1])
@@ -266,6 +269,7 @@ func (c *Canal) prepareSyncer() error {
 		Port:     uint16(port),
 		User:     c.cfg.User,
 		Password: c.cfg.Password,
+		Charset : c.cfg.Charset,
 	}
 
 	c.syncer = replication.NewBinlogSyncer(&cfg)
diff --git a/vendor/github.com/siddontang/go-mysql/canal/config.go b/vendor/github.com/siddontang/go-mysql/canal/config.go
index 50606c8c..8f01d321 100644
--- a/vendor/github.com/siddontang/go-mysql/canal/config.go
+++ b/vendor/github.com/siddontang/go-mysql/canal/config.go
@@ -4,7 +4,7 @@ import (
 	"io/ioutil"
 	"math/rand"
 	"time"
-
+	"github.com/siddontang/go-mysql/mysql"
 	"github.com/BurntSushi/toml"
 	"github.com/juju/errors"
 )
@@ -19,7 +19,7 @@ type DumpConfig struct {
 	TableDB string   `toml:"table_db"`
 
 	Databases []string `toml:"dbs"`
-
+ 
 	// Ignore table format is db.table
 	IgnoreTables []string `toml:"ignore_tables"`
 
@@ -32,6 +32,7 @@ type Config struct {
 	User     string `toml:"user"`
 	Password string `toml:"password"`
 
+	Charset  string `toml:"charset"`
 	ServerID uint32 `toml:"server_id"`
 	Flavor   string `toml:"flavor"`
 
@@ -65,6 +66,7 @@ func NewDefaultConfig() *Config {
 	c.User = "root"
 	c.Password = ""
 
+	c.Charset = mysql.DEFAULT_CHARSET
 	rand.Seed(time.Now().Unix())
 	c.ServerID = uint32(rand.Intn(1000)) + 1001
 
diff --git a/vendor/github.com/siddontang/go-mysql/dump/dump.go b/vendor/github.com/siddontang/go-mysql/dump/dump.go
index a6ff2091..da3e5b3b 100644
--- a/vendor/github.com/siddontang/go-mysql/dump/dump.go
+++ b/vendor/github.com/siddontang/go-mysql/dump/dump.go
@@ -6,7 +6,7 @@ import (
 	"os"
 	"os/exec"
 	"strings"
-
+	. "github.com/siddontang/go-mysql/mysql"
 	"github.com/juju/errors"
 )
 
@@ -25,6 +25,8 @@ type Dumper struct {
 
 	Databases []string
 
+	Charset string
+
 	IgnoreTables map[string][]string
 
 	ErrOut io.Writer
@@ -47,6 +49,7 @@ func NewDumper(executionPath string, addr string, user string, password string)
 	d.Password = password
 	d.Tables = make([]string, 0, 16)
 	d.Databases = make([]string, 0, 16)
+	d.Charset = DEFAULT_CHARSET
 	d.IgnoreTables = make(map[string][]string)
 
 	d.ErrOut = os.Stderr
@@ -54,6 +57,10 @@ func NewDumper(executionPath string, addr string, user string, password string)
 	return d, nil
 }
 
+func (d *Dumper) SetCharset(charset string) {
+	d.Charset = charset
+}
+
 func (d *Dumper) SetErrOut(o io.Writer) {
 	d.ErrOut = o
 }
@@ -133,6 +140,10 @@ func (d *Dumper) Dump(w io.Writer) error {
 		w.Write([]byte(fmt.Sprintf("USE `%s`;\n", d.TableDB)))
 	}
 
+	if len(d.Charset) != 0 {
+		args = append(args, fmt.Sprintf("--default-character-set=%s", d.Charset))
+	}
+
 	cmd := exec.Command(d.ExecutionPath, args...)
 
 	cmd.Stderr = d.ErrOut
diff --git a/vendor/github.com/siddontang/go-mysql/replication/binlogsyncer.go b/vendor/github.com/siddontang/go-mysql/replication/binlogsyncer.go
index b85e7cd2..8f69814a 100644
--- a/vendor/github.com/siddontang/go-mysql/replication/binlogsyncer.go
+++ b/vendor/github.com/siddontang/go-mysql/replication/binlogsyncer.go
@@ -40,6 +40,9 @@ type BinlogSyncerConfig struct {
 	// If not set, use os.Hostname() instead.
 	Localhost string
 
+	// Charset is for MySQL client character set
+	Charset string
+
 	// SemiSyncEnabled enables semi-sync or not.
 	SemiSyncEnabled bool
 
@@ -52,6 +55,8 @@ type BinlogSyncerConfig struct {
 	// Use replication.Time structure for timestamp and datetime.
 	// We will use Local location for timestamp and UTC location for datatime.
 	ParseTime bool
+
+	LogLevel string
 }
 
 // BinlogSyncer syncs binlog event from server.
@@ -76,6 +81,11 @@ type BinlogSyncer struct {
 
 // NewBinlogSyncer creates the BinlogSyncer with cfg.
 func NewBinlogSyncer(cfg *BinlogSyncerConfig) *BinlogSyncer {
+	if cfg.LogLevel == "" {
+		cfg.LogLevel = "info"
+	}
+	log.SetLevelByString(cfg.LogLevel)
+
 	log.Infof("create BinlogSyncer with config %v", cfg)
 
 	b := new(BinlogSyncer)
@@ -143,6 +153,9 @@ func (b *BinlogSyncer) registerSlave() error {
 	if err != nil {
 		return errors.Trace(err)
 	}
+	if len(b.cfg.Charset) != 0 {
+	    b.c.SetCharset(b.cfg.Charset)
+	}
 
 	//for mysql 5.6+, binlog has a crc32 checksum
 	//before mysql 5.6, this will not work, don't matter.:-)
@@ -556,7 +569,7 @@ func (b *BinlogSyncer) parseEvent(s *BinlogStreamer, data []byte) error {
 		data = data[2:]
 	}
 
-	e, err := b.parser.parse(data)
+	e, err := b.parser.Parse(data)
 	if err != nil {
 		return errors.Trace(err)
 	}

From 4bdeb151d007568af3d7504ee9426998b4258148 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 7 May 2017 13:28:38 +0800
Subject: [PATCH 15/65] update go-mysql vendor

---
 .../siddontang/go-mysql/replication/parser.go        | 12 +++++++++---
 1 file changed, 9 insertions(+), 3 deletions(-)

diff --git a/vendor/github.com/siddontang/go-mysql/replication/parser.go b/vendor/github.com/siddontang/go-mysql/replication/parser.go
index 6f727295..056a17b5 100644
--- a/vendor/github.com/siddontang/go-mysql/replication/parser.go
+++ b/vendor/github.com/siddontang/go-mysql/replication/parser.go
@@ -56,10 +56,10 @@ func (p *BinlogParser) ParseFile(name string, offset int64, onEvent OnEventFunc)
 		return errors.Errorf("seek %s to %d error %v", name, offset, err)
 	}
 
-	return p.parseReader(f, onEvent)
+	return p.ParseReader(f, onEvent)
 }
 
-func (p *BinlogParser) parseReader(r io.Reader, onEvent OnEventFunc) error {
+func (p *BinlogParser) ParseReader(r io.Reader, onEvent OnEventFunc) error {
 	p.Reset()
 
 	var err error
@@ -212,7 +212,13 @@ func (p *BinlogParser) parseEvent(h *EventHeader, data []byte) (Event, error) {
 	return e, nil
 }
 
-func (p *BinlogParser) parse(data []byte) (*BinlogEvent, error) {
+// Given the bytes for a a binary log event: return the decoded event.
+// With the exception of the FORMAT_DESCRIPTION_EVENT event type
+// there must have previously been passed a FORMAT_DESCRIPTION_EVENT
+// into the parser for this to work properly on any given event.
+// Passing a new FORMAT_DESCRIPTION_EVENT into the parser will replace
+// an existing one.
+func (p *BinlogParser) Parse(data []byte) (*BinlogEvent, error) {
 	rawData := data
 
 	h, err := p.parseHeader(data)

From 945ddb25fe914a58ba847310f1fc7c1b6b75ef70 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Sun, 7 May 2017 13:33:34 +0800
Subject: [PATCH 16/65] Update sync.go

---
 river/sync.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/river/sync.go b/river/sync.go
index 4217f851..5f914d08 100644
--- a/river/sync.go
+++ b/river/sync.go
@@ -450,5 +450,5 @@ func (r *River) doBulk(reqs []*elastic.BulkRequest) error {
 		}
 	}
 
-	return nil 
+	return nil
 }

From cc7b3144af1fb5cf9abc58c0e32ed8a7c401d132 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Wed, 17 May 2017 20:13:06 +0800
Subject: [PATCH 17/65] add user and password of elasticsearch

---
 elastic/client.go      | 39 ++++++++++++++++++++++++++-------------
 elastic/client_test.go |  2 +-
 river/config.go        |  4 +++-
 river/river.go         |  3 ++-
 river/river_test.go    |  3 ++-
 5 files changed, 34 insertions(+), 17 deletions(-)

diff --git a/elastic/client.go b/elastic/client.go
index 5801368e..da6f480a 100644
--- a/elastic/client.go
+++ b/elastic/client.go
@@ -15,14 +15,19 @@ import (
 // Because we only need some very simple usages.
 type Client struct {
 	Addr string
+	User string
+	Password string
 
 	c *http.Client
 }
 
-func NewClient(addr string) *Client {
+
+func NewClient(addr string, user string, password string) *Client {
 	c := new(Client)
 
 	c.Addr = addr
+	c.User = user
+	c.Password = password
 
 	c.c = &http.Client{}
 
@@ -134,20 +139,33 @@ type BulkResponseItem struct {
 	Found   bool            `json:"found"`
 }
 
-func (c *Client) Do(method string, url string, body map[string]interface{}) (*Response, error) {
-	bodyData, err := json.Marshal(body)
+func (c *Client) DoRequest(method string, url string, body *bytes.Buffer) (*http.Response, error) {
+	req, err := http.NewRequest(method, url, body)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
+	if len(c.User) > 0 && len(c.Password) > 0 {
+		req.SetBasicAuth(c.User, c.Password)
+	}
+	resp, err := c.c.Do(req)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+    if resp.StatusCode > 400 {
+        return nil, errors.New(resp.Status)
+    }
+    return resp, err
+}
 
-	buf := bytes.NewBuffer(bodyData)
-
-	req, err := http.NewRequest(method, url, buf)
+func (c *Client) Do(method string, url string, body map[string]interface{}) (*Response, error) {
+	bodyData, err := json.Marshal(body)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
 
-	resp, err := c.c.Do(req)
+	buf := bytes.NewBuffer(bodyData)
+
+	resp, err := c.DoRequest(method, url, buf)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
@@ -178,12 +196,7 @@ func (c *Client) DoBulk(url string, items []*BulkRequest) (*BulkResponse, error)
 		}
 	}
 
-	req, err := http.NewRequest("POST", url, &buf)
-	if err != nil {
-		return nil, errors.Trace(err)
-	}
-
-	resp, err := c.c.Do(req)
+	resp, err := c.DoRequest("POST", url, &buf)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
diff --git a/elastic/client_test.go b/elastic/client_test.go
index 74214317..0a09510f 100644
--- a/elastic/client_test.go
+++ b/elastic/client_test.go
@@ -22,7 +22,7 @@ type elasticTestSuite struct {
 var _ = Suite(&elasticTestSuite{})
 
 func (s *elasticTestSuite) SetUpSuite(c *C) {
-	s.c = NewClient(fmt.Sprintf("%s:%d", *host, *port))
+	s.c = NewClient(fmt.Sprintf("%s:%d", *host, *port), "", "")
 }
 
 func (s *elasticTestSuite) TearDownSuite(c *C) {
diff --git a/river/config.go b/river/config.go
index c6f626f3..a5777dca 100644
--- a/river/config.go
+++ b/river/config.go
@@ -19,7 +19,9 @@ type Config struct {
 	MyPassword string `toml:"my_pass"`
 	MyCharset  string `toml:"my_charset"`
 
-	ESAddr string `toml:"es_addr"`
+	ESAddr     string `toml:"es_addr"`
+	ESUser     string `toml:"es_user"`
+	ESPassword string `toml:"es_pass"`
 
 	StatAddr string `toml:"stat_addr"`
 
diff --git a/river/river.go b/river/river.go
index 61e7121f..926d22fd 100644
--- a/river/river.go
+++ b/river/river.go
@@ -66,7 +66,8 @@ func NewRiver(c *Config) (*River, error) {
 		return nil, errors.Trace(err)
 	}
 
-	r.es = elastic.NewClient(r.c.ESAddr)
+	r.es = elastic.NewClient(r.c.ESAddr, r.c.ESUser, r.c.ESPassword)
+
 
 	r.st = &stat{r: r}
 	go r.st.Run(r.c.StatAddr)
diff --git a/river/river_test.go b/river/river_test.go
index 0cc7e28d..3a32c649 100644
--- a/river/river_test.go
+++ b/river/river_test.go
@@ -124,7 +124,8 @@ my_user = "root"
 my_pass = ""
 my_charset = "utf8"
 es_addr = "127.0.0.1:9200"
-
+es_user = ""
+es_pass = ""
 data_dir = "./var"
 
 [[source]]

From dc8a11884360c9e0a9be8aaf2798d87cf484573c Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Wed, 17 May 2017 20:21:40 +0800
Subject: [PATCH 18/65] format code

---
 elastic/client.go | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/elastic/client.go b/elastic/client.go
index da6f480a..607ac83b 100644
--- a/elastic/client.go
+++ b/elastic/client.go
@@ -151,10 +151,10 @@ func (c *Client) DoRequest(method string, url string, body *bytes.Buffer) (*http
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
-    if resp.StatusCode > 400 {
-        return nil, errors.New(resp.Status)
-    }
-    return resp, err
+	if resp.StatusCode > 400 {
+		return nil, errors.New(resp.Status)
+    	}
+    	return resp, err
 }
 
 func (c *Client) Do(method string, url string, body map[string]interface{}) (*Response, error) {

From f8e619faf34227dcac60d1aa28410bc78de040e1 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Wed, 17 May 2017 20:28:14 +0800
Subject: [PATCH 19/65] Update river.toml

---
 etc/river.toml | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/etc/river.toml b/etc/river.toml
index 0b9d5d68..553b0062 100644
--- a/etc/river.toml
+++ b/etc/river.toml
@@ -7,6 +7,9 @@ my_charset = "utf8"
 
 # Elasticsearch address
 es_addr = "127.0.0.1:9200"
+# Elasticsearch user and password, maybe set by shield, nginx, or x-pack
+es_user = ""
+es_pass = ""
 
 # Path to store data, like master.info, if not set or empty,
 # we must use this to support breakpoint resume syncing. 

From d516e6c595ac5864b47273893c2f42915719c39f Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Wed, 17 May 2017 20:48:51 +0800
Subject: [PATCH 20/65] Update river_test.go

---
 river/river_test.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/river/river_test.go b/river/river_test.go
index 3a32c649..48e8cd7e 100644
--- a/river/river_test.go
+++ b/river/river_test.go
@@ -103,8 +103,8 @@ func (s *riverTestSuite) SetUpSuite(c *C) {
 	s.r, err = NewRiver(cfg)
 	c.Assert(err, IsNil)
 
-	err = s.r.es.DeleteIndex("river")
-	c.Assert(err, IsNil)
+	//err = s.r.es.DeleteIndex("river")
+	//c.Assert(err, IsNil)
 }
 
 func (s *riverTestSuite) TearDownSuite(c *C) {

From 230b0291e4cdb47d957bc1df44f9bc319c6a753a Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Wed, 17 May 2017 20:57:49 +0800
Subject: [PATCH 21/65] Update client_test.go

---
 elastic/client_test.go | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/elastic/client_test.go b/elastic/client_test.go
index 0a09510f..89477c9a 100644
--- a/elastic/client_test.go
+++ b/elastic/client_test.go
@@ -58,9 +58,6 @@ func (s *elasticTestSuite) TestSimple(c *C) {
 	err = s.c.Delete(index, docType, "1")
 	c.Assert(err, IsNil)
 
-	err = s.c.Delete(index, docType, "1")
-	c.Assert(err, IsNil)
-
 	exists, err = s.c.Exists(index, docType, "1")
 	c.Assert(err, IsNil)
 	c.Assert(exists, Equals, false)

From 84dd57b93632fe28c7dfd12486a58a34f70bb20c Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Wed, 17 May 2017 21:02:24 +0800
Subject: [PATCH 22/65] Update river_test.go

---
 river/river_test.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/river/river_test.go b/river/river_test.go
index 48e8cd7e..3dd3e3e3 100644
--- a/river/river_test.go
+++ b/river/river_test.go
@@ -199,7 +199,7 @@ func (s *riverTestSuite) testElasticGet(c *C, id string) *elastic.Response {
 	docType := "river"
 
 	r, err := s.r.es.Get(index, docType, id)
-	c.Assert(err, IsNil)
+	//c.Assert(err, IsNil) 
 
 	return r
 }

From 8c8bcebed0124e0be815da7651c57c6468eedd34 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Fri, 19 May 2017 21:23:05 +0800
Subject: [PATCH 23/65] add config struct for elasticsearch

---
 elastic/client.go      | 14 ++++++++++----
 elastic/client_test.go |  6 +++++-
 river/river.go         |  6 +++++-
 3 files changed, 20 insertions(+), 6 deletions(-)

diff --git a/elastic/client.go b/elastic/client.go
index da6f480a..7f7ad1c7 100644
--- a/elastic/client.go
+++ b/elastic/client.go
@@ -21,13 +21,19 @@ type Client struct {
 	c *http.Client
 }
 
+type ClientConfig struct {
+    Addr string
+    User string
+    Password string
+}
+
 
-func NewClient(addr string, user string, password string) *Client {
+func NewClient(conf *ClientConfig) *Client {
 	c := new(Client)
 
-	c.Addr = addr
-	c.User = user
-	c.Password = password
+	c.Addr = conf.Addr
+	c.User = conf.User
+	c.Password = conf.Password
 
 	c.c = &http.Client{}
 
diff --git a/elastic/client_test.go b/elastic/client_test.go
index 0a09510f..24fd018e 100644
--- a/elastic/client_test.go
+++ b/elastic/client_test.go
@@ -22,7 +22,11 @@ type elasticTestSuite struct {
 var _ = Suite(&elasticTestSuite{})
 
 func (s *elasticTestSuite) SetUpSuite(c *C) {
-	s.c = NewClient(fmt.Sprintf("%s:%d", *host, *port), "", "")
+    cfg := new(ClientConfig)
+    cfg.Addr = fmt.Sprintf("%s:%d", *host, *port)
+    cfg.User = ""
+    cfg.Password = ""
+	s.c = NewClient(cfg)
 }
 
 func (s *elasticTestSuite) TearDownSuite(c *C) {
diff --git a/river/river.go b/river/river.go
index 926d22fd..b382918c 100644
--- a/river/river.go
+++ b/river/river.go
@@ -66,7 +66,11 @@ func NewRiver(c *Config) (*River, error) {
 		return nil, errors.Trace(err)
 	}
 
-	r.es = elastic.NewClient(r.c.ESAddr, r.c.ESUser, r.c.ESPassword)
+    cfg := new(elastic.ClientConfig)
+    cfg.Addr = r.c.ESAddr
+    cfg.User = r.c.ESUser
+    cfg.Password = r.c.ESPassword
+	r.es = elastic.NewClient(cfg)
 
 
 	r.st = &stat{r: r}

From fbe71a9720d873b8283c716f7710123cf5569715 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Fri, 19 May 2017 21:27:13 +0800
Subject: [PATCH 24/65] Update client_test.go

---
 elastic/client_test.go | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/elastic/client_test.go b/elastic/client_test.go
index d1585da1..616b82de 100644
--- a/elastic/client_test.go
+++ b/elastic/client_test.go
@@ -22,10 +22,10 @@ type elasticTestSuite struct {
 var _ = Suite(&elasticTestSuite{})
 
 func (s *elasticTestSuite) SetUpSuite(c *C) {
-    cfg := new(ClientConfig)
-    cfg.Addr = fmt.Sprintf("%s:%d", *host, *port)
-    cfg.User = ""
-    cfg.Password = ""
+	cfg := new(ClientConfig)
+	cfg.Addr = fmt.Sprintf("%s:%d", *host, *port)
+	cfg.User = ""
+	cfg.Password = ""
 	s.c = NewClient(cfg)
 }
 

From 1e0d0d46756b5ca097fd732a7c36e766103e594c Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Fri, 19 May 2017 21:27:40 +0800
Subject: [PATCH 25/65] Update client.go

---
 elastic/client.go | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/elastic/client.go b/elastic/client.go
index e2020ba1..d54052b7 100644
--- a/elastic/client.go
+++ b/elastic/client.go
@@ -22,9 +22,9 @@ type Client struct {
 }
 
 type ClientConfig struct {
-    Addr string
-    User string
-    Password string
+	Addr string
+	User string
+	Password string
 }
 
 

From 0629e8b9db5246288933327a6c9bc05bb22b703e Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Fri, 19 May 2017 21:31:25 +0800
Subject: [PATCH 26/65] Update river_test.go

---
 river/river_test.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/river/river_test.go b/river/river_test.go
index 3dd3e3e3..bcb3928c 100644
--- a/river/river_test.go
+++ b/river/river_test.go
@@ -199,7 +199,7 @@ func (s *riverTestSuite) testElasticGet(c *C, id string) *elastic.Response {
 	docType := "river"
 
 	r, err := s.r.es.Get(index, docType, id)
-	//c.Assert(err, IsNil) 
+	c.Assert(err, IsNil) 
 
 	return r
 }

From 912e9a8fd1bcbc53b50954aff1aa0157a9bee0d7 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Fri, 19 May 2017 21:34:42 +0800
Subject: [PATCH 27/65] Update river.go

---
 river/river.go | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/river/river.go b/river/river.go
index b382918c..d60507b4 100644
--- a/river/river.go
+++ b/river/river.go
@@ -66,10 +66,10 @@ func NewRiver(c *Config) (*River, error) {
 		return nil, errors.Trace(err)
 	}
 
-    cfg := new(elastic.ClientConfig)
-    cfg.Addr = r.c.ESAddr
-    cfg.User = r.c.ESUser
-    cfg.Password = r.c.ESPassword
+	cfg := new(elastic.ClientConfig)
+	cfg.Addr = r.c.ESAddr
+	cfg.User = r.c.ESUser
+	cfg.Password = r.c.ESPassword
 	r.es = elastic.NewClient(cfg)
 
 

From 5e237e0d4b9c463ab34bb84e22df0be5866d54a0 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Mon, 29 May 2017 20:49:03 +0800
Subject: [PATCH 28/65] if http response code is not 2XX, return error

---
 elastic/client.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/elastic/client.go b/elastic/client.go
index d54052b7..7fe70c13 100644
--- a/elastic/client.go
+++ b/elastic/client.go
@@ -157,7 +157,7 @@ func (c *Client) DoRequest(method string, url string, body *bytes.Buffer) (*http
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
-	if resp.StatusCode > 400 {
+	if resp.StatusCode >= 300 || resp.StatusCode < 200 {
 		return nil, errors.New(resp.Status)
     	}
     	return resp, err

From 5691ecdbcc6f49e6ed0458f7b4a8a99e8967c0f9 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Mon, 29 May 2017 20:58:53 +0800
Subject: [PATCH 29/65] Update client_test.go

---
 elastic/client_test.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/elastic/client_test.go b/elastic/client_test.go
index 616b82de..58146067 100644
--- a/elastic/client_test.go
+++ b/elastic/client_test.go
@@ -63,8 +63,8 @@ func (s *elasticTestSuite) TestSimple(c *C) {
 	c.Assert(err, IsNil)
 
 	exists, err = s.c.Exists(index, docType, "1")
-	c.Assert(err, IsNil)
-	c.Assert(exists, Equals, false)
+	c.Assert(err, NotNil)
+	c.Assert(exists, IsNil)
 
 	items := make([]*BulkRequest, 10)
 

From 37d79616a3b6dccbed078001fd1226b641c62c8b Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Mon, 29 May 2017 21:11:10 +0800
Subject: [PATCH 30/65] Update client_test.go

---
 elastic/client_test.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/elastic/client_test.go b/elastic/client_test.go
index 58146067..310ec024 100644
--- a/elastic/client_test.go
+++ b/elastic/client_test.go
@@ -64,7 +64,7 @@ func (s *elasticTestSuite) TestSimple(c *C) {
 
 	exists, err = s.c.Exists(index, docType, "1")
 	c.Assert(err, NotNil)
-	c.Assert(exists, IsNil)
+	c.Assert(exists, Equals, false)
 
 	items := make([]*BulkRequest, 10)
 

From 1b94f3247cdc4649a6caf51cf00636ba5d8831ff Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Mon, 29 May 2017 21:28:41 +0800
Subject: [PATCH 31/65] Update river_test.go

---
 river/river_test.go | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/river/river_test.go b/river/river_test.go
index bcb3928c..1efba134 100644
--- a/river/river_test.go
+++ b/river/river_test.go
@@ -199,7 +199,11 @@ func (s *riverTestSuite) testElasticGet(c *C, id string) *elastic.Response {
 	docType := "river"
 
 	r, err := s.r.es.Get(index, docType, id)
-	c.Assert(err, IsNil) 
+	if err != nil {
+		r := new(Response)
+		r.Found = false
+		return r
+	}
 
 	return r
 }

From e136df663ed2e75b37851221672675860eb072ae Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Mon, 29 May 2017 21:32:31 +0800
Subject: [PATCH 32/65] Update river_test.go

---
 river/river_test.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/river/river_test.go b/river/river_test.go
index 1efba134..c7a21c77 100644
--- a/river/river_test.go
+++ b/river/river_test.go
@@ -200,7 +200,7 @@ func (s *riverTestSuite) testElasticGet(c *C, id string) *elastic.Response {
 
 	r, err := s.r.es.Get(index, docType, id)
 	if err != nil {
-		r := new(Response)
+		r := new(elastic.Response)
 		r.Found = false
 		return r
 	}

From 72235fd383dd538a3ea403d7580e07480159885d Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Mon, 29 May 2017 21:44:15 +0800
Subject: [PATCH 33/65] Update river_extra_test.go

---
 river/river_extra_test.go | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/river/river_extra_test.go b/river/river_extra_test.go
index 84b4c894..4dd9cb7d 100644
--- a/river/river_extra_test.go
+++ b/river/river_extra_test.go
@@ -97,12 +97,11 @@ func (s *riverTestSuite) testElasticExtraExists(c *C, id string, parent string,
 		url.QueryEscape(parent))
 
 	r, err := s.r.es.Do("HEAD", reqUrl, nil)
-	c.Assert(err, IsNil)
 
 	if exist {
 		c.Assert(r.Code, Equals, http.StatusOK)
 	} else {
-		c.Assert(r.Code, Equals, http.StatusNotFound)
+		c.Assert(err, NotNil)
 	}
 }
 

From 351742e5cc9679dd9527e0b1cf8c8f26830f450b Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Sat, 3 Jun 2017 23:30:10 +0800
Subject: [PATCH 34/65] Update client_test.go

---
 elastic/client_test.go | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git a/elastic/client_test.go b/elastic/client_test.go
index 310ec024..36a6b1a6 100644
--- a/elastic/client_test.go
+++ b/elastic/client_test.go
@@ -98,7 +98,16 @@ func (s *elasticTestSuite) TestSimple(c *C) {
 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++ {

From 66315f307847f2c38d94bea9f2ad4fe9c585b7af Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 4 Jun 2017 17:26:23 +0800
Subject: [PATCH 35/65] modify test code

---
 elastic/client.go         | 31 ++++++++++++++-----------------
 elastic/client_test.go    | 32 +++++++++++++++++---------------
 river/river_extra_test.go |  2 +-
 3 files changed, 32 insertions(+), 33 deletions(-)

diff --git a/elastic/client.go b/elastic/client.go
index 7fe70c13..103d6c69 100644
--- a/elastic/client.go
+++ b/elastic/client.go
@@ -224,29 +224,26 @@ func (c *Client) DoBulk(url string, items []*BulkRequest) (*BulkResponse, error)
 	return ret, errors.Trace(err)
 }
 
-func (c *Client) CreateMapping(index string, docType string, mapping map[string]interface{}) error {
+func (c *Client) CreateMapping(index string, mapping map[string]interface{}) error {
 	reqUrl := fmt.Sprintf("http://%s/%s", c.Addr,
 		url.QueryEscape(index))
 
-	r, err := c.Do("HEAD", reqUrl, nil)
-	if err != nil {
-		return errors.Trace(err)
-	}
+	_, err := c.Do("HEAD", reqUrl, nil)
 
 	// index doesn't exist, create index first
-	if r.Code != http.StatusOK {
-		_, err = c.Do("POST", reqUrl, nil)
-
-		if err != nil {
-			return errors.Trace(err)
-		}
+	//if err != nil {
+	//	_, err = c.Do("POST", reqUrl, nil)
+
+	//	if err != nil {
+	//		return errors.Trace(err)
+	//	}
+	//}
+	if err!= nil {
+		reqUrl = fmt.Sprintf("http://%s/%s", c.Addr,
+		    url.QueryEscape(index))
+
+	    _, err = c.Do("PUT", reqUrl, mapping)
 	}
-
-	reqUrl = fmt.Sprintf("http://%s/%s/%s/_mapping", c.Addr,
-		url.QueryEscape(index),
-		url.QueryEscape(docType))
-
-	_, err = c.Do("POST", reqUrl, mapping)
 	return errors.Trace(err)
 }
 
diff --git a/elastic/client_test.go b/elastic/client_test.go
index 36a6b1a6..3f6b14c3 100644
--- a/elastic/client_test.go
+++ b/elastic/client_test.go
@@ -99,15 +99,17 @@ 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},
+		"mappings": map[string]interface{}{
+			docType: map[string]interface{}{
+				"_parent": map[string]string{"type": ParentType},
 		},
+	    },
 	}
-	err:= s.c.CreateMapping(index, docType, mapping)
+	err := s.c.CreateMapping(index, mapping)
 	c.Assert(err, IsNil)
-	
+
 	items := make([]*BulkRequest, 10)
 
 	for i := 0; i < 10; i++ {
@@ -124,14 +126,14 @@ func (s *elasticTestSuite) TestParent(c *C) {
 	c.Assert(err, IsNil)
 	c.Assert(resp.Errors, Equals, false)
 
-	for i := 0; i < 10; i++ {
-		id := fmt.Sprintf("%d", i)
-		req := new(BulkRequest)
-		req.Action = ActionDelete
-		req.ID = id
-		items[i] = req
-	}
-	resp, err = s.c.IndexTypeBulk(index, docType, items)
-	c.Assert(err, IsNil)
-	c.Assert(resp.Errors, Equals, false)
+	//for i := 0; i < 10; i++ {
+	//	id := fmt.Sprintf("%d", i)
+	//	req := new(BulkRequest)
+	//	req.Action = ActionDelete
+	//	req.ID = id
+	//	items[i] = req
+	//}
+	//resp, err = s.c.IndexTypeBulk(index, docType, items)
+	//c.Assert(err, IsNil)
+	//c.Assert(resp.Errors, Equals, false)
 }
diff --git a/river/river_extra_test.go b/river/river_extra_test.go
index 4dd9cb7d..8e319f6d 100644
--- a/river/river_extra_test.go
+++ b/river/river_extra_test.go
@@ -73,7 +73,7 @@ func (s *riverTestSuite) setupExtra(c *C) (r *River) {
 		},
 	}
 
-	r.es.CreateMapping("river", "river_extra", mapping)
+	r.es.CreateMapping("river", mapping)
 
 	return r
 }

From 1e21b3c29fcdacfdd9a9a023e9d165d5dc5a8ce9 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Tue, 6 Jun 2017 22:41:19 +0800
Subject: [PATCH 36/65] modify code about create mapping

---
 elastic/client.go         | 30 +++++++++++++++---------------
 elastic/client_test.go    |  5 ++---
 river/river_extra_test.go |  2 +-
 3 files changed, 18 insertions(+), 19 deletions(-)

diff --git a/elastic/client.go b/elastic/client.go
index 103d6c69..0fa1f417 100644
--- a/elastic/client.go
+++ b/elastic/client.go
@@ -159,8 +159,8 @@ func (c *Client) DoRequest(method string, url string, body *bytes.Buffer) (*http
 	}
 	if resp.StatusCode >= 300 || resp.StatusCode < 200 {
 		return nil, errors.New(resp.Status)
-    	}
-    	return resp, err
+	}
+	return resp, err
 }
 
 func (c *Client) Do(method string, url string, body map[string]interface{}) (*Response, error) {
@@ -224,26 +224,26 @@ func (c *Client) DoBulk(url string, items []*BulkRequest) (*BulkResponse, error)
 	return ret, errors.Trace(err)
 }
 
-func (c *Client) CreateMapping(index string, mapping map[string]interface{}) error {
+func (c *Client) CreateMapping(index string, docType string, mapping map[string]interface{}) error {
 	reqUrl := fmt.Sprintf("http://%s/%s", c.Addr,
 		url.QueryEscape(index))
 
 	_, err := c.Do("HEAD", reqUrl, nil)
 
-	// index doesn't exist, create index first
-	//if err != nil {
-	//	_, err = c.Do("POST", reqUrl, nil)
-
-	//	if err != nil {
-	//		return errors.Trace(err)
-	//	}
-	//}
-	if err!= nil {
-		reqUrl = fmt.Sprintf("http://%s/%s", c.Addr,
-		    url.QueryEscape(index))
+	// if index doesn't exist, will get 404 not found error, create index first
+	if err != nil {
+		_, err = c.Do("PUT", reqUrl, nil)
 
-	    _, err = c.Do("PUT", reqUrl, mapping)
+		if err != nil {
+			return errors.Trace(err)
+		}
 	}
+
+	reqUrl = fmt.Sprintf("http://%s/%s/%s/_mapping", c.Addr,
+		url.QueryEscape(index),
+		url.QueryEscape(docType))
+
+	_, err = c.Do("POST", reqUrl, mapping)
 	return errors.Trace(err)
 }
 
diff --git a/elastic/client_test.go b/elastic/client_test.go
index 3f6b14c3..c315881d 100644
--- a/elastic/client_test.go
+++ b/elastic/client_test.go
@@ -100,14 +100,13 @@ func (s *elasticTestSuite) TestParent(c *C) {
 	docType := "comment"
 	ParentType := "parent"
 
+	//"mappings": map[string]interface{}{
 	mapping := map[string]interface{}{
-		"mappings": map[string]interface{}{
 			docType: map[string]interface{}{
 				"_parent": map[string]string{"type": ParentType},
 		},
-	    },
 	}
-	err := s.c.CreateMapping(index, mapping)
+	err := s.c.CreateMapping(index, docType,  mapping)
 	c.Assert(err, IsNil)
 
 	items := make([]*BulkRequest, 10)
diff --git a/river/river_extra_test.go b/river/river_extra_test.go
index 8e319f6d..4dd9cb7d 100644
--- a/river/river_extra_test.go
+++ b/river/river_extra_test.go
@@ -73,7 +73,7 @@ func (s *riverTestSuite) setupExtra(c *C) (r *River) {
 		},
 	}
 
-	r.es.CreateMapping("river", mapping)
+	r.es.CreateMapping("river", "river_extra", mapping)
 
 	return r
 }

From 7a7786db8b2a6849ebfb5d6b15f768f1cfae3424 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 18 Jun 2017 14:55:14 +0800
Subject: [PATCH 37/65] judge http error in sync.go

---
 elastic/client.go      | 7 +------
 elastic/client_test.go | 2 +-
 river/sync.go          | 2 +-
 3 files changed, 3 insertions(+), 8 deletions(-)

diff --git a/elastic/client.go b/elastic/client.go
index 0fa1f417..8b9a0224 100644
--- a/elastic/client.go
+++ b/elastic/client.go
@@ -154,12 +154,7 @@ func (c *Client) DoRequest(method string, url string, body *bytes.Buffer) (*http
 		req.SetBasicAuth(c.User, c.Password)
 	}
 	resp, err := c.c.Do(req)
-	if err != nil {
-		return nil, errors.Trace(err)
-	}
-	if resp.StatusCode >= 300 || resp.StatusCode < 200 {
-		return nil, errors.New(resp.Status)
-	}
+
 	return resp, err
 }
 
diff --git a/elastic/client_test.go b/elastic/client_test.go
index c315881d..129b7f76 100644
--- a/elastic/client_test.go
+++ b/elastic/client_test.go
@@ -63,7 +63,7 @@ func (s *elasticTestSuite) TestSimple(c *C) {
 	c.Assert(err, IsNil)
 
 	exists, err = s.c.Exists(index, docType, "1")
-	c.Assert(err, NotNil)
+	c.Assert(err, IsNil)
 	c.Assert(exists, Equals, false)
 
 	items := make([]*BulkRequest, 10)
diff --git a/river/sync.go b/river/sync.go
index 5f914d08..c9f90669 100644
--- a/river/sync.go
+++ b/river/sync.go
@@ -439,7 +439,7 @@ func (r *River) doBulk(reqs []*elastic.BulkRequest) error {
 	if resp, err := r.es.Bulk(reqs); err != nil {
 		log.Errorf("sync docs err %v after binlog %s", err, r.canal.SyncedPosition())
 		return errors.Trace(err)
-	} else if resp.Errors {
+	} else if resp.Code >= 300 || resp.Code < 200 || resp.Errors {
 		for i := 0; i < len(resp.Items); i++ {
 			for action, item := range resp.Items[i] {
 				if len(item.Error) > 0 {

From 767e1c714e7be96e48809056a335544c0413dd0b Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 18 Jun 2017 21:24:22 +0800
Subject: [PATCH 38/65] judge http response status

---
 elastic/client.go | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/elastic/client.go b/elastic/client.go
index 8b9a0224..99b088d6 100644
--- a/elastic/client.go
+++ b/elastic/client.go
@@ -223,10 +223,13 @@ func (c *Client) CreateMapping(index string, docType string, mapping map[string]
 	reqUrl := fmt.Sprintf("http://%s/%s", c.Addr,
 		url.QueryEscape(index))
 
-	_, err := c.Do("HEAD", reqUrl, nil)
+	r, err := c.Do("HEAD", reqUrl, nil)
+	if err != nil {
+		return errors.Trace(err)
+	}
 
 	// if index doesn't exist, will get 404 not found error, create index first
-	if err != nil {
+	if r.Code != http.StatusOK {
 		_, err = c.Do("PUT", reqUrl, nil)
 
 		if err != nil {

From 92ec5ad54291f43faee2b14652e8c388c9ae4288 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 18 Jun 2017 21:30:15 +0800
Subject: [PATCH 39/65] fix bug in test code

---
 river/river_extra_test.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/river/river_extra_test.go b/river/river_extra_test.go
index 4dd9cb7d..b73d9701 100644
--- a/river/river_extra_test.go
+++ b/river/river_extra_test.go
@@ -101,7 +101,7 @@ func (s *riverTestSuite) testElasticExtraExists(c *C, id string, parent string,
 	if exist {
 		c.Assert(r.Code, Equals, http.StatusOK)
 	} else {
-		c.Assert(err, NotNil)
+		c.Assert(r.Code, Equals, http.StatusNotFound)
 	}
 }
 

From dc17f525575c381b4204a64c48d9106cbea17131 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 18 Jun 2017 21:33:18 +0800
Subject: [PATCH 40/65] fix bug in test code

---
 river/river_extra_test.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/river/river_extra_test.go b/river/river_extra_test.go
index b73d9701..84b4c894 100644
--- a/river/river_extra_test.go
+++ b/river/river_extra_test.go
@@ -97,6 +97,7 @@ func (s *riverTestSuite) testElasticExtraExists(c *C, id string, parent string,
 		url.QueryEscape(parent))
 
 	r, err := s.r.es.Do("HEAD", reqUrl, nil)
+	c.Assert(err, IsNil)
 
 	if exist {
 		c.Assert(r.Code, Equals, http.StatusOK)

From b2b1222a1560d8b656f4b07ebb585051e08ebce4 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Mon, 19 Jun 2017 20:39:39 +0800
Subject: [PATCH 41/65] Update river_test.go

---
 river/river_test.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/river/river_test.go b/river/river_test.go
index c7a21c77..07a5a1f8 100644
--- a/river/river_test.go
+++ b/river/river_test.go
@@ -103,8 +103,8 @@ func (s *riverTestSuite) SetUpSuite(c *C) {
 	s.r, err = NewRiver(cfg)
 	c.Assert(err, IsNil)
 
-	//err = s.r.es.DeleteIndex("river")
-	//c.Assert(err, IsNil)
+	err = s.r.es.DeleteIndex("river")
+	c.Assert(err, IsNil)
 }
 
 func (s *riverTestSuite) TearDownSuite(c *C) {

From 9ce43fb4ed425f80984340acf7d497c8cc54f927 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Mon, 19 Jun 2017 20:40:18 +0800
Subject: [PATCH 42/65] Update sync.go

---
 river/sync.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/river/sync.go b/river/sync.go
index c9f90669..2c48d1a4 100644
--- a/river/sync.go
+++ b/river/sync.go
@@ -439,7 +439,7 @@ func (r *River) doBulk(reqs []*elastic.BulkRequest) error {
 	if resp, err := r.es.Bulk(reqs); err != nil {
 		log.Errorf("sync docs err %v after binlog %s", err, r.canal.SyncedPosition())
 		return errors.Trace(err)
-	} else if resp.Code >= 300 || resp.Code < 200 || resp.Errors {
+	} else if resp.Code / 100 == 2 || resp.Errors {
 		for i := 0; i < len(resp.Items); i++ {
 			for action, item := range resp.Items[i] {
 				if len(item.Error) > 0 {

From c98d66586ed5ac9932cea4cc4b8fa8f7890d069f Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Mon, 19 Jun 2017 20:49:15 +0800
Subject: [PATCH 43/65] Update river_test.go

---
 river/river_test.go | 6 +-----
 1 file changed, 1 insertion(+), 5 deletions(-)

diff --git a/river/river_test.go b/river/river_test.go
index 07a5a1f8..3a32c649 100644
--- a/river/river_test.go
+++ b/river/river_test.go
@@ -199,11 +199,7 @@ func (s *riverTestSuite) testElasticGet(c *C, id string) *elastic.Response {
 	docType := "river"
 
 	r, err := s.r.es.Get(index, docType, id)
-	if err != nil {
-		r := new(elastic.Response)
-		r.Found = false
-		return r
-	}
+	c.Assert(err, IsNil)
 
 	return r
 }

From 447dc8228a4167df9f9547715ebae2221841b348 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Mon, 19 Jun 2017 21:17:50 +0800
Subject: [PATCH 44/65] Update client.go

---
 elastic/client.go | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/elastic/client.go b/elastic/client.go
index 99b088d6..02f13067 100644
--- a/elastic/client.go
+++ b/elastic/client.go
@@ -229,12 +229,14 @@ func (c *Client) CreateMapping(index string, docType string, mapping map[string]
 	}
 
 	// if index doesn't exist, will get 404 not found error, create index first
-	if r.Code != http.StatusOK {
+	if r.Code == StatusNotFound {
 		_, err = c.Do("PUT", reqUrl, nil)
 
 		if err != nil {
 			return errors.Trace(err)
 		}
+	} else if r.Code != StatusOK {
+		return errors.Trace(err)
 	}
 
 	reqUrl = fmt.Sprintf("http://%s/%s/%s/_mapping", c.Addr,

From f5b227965dd2fdfab257e0852c64b1edfe44ecbf Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Mon, 19 Jun 2017 21:18:22 +0800
Subject: [PATCH 45/65] Update client_test.go

---
 elastic/client_test.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/elastic/client_test.go b/elastic/client_test.go
index 2e56cd92..a9650b98 100644
--- a/elastic/client_test.go
+++ b/elastic/client_test.go
@@ -102,7 +102,7 @@ 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},

From 0ca37e39bfe0259b13876d0db6824cca12eaed21 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Mon, 19 Jun 2017 21:20:27 +0800
Subject: [PATCH 46/65] Update client.go

---
 elastic/client.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/elastic/client.go b/elastic/client.go
index 02f13067..3f355d1a 100644
--- a/elastic/client.go
+++ b/elastic/client.go
@@ -228,7 +228,7 @@ func (c *Client) CreateMapping(index string, docType string, mapping map[string]
 		return errors.Trace(err)
 	}
 
-	// if index doesn't exist, will get 404 not found error, create index first
+	// if index doesn't exist, will get 404 not found, create index first
 	if r.Code == StatusNotFound {
 		_, err = c.Do("PUT", reqUrl, nil)
 

From 4f403dbfa615184f52de85eb021ad0f4896b1232 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Mon, 19 Jun 2017 21:22:49 +0800
Subject: [PATCH 47/65] Update client.go

---
 elastic/client.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/elastic/client.go b/elastic/client.go
index 3f355d1a..f16cf88b 100644
--- a/elastic/client.go
+++ b/elastic/client.go
@@ -229,13 +229,13 @@ func (c *Client) CreateMapping(index string, docType string, mapping map[string]
 	}
 
 	// if index doesn't exist, will get 404 not found, create index first
-	if r.Code == StatusNotFound {
+	if r.Code == http.StatusNotFound {
 		_, err = c.Do("PUT", reqUrl, nil)
 
 		if err != nil {
 			return errors.Trace(err)
 		}
-	} else if r.Code != StatusOK {
+	} else if r.Code != http.StatusOK {
 		return errors.Trace(err)
 	}
 

From 6c68f844c8fc8aeef2dad3991dc7247c901b13be Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Mon, 19 Jun 2017 22:30:17 +0800
Subject: [PATCH 48/65] Update client.go

---
 elastic/client.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/elastic/client.go b/elastic/client.go
index f16cf88b..204b4f75 100644
--- a/elastic/client.go
+++ b/elastic/client.go
@@ -236,7 +236,7 @@ func (c *Client) CreateMapping(index string, docType string, mapping map[string]
 			return errors.Trace(err)
 		}
 	} else if r.Code != http.StatusOK {
-		return errors.Trace(err)
+		return errors.New(resp.Status)
 	}
 
 	reqUrl = fmt.Sprintf("http://%s/%s/%s/_mapping", c.Addr,

From 6f3c08067200c9f58e69750b3da6c80ad67fdec4 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Tue, 20 Jun 2017 14:18:45 +0800
Subject: [PATCH 49/65] Update client.go

---
 elastic/client.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/elastic/client.go b/elastic/client.go
index 204b4f75..25935321 100644
--- a/elastic/client.go
+++ b/elastic/client.go
@@ -236,7 +236,7 @@ func (c *Client) CreateMapping(index string, docType string, mapping map[string]
 			return errors.Trace(err)
 		}
 	} else if r.Code != http.StatusOK {
-		return errors.New(resp.Status)
+		return errors.Errorf("Error: %s, code: %d", http.StatusText(r.Code), r.Code)
 	}
 
 	reqUrl = fmt.Sprintf("http://%s/%s/%s/_mapping", c.Addr,

From 8ab17231e75e42db0530b934909afa122ad22381 Mon Sep 17 00:00:00 2001
From: gzwangxiang2015 <gzwangxiang2015@bj48-176.i.netease.com>
Date: Fri, 23 Jun 2017 22:44:36 +0800
Subject: [PATCH 50/65] add support of parse json field

---
 river/sync.go | 13 +++++++++++++
 1 file changed, 13 insertions(+)

diff --git a/river/sync.go b/river/sync.go
index 2c48d1a4..6ddaf114 100644
--- a/river/sync.go
+++ b/river/sync.go
@@ -6,6 +6,7 @@ import (
 	"reflect"
 	"strings"
 	"time"
+	"encoding/json"
 
 	"github.com/juju/errors"
 	"github.com/ngaut/log"
@@ -290,6 +291,18 @@ func (r *River) makeReqColumnData(col *schema.TableColumn, value interface{}) in
 		case []byte:
 			return string(value[:])
 		}
+	case schema.TYPE_JSON:
+		var f interface{}
+		var err error
+		switch v := value.(type) {
+		case string:
+			err = json.Unmarshal([]byte(v), &f)
+		case []byte:
+			err = json.Unmarshal(v, &f)
+		}
+		if err == nil && f != nil {
+			return f
+		}
 	}
 
 	return value

From 154366a89568e9f53265f6631224e80d15c25fb4 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 25 Jun 2017 13:57:10 +0800
Subject: [PATCH 51/65] add test of json field

---
 river/river_test.go | 44 ++++++++++++++++++++++++++++++++++++--------
 1 file changed, 36 insertions(+), 8 deletions(-)

diff --git a/river/river_test.go b/river/river_test.go
index 3a32c649..c90c96ae 100644
--- a/river/river_test.go
+++ b/river/river_test.go
@@ -45,9 +45,20 @@ func (s *riverTestSuite) SetUpSuite(c *C) {
             PRIMARY KEY(id)) ENGINE=INNODB;
     `
 
+	schema_json := `
+	CREATE TABLE IF NOT EXISTS %s (
+	    id INT,
+	    info JSON,
+	    PRIMARY KEY(id)) ENGINE=INNODB;
+    `
+
+	s.testExecute(c, "DROP TABLE IF EXISTS test_river")
+	s.testExecute(c, "DROP TABLE IF EXISTS test_for_id")
+	s.testExecute(c, "DROP TABLE IF EXISTS test_for_json")
 	s.testExecute(c, "DROP TABLE IF EXISTS test_river")
 	s.testExecute(c, fmt.Sprintf(schema, "test_river"))
 	s.testExecute(c, fmt.Sprintf(schema, "test_for_id"))
+	s.testExecute(c, fmt.Sprintf(schema_json, "test_for_json"))
 
 	for i := 0; i < 10; i++ {
 		table := fmt.Sprintf("test_river_%04d", i)
@@ -74,7 +85,7 @@ func (s *riverTestSuite) SetUpSuite(c *C) {
 
 	os.RemoveAll(cfg.DataDir)
 
-	cfg.Sources = []SourceConfig{SourceConfig{Schema: "test", Tables: []string{"test_river", "test_river_[0-9]{4}", "test_for_id"}}}
+	cfg.Sources = []SourceConfig{SourceConfig{Schema: "test", Tables: []string{"test_river", "test_river_[0-9]{4}", "test_for_id", "test_for_json"}}}
 
 	cfg.Rules = []*Rule{
 		&Rule{Schema: "test",
@@ -83,7 +94,7 @@ func (s *riverTestSuite) SetUpSuite(c *C) {
 			Type:         "river",
 			FieldMapping: map[string]string{"title": "es_title", "mylist": "es_mylist,list"},
 		},
-		
+
 		&Rule{Schema: "test",
 			Table:        "test_for_id",
 			Index:        "river",
@@ -98,6 +109,12 @@ func (s *riverTestSuite) SetUpSuite(c *C) {
 			Type:         "river",
 			FieldMapping: map[string]string{"title": "es_title", "mylist": "es_mylist,list"},
 		},
+
+		&Rule{Schema: "test",
+			Table:        "test_for_json",
+			Index:        "river",
+			Type:         "river",
+		},
 	}
 
 	s.r, err = NewRiver(cfg)
@@ -131,7 +148,7 @@ data_dir = "./var"
 [[source]]
 schema = "test"
 
-tables = ["test_river", "test_river_[0-9]{4}", "test_for_id"]
+tables = ["test_river", "test_river_[0-9]{4}", "test_for_id", "test_for_json"]
 
 [[rule]]
 schema = "test"
@@ -167,13 +184,18 @@ type = "river"
     title = "es_title"
     mylist = "es_mylist,list"
 
+[[rule]]
+schema = "test"
+table = "test_for_json"
+index = "river"
+type = "river"
 `
 
 	cfg, err := NewConfig(str)
 	c.Assert(err, IsNil)
 	c.Assert(cfg.Sources, HasLen, 1)
-	c.Assert(cfg.Sources[0].Tables, HasLen, 3)
-	c.Assert(cfg.Rules, HasLen, 3)
+	c.Assert(cfg.Sources[0].Tables, HasLen, 4)
+	c.Assert(cfg.Rules, HasLen, 4)
 }
 
 func (s *riverTestSuite) testExecute(c *C, query string, args ...interface{}) {
@@ -187,7 +209,8 @@ func (s *riverTestSuite) testPrepareData(c *C) {
 	s.testExecute(c, "INSERT INTO test_river (id, title, content, tenum, tset) VALUES (?, ?, ?, ?, ?)", 3, "third", "hello elaticsearch 3", "e3", "c")
 	s.testExecute(c, "INSERT INTO test_river (id, title, content, tenum, tset, tbit) VALUES (?, ?, ?, ?, ?, ?)", 4, "fouth", "hello go-mysql-elasticserach 4", "e1", "a,b,c", 0)
 	s.testExecute(c, "INSERT INTO test_for_id (id, title, content, tenum, tset) VALUES (?, ?, ?, ?, ?)", 1, "first", "hello go 1", "e1", "a,b")
-	
+	s.testExecute(c, "INSERT INTO test_for_json (id, json) VALUES (?, ?)", 9200, "{\"first\": \"a\", \"second\": \"b\"}")
+
 	for i := 0; i < 10; i++ {
 		table := fmt.Sprintf("test_river_%04d", i)
 		s.testExecute(c, fmt.Sprintf("INSERT INTO %s (id, title, content, tenum, tset) VALUES (?, ?, ?, ?, ?)", table), 5+i, "abc", "hello", "e1", "a,b,c")
@@ -233,10 +256,15 @@ func (s *riverTestSuite) TestRiver(c *C) {
 	c.Assert(r.Found, Equals, true)
 	c.Assert(r.Source["tenum"], Equals, "e1")
 	c.Assert(r.Source["tset"], Equals, "a,b")
-	
+
 	r = s.testElasticGet(c, "1:first")
 	c.Assert(r.Found, Equals, true)
-	
+
+	r = s.testElasticGet(c, "9200")
+	c.Assert(r.Found, Equals, true)
+	c.Assert(r.Source["info"]["first"], Equals, "a")
+	c.Assert(r.Source["info"]["second"], Equals, "b")
+
 	r = s.testElasticGet(c, "100")
 	c.Assert(r.Found, Equals, false)
 

From eb4b28d8ba7bc2df9638fb9d65e5e4b819c6cffe Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 25 Jun 2017 14:07:16 +0800
Subject: [PATCH 52/65] fix the bug of test

---
 river/river_test.go | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/river/river_test.go b/river/river_test.go
index c90c96ae..4f465fdf 100644
--- a/river/river_test.go
+++ b/river/river_test.go
@@ -262,8 +262,13 @@ func (s *riverTestSuite) TestRiver(c *C) {
 
 	r = s.testElasticGet(c, "9200")
 	c.Assert(r.Found, Equals, true)
-	c.Assert(r.Source["info"]["first"], Equals, "a")
-	c.Assert(r.Source["info"]["second"], Equals, "b")
+	switch v := r.source["info"].(type) {
+	case map[string]string:
+	    c.Assert(v["first"], Equals, "a")
+	    c.Assert(v["second"], Equals, "b")
+	default:
+	    c.Assert(true, Equals, false)
+	}
 
 	r = s.testElasticGet(c, "100")
 	c.Assert(r.Found, Equals, false)

From 5a4d85066f7c98a8efddfacffa524a5314b8f4d4 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 25 Jun 2017 14:21:40 +0800
Subject: [PATCH 53/65] fix the bug of test

---
 river/river_test.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/river/river_test.go b/river/river_test.go
index 4f465fdf..4aefc216 100644
--- a/river/river_test.go
+++ b/river/river_test.go
@@ -262,7 +262,7 @@ func (s *riverTestSuite) TestRiver(c *C) {
 
 	r = s.testElasticGet(c, "9200")
 	c.Assert(r.Found, Equals, true)
-	switch v := r.source["info"].(type) {
+	switch v := r.Source["info"].(type) {
 	case map[string]string:
 	    c.Assert(v["first"], Equals, "a")
 	    c.Assert(v["second"], Equals, "b")

From e417e927029a89a3f0bd5939640a5c1404bd3911 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 25 Jun 2017 14:35:36 +0800
Subject: [PATCH 54/65] fix the bug of test

---
 river/river_test.go | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/river/river_test.go b/river/river_test.go
index 4aefc216..e7a3a734 100644
--- a/river/river_test.go
+++ b/river/river_test.go
@@ -28,7 +28,7 @@ var _ = Suite(&riverTestSuite{})
 
 func (s *riverTestSuite) SetUpSuite(c *C) {
 	var err error
-	s.c, err = client.Connect(*my_addr, "root", "", "test")
+	s.c, err = client.Connect(*my_addr, "root", "123456", "test")
 	c.Assert(err, IsNil)
 
 	s.testExecute(c, "SET SESSION binlog_format = 'ROW'")
@@ -55,7 +55,6 @@ func (s *riverTestSuite) SetUpSuite(c *C) {
 	s.testExecute(c, "DROP TABLE IF EXISTS test_river")
 	s.testExecute(c, "DROP TABLE IF EXISTS test_for_id")
 	s.testExecute(c, "DROP TABLE IF EXISTS test_for_json")
-	s.testExecute(c, "DROP TABLE IF EXISTS test_river")
 	s.testExecute(c, fmt.Sprintf(schema, "test_river"))
 	s.testExecute(c, fmt.Sprintf(schema, "test_for_id"))
 	s.testExecute(c, fmt.Sprintf(schema_json, "test_for_json"))
@@ -209,7 +208,7 @@ func (s *riverTestSuite) testPrepareData(c *C) {
 	s.testExecute(c, "INSERT INTO test_river (id, title, content, tenum, tset) VALUES (?, ?, ?, ?, ?)", 3, "third", "hello elaticsearch 3", "e3", "c")
 	s.testExecute(c, "INSERT INTO test_river (id, title, content, tenum, tset, tbit) VALUES (?, ?, ?, ?, ?, ?)", 4, "fouth", "hello go-mysql-elasticserach 4", "e1", "a,b,c", 0)
 	s.testExecute(c, "INSERT INTO test_for_id (id, title, content, tenum, tset) VALUES (?, ?, ?, ?, ?)", 1, "first", "hello go 1", "e1", "a,b")
-	s.testExecute(c, "INSERT INTO test_for_json (id, json) VALUES (?, ?)", 9200, "{\"first\": \"a\", \"second\": \"b\"}")
+	s.testExecute(c, "INSERT INTO test_for_json (id, info) VALUES (?, ?)", 9200, "{\"first\": \"a\", \"second\": \"b\"}")
 
 	for i := 0; i < 10; i++ {
 		table := fmt.Sprintf("test_river_%04d", i)

From deee9d17bb49404db9d5504311d3f787fcb73edf Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Sun, 25 Jun 2017 14:40:38 +0800
Subject: [PATCH 55/65] Update river_test.go

---
 river/river_test.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/river/river_test.go b/river/river_test.go
index e7a3a734..0badd104 100644
--- a/river/river_test.go
+++ b/river/river_test.go
@@ -28,7 +28,7 @@ var _ = Suite(&riverTestSuite{})
 
 func (s *riverTestSuite) SetUpSuite(c *C) {
 	var err error
-	s.c, err = client.Connect(*my_addr, "root", "123456", "test")
+	s.c, err = client.Connect(*my_addr, "root", "", "test")
 	c.Assert(err, IsNil)
 
 	s.testExecute(c, "SET SESSION binlog_format = 'ROW'")

From feb0ad56852c3b0d9048d08a00c29bd595c554be Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 25 Jun 2017 19:27:32 +0800
Subject: [PATCH 56/65] try to support mysql 5.7 in travis ci

---
 .travis.install-mysql-5.7.sh | 6 ++++++
 .travis.yml                  | 4 ++++
 2 files changed, 10 insertions(+)
 create mode 100644 .travis.install-mysql-5.7.sh

diff --git a/.travis.install-mysql-5.7.sh b/.travis.install-mysql-5.7.sh
new file mode 100644
index 00000000..783c5bb0
--- /dev/null
+++ b/.travis.install-mysql-5.7.sh
@@ -0,0 +1,6 @@
+echo mysql-apt-config mysql-apt-config/select-server select mysql-5.7 | sudo debconf-set-selections
+wget http://dev.mysql.com/get/mysql-apt-config_0.7.3-1_all.deb
+sudo dpkg --install mysql-apt-config_0.7.3-1_all.deb
+sudo apt-get update -q
+sudo apt-get install -q -y -o Dpkg::Options::=--force-confnew mysql-server
+sudo mysql_upgrade
diff --git a/.travis.yml b/.travis.yml
index c4185519..5dc287a6 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -6,6 +6,7 @@ go:
 
 services:
   - elasticsearch
+  - mysql
 
 before_install:
   - go install -race std
@@ -25,3 +26,6 @@ before_install:
 
 script:
   - go test --race ./...
+
+before_script:
+  - bash .travis.install-mysql-5.7.sh

From 923f83aabc8975815d5598b7972e0ae80b51f3bc Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 25 Jun 2017 19:31:20 +0800
Subject: [PATCH 57/65] try to support mysql 5.7 in travis ci

---
 .travis.install-mysql-5.7.sh | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.travis.install-mysql-5.7.sh b/.travis.install-mysql-5.7.sh
index 783c5bb0..dfb7b894 100644
--- a/.travis.install-mysql-5.7.sh
+++ b/.travis.install-mysql-5.7.sh
@@ -3,4 +3,4 @@ wget http://dev.mysql.com/get/mysql-apt-config_0.7.3-1_all.deb
 sudo dpkg --install mysql-apt-config_0.7.3-1_all.deb
 sudo apt-get update -q
 sudo apt-get install -q -y -o Dpkg::Options::=--force-confnew mysql-server
-sudo mysql_upgrade
+sudo mysql_upgrade --force

From f2ab181fc4af33f7ccd5871103e1e36f01084ae1 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 25 Jun 2017 19:36:51 +0800
Subject: [PATCH 58/65] try to support mysql 5.7 in travis ci

---
 .travis.install-mysql-5.7.sh | 1 +
 1 file changed, 1 insertion(+)

diff --git a/.travis.install-mysql-5.7.sh b/.travis.install-mysql-5.7.sh
index dfb7b894..489f2cb4 100644
--- a/.travis.install-mysql-5.7.sh
+++ b/.travis.install-mysql-5.7.sh
@@ -4,3 +4,4 @@ sudo dpkg --install mysql-apt-config_0.7.3-1_all.deb
 sudo apt-get update -q
 sudo apt-get install -q -y -o Dpkg::Options::=--force-confnew mysql-server
 sudo mysql_upgrade --force
+mysql --version

From 85c23cf933c5ae2e1bbf9dfdca708c891e51579e Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 25 Jun 2017 19:47:59 +0800
Subject: [PATCH 59/65] try to support mysql 5.7 in travis ci

---
 .travis.yml | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/.travis.yml b/.travis.yml
index 5dc287a6..99075d23 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -12,6 +12,7 @@ before_install:
   - go install -race std
 
   # stop mysql and use row-based format binlog
+  - "bash .travis.install-mysql-5.7.sh"
   - "sudo /etc/init.d/mysql stop || true"
   - "echo '[mysqld]'            | sudo tee /etc/mysql/conf.d/replication.cnf"
   - "echo 'server-id=1'         | sudo tee -a /etc/mysql/conf.d/replication.cnf"
@@ -27,5 +28,3 @@ before_install:
 script:
   - go test --race ./...
 
-before_script:
-  - bash .travis.install-mysql-5.7.sh

From 500e4d41637aca4050297b7faca1aa6ea1b56486 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 25 Jun 2017 19:59:37 +0800
Subject: [PATCH 60/65] try to support mysql 5.7 in travis ci

---
 .travis.install-mysql-5.7.sh | 1 +
 1 file changed, 1 insertion(+)

diff --git a/.travis.install-mysql-5.7.sh b/.travis.install-mysql-5.7.sh
index 489f2cb4..c7500765 100644
--- a/.travis.install-mysql-5.7.sh
+++ b/.travis.install-mysql-5.7.sh
@@ -1,3 +1,4 @@
+mysql --version
 echo mysql-apt-config mysql-apt-config/select-server select mysql-5.7 | sudo debconf-set-selections
 wget http://dev.mysql.com/get/mysql-apt-config_0.7.3-1_all.deb
 sudo dpkg --install mysql-apt-config_0.7.3-1_all.deb

From c82a1d6af9e41a5b0a40d4b588c8d7e84d623941 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 25 Jun 2017 20:18:44 +0800
Subject: [PATCH 61/65] try to support mysql 5.7 in travis ci

---
 .travis.install-mysql-5.7.sh | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.travis.install-mysql-5.7.sh b/.travis.install-mysql-5.7.sh
index c7500765..95b74a8e 100644
--- a/.travis.install-mysql-5.7.sh
+++ b/.travis.install-mysql-5.7.sh
@@ -1,5 +1,5 @@
 mysql --version
-echo mysql-apt-config mysql-apt-config/select-server select mysql-5.7 | sudo debconf-set-selections
+echo mysql-apt-config mysql-apt-config/select-server select mysql-5.6 | sudo debconf-set-selections
 wget http://dev.mysql.com/get/mysql-apt-config_0.7.3-1_all.deb
 sudo dpkg --install mysql-apt-config_0.7.3-1_all.deb
 sudo apt-get update -q

From e4126d65217bc30f74e7dfd1941424d2e09ce4ba Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 25 Jun 2017 20:32:39 +0800
Subject: [PATCH 62/65] try to support mysql 5.7 in travis ci

---
 .travis.install-mysql-5.7.sh | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/.travis.install-mysql-5.7.sh b/.travis.install-mysql-5.7.sh
index 95b74a8e..c6df48cf 100644
--- a/.travis.install-mysql-5.7.sh
+++ b/.travis.install-mysql-5.7.sh
@@ -1,7 +1,7 @@
 mysql --version
 echo mysql-apt-config mysql-apt-config/select-server select mysql-5.6 | sudo debconf-set-selections
-wget http://dev.mysql.com/get/mysql-apt-config_0.7.3-1_all.deb
-sudo dpkg --install mysql-apt-config_0.7.3-1_all.deb
+wget https://dev.mysql.com/get/mysql-apt-config_0.8.6-1_all.deb
+sudo dpkg --install mysql-apt-config_0.8.6-1_all.deb
 sudo apt-get update -q
 sudo apt-get install -q -y -o Dpkg::Options::=--force-confnew mysql-server
 sudo mysql_upgrade --force

From daafd0f1027971981c99ad44405f8aa56bc5d4af Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <347249478@qq.com>
Date: Sun, 25 Jun 2017 20:36:25 +0800
Subject: [PATCH 63/65] try to support mysql 5.7 in travis ci

---
 .travis.install-mysql-5.7.sh | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.travis.install-mysql-5.7.sh b/.travis.install-mysql-5.7.sh
index c6df48cf..a863cc74 100644
--- a/.travis.install-mysql-5.7.sh
+++ b/.travis.install-mysql-5.7.sh
@@ -1,5 +1,5 @@
 mysql --version
-echo mysql-apt-config mysql-apt-config/select-server select mysql-5.6 | sudo debconf-set-selections
+echo mysql-apt-config mysql-apt-config/select-server select mysql-5.7 | sudo debconf-set-selections
 wget https://dev.mysql.com/get/mysql-apt-config_0.8.6-1_all.deb
 sudo dpkg --install mysql-apt-config_0.8.6-1_all.deb
 sudo apt-get update -q

From 0accf567de8a59de40332aca0c0b513353603591 Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Sun, 25 Jun 2017 21:01:21 +0800
Subject: [PATCH 64/65] Update river_test.go

---
 river/river_test.go | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/river/river_test.go b/river/river_test.go
index 0badd104..e31791b8 100644
--- a/river/river_test.go
+++ b/river/river_test.go
@@ -263,10 +263,11 @@ func (s *riverTestSuite) TestRiver(c *C) {
 	c.Assert(r.Found, Equals, true)
 	switch v := r.Source["info"].(type) {
 	case map[string]string:
-	    c.Assert(v["first"], Equals, "a")
-	    c.Assert(v["second"], Equals, "b")
+	    	c.Assert(v["first"], Equals, "a")
+	    	c.Assert(v["second"], Equals, "b")
 	default:
-	    c.Assert(true, Equals, false)
+		c.Assert(v, Equals, nil)
+	    	c.Assert(true, Equals, false)
 	}
 
 	r = s.testElasticGet(c, "100")

From 629cc8e538c8f9483c82423d40971c6cdebf9f1d Mon Sep 17 00:00:00 2001
From: WangXiangUSTC <xiang13225080@163.com>
Date: Sun, 25 Jun 2017 21:56:03 +0800
Subject: [PATCH 65/65] Update river_test.go

---
 river/river_test.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/river/river_test.go b/river/river_test.go
index e31791b8..5c555a24 100644
--- a/river/river_test.go
+++ b/river/river_test.go
@@ -262,7 +262,7 @@ func (s *riverTestSuite) TestRiver(c *C) {
 	r = s.testElasticGet(c, "9200")
 	c.Assert(r.Found, Equals, true)
 	switch v := r.Source["info"].(type) {
-	case map[string]string:
+	case map[string]interface{}:
 	    	c.Assert(v["first"], Equals, "a")
 	    	c.Assert(v["second"], Equals, "b")
 	default: