Integry/IntegryManagementSystem
19
0
Fork 0
Code Actions Packages Projects Releases Activity

Merge branch 'master' into develop
All checks were successful
IntegryManagementSystem_Multi/pipeline/head This commit looks good
Details

Browse Source
This commit is contained in:
Giuseppe Scorrano
2025-11-20 18:02:55 +01:00
parent 4385520480 974b5f61af
commit cd389dc83f
1 changed files with 20 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
ems-core/src/main/java/it/integry/ems/dbms_change_tracker/component/DbmsChangeTracker.java
View File

@@ -132,7 +132,7 @@ public class DbmsChangeTracker {
final List<String> trackedTables = retrieveTrackedTables(); final List<String> trackedTables = retrieveTrackedTables();
HashMap<Long, DetectedChangeDataDTO> changesByVersion = new HashMap<>(); HashMap<Long, List<DetectedChangeDataDTO>> changesByVersion = new HashMap<>();
for (String trackedTable : trackedTables) { for (String trackedTable : trackedTables) {
long minValidVersion = getMinValidVersion(trackedTable); long minValidVersion = getMinValidVersion(trackedTable);
@@ -140,13 +140,12 @@ public class DbmsChangeTracker {
throw new SQLException("Change Tracking on table " + trackedTable + " has been reset. Current version: " + currentVersion + ", Min valid version: " + minValidVersion); throw new SQLException("Change Tracking on table " + trackedTable + " has been reset. Current version: " + currentVersion + ", Min valid version: " + minValidVersion);
} }
final List<DetectedChangeDataDTO> detectedChanges = retrieveChangeList(trackedTable); final List<DetectedChangeDataDTO> detectedChanges = retrieveChangeList(trackedTable);
detectedChanges.forEach(detectedChangeDataDTO -> { detectedChanges.forEach(detectedChangeDataDTO -> {
changesByVersion.put(detectedChangeDataDTO.getSysChangeVersion(), detectedChangeDataDTO); changesByVersion.putIfAbsent(detectedChangeDataDTO.getSysChangeVersion(), new ArrayList<>());
changesByVersion.get(detectedChangeDataDTO.getSysChangeVersion()).add(detectedChangeDataDTO);
}); });
} }
logTrace("Detected " + changesByVersion.size() + " changes since version " + currentVersion); logTrace("Detected " + changesByVersion.size() + " changes since version " + currentVersion);
@@ -154,32 +153,33 @@ public class DbmsChangeTracker {
dispatchChanges(changesByVersion); dispatchChanges(changesByVersion);
} }
private void dispatchChanges(HashMap<Long, DetectedChangeDataDTO> changesByVersion) throws Exception { private void dispatchChanges(HashMap<Long, List<DetectedChangeDataDTO>> changesByVersion) throws Exception {
final List<Long> sortedChanges = changesByVersion.keySet().stream() final List<Long> sortedChanges = changesByVersion.keySet().stream()
.sorted() .sorted()
.collect(Collectors.toList()); .collect(Collectors.toList());
for (Long changeId : sortedChanges) { for (Long changeId : sortedChanges) {
final DetectedChangeDataDTO detectedChange = changesByVersion.get(changeId); final List<DetectedChangeDataDTO> detectedChanges = changesByVersion.get(changeId);
switch (detectedChange.getSysChangeOperation()) { for (DetectedChangeDataDTO detectedChange : detectedChanges) {
case INSERT: switch (detectedChange.getSysChangeOperation()) {
if (eventListener != null) case INSERT:
eventListener.onInsertDetected(detectedChange.getTableName(), detectedChange.getPrimaryKey()); if (eventListener != null)
break; eventListener.onInsertDetected(detectedChange.getTableName(), detectedChange.getPrimaryKey());
break;
case UPDATE: case UPDATE:
if (eventListener != null) if (eventListener != null)
eventListener.onUpdateDetected(detectedChange.getTableName(), detectedChange.getPrimaryKey()); eventListener.onUpdateDetected(detectedChange.getTableName(), detectedChange.getPrimaryKey());
break; break;
case DELETE: case DELETE:
if (eventListener != null) if (eventListener != null)
eventListener.onDeleteDetected(detectedChange.getTableName(), detectedChange.getPrimaryKey()); eventListener.onDeleteDetected(detectedChange.getTableName(), detectedChange.getPrimaryKey());
break; break;
}
} }
currentVersion = changeId; currentVersion = changeId;
} }
} }