Merge pull request #1 from codingapi/dev

add Listener order

Author: xlorne

README.md

@@ -106,6 +106,12 @@ import com.codingapi.dbstream.interceptor.SQLExecuteState;
 import com.codingapi.dbstream.listener.SQLExecuteListener;
 
 public class MySQLListener implements SQLExecuteListener {
+
+    @Override
+    public int order() {
+        return 0;
+    }
+    
     @Override
     public void before(SQLExecuteState executeState) {
         System.out.println("执行前 - SQL: " + executeState.getSql());
@@ -259,8 +265,9 @@ mvn clean test -P travis
    - 手动事务模式下，事件会在 `commit()` 时批量推送
    - 事务回滚时，相关事件会被丢弃
 
-3. **数据表限制**：
-   - 执行数据拦截事件的分析，要求表必须存在主键的定义
+3. **使用场景限制**：
+   - 数据库表必须有主键的定义，在DELETE事件需要明确主键信息，主键物理表不存在时可通过外部key文件配置的方式添加。
+   - 若INSERT INTO SELECT 语句中，采用主键自增模式，受限于JDBC的支持将无法解析到自增ID，建议修改单条保存或修改ID为手动传递。
 
 4. **元数据缓存**：
    - 数据库元数据会在首次连接时自动扫描并缓存

pom.xml

@@ -4,7 +4,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.codingapi.dbstream</groupId>
     <artifactId>dbstream-driver</artifactId>
-    <version>1.0.6</version>
+    <version>1.0.7</version>
 
     <url>https://github.com/codingapi/dbstream-driver</url>
     <name>dbstream-driver</name>

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

@@ -4,10 +4,11 @@
 import lombok.Getter;
 
 import java.sql.SQLException;
+import java.util.Comparator;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 
-public class SQLRunningContext implements SQLExecuteListener {
+public class SQLRunningContext   {
 
     @Getter
     private final static SQLRunningContext instance = new SQLRunningContext();
@@ -23,10 +24,11 @@ public void addListener(SQLExecuteListener listener) {
         if (listener != null) {
             listeners.add(listener);
         }
+
+        listeners.sort(Comparator.comparingInt(SQLExecuteListener::order));
     }
 
 
-    @Override
     public void after(SQLExecuteState executeState, Object result) throws SQLException {
         executeState.setResult(result);
         executeState.after();
@@ -36,7 +38,6 @@ public void after(SQLExecuteState executeState, Object result) throws SQLExcepti
     }
 
 
-    @Override
     public void before(SQLExecuteState executeState) throws SQLException {
         executeState.begin();
         for (SQLExecuteListener listener : listeners) {

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

@@ -16,6 +16,11 @@ public class SQLDeleteExecuteListener implements SQLExecuteListener {
 
     private final static ThreadLocal<DeleteDBEventParser> threadLocal = new ThreadLocal<>();
 
+    @Override
+    public int order() {
+        return 100;
+    }
+
     @Override
     public void before(SQLExecuteState executeState) throws SQLException {
         String sql = executeState.getSql();

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

@@ -6,6 +6,12 @@
 
 public interface SQLExecuteListener {
 
+    /**
+     * 执行顺序，越小越靠前
+     * @return index
+     */
+    int order();
+
     /**
      * before sql execute
      *

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

@@ -16,6 +16,11 @@ public class SQLInsertExecuteListener implements SQLExecuteListener {
 
     private final static ThreadLocal<InsertDBEventParser> threadLocal = new ThreadLocal<>();
 
+    @Override
+    public int order() {
+        return 100;
+    }
+
     @Override
     public void before(SQLExecuteState executeState) throws SQLException {
         String sql = executeState.getSql();

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

@@ -16,6 +16,11 @@ public class SQLUpdateExecuteListener implements SQLExecuteListener {
 
     private final static ThreadLocal<UpdateDBEventParser> threadLocal = new ThreadLocal<>();
 
+    @Override
+    public int order() {
+        return 100;
+    }
+
     @Override
     public void before(SQLExecuteState executeState) throws SQLException {
         String sql = executeState.getSql();

src/main/java/com/codingapi/dbstream/stream/DBEvent.java

@@ -81,7 +81,14 @@ void setTransactionKey(String transactionKey) {
     }
 
     public void addPrimaryKey(String primaryKey) {
-        this.primaryKeys.add(primaryKey);
+        if(!this.primaryKeys.contains(primaryKey)) {
+            this.primaryKeys.add(primaryKey);
+        }
+    }
+
+
+    public boolean hasPrimaryKeys(){
+        return this.primaryKeys!=null && !this.primaryKeys.isEmpty();
     }
 
 }

src/main/java/com/codingapi/dbstream/stream/DefaultDBEventPusher.java

@@ -6,6 +6,9 @@ public class DefaultDBEventPusher implements DBEventPusher {
 
     @Override
     public void push(List<DBEvent> events) {
-        System.out.println(events);
+        System.out.println("<=== DBStream DBEvent Total "+events.size()+" ===> ");
+        for(DBEvent event:events){
+            System.out.println(event);
+        }
     }
 }

src/test/java/com/example/dbstream/listener/MySQLListener.java

@@ -9,6 +9,11 @@
 @Slf4j
 public class MySQLListener implements SQLExecuteListener {
 
+    @Override
+    public int order() {
+        return 0;
+    }
+
     @Override
     public void after(SQLExecuteState executeState, Object result) throws SQLException {
         log.info("after sql:{},params:{},execute timestamp:{}", executeState.getSql(), executeState.getListParams(), executeState.getExecuteTimestamp());