update vendor (#171)

Author: siddontang

cmd/go-mysql-elasticsearch/main.go

@@ -83,7 +83,11 @@ func main() {
 		return
 	}
 
-	r.Start()
+	done := make(chan struct{}, 1)
+	go func() {
+		r.Run()
+		done <- struct{}{}
+	}()
 
 	select {
 	case n := <-sc:
@@ -93,4 +97,5 @@ func main() {
 	}
 
 	r.Close()
+	<-done
 }

river/river.go

@@ -7,9 +7,9 @@ import (
 	"sync"
 
 	"github.com/juju/errors"
-	log "github.com/sirupsen/logrus"
 	"github.com/siddontang/go-mysql-elasticsearch/elastic"
 	"github.com/siddontang/go-mysql/canal"
+	log "github.com/sirupsen/logrus"
 )
 
 // In Elasticsearch, river is a pluggable service within Elasticsearch pulling data then indexing it into Elasticsearch.
@@ -249,7 +249,7 @@ func (r *River) prepareRule() error {
 				log.Errorf("ignored table without a primary key: %s\n", rule.TableInfo.Name)
 			}
 		} else {
-			rules[key] = rule;
+			rules[key] = rule
 		}
 	}
 	r.rules = rules
@@ -261,12 +261,13 @@ func ruleKey(schema string, table string) string {
 	return fmt.Sprintf("%s:%s", schema, table)
 }
 
-func (r *River) Start() error {
+// Run syncs the data from MySQL and inserts to ES.
+func (r *River) Run() error {
 	r.wg.Add(1)
 	go r.syncLoop()
 
 	pos := r.master.Position()
-	if err := r.canal.StartFrom(pos); err != nil {
+	if err := r.canal.RunFrom(pos); err != nil {
 		log.Errorf("start canal err %v", err)
 		return errors.Trace(err)
 	}
@@ -290,7 +291,7 @@ func (r *River) Close() {
 	r.wg.Wait()
 }
 
-func isValidTables(tables[] string) bool {
+func isValidTables(tables []string) bool {
 	if len(tables) > 1 {
 		for _, table := range tables {
 			if table == "*" {

river/river_extra_test.go

@@ -113,7 +113,7 @@ func (s *riverTestSuite) TestRiverWithParent(c *C) {
 
 	s.testPrepareExtraData(c)
 
-	river.Start()
+	go func() { river.Run() }()
 
 	testWaitSyncDone(c, river)
 

river/river_test.go

@@ -267,7 +267,7 @@ func testWaitSyncDone(c *C, r *River) {
 func (s *riverTestSuite) TestRiver(c *C) {
 	s.testPrepareData(c)
 
-	s.r.Start()
+	go func() { s.r.Run() }()
 
 	testWaitSyncDone(c, s.r)
 
@@ -356,7 +356,7 @@ func TestTableValidation(t *testing.T) {
 	tables := []struct {
 		Tables []string
 		Expect bool
-	} {
+	}{
 		{[]string{"*"}, true},
 		{[]string{"table", "table2"}, true},
 		{[]string{"*", "table"}, false},
@@ -371,9 +371,9 @@ func TestTableValidation(t *testing.T) {
 
 func TestBuildTable(t *testing.T) {
 	tables := []struct {
-		Table string
+		Table  string
 		Expect string
-	} {
+	}{
 		{"*", ".*"},
 		{"table2", "table2"},
 	}