updaate

parse.go

@@ -8,6 +8,7 @@ import (
 	"github.com/starainrt/go-mysql/replication"
 	"io"
 	"os"
+	"strings"
 	"time"
 )
 
@@ -311,20 +312,44 @@ func ParseBinlogEvent(ev *replication.BinlogEvent) []BinlogEvent {
 }
 
 type BinlogFilter struct {
-	IncludeGtid  string
+	IncludeGtid   string
-	ExcludeGtid  string
+	ExcludeGtid   string
-	StartPos     int
+	IncludeTables []string
-	EndPos       int
+	ExcludeTables []string
-	StartDate    time.Time
+	StartPos      int
-	EndDate      time.Time
+	EndPos        int
-	BigThan      int
+	StartDate     time.Time
-	SmallThan    int
+	EndDate       time.Time
-	OnlyShowGtid bool
+	BigThan       int
+	SmallThan     int
+	OnlyShowGtid  bool
 }
 
 func parseBinlogWithFilter(r io.Reader, parse *replication.BinlogParser, filter BinlogFilter, fn func(Transaction) bool) error {
 	var subGtid, inGtid, exGtid *gtid.Gtid
 	var err error
+	var includeMap = make(map[string]bool)
+	var excludeMap = make(map[string]bool)
+	if len(filter.IncludeTables) != 0 {
+		for _, v := range filter.IncludeTables {
+			if len(strings.Split(v, ".")) != 2 {
+				return fmt.Errorf("IncludeTable Name Is Invalid:%s", v)
+			}
+			includeMap[v] = true
+		}
+	} else {
+		includeMap["*.*"] = true
+	}
+	if len(filter.ExcludeTables) != 0 {
+		for _, v := range filter.ExcludeTables {
+			if len(strings.Split(v, ".")) != 2 {
+				return fmt.Errorf("ExcludeTable Name Is Invalid:%s", v)
+			}
+			excludeMap[v] = true
+		}
+	} else {
+		excludeMap["*.*"] = true
+	}
 	if filter.IncludeGtid != "" {
 		inGtid, err = gtid.Parse(filter.IncludeGtid)
 		if err != nil {
@@ -338,7 +363,6 @@ func parseBinlogWithFilter(r io.Reader, parse *replication.BinlogParser, filter
 			return err
 		}
 	}
-
 	// process: 0, continue: 1, break: 2, EOF: 3
 	var (
 		n            int64