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

INT-947: cleaning out some obsolete code, splitting locking and filtering a bit better.

Browse Source
This commit is contained in:
Iwein Fuld
2010-02-12 19:31:10 +00:00
parent fec08008ea
commit e6d1b2f9c3
13 changed files with 147 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
org.springframework.integration.file/src/main/java/org/springframework/integration/file/CompositeFileListFilter.java
View File

@@ -16,71 +16,66 @@
package org.springframework.integration.file;
import java.io.File;
import java.io.FileFilter;
import java.util.Arrays;
import java.util.Collection;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import org.springframework.util.Assert;
import java.io.File;
import java.io.FileFilter;
import java.util.*;
/**
* Composition that delegates to multiple {@link FileFilter}s. The composition
* is AND based, meaning that a file must pass through each filter's
* {@link #filterFiles(File)} method in order to be accepted by the composite.
*
* Composition that delegates to multiple {@link FileFilter}s. The composition is AND based, meaning that a file must
* pass through each filter's {@link #filterFiles(java.io.File[])} method in order to be accepted by the composite.
*
* @author Iwein Fuld
* @author Mark Fisher
*/
public class CompositeFileListFilter implements FileListFilter {
private final Set<FileListFilter> fileFilters;
private final Set<FileListFilter> fileFilters;
public CompositeFileListFilter(FileListFilter... fileFilters) {
this.fileFilters = new LinkedHashSet<FileListFilter>(Arrays.asList(fileFilters));
}
public CompositeFileListFilter(FileListFilter... fileFilters) {
this.fileFilters = new LinkedHashSet<FileListFilter>(Arrays.asList(fileFilters));
}
public CompositeFileListFilter(Collection<FileListFilter> fileFilters) {
this.fileFilters = new LinkedHashSet<FileListFilter>(fileFilters);
}
public CompositeFileListFilter(Collection<FileListFilter> fileFilters) {
this.fileFilters = new LinkedHashSet<FileListFilter>(fileFilters);
}
/**
* {@inheritDoc}
*
* This implementation delegates to a collection of filters and returns
* only files that pass all the filters.
*/
public List<File> filterFiles(File[] files) {
Assert.notNull(files, "'files' should not be null");
List<File> leftOver = Arrays.asList(files);
for (FileListFilter fileFilter : this.fileFilters) {
leftOver = fileFilter.filterFiles(leftOver.toArray(new File[] {}));
}
return leftOver;
}
/**
* {@inheritDoc}
* <p/>
* This implementation delegates to a collection of filters and returns only files that pass all the filters.
*/
public List<File> filterFiles(File[] files) {
Assert.notNull(files, "'files' should not be null");
List<File> leftOver = Arrays.asList(files);
for (FileListFilter fileFilter : this.fileFilters) {
leftOver = fileFilter.filterFiles(leftOver.toArray(new File[]{}));
}
return leftOver;
}
/**
* @see #addFilters(Collection)
* @param filters one or more new filters to add
* @return this CompositeFileFilter instance with the added filters
*/
public CompositeFileListFilter addFilter(FileListFilter... filters) {
return addFilters(Arrays.asList(filters));
}
/**
* @param filters one or more new filters to add
* @return this CompositeFileFilter instance with the added filters
* @see #addFilters(Collection)
*/
public CompositeFileListFilter addFilter(FileListFilter... filters) {
return addFilters(Arrays.asList(filters));
}
/**
* Add the new filters to this CompositeFileFilter while maintaining the
* existing filters.
*
* @param filtersToAdd a list of filters to add
* @return this CompositeFileFilter instance with the added filters
*/
public CompositeFileListFilter addFilters(Collection<FileListFilter> filtersToAdd) {
this.fileFilters.addAll(filtersToAdd);
/**
* Not thread safe. Only a single thread may add filters at a time.
*
* Add the new filters to this CompositeFileFilter while maintaining the existing filters.
*
* @param filtersToAdd a list of filters to add
* @return this CompositeFileFilter instance with the added filters
*/
public CompositeFileListFilter addFilters(Collection<FileListFilter> filtersToAdd) {
this.fileFilters.addAll(filtersToAdd);
return this;
}

21
org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
View File

@@ -143,9 +143,7 @@ public class FileReadingMessageSource implements MessageSource<File>,
/**
* Optional. Sets a
* {@link org.springframework.integration.file.locking.FileLocker} to be
* used instead of the default NoopFileLocker. Note that the locker is not
* queried by this FileReadingMessageSource: integration with a
* FileListFilter is an external concern.
* used to guard files against duplicate processing.
* <p/>
* <b>The supplied FileLocker must be thread safe</b>
*/
@@ -216,7 +214,7 @@ public class FileReadingMessageSource implements MessageSource<File>,
}
/**
* Adds the failed message back to the 'toBeReceived' queue.
* Adds the failed message back to the 'toBeReceived' queue if there is room.
*/
public void onFailure(Message<File> failedMessage, Throwable t) {
if (logger.isWarnEnabled()) {
@@ -234,19 +232,4 @@ public class FileReadingMessageSource implements MessageSource<File>,
logger.debug("Sent: " + sentMessage);
}
}
/**
* Implementation of FileLocker that doesn't provide any protection against
* duplicate listing.
*/
private static class NoopFileLocker implements FileLocker {
public boolean lock(File fileToLock) {
return true;
}
public void unlock(File fileToUnlock) {
// noop
}
}
}

