fix #2

Author: xlorne

scripts/.gitignore

@@ -1,9 +1,9 @@
 package com.codingapi.dbstream.driver;
 
 import com.codingapi.dbstream.interceptor.SQLRunningContext;
-import com.codingapi.dbstream.listener.SQLDeleteExecuteListener;
-import com.codingapi.dbstream.listener.SQLInsertExecuteListener;
-import com.codingapi.dbstream.listener.SQLUpdateExecuteListener;
+import com.codingapi.dbstream.listener.stream.SQLDeleteExecuteListener;
+import com.codingapi.dbstream.listener.stream.SQLInsertExecuteListener;
+import com.codingapi.dbstream.listener.stream.SQLUpdateExecuteListener;
 import com.codingapi.dbstream.proxy.ConnectionProxy;
 import com.codingapi.dbstream.scanner.DBMetaContext;
 import com.codingapi.dbstream.scanner.DBMetaData;

scripts/db.yaml

@@ -201,22 +201,6 @@ public List<Object> getListParams() {
         return new ArrayList<>();
     }
 
-    /**
-     * 获取执行的SQL队列
-     * @return List
-     */
-    public List<SQLExecuteParam> getBatchExecuteSQLParamList(){
-        if(this.batchMode){
-            if(this.sqlExecuteParams.isEmpty()){
-                return new ArrayList<>();
-            }
-            int size = this.sqlExecuteParams.size();
-            return this.sqlExecuteParams.subList(0,size-1);
-
-        }
-        return new ArrayList<>();
-    }
-
 
     /**
      * 获取Batch的SQLExecuteState

src/main/java/com/codingapi/dbstream/driver/DBStreamProxyDriver.java

@@ -0,0 +1,33 @@
+package com.codingapi.dbstream.listener.stream;
+
+import com.codingapi.dbstream.interceptor.SQLExecuteState;
+import com.codingapi.dbstream.parser.DBEventParser;
+import com.codingapi.dbstream.parser.DeleteDBEventParser;
+import com.codingapi.dbstream.parser.DeleteSQLParser;
+import com.codingapi.dbstream.parser.SQLParser;
+import com.codingapi.dbstream.scanner.DbTable;
+import com.codingapi.dbstream.utils.SQLUtils;
+
+public class SQLDeleteExecuteListener extends SQLStreamExecuteListener {
+
+    @Override
+    public int order() {
+        return 100;
+    }
+
+    @Override
+    public boolean isSupport(String sql) {
+        return SQLUtils.isDeleteSQL(sql);
+    }
+
+    @Override
+    public SQLParser createSQLParser(String sql) {
+        return new DeleteSQLParser(sql);
+    }
+
+    @Override
+    public DBEventParser createDbEventParser(SQLExecuteState sqlExecuteState, SQLParser sqlParser, DbTable dbTable) {
+        return new DeleteDBEventParser(sqlExecuteState, (DeleteSQLParser) sqlParser, dbTable);
+    }
+
+}

src/main/java/com/codingapi/dbstream/interceptor/SQLExecuteState.java

@@ -0,0 +1,34 @@
+package com.codingapi.dbstream.listener.stream;
+
+import com.codingapi.dbstream.interceptor.SQLExecuteState;
+import com.codingapi.dbstream.parser.DBEventParser;
+import com.codingapi.dbstream.parser.InsertDBEventParser;
+import com.codingapi.dbstream.parser.InsertSQLParser;
+import com.codingapi.dbstream.parser.SQLParser;
+import com.codingapi.dbstream.scanner.DbTable;
+import com.codingapi.dbstream.utils.SQLUtils;
+
+public class SQLInsertExecuteListener extends SQLStreamExecuteListener {
+
+    @Override
+    public int order() {
+        return 100;
+    }
+
+    @Override
+    public boolean isSupport(String sql) {
+        return SQLUtils.isInsertSQL(sql);
+    }
+
+    @Override
+    public SQLParser createSQLParser(String sql) {
+        return new InsertSQLParser(sql);
+    }
+
+    @Override
+    public DBEventParser createDbEventParser(SQLExecuteState sqlExecuteState, SQLParser sqlParser, DbTable dbTable) {
+        return new InsertDBEventParser(sqlExecuteState, (InsertSQLParser) sqlParser, dbTable);
+    }
+
+
+}

src/main/java/com/codingapi/dbstream/listener/SQLDeleteExecuteListener.java

@@ -0,0 +1,80 @@
+package com.codingapi.dbstream.listener.stream;
+
+import com.codingapi.dbstream.DBStreamContext;
+import com.codingapi.dbstream.interceptor.SQLExecuteState;
+import com.codingapi.dbstream.listener.SQLExecuteListener;
+import com.codingapi.dbstream.parser.*;
+import com.codingapi.dbstream.scanner.DbTable;
+import com.codingapi.dbstream.stream.DBEvent;
+import com.codingapi.dbstream.stream.TransactionEventPools;
+
+import java.sql.SQLException;
+import java.util.List;
+
+public abstract class SQLStreamExecuteListener implements SQLExecuteListener {
+
+    public abstract boolean isSupport(String sql);
+
+    public abstract SQLParser createSQLParser(String sql);
+
+    public abstract DBEventParser createDbEventParser(SQLExecuteState sqlExecuteState,SQLParser sqlParser,DbTable dbTable);
+
+    @Override
+    public void before(SQLExecuteState executeState) throws SQLException {
+        String sql = executeState.getSql();
+        if (this.isSupport(sql)) {
+            try {
+                ThreadLocalContext.getInstance().remove();
+                SQLParser sqlParser = this.createSQLParser(sql);
+                String tableName = sqlParser.getTableName();
+                executeState.updateMetaData(tableName);
+                DbTable dbTable = executeState.getDbTable(tableName);
+                if (dbTable != null && DBStreamContext.getInstance().support(executeState.getDriverProperties(), dbTable)) {
+                    if (executeState.isBatchMode()) {
+                        List<SQLExecuteState> executeStateList = executeState.getBatchSQLExecuteStateList();
+                        for (int i = 0; i < executeStateList.size(); i++) {
+                            SQLExecuteState sqlExecuteState = executeStateList.get(i);
+                            DBEventParser dataParser = this.createDbEventParser(sqlExecuteState,sqlParser,dbTable);
+                            dataParser.prepare();
+                            ThreadLocalContext.getInstance().push(i, dataParser);
+                        }
+                    } else {
+                        DBEventParser dataParser = this.createDbEventParser(executeState,sqlParser,dbTable);
+                        dataParser.prepare();
+                        ThreadLocalContext.getInstance().push(dataParser);
+                    }
+                }
+            } catch (Exception e) {
+                ThreadLocalContext.getInstance().remove();
+                throw new SQLException(e);
+            }
+        }
+    }
+
+    @Override
+    public void after(SQLExecuteState executeState, Object result) throws SQLException {
+        String sql = executeState.getSql();
+        String transactionKey = executeState.getTransactionKey();
+        if (this.isSupport(sql)) {
+            if (executeState.isBatchMode()) {
+                List<SQLExecuteState> executeStateList = executeState.getBatchSQLExecuteStateList();
+                for (int i = 0; i < executeStateList.size(); i++) {
+                    DBEventParser dataParser = ThreadLocalContext.getInstance().get(i);
+                    if (dataParser != null) {
+                        List<DBEvent> eventList = dataParser.loadEvents(result);
+                        TransactionEventPools.getInstance().addEvents(transactionKey, eventList);
+                    }
+                }
+                ThreadLocalContext.getInstance().remove();
+            } else {
+                DBEventParser dataParser = ThreadLocalContext.getInstance().get();
+                if (dataParser != null) {
+                    List<DBEvent> eventList = dataParser.loadEvents(result);
+                    TransactionEventPools.getInstance().addEvents(transactionKey, eventList);
+                }
+                ThreadLocalContext.getInstance().remove();
+            }
+        }
+    }
+
+}