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
index 57b947b7ab..ffe13214ad 100644
--- 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
@@ -47,6 +47,9 @@ public final class VFSConstants {
     public static final String TRANSPORT_FILE_LOCKING_ENABLED = "enable";
     public static final String TRANSPORT_FILE_LOCKING_DISABLED = "disable";
 
+    public static final String TRANSPORT_CHECK_SIZE_INTERVAL = "transport.vfs.CheckSizeInterval";
+    public static final String TRANSPORT_CHECK_SIZE_IGNORE_EMPTY = "transport.vfs.CheckSizeIgnoreEmpty";
+
     /**
      * This parameter is used decide whether the resolving hostname IP of URIs are done at deployment or dynamically.
      * At usage default id 'false' which lead hostname resolution at deployment
diff --git a/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/PollTableEntry.java b/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/PollTableEntry.java
index 8953838fa1..c9c94b58c3 100644
--- a/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/PollTableEntry.java
+++ b/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/PollTableEntry.java
@@ -80,6 +80,12 @@ public class PollTableEntry extends AbstractPollTableEntry {
     /** moved file will have this formatted timestamp prefix */    
     private DateFormat moveTimestampFormat;
 
+    /** containing the time in [ms] between the size check on files (to avoid reading files which are currently written) */
+    private String checkSizeInterval = null;
+    /** does the checkSize Lock mechanisme take empty files or not, default = false */
+    private String checkSizeIgnoreEmpty = "false";
+
+
     private boolean streaming;
 
     private int maxRetryCount;
@@ -311,6 +317,30 @@ private void setMoveAfterFailure(String moveAfterFailure) throws AxisFault {
         }
     }
 
+    public void setCheckSizeInterval(String checkSizeInterval) {
+        this.checkSizeInterval = checkSizeInterval;
+    }
+
+    public String getCheckSizeInterval() {
+        return checkSizeInterval;
+    }
+
+    public boolean hasCheckSizeInterval() {
+        return (checkSizeInterval != null && checkSizeInterval.length() > 0);
+    }
+
+    public void setCheckSizeIgnoreEmpty(String checkSizeIgnoreEmpty) {
+        this.checkSizeIgnoreEmpty = checkSizeIgnoreEmpty;
+    }
+
+    public String getCheckSizeIgnoreEmpty() {
+        return checkSizeIgnoreEmpty;
+    }
+
+    public boolean isCheckSizeIgnoreEmpty() {
+        return "true".equals(checkSizeIgnoreEmpty);
+    }
+
     public boolean isStreaming() {
         return streaming;
     }
@@ -609,6 +639,17 @@ protected boolean loadConfigurationsFromService(ParameterInclude params) throws
             Map<String, String> schemeFileOptions = VFSUtils.parseSchemeFileOptions(fileURI, params);
             setVfsSchemeProperties(schemeFileOptions);
 