38
org.springframework.integration.file/src/main/java/org/springframework/integration/file/NoopFileLocker.java
View File

@@ -1,38 +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;
import org.springframework.integration.file.locking.FileLocker;
import java.io.File;
/**
* Implementation of FileLocker that doesn't provide any protection against duplicate listing. This is the default used
* by the FileReadingMessageSource.
*
* @author Iwein Fuld
*/
final class NoopFileLocker implements FileLocker {
public boolean lock(File fileToLock) {
return true;
}
public void unlock(File fileToUnlock) {
//noop
}
}

10
org.springframework.integration.file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java
View File

@@ -21,7 +21,7 @@ import org.springframework.integration.file.CompositeFileListFilter;
import org.springframework.integration.file.DirectoryScanner;
import org.springframework.integration.file.FileListFilter;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.integration.file.locking.AbstractLockingFilter;
import org.springframework.integration.file.locking.AbstractFileLockerFilter;
import java.io.File;
import java.util.Comparator;
@@ -39,7 +39,7 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean {
private volatile FileListFilter filter;
private volatile AbstractLockingFilter locker;
private volatile AbstractFileLockerFilter locker;
private volatile Comparator<File> comparator;
@@ -64,8 +64,8 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean {
}
public void setFilter(FileListFilter filter) {
if (filter instanceof AbstractLockingFilter && this.locker == null) {
this.setLocker((AbstractLockingFilter) filter);
if (filter instanceof AbstractFileLockerFilter && this.locker == null) {
this.setLocker((AbstractFileLockerFilter) filter);
}
this.filter = filter;
}
@@ -78,7 +78,7 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean {
this.autoCreateDirectory = autoCreateDirectory;
}
public void setLocker(AbstractLockingFilter locker) {
public void setLocker(AbstractFileLockerFilter locker) {
this.locker = locker;
}

6
org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/AbstractLockingFilter.java → org.springframework.integration.file/src/main/java/org/springframework/integration/file/locking/AbstractFileLockerFilter.java
View File

@@ -21,16 +21,16 @@ import org.springframework.integration.file.AbstractFileListFilter;
import java.io.File;
/**
* Convenience base class for implementing FileLockers that acquire a lock upon accepting a file. This is needed
* Convenience base class for implementing FileLockers that check a lock before accepting a file. This is needed
* when used in combination with a FileReadingMessageSource through a DirectoryScanner.
*
* @author Iwein Fuld
* @since 2.0
*
*/
public abstract class AbstractLockingFilter extends AbstractFileListFilter implements FileLocker {
public abstract class AbstractFileLockerFilter extends AbstractFileListFilter implements FileLocker {
protected final boolean accept(File file) {
return lock(file);
return isLockable(file);
}
}

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

@@ -25,7 +25,7 @@ import java.io.File;
* contract, defining these guarantees is up to the implementation.
*
* If a filter that respects locks is required extend
* {@link org.springframework.integration.file.locking.AbstractLockingFilter} instead.
* {@link AbstractFileLockerFilter} instead.
*
* @author Iwein Fuld
* @since 2.0
@@ -40,6 +40,13 @@ public interface FileLocker {
*/
boolean lock(File fileToLock);
/**
* Checks whether the file passed in can be locked by this locker. This method never changes the locked state.
*
* @return true if the file was locked by another locker than this locker
*/
boolean isLockable(File file);
/**
* Unlocks the given file.
*

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

@@ -26,56 +26,56 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
/**
* 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.
* {@link FileReadingMessageSource}s sharing a Locker will not pick up
* the same files.
*
* This implementation will acquire or create a {@link FileLock} for the given
* file. Caching locks might be expensive, so this locking strategy is not
* recommended for scenarios where many files are accessed in parallel.
*
* 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.
* {@link FileReadingMessageSource}s sharing a Locker will not pick up the same files.
* <p/>
* This implementation will acquire or create a {@link FileLock} for the given file. Caching locks might be expensive,
* so this locking strategy is not recommended for scenarios where many files are accessed in parallel.
*
* @author Iwein Fuld
* @author Mark Fisher
* @since 2.0
*/
public class NioFileLocker extends AbstractLockingFilter {
public class NioFileLocker extends AbstractFileLockerFilter {
private final ConcurrentMap<File, FileLock> lockCache = new ConcurrentHashMap<File, FileLock>();
private final ConcurrentMap<File, FileLock> lockCache = new ConcurrentHashMap<File, FileLock>();
/**
* {@inheritDoc}
*
*/
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;
}
* {@inheritDoc}
*/
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);
}
public boolean isLockable(File file) {
return lockCache.containsKey(file) || !FileChannelCache.isLocked(file);
}
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);
}
}
}

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

