Skip to content

Commit 10ae07d

Browse files
xlornexlorne
committed
add clean api & update test
1 parent 827d46c commit 10ae07d

10 files changed

Lines changed: 308 additions & 22 deletions

File tree

‎README.md‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -202,13 +202,23 @@ USERNAME,ID
202202
DBStreamContext.getInstance().addEventPusher(DBEventPusher pusher);
203203
```
204204

205+
```java
206+
// 清空所有的事件推送器
207+
DBStreamContext.getInstance().cleanEventPushers();
208+
```
209+
205210
#### 监听器相关
206211

207212
```java
208213
// 添加 SQL 执行监听器
209214
DBStreamContext.getInstance().addListener(SQLExecuteListener listener);
210215
```
211216

217+
```java
218+
// 清空所有自定义的SQL 执行监听器
219+
DBStreamContext.getInstance().cleanCustomListeners();
220+
```
221+
212222
#### 数据库表支持判断
213223

214224
```java

‎src/main/java/com/codingapi/dbstream/DBStreamContext.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,15 @@ public void addListener(SQLExecuteListener sqlExecuteListener) {
4040
SQLRunningContext.getInstance().addListener(sqlExecuteListener);
4141
}
4242

43+
44+
/**
45+
* 清空所有自定义的SQL执行订阅
46+
*/
47+
public void cleanCustomListeners(){
48+
SQLRunningContext.getInstance().cleanCustomListeners();
49+
}
50+
51+
4352
/**
4453
* 是否支持SQL 拦截代理分析
4554
*
@@ -68,6 +77,14 @@ public void addEventPusher(DBEventPusher dbEventPusher) {
6877
DBEventContext.getInstance().addPusher(dbEventPusher);
6978
}
7079

80+
81+
/**
82+
* 清空DB事件推送
83+
*/
84+
public void cleanEventPushers() {
85+
DBEventContext.getInstance().clean();
86+
}
87+
7188
/**
7289
* 获取元数据库的信息
7390
*
@@ -117,4 +134,5 @@ public void clear(String jdbcKey) {
117134
DBMetaContext.getInstance().clear(jdbcKey);
118135
}
119136

137+
120138
}

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

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,6 @@
11
package com.codingapi.dbstream.driver;
22

33
import com.codingapi.dbstream.listener.SQLRunningContext;
4-
import com.codingapi.dbstream.listener.dbevent.DeleteEventListener;
5-
import com.codingapi.dbstream.listener.dbevent.InsertEventListener;
6-
import com.codingapi.dbstream.listener.dbevent.UpdateEventListener;
74
import com.codingapi.dbstream.proxy.ConnectionProxy;
85
import com.codingapi.dbstream.scanner.DBMetaContext;
96
import com.codingapi.dbstream.scanner.DBMetaData;
@@ -26,16 +23,14 @@ public class DBStreamProxyDriver implements Driver {
2623

2724
static {
2825
String version = VersionUtils.getDriverVersion();
29-
System.out.println("<--- dbstream-driver successfully loaded, version: "+version+" --->");
26+
System.out.println("\n<--- dbstream-driver successfully loaded, version: "+version+" --->\n");
3027
try {
3128
DriverManager.registerDriver(new DBStreamProxyDriver());
3229
} catch (Exception e) {
3330
LOGGER.log(Level.SEVERE, "Failed to register DBStreamProxyDriver", e);
3431
throw new RuntimeException("Failed to register DBStreamProxyDriver", e);
3532
}
36-
SQLRunningContext.getInstance().addListener(new DeleteEventListener());
37-
SQLRunningContext.getInstance().addListener(new InsertEventListener());
38-
SQLRunningContext.getInstance().addListener(new UpdateEventListener());
33+
SQLRunningContext.getInstance().addSystemListeners();
3934
LOGGER.info("DBStreamProxyDriver initialized and registered");
4035
}
4136

‎src/main/java/com/codingapi/dbstream/event/DBEvent.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,18 @@ public class DBEvent {
6060
@Getter
6161
private String transactionKey;
6262

63+
public boolean isInsert(){
64+
return this.type == EventType.INSERT;
65+
}
66+
67+
public boolean isDelete(){
68+
return this.type == EventType.DELETE;
69+
}
70+
71+
public boolean isUpdate(){
72+
return this.type == EventType.UPDATE;
73+
}
74+
6375
@Override
6476
public String toString() {
6577
return "DBEvent{" +

‎src/main/java/com/codingapi/dbstream/event/DBEventContext.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,4 +44,7 @@ public void addPusher(DBEventPusher pusher) {
4444
}
4545
}
4646

47+
public void clean() {
48+
this.pushers.clear();
49+
}
4750
}

‎src/main/java/com/codingapi/dbstream/listener/SQLRunningContext.java‎

Lines changed: 31 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,8 @@
11
package com.codingapi.dbstream.listener;
22

3+
import com.codingapi.dbstream.listener.dbevent.DeleteEventListener;
4+
import com.codingapi.dbstream.listener.dbevent.InsertEventListener;
5+
import com.codingapi.dbstream.listener.dbevent.UpdateEventListener;
36
import lombok.Getter;
47

58
import java.sql.SQLException;
@@ -18,6 +21,19 @@ public class SQLRunningContext {
1821
@Getter
1922
private final List<SQLExecuteListener> listeners = new CopyOnWriteArrayList<>();
2023

24+
private final InsertEventListener insertEventListener = new InsertEventListener();
25+
private final DeleteEventListener deleteEventListener = new DeleteEventListener();
26+
private final UpdateEventListener updateEventListener = new UpdateEventListener();
27+
28+
/**
29+
* 添加系统默认订阅器
30+
*/
31+
public void addSystemListeners() {
32+
this.addListener(insertEventListener);
33+
this.addListener(deleteEventListener);
34+
this.addListener(updateEventListener);
35+
}
36+
2137
private SQLRunningContext() {
2238

2339
}
@@ -27,9 +43,10 @@ private SQLRunningContext() {
2743
*/
2844
public void addListener(SQLExecuteListener listener) {
2945
if (listener != null) {
30-
listeners.add(listener);
46+
if (!listeners.contains(listener)) {
47+
listeners.add(listener);
48+
}
3149
}
32-
3350
listeners.sort(Comparator.comparingInt(SQLExecuteListener::order));
3451
}
3552

@@ -55,4 +72,16 @@ public void before(SQLRunningState executeState) throws SQLException {
5572
listener.before(executeState);
5673
}
5774
}
75+
76+
77+
/**
78+
* 清空自定义的订阅
79+
*/
80+
public void cleanCustomListeners() {
81+
listeners.removeIf(listener ->
82+
!(listener.equals(insertEventListener)
83+
|| listener.equals(updateEventListener)
84+
|| listener.equals(deleteEventListener))
85+
);
86+
}
5887
}

