diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/FileChannelCache.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/FileChannelCache.java new file mode 100644 index 0000000000..38c1cfc525 --- /dev/null +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/FileChannelCache.java @@ -0,0 +1,73 @@ +/* + * Copyright 2002-2008 the original author or authors. + * + * Licensed 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.springframework.integration.file.locking; + +import java.io.File; +import java.io.IOException; +import java.io.RandomAccessFile; +import java.nio.channels.FileLock; +import java.nio.channels.OverlappingFileLockException; +import java.nio.channels.FileChannel; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +/** + * @author Iwein Fuld + */ + +final class FileChannelCache { + private static ConcurrentMap channelCache = new ConcurrentHashMap(); + + + /** + * Try to get a lock for this file while guaranteeing that the same channel will be used for all file locks in this + * VM. If the lock could not be aquired this method will return null. + *

+ * Locks aquired through this method should be passed back to #closeChannelFor to prevent memory leak. + *

+ * Thread safe. + */ + public static FileLock tryLockFor(File fileToLock) throws IOException { + FileChannel channel = channelCache.get(fileToLock); + if (channel == null) { + FileChannel newChannel = new RandomAccessFile(fileToLock, "rw").getChannel(); + FileChannel original = channelCache.putIfAbsent(fileToLock, newChannel); + channel = original != null ? original : newChannel; + } + FileLock lock = null; + if (channel != null) { + try { + lock = channel.tryLock(); + } catch (OverlappingFileLockException e) { + // File is already locked in this thread or virtual machine + } + } + return lock; + } + + /** + * Close the channel for the file passed in. + *

+ * Thread safe. + */ + public static void closeChannelFor(File fileToUnlock) throws IOException { + channelCache.remove(fileToUnlock).close(); + } + + public static boolean isLocked(File file) { + return channelCache.containsKey(file); + } +} diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/FileLocker.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/FileLocker.java index 30b8ec1bae..94b839c2de 100644 --- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/FileLocker.java +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/FileLocker.java @@ -16,6 +16,7 @@ package org.springframework.integration.file.locking; import java.io.File; +import java.io.IOException; /** * A FileLocker is a strategy that can ensure that files are only processed a single time. diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/LockFileFileListFilter.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/LockFileFileListFilter.java index 5b0825c322..8e42535519 100644 --- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/LockFileFileListFilter.java +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/LockFileFileListFilter.java @@ -18,25 +18,32 @@ package org.springframework.integration.file.locking; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.integration.file.AbstractFileListFilter; -import static org.springframework.integration.file.locking.LockFileFileLocker.*; +import org.springframework.util.Assert; import java.io.File; +import java.io.IOException; /** * LockFileFileListFilter keeps track of the files it should pick up by adding lock files in a directory. If this * directory is the same as the directory of the original files addition precautions need to be taken to avoid treating * the lock files as normal input. * - * If different LockFileFileListFilters share their lock directory they are guaranteed not to pass the same file. + * If different LockFileFileListFilters share their lock directory they are unlikely to pass the same file. * * @author Iwein Fuld */ -public class LockFileFileListFilter extends AbstractFileListFilter { +public class LockFileFileListFilter extends AbstractFileListFilter implements FileLocker{ private static final Log logger = LogFactory.getLog(LockFileFileListFilter.class); + private static final String LOCK_SUFFIX = ".lock"; + private static final String PRELOCK_SUFFIX = ".prelock"; + private File workdir; public LockFileFileListFilter(File workdir) { + Assert.notNull(workdir, "Work directy must not be null."); + Assert.isTrue(workdir.isDirectory(),"Work directy must be a directory."); + Assert.isTrue(workdir.canWrite(), "Work directy must be write accessible."); this.workdir = workdir; } @@ -49,4 +56,43 @@ public class LockFileFileListFilter extends AbstractFileListFilter { String name = file.getName(); return !name.endsWith(LOCK_SUFFIX)&&!name.endsWith(PRELOCK_SUFFIX); } + + /** + * Makes a best effort attempt at locking the file atomically. + * + * The locking mechanism will create a prelock file and move it to a lock + * file location. + */ + public boolean lock(File fileToLock) { + File lockFile = new File(workdir, fileToLock.getName() + LOCK_SUFFIX); + File preLockFile = new File(workdir, fileToLock.getName() + PRELOCK_SUFFIX); + if (lockFile.exists() || preLockFile.exists()) { + return false; + } + try { + preLockFile.createNewFile(); + } catch (IOException e) { + logger.warn("Failed to lock file", e); + return false; + } + //there is still a small chance that the next line will be done concurrently and the file will be picked up twice + if (!lockFile.exists()&&preLockFile.renameTo(lockFile)) { + lockFile.deleteOnExit(); + return true; + } + preLockFile.delete(); + logger.warn("Failed to lock file [" + fileToLock + "]"); + return false; + } + + /** + * Deletes the lockFile for this file. + */ + public void unlock(File fileToUnlock) { + File lockFile = new File(workdir, fileToUnlock.getName() + LOCK_SUFFIX); + File preLockFile = new File(workdir, fileToUnlock.getName() + PRELOCK_SUFFIX); + if (!preLockFile.exists()) { + lockFile.delete(); + } + } } diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/LockFileFileLocker.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/LockFileFileLocker.java deleted file mode 100644 index dcc3fdf7e1..0000000000 --- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/LockFileFileLocker.java +++ /dev/null @@ -1,78 +0,0 @@ -/* - * Copyright 2002-2008 the original author or authors. - * - * Licensed 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.springframework.integration.file.locking; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.springframework.util.Assert; - -import java.io.File; -import java.io.IOException; - -/** - * @author Iwein Fuld - */ -public class LockFileFileLocker implements FileLocker { - private static final Log logger = LogFactory.getLog(LockFileFileLocker.class); - - private File workdir; - static final String LOCK_SUFFIX = ".lock"; - static final String PRELOCK_SUFFIX = ".prelock"; - - public LockFileFileLocker(File workdir) { - Assert.notNull(workdir, "Work directy must not be null."); - Assert.isTrue(workdir.isDirectory(),"Work directy must be a directory."); - Assert.isTrue(workdir.canWrite(), "Work directy must be write accessible."); - this.workdir = workdir; - } - - /** - * Makes a best effort attempt at locking the file atomically. The chances of success are wholly dependant on the - * underlying operating system. - * - * The locking mechanism will create a prelock file and move it to a lock - * file location. - */ - public boolean lock(File fileToLock) { - File lockFile = new File(workdir, fileToLock.getName() + LOCK_SUFFIX); - File preLockFile = new File(workdir, fileToLock.getName() + PRELOCK_SUFFIX); - if (lockFile.exists() || preLockFile.exists()) { - return false; - } - try { - preLockFile.createNewFile(); - } catch (IOException e) { - logger.warn("Failed to lock file", e); - return false; - } - //there is still a small chance that the next line will be done concurrently and the file will be picked up twice - if (!lockFile.exists()&&preLockFile.renameTo(lockFile)) { - return true; - } - preLockFile.delete(); - logger.warn("Failed to lock file [" + fileToLock + "]"); - return false; - } - - /** - * Sets the lockFile for this file to be writable and then deletes it. If a lock happens concurrently, the delete - * should not succeed, so the lock will remain in place. This might be different on various operating systems. - */ - public void unlock(File fileToUnlock) { - File lockFile = new File(workdir, fileToUnlock.getName() + LOCK_SUFFIX); - lockFile.delete(); - } -} diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/NioFileLocker.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/NioFileLocker.java new file mode 100644 index 0000000000..889fee3898 --- /dev/null +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/NioFileLocker.java @@ -0,0 +1,71 @@ +/* + * Copyright 2002-2008 the original author or authors. + * + * Licensed 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.springframework.integration.file.locking; + +import org.springframework.integration.file.FileListFilter; +import org.springframework.integration.file.AbstractFileListFilter; +import org.springframework.integration.core.MessagingException; + +import java.io.File; +import java.io.IOException; +import java.nio.channels.FileLock; +import java.util.List; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ConcurrentHashMap; + +/** + * File locking strategy that uses java.nio. The locks taken by FileChannel are shared with all the threads in a single + * JVM, so this locking strategy does not prevent files being picked up multiple times within the same JVM. + * + * @author Iwein Fuld + * @author Mark Fisher + */ +public class NioFileLocker extends AbstractFileListFilter implements FileLocker { + private final ConcurrentMap lockCache = new ConcurrentHashMap(); + + public boolean lock(File fileToLock) { + FileLock lock = lockCache.get(fileToLock); + if (lock == null) { + FileLock newLock = null; + try { + newLock = FileChannelCache.tryLockFor(fileToLock); + } catch (IOException e) { + throw new MessagingException("Failed to lock file: " + fileToLock, e); + } + if (newLock != null) { + FileLock original = lockCache.putIfAbsent(fileToLock, newLock); + lock = original != null ? original : newLock; + } + } + return lock != null; + } + + public void unlock(File fileToUnlock) { + FileLock fileLock = lockCache.get(fileToUnlock); + try { + if (fileLock != null) { + fileLock.release(); + } + FileChannelCache.closeChannelFor(fileToUnlock); + } catch (IOException e) { + throw new MessagingException("Failed to unlock file: " + fileToUnlock, e); + } + } + + protected boolean accept(File file) { + return this.lock(file); + } +} diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java index 7c41d1785e..b1ea3833c3 100644 --- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java @@ -28,6 +28,7 @@ import org.springframework.integration.core.Message; import org.springframework.integration.file.locking.FileLocker; import java.io.File; +import java.io.IOException; import java.util.Comparator; /** @@ -113,7 +114,7 @@ public class FileReadingMessageSourceTests { } @Test - public void lockIsAquired() { + public void lockIsAquired() throws IOException { when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock}); Message received = source.receive(); assertNotNull(received); @@ -122,7 +123,7 @@ public class FileReadingMessageSourceTests { } @Test - public void lockedFilesAreIgnored() { + public void lockedFilesAreIgnored() throws IOException { when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock}); when(locker.lock(fileMock)).thenReturn(false); Message received = source.receive(); diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests-context.xml b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests-context.xml index b7a1ba5463..850f4e20b3 100644 --- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests-context.xml +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests-context.xml @@ -8,14 +8,11 @@ + p:locker-ref="filter"/> + p:locker-ref="filter" p:filter-ref="filter"/> - - - diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/LockFileFileListFilterTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/LockFileFileListFilterTests.java index febcc1ed19..ef81817e04 100644 --- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/LockFileFileListFilterTests.java +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/LockFileFileListFilterTests.java @@ -38,7 +38,7 @@ public class LockFileFileListFilterTests { workdir.mkdir(); } - private FileLocker locker = new LockFileFileLocker(workdir); + private FileLocker locker = new LockFileFileListFilter(workdir); @Before public void cleanDirectory() { diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/NioFileLockerTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/NioFileLockerTests.java new file mode 100644 index 0000000000..0408959136 --- /dev/null +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/NioFileLockerTests.java @@ -0,0 +1,69 @@ +/* + * Copyright 2002-2008 the original author or authors. + * + * Licensed 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.springframework.integration.file.locking; + +import static org.hamcrest.CoreMatchers.is; +import static org.junit.Assert.assertThat; +import org.junit.Before; +import org.junit.Test; +import org.junit.BeforeClass; +import org.springframework.integration.file.FileListFilter; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +/** + * @author Iwein Fuld + */ +public class NioFileLockerTests { + + private static File workdir = new File(new File(System.getProperty("java.io.tmpdir")), NioFileLockerTests.class.getSimpleName()); + + @BeforeClass + public static void setupWorkDir() { + workdir.mkdir(); + } + + @Before + public void cleanDirectory() { + File[] files = workdir.listFiles(); + for (File file : files) { + file.delete(); + } + } + + @Test + public void fileListedByFirstFilter() throws IOException { + FileListFilter filter = new NioFileLocker(); + File testFile = new File(workdir, "test0"); + testFile.createNewFile(); + assertThat(filter.filterFiles(workdir.listFiles()).get(0), is(testFile)); + assertThat(filter.filterFiles(workdir.listFiles()).get(0), is(testFile)); + } + + @Test + public void fileListedByOneFilterOnly() throws IOException { + FileListFilter filter1 = new NioFileLocker(); + FileListFilter filter2 = new NioFileLocker(); + File testFile = new File(workdir, "test1"); + testFile.createNewFile(); + assertThat(filter1.filterFiles(workdir.listFiles()).get(0), is(testFile)); + assertThat(filter2.filterFiles(workdir.listFiles()), is((List)new ArrayList())); + } + +} \ No newline at end of file