@@ -28,6 +28,7 @@ import java.io.File;
import java.io.IOException;
import java.util.Comparator;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;
@@ -70,7 +71,7 @@ public class FileReadingMessageSourceTests {
@Test
public void straightProcess() throws Exception {
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock});
source.onSend(source.receive());
assertThat(source.receive().getPayload(), is(fileMock));
}
@Test

6
org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingNamespaceTests.java
View File

@@ -77,11 +77,15 @@ public class FileLockingNamespaceTests {
assertThat(extractFromScanner("filter", nioLockingSource), is(CompositeFileListFilter.class));
}
public static class StubLocker extends AbstractLockingFilter {
public static class StubLocker extends AbstractFileLockerFilter {
public boolean lock(File fileToLock) {
return true;
}
public boolean isLockable(File file) {
return true;
}
public void unlock(File fileToUnlock) {
//
}

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

@@ -8,15 +8,15 @@
<!-- under test -->
<bean id="fileSource1" class="org.springframework.integration.file.FileReadingMessageSource"
p:directory="file:${java.io.tmpdir}/FileLockingWithMultipleSourcesIntegrationTests"
p:filter-ref="filter1"/>
p:filter-ref="filter1" p:locker-ref="filter1"/>
<bean id="fileSource2" class="org.springframework.integration.file.FileReadingMessageSource"
p:directory="file:${java.io.tmpdir}/FileLockingWithMultipleSourcesIntegrationTests"
p:filter-ref="filter2"/>
p:filter-ref="filter2" p:locker-ref="filter2"/>
<bean id="fileSource3" class="org.springframework.integration.file.FileReadingMessageSource"
p:directory="file:${java.io.tmpdir}/FileLockingWithMultipleSourcesIntegrationTests"
p:filter-ref="filter1"/>
p:filter-ref="filter1" p:locker-ref="filter1"/>
<bean id="filter1" class="org.springframework.integration.file.locking.NioFileLocker"/>
<bean id="filter2" class="org.springframework.integration.file.locking.NioFileLocker"/>

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

@@ -15,11 +15,9 @@
*/
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.junit.Test;
import org.springframework.integration.file.FileListFilter;
import java.io.File;
@@ -27,6 +25,9 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertThat;
/**
* @author Iwein Fuld
*/
@@ -49,20 +50,22 @@ public class NioFileLockerTests {
@Test
public void fileListedByFirstFilter() throws IOException {
FileListFilter filter = new NioFileLocker();
NioFileLocker filter = new NioFileLocker();
File testFile = new File(workdir, "test0");
testFile.createNewFile();
assertThat(filter.filterFiles(workdir.listFiles()).get(0), is(testFile));
filter.lock(testFile);
assertThat(filter.filterFiles(workdir.listFiles()).get(0), is(testFile));
}
@Test
public void fileListedByOneFilterOnly() throws IOException {
FileListFilter filter1 = new NioFileLocker();
public void fileNotListedWhenLockedByOtherFilter() throws IOException {
NioFileLocker filter1 = new NioFileLocker();
FileListFilter filter2 = new NioFileLocker();
File testFile = new File(workdir, "test1");
testFile.createNewFile();
assertThat(filter1.filterFiles(workdir.listFiles()).get(0), is(testFile));
filter1.lock(testFile);
assertThat(filter2.filterFiles(workdir.listFiles()), is((List)new ArrayList<File>()));
}