‎src/test/java/com/example/dbstream/tests/User1RepositoryTest.java‎

Lines changed: 105 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,9 @@
22

33

44
import com.codingapi.dbstream.DBStreamContext;
5+
import com.codingapi.dbstream.event.DBEvent;
6+
import com.codingapi.dbstream.event.DBEventPusher;
7+
import com.codingapi.dbstream.query.JdbcQuery;
58
import com.example.dbstream.entity.User1;
69
import com.example.dbstream.listener.MySQLListener;
710
import com.example.dbstream.repository.User1Repository;
@@ -13,8 +16,9 @@
1316
import javax.persistence.EntityManager;
1417
import javax.transaction.Transactional;
1518
import java.util.List;
19+
import java.util.concurrent.atomic.AtomicBoolean;
1620

17-
import static org.junit.jupiter.api.Assertions.assertThrows;
21+
import static org.junit.jupiter.api.Assertions.*;
1822

1923

2024
@SpringBootTest
@@ -33,9 +37,19 @@ class User1RepositoryTest {
3337
@Transactional
3438
@Rollback(false)
3539
void test1() {
40+
3641
DBStreamContext.getInstance().addListener(new MySQLListener());
3742

38-
userRepository.deleteAll();
43+
DBStreamContext.getInstance().cleanEventPushers();
44+
DBStreamContext.getInstance().addEventPusher(new DBEventPusher() {
45+
@Override
46+
public void push(JdbcQuery jdbcQuery, List<DBEvent> events) {
47+
assertTrue(events.size()>=4);
48+
for (DBEvent event:events){
49+
assertTrue(event.hasPrimaryKeys());
50+
}
51+
}
52+
});
3953
User1 user = new User1();
4054
user.setUsername("admin");
4155
user.setPassword("admin");
@@ -59,7 +73,6 @@ void test1() {
5973
int deleteRows = userRepository.deleteByUsername("admin");
6074
System.out.println("deleteRows:" + deleteRows);
6175

62-
userRepository.deleteAll();
6376
}
6477

6578
/**
@@ -69,6 +82,18 @@ void test1() {
6982
@Transactional
7083
@Rollback(false)
7184
void test2() {
85+
DBStreamContext.getInstance().cleanEventPushers();
86+
87+
DBStreamContext.getInstance().addEventPusher(new DBEventPusher() {
88+
@Override
89+
public void push(JdbcQuery jdbcQuery, List<DBEvent> events) {
90+
assertEquals(1, events.size());
91+
for (DBEvent event:events){
92+
assertTrue(event.hasPrimaryKeys());
93+
assertTrue(event.isInsert());
94+
}
95+
}
96+
});
7297
User1 user = new User1();
7398
user.setUsername("admin");
7499
user.setPassword("admin");
@@ -86,6 +111,16 @@ void test2() {
86111
@Transactional
87112
@Rollback(false)
88113
void test3() {
114+
DBStreamContext.getInstance().cleanEventPushers();
115+
DBStreamContext.getInstance().addEventPusher(new DBEventPusher() {
116+
@Override
117+
public void push(JdbcQuery jdbcQuery, List<DBEvent> events) {
118+
for (DBEvent event:events){
119+
assertTrue(event.hasPrimaryKeys());
120+
assertTrue(event.isUpdate());
121+
}
122+
}
123+
});
89124
userRepository.resetPassword("123456");
90125
}
91126

@@ -96,6 +131,16 @@ void test3() {
96131
@Transactional
97132
@Rollback(false)
98133
void test4() {
134+
DBStreamContext.getInstance().cleanEventPushers();
135+
DBStreamContext.getInstance().addEventPusher(new DBEventPusher() {
136+
@Override
137+
public void push(JdbcQuery jdbcQuery, List<DBEvent> events) {
138+
assertTrue(events.size()>=2);
139+
for (DBEvent event:events){
140+
assertTrue(event.hasPrimaryKeys());
141+
}
142+
}
143+
});
99144
User1 user = new User1();
100145
user.setUsername("admin");
101146
user.setPassword("admin");
@@ -113,15 +158,23 @@ void test4() {
113158
@Transactional
114159
@Rollback(false)
115160
void test5() {
161+
DBStreamContext.getInstance().cleanEventPushers();
162+
DBStreamContext.getInstance().addEventPusher(new DBEventPusher() {
163+
@Override
164+
public void push(JdbcQuery jdbcQuery, List<DBEvent> events) {
165+
for (DBEvent event:events){
166+
assertTrue(event.isInsert());
167+
assertTrue(event.hasPrimaryKeys());
168+
}
169+
}
170+
});
116171
User1 user = new User1();
117172
user.setUsername("admin");
118173
user.setPassword("admin");
119174
user.setEmail("admin@example.com");
120175
user.setNickname("admin");
121176

122177
userRepository.save(user);
123-
124-
userRepository.deleteAll();
125178
}
126179

127180

@@ -132,6 +185,29 @@ void test5() {
132185
@Transactional
133186
@Rollback(false)
134187
void test6() {
188+
DBStreamContext.getInstance().cleanEventPushers();
189+
DBStreamContext.getInstance().addEventPusher(new DBEventPusher() {
190+
@Override
191+
public void push(JdbcQuery jdbcQuery, List<DBEvent> events) {
192+
assertTrue(events.size()>=2);
193+
int hasPrimaryEventCount = 0;
194+
for (DBEvent event:events){
195+
assertTrue(event.isInsert());
196+
if(event.hasPrimaryKeys()){
197+
hasPrimaryEventCount++;
198+
}
199+
}
200+
assertEquals(1,hasPrimaryEventCount);
201+
}
202+
});
203+
User1 user = new User1();
204+
user.setUsername("admin");
205+
user.setPassword("admin");
206+
user.setEmail("admin@example.com");
207+
user.setNickname("admin");
208+
209+
userRepository.save(user);
210+
135211
userRepository.insertIntoFromSelect();
136212
}
137213

@@ -142,6 +218,16 @@ void test6() {
142218
@Test
143219
@Transactional
144220
void test7() {
221+
222+
DBStreamContext.getInstance().cleanEventPushers();
223+
AtomicBoolean running = new AtomicBoolean(false);
224+
DBStreamContext.getInstance().addEventPusher(new DBEventPusher() {
225+
@Override
226+
public void push(JdbcQuery jdbcQuery, List<DBEvent> events) {
227+
running.set(true);
228+
}
229+
});
230+
145231
User1 user = new User1();
146232
user.setUsername("admin");
147233
user.setPassword("admin");
@@ -153,6 +239,8 @@ void test7() {
153239
int result = 100 / 0;
154240
System.out.println(result);
155241
});
242+
243+
assertFalse(running.get());
156244
}
157245

158246
/**
@@ -162,6 +250,17 @@ void test7() {
162250
@Transactional
163251
@Rollback(value = false)
164252
void test8() {
253+
DBStreamContext.getInstance().cleanEventPushers();
254+
DBStreamContext.getInstance().addEventPusher(new DBEventPusher() {
255+
@Override
256+
public void push(JdbcQuery jdbcQuery, List<DBEvent> events) {
257+
assertTrue(events.size()>=15);
258+
for (DBEvent event:events){
259+
assertTrue(event.hasPrimaryKeys());
260+
}
261+
}
262+
});
263+
165264
List<User1> list = userRepository.findAll();
166265
for (User1 user : list) {
167266
user.setEmail("111");
@@ -180,6 +279,7 @@ void test8() {
180279
entityManager.flush();
181280
entityManager.clear();
182281

282+
183283
}
184284

185285

0 commit comments

Comments
 (0)