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 d3a7373cc..d87463626 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 @@ -72,6 +72,8 @@ public class PollTableEntry extends AbstractPollTableEntry { private int maxRetryCount; private long reconnectTimeout; private boolean fileLocking; + private boolean autoLockRelease; + private Long autoLockReleaseInterval; private String moveAfterMoveFailure; @@ -196,6 +198,22 @@ public boolean isFileLockingEnabled() { return fileLocking; } + public boolean getAutoLockRelease() { + return autoLockRelease; + } + + public void setAutoLockRelease(boolean autoLockRelease) { + this.autoLockRelease = autoLockRelease; + } + + public Long getAutoLockReleaseInterval() { + return autoLockReleaseInterval; + } + + public void setAutoLockReleaseInterval(Long autoLockReleaseInterval) { + this.autoLockReleaseInterval = autoLockReleaseInterval; + } + public long getReconnectTimeout() { return reconnectTimeout; } @@ -281,6 +299,28 @@ public boolean loadConfiguration(ParameterInclude params) throws AxisFault { fileLocking = false; } + String strAutoLockRelease = ParamUtils.getOptionalParam( + params, VFSConstants.TRANSPORT_AUTO_LOCK_RELEASE); + autoLockRelease = false; + autoLockReleaseInterval = VFSConstants.DEFAULT_AUTO_LOCK_RELEASE_INTERVAL; + if (strAutoLockRelease != null) { + autoLockRelease = Boolean.parseBoolean(strAutoLockRelease); + if (autoLockRelease) { + String strAutoLockReleaseInterval = ParamUtils.getOptionalParam( + params, VFSConstants.TRANSPORT_AUTO_LOCK_RELEASE_INTERVAL); + if (strAutoLockReleaseInterval != null) { + try { + autoLockReleaseInterval = Long.parseLong(strAutoLockReleaseInterval); + } catch (NumberFormatException nfe) { + log.warn("VFS " + VFSConstants.TRANSPORT_AUTO_LOCK_RELEASE_INTERVAL + + " value is invalid : " + strAutoLockReleaseInterval, nfe); + autoLockReleaseInterval = + VFSConstants.DEFAULT_AUTO_LOCK_RELEASE_INTERVAL; + } + } + } + } + moveAfterMoveFailure = ParamUtils.getOptionalParam(params, VFSConstants.TRANSPORT_FILE_MOVE_AFTER_FAILED_MOVE); diff --git a/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSConstants.java b/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSConstants.java index 56326b58d..eecc100b8 100644 --- a/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSConstants.java +++ b/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSConstants.java @@ -45,6 +45,23 @@ public final class VFSConstants { public static final String TRANSPORT_FILE_LOCKING_ENABLED = "enable"; public static final String TRANSPORT_FILE_LOCKING_DISABLED = "disable"; + /** + * Lock release true/false + * */ + public static final String TRANSPORT_AUTO_LOCK_RELEASE = "transport.vfs.AutoLockRelease"; + + /** + * Lock release interval + * */ + public static final String TRANSPORT_AUTO_LOCK_RELEASE_INTERVAL + = "transport.vfs.AutoLockReleaseInterval"; + + /** + * Age in milliseconds beyond which an existing lock file is treated as abandoned, used when + * auto lock release is enabled without an explicit interval. + */ + public static final long DEFAULT_AUTO_LOCK_RELEASE_INTERVAL = 3600000L; + public static final String REPLY_FILE_URI = "transport.vfs.ReplyFileURI"; public static final String REPLY_FILE_NAME = "transport.vfs.ReplyFileName"; diff --git a/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSParamDTO.java b/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSParamDTO.java new file mode 100644 index 000000000..f922ba156 --- /dev/null +++ b/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSParamDTO.java @@ -0,0 +1,63 @@ +/* +* Licensed to the Apache Software Foundation (ASF) under one +* or more contributor license agreements. See the NOTICE file +* distributed with this work for additional information +* regarding copyright ownership. The ASF licenses this file +* to you under the Apache License, Version 2.0 (the +* "License"); you may not use this file except in compliance +* with the License. You may obtain a copy of the License at +* +* http://www.apache.org/licenses/LICENSE-2.0 +* +* Unless required by applicable law or agreed to in writing, +* software distributed under the License is distributed on an +* * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +* KIND, either express or implied. See the License for the +* specific language governing permissions and limitations +* under the License. +*/ +package org.apache.synapse.transport.vfs; + +/** + * Carries the auto lock release settings of a poll table entry down to + * {@link VFSUtils#acquireLock(org.apache.commons.vfs2.FileSystemManager, + * org.apache.commons.vfs2.FileObject, VFSParamDTO)}. + */ +public class VFSParamDTO { + + private boolean autoLockRelease; + private Long autoLockReleaseInterval; + + public VFSParamDTO() { + autoLockRelease = false; + autoLockReleaseInterval = VFSConstants.DEFAULT_AUTO_LOCK_RELEASE_INTERVAL; + } + + /** + * @return the autoLockRelease + */ + public boolean isAutoLockRelease() { + return autoLockRelease; + } + + /** + * @param autoLockRelease the autoLockRelease to set + */ + public void setAutoLockRelease(boolean autoLockRelease) { + this.autoLockRelease = autoLockRelease; + } + + /** + * @return the autoLockReleaseInterval + */ + public Long getAutoLockReleaseInterval() { + return autoLockReleaseInterval; + } + + /** + * @param autoLockReleaseInterval the autoLockReleaseInterval to set + */ + public void setAutoLockReleaseInterval(Long autoLockReleaseInterval) { + this.autoLockReleaseInterval = autoLockReleaseInterval; + } +} 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 4eb06cb21..9d196a64a 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 @@ -227,7 +227,7 @@ private void scanFileOrDirectory(final PollTableEntry entry, String fileURI) { if (fileObject.getType() == FileType.FILE && !isFailedRecord) { if (!entry.isFileLockingEnabled() || (entry.isFileLockingEnabled() && - VFSUtils.acquireLock(fsManager, fileObject))) { + acquireLock(fsManager, fileObject, entry))) { try { processFile(entry, fileObject); entry.setLastPollState(PollTableEntry.SUCCSESSFUL); @@ -304,7 +304,7 @@ private void scanFileOrDirectory(final PollTableEntry entry, String fileURI) { } if ((!entry.isFileLockingEnabled() - || (entry.isFileLockingEnabled() && VFSUtils.acquireLock(fsManager, child))) + || (entry.isFileLockingEnabled() && acquireLock(fsManager, child, entry))) && !isFailedRecord){ //process the file try { @@ -395,6 +395,23 @@ private void scanFileOrDirectory(final PollTableEntry entry, String fileURI) { } } + /** + * Acquire the lock of the given file, passing the auto lock release settings configured on + * the poll table entry so that an abandoned lock does not keep the file unprocessed. + * + * @param fsManager used to resolve the lock file + * @param fileObject representing the file to be processed + * @param entry the PollTableEntry carrying the auto lock release configuration + * @return true if the lock has been acquired, false if not + */ + protected boolean acquireLock(FileSystemManager fsManager, FileObject fileObject, + final PollTableEntry entry) { + VFSParamDTO vfsParamDTO = new VFSParamDTO(); + vfsParamDTO.setAutoLockRelease(entry.getAutoLockRelease()); + vfsParamDTO.setAutoLockReleaseInterval(entry.getAutoLockReleaseInterval()); + return VFSUtils.acquireLock(fsManager, fileObject, vfsParamDTO); + } + /** * Take specified action to either move or delete the processed file, depending on the outcome * @param entry the PollTableEntry for the file that has been processed diff --git a/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSUtils.java b/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSUtils.java index a38507609..f213a1fb9 100644 --- a/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSUtils.java +++ b/modules/transports/core/vfs/src/main/java/org/apache/synapse/transport/vfs/VFSUtils.java @@ -24,6 +24,7 @@ import org.apache.axis2.transport.base.BaseUtils; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.commons.vfs2.Capability; import org.apache.commons.vfs2.FileContent; import org.apache.commons.vfs2.FileObject; import org.apache.commons.vfs2.FileSystemException; @@ -111,6 +112,21 @@ public static String getFileName(MessageContext msgCtx, VFSOutTransportInfo vfsO * @return boolean true if the lock has been acquired or false if not */ public synchronized static boolean acquireLock(FileSystemManager fsManager, FileObject fo) { + return acquireLock(fsManager, fo, null); + } + + /** + * Acquires a file item lock before processing the item, guaranteeing that the file is not + * processed while it is being uploaded and/or the item is not processed by two listeners + * + * @param fsManager used to resolve the processing file + * @param fo representing the processing file item + * @param paramDTO the auto lock release settings of the poll table entry, or null to leave + * an existing lock in place regardless of its age + * @return boolean true if the lock has been acquired or false if not + */ + public synchronized static boolean acquireLock(FileSystemManager fsManager, FileObject fo, + VFSParamDTO paramDTO) { // generate a random lock value to ensure that there are no two parties // processing the same file @@ -128,10 +144,16 @@ public synchronized static boolean acquireLock(FileSystemManager fsManager, File } FileObject lockObject = fsManager.resolveFile(fullPath + ".lock"); if (lockObject.exists()) { - log.debug("There seems to be an external lock, aborting the processing of the file " + long lockAge = getLockAge(lockObject); + log.warn("There seems to be an external lock, aborting the processing of the file " + maskURLPassword(fo.getName().getURI()) + ". This could possibly be due to some other party already " - + "processing this file or the file is still being uploaded"); + + "processing this file or the file is still being uploaded" + + (lockAge >= 0 ? ". The lock file is " + lockAge + "ms old" : "")); + if (paramDTO != null && paramDTO.isAutoLockRelease() + && isLockAbandoned(lockAge, paramDTO.getAutoLockReleaseInterval())) { + releaseLock(lockObject, lockAge); + } } else { // write a lock file before starting of the processing, to ensure that the @@ -196,6 +218,63 @@ public static void releaseLock(FileSystemManager fsManager, FileObject fo) { } } + /** + * Decide whether an existing lock file is old enough to be treated as abandoned. + * + * @param lockAge the age of the lock file, or -1 if it could not be determined + * @param autoLockReleaseInterval age in milliseconds beyond which the lock is abandoned + * @return true if the lock has not been updated for at least the configured interval + */ + private static boolean isLockAbandoned(long lockAge, Long autoLockReleaseInterval) { + if (lockAge < 0) { + // the file system cannot report a last modified time, so the age of the lock is + // unknown; leave it in place rather than removing a lock that may still be held + return false; + } + return autoLockReleaseInterval == null || autoLockReleaseInterval <= lockAge; + } + + /** + * Remove a lock file that {@link #isLockAbandoned} has established is no longer held. + * + * @param lockObject the existing lock file + * @param lockAge the age of the lock file, reported in the log + */ + private static void releaseLock(FileObject lockObject, long lockAge) { + try { + log.warn("Removing the lock file " + + maskURLPassword(lockObject.getName().getURI()) + " after " + lockAge + + "ms without an update"); + lockObject.delete(); + } catch (Exception e) { + log.warn("Unable to delete the lock file during auto release cycle.", e); + } finally { + try { + lockObject.close(); + } catch (FileSystemException e) { + log.error("Couldn't release the lock file " + + maskURLPassword(lockObject.getName().getURI()), e); + } + } + } + + /** + * @param lockObject the lock file whose age is required + * @return the age of the lock file in milliseconds, or -1 if the file system cannot report + * a last modified time + */ + private static long getLockAge(FileObject lockObject) { + try { + if (!lockObject.getFileSystem().hasCapability(Capability.GET_LAST_MODIFIED)) { + return -1; + } + return System.currentTimeMillis() - lockObject.getContent().getLastModifiedTime(); + } catch (FileSystemException e) { + log.debug("Couldn't read the last modified time of the lock file", e); + return -1; + } + } + /** * Mask the password of the connection url with *** * @param url the actual url