Add transport parameters transport.vfs.MinimumAge and transport.vfs.MaximumAge

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";

modules/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,14 @@ public void setSubfolderTimestamp(String subfolderTimestamp) {
         this.subfolderTimestamp = subfolderTimestamp;
     }
 
+    public Long getMinimumAge() {
+        return minimumAge;
+    }
+
+    public Long getMaximumAge() {
+        return maximumAge;
+    }
+
     @Override
     public boolean loadConfiguration(ParameterInclude params) throws AxisFault {
 
@@ -690,6 +702,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;

modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSTransportListener.java

@@ -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