Add transport parameters transport.vfs.MinimumAge and transport.vfs.M…

…aximumAge which check file age (seconds)
wso2 · Aug 8, 2023 · 4828ec5 · 4828ec5
1 parent 728985f
commit 4828ec5
Show file tree

Hide file tree

Showing 3 changed files with 56 additions and 1 deletion.
diff --git a/modules/commons/src/main/java/org/apache/synapse/commons/vfs/VFSConstants.java b/modules/commons/src/main/java/org/apache/synapse/commons/vfs/VFSConstants.java
@@ -200,6 +200,11 @@ public final class VFSConstants {
     public static final String SCHEME_FTPS = "ftps";
     // sftp scheme file option list
     public static enum SFTP_FILE_OPTION {Identities, UserDirIsRoot, IdentityPassPhrase};
+
+    /** Parameter for minimum age **/
+    public static final String TRANSPORT_FILE_MINIMUM_AGE = "transport.vfs.MinimumAge";
+    /** Parameter for maximum age **/
+    public static final String TRANSPORT_FILE_MAXIMUM_AGE = "transport.vfs.MaximumAge";
 
     public static final String FILE_TYPE_PREFIX = "transport.vfs.fileType";
     public static final String FILE_TYPE = "filetype";

diff --git a/...es/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/PollTableEntry.java b/...es/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/PollTableEntry.java
@@ -143,6 +143,10 @@ public class PollTableEntry extends AbstractPollTableEntry {
 
     private static final Log log = LogFactory.getLog(PollTableEntry.class);
 
+    private Long minimumAge = null; //defines a minimum age of a file before being consumed. Use to avoid just written files to be consumed
+    private Long maximumAge = null; //defines a maximum age of a file being consumed. Old files will stay in the directory
+
+
     public PollTableEntry(boolean fileLocking) {
         this.fileLocking = fileLocking;
     }
@@ -499,6 +503,18 @@ public void setSubfolderTimestamp(String subfolderTimestamp) {
         this.subfolderTimestamp = subfolderTimestamp;
     }
 
+    public Long getMinimumAge() {
+        return minimumAge;
+    }
+
+    public Long getMaximumAge() {
+        return maximumAge;
+    }
+
+    public boolean hasAgeCheck(){
+        return minimumAge != null||maximumAge != null;
+    }
+
     @Override
     public boolean loadConfiguration(ParameterInclude params) throws AxisFault {
 
@@ -515,6 +531,7 @@ public boolean loadConfiguration(ParameterInclude params) throws AxisFault {
     }
 
     protected boolean loadConfigurationsFromService(ParameterInclude params) throws AxisFault {
+        System.out.println("loading config from service");
         fileURI = ParamUtils.getOptionalParam(params, VFSConstants.TRANSPORT_FILE_FILE_URI);
         if (fileURI == null) {
             log.warn("transport.vfs.FileURI parameter is missing in the proxy service configuration");
@@ -690,6 +707,23 @@ protected boolean loadConfigurationsFromService(ParameterInclude params) throws
                 }
             }
 
+            String strMinimumAge = ParamUtils.getOptionalParam(params, VFSConstants.TRANSPORT_FILE_MINIMUM_AGE);
+            if(strMinimumAge != null){
+                try {
+                    minimumAge = Long.parseLong(strMinimumAge);
+                } catch (NumberFormatException nfe) {
+                    log.warn("VFS File MinimumAge value is invalid : " + strMinimumAge, nfe);
+                }
+            }
+            String strMaximumAge = ParamUtils.getOptionalParam(params, VFSConstants.TRANSPORT_FILE_MAXIMUM_AGE);
+            if(strMaximumAge != null){
+                try {
+                    maximumAge = Long.parseLong(strMaximumAge);
+                } catch (NumberFormatException nfe) {
+                    log.warn("VFS File MaximumAge value is invalid : " + strMinimumAge, nfe);
+                }
+            }
+
             String strAutoLock = ParamUtils.getOptionalParam(params,
                                                              VFSConstants.TRANSPORT_AUTO_LOCK_RELEASE);
             autoLockRelease = false;

diff --git a/...nsports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSTransportListener.java b/...nsports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSTransportListener.java
@@ -37,7 +37,7 @@
 import org.apache.axis2.transport.base.BaseConstants;
 import org.apache.axis2.transport.base.BaseUtils;
 import org.apache.axis2.transport.base.ManagementSupport;
-import org.apache.axis2.transport.base.threads.WorkerPool;
+import org.apache.axis2.transport.base.threads.WorkerPool; 
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.input.AutoCloseInputStream;
 import org.apache.commons.lang3.StringUtils;
@@ -452,6 +452,22 @@ protected void scanFileOrDirectory(final PollTableEntry entry, String fileURI) {
                             isFailedRecord = isFailedRecord(child, entry);
                         }
 
+                        if(entry.getMinimumAge() != null){
+                            long age = child.getContent().getLastModifiedTime();
+                            long time = System.currentTimeMillis();
+                            if((time-age)/1000 <= entry.getMinimumAge()){
+                                continue;
+                            }
+                        }
+
+                        if(entry.getMaximumAge() != null){
+                            long age = child.getContent().getLastModifiedTime();
+                            long time = System.currentTimeMillis();
+                            if((time-age)/1000 >= entry.getMaximumAge()){
+                                continue;
+                            }
+                        }
+
                         if(entry.getFileNamePattern()!=null &&
                                 child.getName().getBaseName().matches(entry.getFileNamePattern())){
                             //child's file name matches the file name pattern