+            //get check size intervall for locking 
+            String checkSizeIntervalString = ParamUtils.getOptionalParam(params, VFSConstants.TRANSPORT_CHECK_SIZE_INTERVAL);
+            setCheckSizeInterval(checkSizeIntervalString);
+
+            //get check size ignore emtpy for locking 
+            String checkSizeIgnoreEmptyString = ParamUtils.getOptionalParam(params, VFSConstants.TRANSPORT_CHECK_SIZE_IGNORE_EMPTY);
+            //set parameter only if it is set, default value is true
+            if (checkSizeIgnoreEmptyString != null) {
+                setCheckSizeIgnoreEmpty(checkSizeIgnoreEmptyString);
+            }
+
             String strStreaming = ParamUtils.getOptionalParam(params, VFSConstants.STREAMING);
             if (strStreaming != null) {
                 streaming = Boolean.parseBoolean(strStreaming);
diff --git a/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSTransportListener.java b/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSTransportListener.java
index 5e799166e9..693d64a5f3 100644
--- a/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSTransportListener.java
+++ b/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSTransportListener.java
@@ -18,7 +18,6 @@
 */
 package org.apache.synapse.transport.vfs;
 
-import org.apache.axiom.om.OMAttribute;
 import org.apache.axiom.om.OMElement;
 import org.apache.axis2.AxisFault;
 import org.apache.axis2.Constants;
@@ -41,6 +40,7 @@
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.input.AutoCloseInputStream;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.apache.commons.vfs2.FileContent;
 import org.apache.commons.vfs2.FileNotFolderException;
 import org.apache.commons.vfs2.FileNotFoundException;
@@ -65,6 +65,7 @@
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
+import java.security.MessageDigest;
 import java.text.SimpleDateFormat;
 import java.util.Arrays;
 import java.util.Comparator;
@@ -140,6 +141,8 @@ public class VFSTransportListener extends AbstractPollingTransportListener<PollT
     public static final String MOVE = "MOVE";
     public static final String NONE = "NONE";
 
+    public static final String EMPTY_MD5 = "d41d8cd98f00b204e9800998ecf8427e";
+
     /** The VFS file system manager */
     private DefaultFileSystemManager fsManager = null;
 
@@ -305,78 +308,82 @@ protected void scanFileOrDirectory(final PollTableEntry entry, String fileURI) {
                     if (fileObject.getType() == FileType.FILE &&
                             !isFailedRecord) {
                         boolean runPostProcess = true;
-                        if (!entry.isFileLockingEnabled() || (entry.isFileLockingEnabled() &&
-                                acquireLock(getFsManager(), fileObject, entry, fso, true))) {
-                            try {
-                                if (fileObject.getType() == FileType.FILE) {
-                                    boolean status = processFile(entry, fileObject);
-                                    if (status) {
-                                        entry.setLastPollState(PollTableEntry.SUCCSESSFUL);
+                        // Cancel the process if the file is still uploading (not complete)
+                        if (!isFileStillUploading(entry, fileObject)) {
+                            if (!entry.isFileLockingEnabled() || (entry.isFileLockingEnabled() &&
+                                    acquireLock(getFsManager(), fileObject, entry, fso, true))) {
+                                try {
+                                    if (fileObject.getType() == FileType.FILE) {
+                                        boolean status = processFile(entry, fileObject);
+                                        if (status) {
+                                            entry.setLastPollState(PollTableEntry.SUCCSESSFUL);
+                                        } else {
+                                            entry.setLastPollState(PollTableEntry.FAILED);
+                                        }
+                                        metrics.incrementMessagesReceived();
+                                    } else {
+                                        runPostProcess = false;
+                                    }
+
+                                } catch (AxisFault e) {
+                                    if (e.getCause() instanceof FileNotFoundException) {
+                                        String message = "Error processing File URI : " +
+                                                VFSUtils.maskURLPassword(fileObject.getName().toString()) +
+                                                ". This can be due to file moved from another process.";
+                                        VFSTransportErrorHandler.logException(log, LogType.WARN, message, serviceName);
+                                        runPostProcess = false;
                                     } else {
+                                        String message = "Error processing File URI : " +
+                                                VFSUtils.maskURLPassword(fileObject.getName().getURI());
+                                        VFSTransportErrorHandler.logException(log, LogType.ERROR, message, serviceName, e);
                                         entry.setLastPollState(PollTableEntry.FAILED);
+                                        metrics.incrementFaultsReceiving();
                                     }
-                                    metrics.incrementMessagesReceived();
-                                } else {
-                                    runPostProcess = false;
+                                    closeFileSystem(fileObject);
                                 }
 
-                            } catch (AxisFault e) {
-                                if (e.getCause() instanceof FileNotFoundException) {
-                                    String message = "Error processing File URI : " +
-                                            VFSUtils.maskURLPassword(fileObject.getName().toString()) +
-                                            ". This can be due to file moved from another process.";
-                                    VFSTransportErrorHandler.logException(log, LogType.WARN, message, serviceName);
-                                    runPostProcess = false;
-                                } else {
-                                    String message = "Error processing File URI : " +
-                                            VFSUtils.maskURLPassword(fileObject.getName().getURI());
-                                    VFSTransportErrorHandler.logException(log, LogType.ERROR, message, serviceName, e);
-                                    entry.setLastPollState(PollTableEntry.FAILED);
-                                    metrics.incrementFaultsReceiving();
-                                }
-                                closeFileSystem(fileObject);
-                            }
-                            if (runPostProcess) {
-                                try {
-                                    moveOrDeleteAfterProcessing(entry, fileObject, fso);
-                                } catch (AxisFault axisFault) {
-                                    String message = "File object '" +
+                                if (runPostProcess) {
+                                    try {
+                                        moveOrDeleteAfterProcessing(entry, fileObject, fso);
+                                    } catch (AxisFault axisFault) {
+                                        String message = "File object '" +
                                             VFSUtils.maskURLPassword(fileObject.getURL().toString()) +
                                             "' " + "cloud not be moved";
-                                    VFSTransportErrorHandler.logException(log, LogType.ERROR, message, serviceName,
-                                            axisFault);
-                                    entry.setLastPollState(PollTableEntry.FAILED);
-                                    String timeStamp = VFSUtils.getSystemTime(entry.getFailedRecordTimestampFormat());
-                                    addFailedRecord(entry, fileObject, timeStamp);
+                                        VFSTransportErrorHandler.logException(log, LogType.ERROR, message, serviceName,
+                                                axisFault);
+                                        entry.setLastPollState(PollTableEntry.FAILED);
+                                        String timeStamp = VFSUtils.getSystemTime(entry.getFailedRecordTimestampFormat());
+                                        addFailedRecord(entry, fileObject, timeStamp);
+                                    }
+                                }
+                                if (entry.isFileLockingEnabled()) {
+                                    VFSUtils.releaseLock(getFsManager(), fileObject, fso);
+                                    if (log.isDebugEnabled()) {
+                                        log.debug("Removed the lock file '"
+                                                + VFSUtils.maskURLPassword(fileObject.toString())
+                                                + ".lock' of the file '"
+                                                + VFSUtils.maskURLPassword(fileObject.toString()));
+                                    }
+                                }
+                            } else if (log.isDebugEnabled()) {
+                                log.debug("Couldn't get the lock for processing the file : "
+                                        + VFSUtils.maskURLPassword(fileObject.getName().getURI()));
+                            } else if (isFailedRecord) {
+                                if (entry.isFileLockingEnabled()) {
+                                    VFSUtils.releaseLock(getFsManager(), fileObject, fso);
+                                }
+                                // schedule a cleanup task if the file is there
+                                if (getFsManager().resolveFile(fileObject.getURL().toString(), fso) != null &&
+                                        removeTaskState == STATE_STOPPED && entry.getMoveAfterMoveFailure() != null) {
+                                    workerPool.execute(new FileRemoveTask(entry, fileObject, fso));
                                 }
-                            }
-                            if (entry.isFileLockingEnabled()) {
-                                VFSUtils.releaseLock(getFsManager(), fileObject, fso);
                                 if (log.isDebugEnabled()) {
-                                    log.debug("Removed the lock file '"
-                                    		+ VFSUtils.maskURLPassword(fileObject.toString())
-                                    		+ ".lock' of the file '"
-                                    		+ VFSUtils.maskURLPassword(fileObject.toString()));
+                                    log.debug("File '"
+                                            + VFSUtils.maskURLPassword(fileObject.getURL().toString())
+                                            + "' has been marked as a failed"
+                                            + " record, it will not process");
                                 }
                             }
-                        } else if (log.isDebugEnabled()) {
-                            log.debug("Couldn't get the lock for processing the file : "
-                                    + VFSUtils.maskURLPassword(fileObject.getName().getURI()));
-                        } else if (isFailedRecord) {
-                            if (entry.isFileLockingEnabled()) {
-                                VFSUtils.releaseLock(getFsManager(), fileObject, fso);
-                            }
-                            // schedule a cleanup task if the file is there
-                            if (getFsManager().resolveFile(fileObject.getURL().toString(), fso) != null &&
-                                    removeTaskState == STATE_STOPPED && entry.getMoveAfterMoveFailure() != null) {
-                                workerPool.execute(new FileRemoveTask(entry, fileObject, fso));
-                            }
-                            if (log.isDebugEnabled()) {
-                                log.debug("File '"
-                                		+ VFSUtils.maskURLPassword(fileObject.getURL().toString())
-                                		+ "' has been marked as a failed"
-                                		+ " record, it will not process");
-                            }
                         }
                     }
 
@@ -459,6 +466,14 @@ protected void scanFileOrDirectory(final PollTableEntry entry, String fileURI) {
                         if(entry.getFileNamePattern()!=null &&
                                 child.getName().getBaseName().matches(entry.getFileNamePattern())){
                             //child's file name matches the file name pattern
+
+                            //check if file is still uploading (only used when checkSizeInterval is set)
+                            if (isFileStillUploading(entry, child)) {
+                                //continiue with next file this one is still uploading
+                                log.debug("Skipped file " + VFSUtils.maskURLPassword(child.getName().getBaseName()) + " is still uploading.");
+                                continue;
+                            }
+
                             //now we try to get the lock and process
                             if (log.isDebugEnabled()) {
                                 log.debug("Matching file : " + child.getName().getBaseName());
@@ -1216,5 +1231,148 @@ public int compare(FileObject o1, FileObject o2) {
             }
             return lDiff.intValue();
         }
-    }  
+    }
+
+    /**
+     * Verifies if the given md5 is the md5 of an Empty File
+     * @param entry current poll
+     * @param md5 of the given file
+     * @return true if configuration is set and file is empty
+     */
+    private boolean isFileEmpty(PollTableEntry entry, String md5) {
+        if (entry.isCheckSizeIgnoreEmpty()) {
+            return EMPTY_MD5.equals(md5);
+        }
+        return false;
+    }
+    
+    /**
+     * This Function calculates the MD5 Hash of the FileObject, waits
+     * checkSizeInterval [ms] time, and calculates the MD5 Hash again. If they
+     * are the same, the file is finished uploading and can be consumed.
+     *
+     * @param entry current poll
+     * @param child the fileobject currently read
+     * @return if file is empty or the filesize is changing
+     */
+    private boolean isFileStillChangingSize(PollTableEntry entry, FileObject child, String md5) {
+        //check if the lock mechanism is activated
+        if (!entry.hasCheckSizeInterval()) {
+            //not checking the file size changing
+            return false;
+        }
+        try {
+            //get interval time
+            String checkSizeIntervalString = entry.getCheckSizeInterval();
+            Long checkSizeInterval = Long.valueOf(checkSizeIntervalString);
+
+            //wait interval time 
+            log.debug("Check if file is still uploading. Now sleep "+checkSizeInterval+" ms");
+            Thread.sleep(checkSizeInterval);
+            //get second MD5
+            //clear cache and refresh
+            fsManager.getFilesCache().close();
+            child.refresh();
+            String md5AfterSleep = getMD5Checksum(child);
+            if (!md5.equals(md5AfterSleep)) {
+                //file is still uploading
+                log.debug("File ist still uploading. md5 Hashcode Before="+md5 + " After="+md5AfterSleep);
+                return true;
+            }
+        } catch (Exception e) {
+            try {
+                if (ExceptionUtils.getStackTrace(e).contains("The file is being used by another process")) {
+                    return true;
+                }
+            } catch (Exception e2) {
+            }
+            return true;
+        }
+        return false;
+    }
+    
+    private String getMD5Checksum(FileObject child) throws Exception {
+        InputStream inputStream = null;
+        try {
+            //get first MD5
+            inputStream = child.getContent().getInputStream();
+            String md5 = getMD5Checksum(inputStream);
+            return md5;
+        } finally {
+            if(inputStream != null){
+                try {
+                    inputStream.close();
+                } catch (Exception ex) {
+                }
+            }
+        }
+    }
+    
+    /**
+     * Return the MD5Checksum of a InputStream as String
+     *
+     * @param fis inputStream of the current file
+     * @return MD5 Checksum
+     * @throws Exception
+     */
+    private String getMD5Checksum(InputStream fis) throws Exception {
+        byte[] b = createChecksum(fis);
+        String result = "";
+        for (int i = 0; i < b.length; i++) {
+            result += Integer.toString((b[i] & 0xff) + 0x100, 16).substring(1);
+        }
+        return result;
+    }
+
+    /**
+     * Used by getMD5Checksum to get the MD5 as byte array
+     *
+     * @param fis inputStream of the current file
+     * @return MD5 Checksum
+     * @throws Exception
+     */
+    private byte[] createChecksum(InputStream fis) throws Exception {
+        byte[] buffer = new byte[1024];
+        MessageDigest complete = MessageDigest.getInstance("MD5");
+        int numRead;
+        do {
+            numRead = fis.read(buffer);
+            if (numRead > 0) {
+                complete.update(buffer, 0, numRead);
+            }
+        } while (numRead != -1);
+        fis.close();
+        return complete.digest();
+    }
+
+    private boolean isFileStillUploading(PollTableEntry entry, FileObject child) {
+        if(!entry.isCheckSizeIgnoreEmpty() && !entry.hasCheckSizeInterval()) {
+            //CheckEmpty and CheckSize are not active - return false (file is not uploading)
+            return false;
+        }
+        InputStream inputStream = null;
+        try {
+            //get first MD5
+            log.debug("Create MD5 Checksum of File: "+VFSUtils.maskURLPassword(child.getName().toString()));
+            inputStream = child.getContent().getInputStream();
+            String md5 = getMD5Checksum(inputStream);
+            return isFileEmpty(entry, md5) || isFileStillChangingSize(entry, child, md5);
+        } catch (Exception e) {
+            try {
+                if (ExceptionUtils.getStackTrace(e).contains("The file is being used by another process")) {
+                    return true;
+                }
+            } catch (Exception e2) {
+            }
+            //return true when any exception occurs
+            return true;
+        } finally {
+            if(inputStream != null){
+                try {
+                    inputStream.close();
+                } catch (Exception ex) {
+                }
+            }
+        }
+    }
 }