[issue #7]support source-connector database white-list and table white-list

README.md

@@ -84,7 +84,8 @@ Note to replace the arguments in "()" with your own mysql setting.
 |mysqlUsername         |false   |           |Username of MySQL account|
 |mysqlPassword         |false   |           |Password of MySQL account|
 |source-record-converter         |false   |           |Full class name of the impl of the converter used to convert SourceDataEntry to byte[]|
-
+|whiteDataBase      |false  |           |DataBase Name which you want to source data
+|whiteTable         |true   |           |Table Name which you want to source data (choosable)
 
 ### Verify Mysql Source Connector Started Successfully
 

odar-mysql-connector/src/main/java/io/openmessaging/mysql/Config.java

@@ -31,6 +31,8 @@ public class Config {
     public Integer mysqlPort;
     public String mysqlUsername;
     public String mysqlPassword;
+    public String whiteDataBase;
+    public String whiteTable;
 
     public String queueName;
 
@@ -45,6 +47,7 @@ public class Config {
             add("mysqlPort");
             add("mysqlUsername");
             add("mysqlPassword");
+            add("whiteDataBase");
         }
     };
 
@@ -146,6 +149,22 @@ public String getMysqlPassword() {
         return mysqlPassword;
     }
 
+    public String getWhiteDataBase() {
+        return whiteDataBase;
+    }
+
+    public void setWhiteDataBase(String whiteDataBase) {
+        this.whiteDataBase = whiteDataBase;
+    }
+
+    public String getWhiteTable() {
+        return whiteTable;
+    }
+
+    public void setWhiteTable(String whiteTable) {
+        this.whiteTable = whiteTable;
+    }
+
     public String getQueueName() {
         return queueName;
     }

odar-mysql-connector/src/main/java/io/openmessaging/mysql/binlog/EventProcessor.java

@@ -18,6 +18,7 @@
 package io.openmessaging.mysql.binlog;
 
 import com.alibaba.druid.pool.DruidDataSourceFactory;
+import com.alibaba.druid.util.StringUtils;
 import com.github.shyiko.mysql.binlog.BinaryLogClient;
 import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData;
 import com.github.shyiko.mysql.binlog.event.Event;
@@ -36,9 +37,7 @@
 import io.openmessaging.mysql.schema.Schema;
 import io.openmessaging.mysql.schema.Table;
 import java.io.Serializable;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
@@ -85,6 +84,20 @@ public void start() throws Exception {
         binlogPositionManager.initBeginPosition();
 
         schema = new Schema(dataSource);
+        String whiteDataBases = config.getWhiteDataBase();
+        String whiteTables = config.getWhiteTable();
+
+        if (!StringUtils.isEmpty(whiteDataBases)){
+            Arrays.asList(whiteDataBases.trim().split(",")).forEach(whiteDataBase ->{
+                Collections.addAll(schema.dataBaseWhiteList, whiteDataBase);
+            });
+        }
+
+        if (!StringUtils.isEmpty(whiteTables)){
+            Arrays.asList(whiteTables.trim().split(",")).forEach(whiteTable ->{
+                Collections.addAll(schema.tableWhiteList, whiteTable);
+            });
+        }
         schema.load();
 
         eventListener = new EventListener(queue);
@@ -190,8 +203,9 @@ private void processTableMapEvent(Event event) {
         Long tableId = data.getTableId();
 
         Table table = schema.getTable(dbName, tableName);
-
-        tableMap.put(tableId, table);
+        if (table != null){
+            tableMap.put(tableId, table);
+        }
     }
 
     private void processWriteEvent(Event event) {

odar-mysql-connector/src/main/java/io/openmessaging/mysql/schema/Database.java

@@ -25,6 +25,7 @@
 import java.sql.SQLException;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
 import javax.sql.DataSource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -41,11 +42,19 @@ public class Database {
 
     private Map<String, Table> tableMap = new HashMap<String, Table>();
 
+    public Set<String> tableWhiteList;
+
     public Database(String name, DataSource dataSource) {
         this.name = name;
         this.dataSource = dataSource;
     }
 
+    public Database(String name, DataSource dataSource, Set<String> tableWhiteList){
+        this.name = name;
+        this.dataSource = dataSource;
+        this.tableWhiteList = tableWhiteList;
+    }
+
     public void init() throws SQLException {
         Connection conn = null;
         PreparedStatement ps = null;
@@ -66,7 +75,9 @@ public void init() throws SQLException {
                 String charset = rs.getString(5);
 
                 ColumnParser columnParser = ColumnParser.getColumnParser(dataType, colType, charset);
-
+                if (tableWhiteList.size() != 0 && !tableWhiteList.contains(tableName)){
+                    continue;
+                }
                 if (!tableMap.containsKey(tableName)) {
                     addTable(tableName);
                 }

odar-mysql-connector/src/main/java/io/openmessaging/mysql/schema/Schema.java

@@ -21,11 +21,7 @@
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import javax.sql.DataSource;
 import io.openmessaging.mysql.binlog.EventProcessor;
 import org.slf4j.Logger;
@@ -40,12 +36,18 @@ public class Schema {
         Arrays.asList(new String[] {"information_schema", "mysql", "performance_schema", "sys"})
     );
 
+    public Set<String> dataBaseWhiteList;
+
+    public Set<String> tableWhiteList;
+
     private DataSource dataSource;
 
     private Map<String, Database> dbMap;
 
     public Schema(DataSource dataSource) {
         this.dataSource = dataSource;
+        this.dataBaseWhiteList = new HashSet<>();
+        this.tableWhiteList = new HashSet<>();
     }
 
     public void load() throws SQLException {
@@ -64,8 +66,8 @@ public void load() throws SQLException {
 
             while (rs.next()) {
                 String dbName = rs.getString(1);
-                if (!IGNORED_DATABASES.contains(dbName)) {
-                    Database database = new Database(dbName, dataSource);
+                if (!IGNORED_DATABASES.contains(dbName) && dataBaseWhiteList.contains(dbName)) {
+                    Database database = new Database(dbName, dataSource, tableWhiteList);
                     dbMap.put(dbName, database);
                 }
             }