add support of parse json field

.travis.install-mysql-5.7.sh

@@ -0,0 +1,8 @@
+mysql --version
+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
+sudo apt-get install -q -y -o Dpkg::Options::=--force-confnew mysql-server
+sudo mysql_upgrade --force
+mysql --version

.travis.yml

@@ -6,11 +6,13 @@ go:
 
 services:
   - elasticsearch
+  - mysql
 
 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"
@@ -25,3 +27,4 @@ before_install:
 
 script:
   - go test --race ./...
+

river/river_test.go

@@ -45,9 +45,19 @@ 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, 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 +84,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 +93,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 +108,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 +147,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 +183,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 +208,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, info) 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 +255,21 @@ 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)
+	switch v := r.Source["info"].(type) {
+	case map[string]interface{}:
+	    	c.Assert(v["first"], Equals, "a")
+	    	c.Assert(v["second"], Equals, "b")
+	default:
+		c.Assert(v, Equals, nil)
+	    	c.Assert(true, Equals, false)
+	}
+
 	r = s.testElasticGet(c, "100")
 	c.Assert(r.Found, Equals, false)
 

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