0005945: compare and repair dev checkin

JumpMind · Aug 8, 2023 · d9cff2d · d9cff2d
1 parent ffde525
commit d9cff2d
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 1 deletion.
diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/ext/IBatchStagingExtension.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/ext/IBatchStagingExtension.java
@@ -0,0 +1,14 @@
+package org.jumpmind.symmetric.ext;
+
+import org.jumpmind.extension.IExtensionPoint;
+import org.jumpmind.symmetric.io.stage.IStagedResource;
+import org.jumpmind.symmetric.io.stage.StagingPurgeContext;
+
+public interface IBatchStagingExtension extends IExtensionPoint {
+    public void beforeClean(StagingPurgeContext context);
+
+    public boolean isValidPath(String category);
+
+    public boolean shouldCleanPath(IStagedResource resource, long ttlInMs, StagingPurgeContext context, String[] path, boolean resourceIsOld,
+            boolean resourceClearsMinTimeHurdle);
+}
diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/io/stage/BatchStagingManager.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/io/stage/BatchStagingManager.java
@@ -32,6 +32,7 @@
 
 import org.jumpmind.symmetric.ISymmetricEngine;
 import org.jumpmind.symmetric.common.ParameterConstants;
+import org.jumpmind.symmetric.ext.IBatchStagingExtension;
 import org.jumpmind.symmetric.model.BatchId;
 import org.jumpmind.symmetric.service.ClusterConstants;
 import org.slf4j.Logger;
@@ -90,6 +91,11 @@ public long clean(long ttlInMs) {
             context.putContextValue("outgoingBatches", outgoingBatches);
             context.putContextValue("incomingBatches", incomingBatches);
             context.putContextValue("biggestIncomingByNode", biggestIncomingByNode);
+            IBatchStagingExtension ext = engine.getExtensionService().getExtensionPoint(IBatchStagingExtension.class);
+            if (ext != null) {
+                context.putContextValue("extension", ext);
+                ext.beforeClean(context);
+            }
             return super.clean(ttlInMs, context);
         } finally {
             if (isLockAcquired) {
@@ -119,7 +125,12 @@ protected boolean shouldCleanPath(IStagedResource resource, long ttlInMs, Stagin
         } else if (path[0].equals(STAGING_CATEGORY_BULK_LOAD)) {
             return false;
         } else {
-            log.warn("Unrecognized path: " + resource.getPath());
+            IBatchStagingExtension ext = (IBatchStagingExtension) context.getContextValue("extension");
+            if (ext != null && ext.isValidPath(path[0])) {
+                return ext.shouldCleanPath(resource, ttlInMs, context, path, resourceIsOld, resourceClearsMinTimeHurdle);
+            } else {
+                log.warn("Unrecognized path: " + resource.getPath());
+            }
         }
         return false;
     }