Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-768 added nio support draft, put locking and filtering implenentations in the same class

Browse Source
This commit is contained in:
Iwein Fuld
2009-09-21 11:24:03 +00:00
parent 83eba0bb4a
commit 2bfa7cb42a
9 changed files with 269 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/FileChannelCache.java Normal file
View File

@@ -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<File, FileChannel> channelCache = new ConcurrentHashMap<File, FileChannel>();
/**
* 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 <code>null</code>.
* <p/>
* Locks aquired through this method should be passed back to #closeChannelFor to prevent memory leak.
* <p/>
* 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.
* <p/>
* Thread safe.
*/
public static void closeChannelFor(File fileToUnlock) throws IOException {
channelCache.remove(fileToUnlock).close();
}
public static boolean isLocked(File file) {
return channelCache.containsKey(file);
}
}

1
org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/FileLocker.java
View File

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

52
org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/LockFileFileListFilter.java
View File

@@ -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();
}
}
}

78
org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/LockFileFileLocker.java
View File

@@ -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();
}
}

71
org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/NioFileLocker.java Normal file
View File

@@ -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 <b>does not</b> 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<File, FileLock> lockCache = new ConcurrentHashMap<File, FileLock>();
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);
}
}

5
org.springframework.integration.file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java
View File

@@ -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<File> 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<File> received = source.receive();

7
org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests-context.xml
View File

@@ -8,14 +8,11 @@
<!-- under test -->
<bean id="fileSource1" class="org.springframework.integration.file.FileReadingMessageSource"
p:inputDirectory="file:${java.io.tmpdir}/FileLockingWithMultipleSourcesIntegrationTests"
p:locker-ref="locker"/>
p:locker-ref="filter"/>
<bean id="fileSource2" class="org.springframework.integration.file.FileReadingMessageSource"
p:inputDirectory="file:${java.io.tmpdir}/FileLockingWithMultipleSourcesIntegrationTests"
p:locker-ref="locker" p:filter-ref="filter"/>
p:locker-ref="filter" p:filter-ref="filter"/>
<bean id="locker" class="org.springframework.integration.file.locking.LockFileFileLocker">
<constructor-arg value="file:${java.io.tmpdir}/FileLockingWithMultipleSourcesIntegrationTests"/>
</bean>
<bean id="filter" class="org.springframework.integration.file.locking.LockFileFileListFilter">
<constructor-arg value="file:${java.io.tmpdir}/FileLockingWithMultipleSourcesIntegrationTests"/>

2
org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/LockFileFileListFilterTests.java
View File

@@ -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() {

69
org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/NioFileLockerTests.java Normal file
View File

@@ -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<File>()));
}
}